Two milestones: a note can carry its own check (317), and a rule keeps what it used to say (323) #135
@@ -548,6 +548,66 @@ async def update_rule(
|
||||
return await rulebooks_svc.rule_detail(uid, rule, system_ids)
|
||||
|
||||
|
||||
async def rule_history(rule_id: int, version_id: int = 0) -> dict:
|
||||
"""What a rule USED TO SAY, newest change first.
|
||||
|
||||
Read this before you argue with a rule, and before you rewrite one. A
|
||||
rule that has been reworded may have been reworded for a reason you are
|
||||
about to rediscover the hard way — and the wording it replaced is often
|
||||
the fastest way to see what the current one is guarding against. The
|
||||
rescoping of rule 79 is the case this exists for: the superseded
|
||||
statement had to be hand-copied into a task log to survive the edit.
|
||||
|
||||
EACH ENTRY HOLDS THE TEXT THE EDIT REPLACED, not the text it introduced.
|
||||
So "what did this say before the most recent change?" is the first entry,
|
||||
and the text the change PRODUCED is the rule as it stands now — read that
|
||||
with get_rule. Pair the two and you have the diff.
|
||||
|
||||
An empty history is ordinary and means the rule has never been reworded,
|
||||
not that its history was lost. Nothing is written before milestone 323,
|
||||
so a rule edited before then starts empty too.
|
||||
|
||||
Args:
|
||||
rule_id: The rule whose history to read.
|
||||
version_id: 0 (default) lists the history — when each change
|
||||
happened, by whom, and the title as it then stood. Pass an id
|
||||
from that list to read that snapshot IN FULL. The list omits
|
||||
statement and why on purpose: a rule's statement runs to
|
||||
thousands of characters, and a history carrying every field would
|
||||
cost more to read than the answer is worth.
|
||||
|
||||
There is deliberately no restore. Putting an old wording back is a
|
||||
decision, so it goes through update_rule — which snapshots what it
|
||||
replaces, leaving the undo visible in the history like any other edit. A
|
||||
one-click revert would erase the only record of why the rewrite happened.
|
||||
"""
|
||||
uid = current_user_id()
|
||||
if version_id:
|
||||
version = await rulebooks_svc.get_rule_version(rule_id, version_id, uid)
|
||||
if version is None:
|
||||
raise ValueError(
|
||||
f"version {version_id} not found on rule {rule_id}"
|
||||
)
|
||||
return version.to_dict(include_text=True)
|
||||
|
||||
versions = await rulebooks_svc.list_rule_versions(rule_id, uid)
|
||||
if versions is None:
|
||||
raise ValueError(f"rule {rule_id} not found")
|
||||
return {
|
||||
"rule_id": rule_id,
|
||||
"versions": [v.to_dict(include_text=False) for v in versions],
|
||||
"total": len(versions),
|
||||
# Said in-band because an empty list is the ordinary case and reads
|
||||
# like a missing feature otherwise.
|
||||
"note": (
|
||||
"Each entry holds the text the edit REPLACED. The current wording "
|
||||
"is on the rule itself — get_rule(%d)." % rule_id
|
||||
if versions else
|
||||
"This rule has never been reworded."
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
async def delete_rule(rule_id: int, confirmed: bool = False) -> dict:
|
||||
"""Move a rule to the trash (recoverable). Requires confirmed=True."""
|
||||
uid = current_user_id()
|
||||
@@ -832,5 +892,6 @@ def register(mcp) -> None:
|
||||
suppress_topic_for_project, unsuppress_topic_for_project,
|
||||
exclude_always_on_rulebook, include_always_on_rulebook,
|
||||
rules_due_for_verification, mark_rule_verified,
|
||||
rule_history,
|
||||
):
|
||||
mcp.tool(name=fn.__name__)(fn)
|
||||
|
||||
@@ -206,6 +206,42 @@ async def update_rule(rule_id: int):
|
||||
return jsonify(await rulebooks_svc.rule_detail(uid, rule, data.get("system_ids")))
|
||||
|
||||
|
||||
@rulebooks_bp.get("/rules/<int:rule_id>/versions")
|
||||
@login_required
|
||||
async def list_rule_versions(rule_id: int):
|
||||
"""A rule's edit history, newest first.
|
||||
|
||||
Listing form only — a rule's `statement` and `why` run to thousands of
|
||||
characters, so a history list carrying every field would be unreadable
|
||||
and expensive to send. Open one for the text.
|
||||
"""
|
||||
uid = get_current_user_id()
|
||||
versions = await rulebooks_svc.list_rule_versions(rule_id, uid)
|
||||
if versions is None:
|
||||
return jsonify({"error": "rule not found"}), 404
|
||||
return jsonify({
|
||||
"versions": [v.to_dict(include_text=False) for v in versions],
|
||||
})
|
||||
|
||||
|
||||
@rulebooks_bp.get("/rules/<int:rule_id>/versions/<int:version_id>")
|
||||
@login_required
|
||||
async def get_rule_version(rule_id: int, version_id: int):
|
||||
"""One snapshot in full — what the rule said before that edit."""
|
||||
uid = get_current_user_id()
|
||||
version = await rulebooks_svc.get_rule_version(rule_id, version_id, uid)
|
||||
if version is None:
|
||||
return jsonify({"error": "version not found"}), 404
|
||||
return jsonify(version.to_dict(include_text=True))
|
||||
|
||||
|
||||
# NO restore route, deliberately (milestone 323). A note version can be
|
||||
# restored; a binding instruction should not be revertible in one click.
|
||||
# Putting a rewrite back goes through update_rule, which takes its own
|
||||
# snapshot and leaves the undo in the history like any other edit — a silent
|
||||
# revert would erase the only record of why the rewrite happened.
|
||||
|
||||
|
||||
@rulebooks_bp.post("/rules/<int:rule_id>/relations")
|
||||
@login_required
|
||||
async def relate_rules(rule_id: int):
|
||||
|
||||
@@ -21,6 +21,7 @@ from scribe.services.verification import (
|
||||
last_verified_label as _last_verified_label,
|
||||
)
|
||||
from scribe.services import rule_versions
|
||||
from scribe.models.rule_version import RuleVersion
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -787,6 +788,48 @@ async def update_rule(
|
||||
return rule
|
||||
|
||||
|
||||
# ── Edit history (milestone 323) ───────────────────────────────────────
|
||||
#
|
||||
# The ACL-scoped reads live HERE rather than in services/rule_versions.py,
|
||||
# and not by preference: rulebooks imports rule_versions for the write path,
|
||||
# so the reverse import would be a cycle. The split is also the honest one —
|
||||
# rule_versions owns what a version IS, this module owns who may read one.
|
||||
|
||||
|
||||
async def list_rule_versions(rule_id: int, user_id: int):
|
||||
"""A rule's history, newest first. None when the rule is not readable.
|
||||
|
||||
Scoped through the rule itself, never through the version's `user_id`:
|
||||
that column is the ACTOR. Reading a rule's history is a question about
|
||||
the RULE, so anyone who can read the rule can read what it used to say,
|
||||
and anyone who cannot read the rule gets nothing — including the versions
|
||||
they personally wrote, if the rule has since moved out of their reach.
|
||||
"""
|
||||
async with async_session() as session:
|
||||
if await _fetch_owned_rule(session, rule_id, user_id) is None:
|
||||
return None
|
||||
return await rule_versions.list_versions(rule_id)
|
||||
|
||||
|
||||
async def get_rule_version(rule_id: int, version_id: int, user_id: int):
|
||||
"""One snapshot in full. None when the rule or the version is not found.
|
||||
|
||||
Takes the rule id as well as the version id so the ownership check has
|
||||
something to run against BEFORE the version is read, and so a version id
|
||||
from another rule cannot be read through a rule the caller does happen to
|
||||
own — the check and the fetch have to agree about which rule is in play.
|
||||
"""
|
||||
async with async_session() as session:
|
||||
if await _fetch_owned_rule(session, rule_id, user_id) is None:
|
||||
return None
|
||||
return (await session.execute(
|
||||
select(RuleVersion).where(
|
||||
RuleVersion.id == version_id,
|
||||
RuleVersion.rule_id == rule_id,
|
||||
)
|
||||
)).scalar_one_or_none()
|
||||
|
||||
|
||||
# ── Canon tags + typed edges (milestone 307) ───────────────────────────
|
||||
|
||||
async def set_rule_systems(
|
||||
|
||||
@@ -196,3 +196,98 @@ async def test_history_reads_newest_first(constraint):
|
||||
assert [v.statement for v in listed] == [
|
||||
"first change", "Write every `run:` step in POSIX sh.",
|
||||
]
|
||||
|
||||
|
||||
# ── The read path is ACL-scoped (milestone 323 step 3) ─────────────────
|
||||
#
|
||||
# Rule 47: every read of user data is scoped by owner. A version carries a
|
||||
# `user_id`, which makes it tempting to scope the history by it — that would
|
||||
# be wrong in both directions, and these pin which way round it goes.
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def stranger():
|
||||
"""A second user who owns nothing in the fixture above."""
|
||||
async with async_session() as s:
|
||||
other = await ensure_user(s, "rule_history_stranger")
|
||||
uid = other.id
|
||||
await s.commit()
|
||||
return uid
|
||||
|
||||
|
||||
async def test_the_owner_reads_the_history(constraint):
|
||||
await rulebooks_svc.update_rule(
|
||||
constraint["rule_id"], constraint["uid"], statement="reworded",
|
||||
)
|
||||
versions = await rulebooks_svc.list_rule_versions(
|
||||
constraint["rule_id"], constraint["uid"],
|
||||
)
|
||||
assert [v.statement for v in versions] == [
|
||||
"Write every `run:` step in POSIX sh.",
|
||||
]
|
||||
|
||||
|
||||
async def test_a_stranger_reads_nothing(constraint, stranger):
|
||||
"""None, not an empty list. The two mean different things — "not your
|
||||
rule" versus "this rule has never been reworded" — and collapsing them
|
||||
would tell a caller the rule exists and is unedited."""
|
||||
await rulebooks_svc.update_rule(
|
||||
constraint["rule_id"], constraint["uid"], statement="reworded",
|
||||
)
|
||||
assert await rulebooks_svc.list_rule_versions(
|
||||
constraint["rule_id"], stranger,
|
||||
) is None
|
||||
|
||||
|
||||
async def test_a_version_cannot_be_read_through_a_DIFFERENT_rule(constraint):
|
||||
"""The check and the fetch have to agree about which rule is in play.
|
||||
|
||||
The second rule is owned by the SAME user on purpose — an id that simply
|
||||
does not exist would pass on the ownership check alone and prove nothing
|
||||
about the `rule_id` clause. Here the caller genuinely owns the rule they
|
||||
name and genuinely owns the version's rule, and the read must still
|
||||
refuse, because scoping by version id alone is how a snapshot leaks
|
||||
through whichever rule the caller happens to be able to see.
|
||||
"""
|
||||
await rulebooks_svc.update_rule(
|
||||
constraint["rule_id"], constraint["uid"], statement="reworded",
|
||||
)
|
||||
[version] = await _versions(constraint["rule_id"])
|
||||
|
||||
async with async_session() as s:
|
||||
rule = await s.get(Rule, constraint["rule_id"])
|
||||
topic_id = rule.topic_id
|
||||
sibling = await rulebooks_svc.create_rule(
|
||||
topic_id, constraint["uid"], "A different rule", "Unrelated.",
|
||||
)
|
||||
|
||||
assert await rulebooks_svc.get_rule_version(
|
||||
sibling.id, version.id, constraint["uid"],
|
||||
) is None
|
||||
# And through its own rule it reads fine — or the assertion above would
|
||||
# pass for the wrong reason.
|
||||
assert await rulebooks_svc.get_rule_version(
|
||||
constraint["rule_id"], version.id, constraint["uid"],
|
||||
) is not None
|
||||
|
||||
|
||||
async def test_the_actor_does_not_grant_the_read(constraint, stranger):
|
||||
"""The inverse mistake. `user_id` on a version is the ACTOR, so scoping
|
||||
the history by it would hand someone the snapshots they personally wrote
|
||||
on a rule that has since moved out of their reach — and would hide, from
|
||||
the rule's owner, every edit somebody else made."""
|
||||
async with async_session() as s:
|
||||
s.add(RuleVersion(
|
||||
rule_id=constraint["rule_id"], user_id=stranger,
|
||||
title="written by someone else", statement="a stranger's edit",
|
||||
))
|
||||
await s.commit()
|
||||
|
||||
assert await rulebooks_svc.list_rule_versions(
|
||||
constraint["rule_id"], stranger,
|
||||
) is None
|
||||
|
||||
owner_view = await rulebooks_svc.list_rule_versions(
|
||||
constraint["rule_id"], constraint["uid"],
|
||||
)
|
||||
assert "a stranger's edit" in [v.statement for v in owner_view]
|
||||
|
||||
@@ -229,8 +229,9 @@ def test_register_attaches_every_tool():
|
||||
mcp = FakeMCP()
|
||||
|
||||
register(mcp)
|
||||
# 26 through milestone 307, +2 for the staleness sweep (milestone 312).
|
||||
assert len(mcp.names) == 28
|
||||
# 26 through milestone 307, +2 for the staleness sweep (milestone 312),
|
||||
# +1 for a rule's edit history (milestone 323).
|
||||
assert len(mcp.names) == 29
|
||||
# spot-check a few names
|
||||
assert "list_rulebooks" in mcp.names
|
||||
assert "create_rule" in mcp.names
|
||||
@@ -244,6 +245,8 @@ def test_register_attaches_every_tool():
|
||||
# milestone 312: the sweep, and the stamp that answers it
|
||||
assert "rules_due_for_verification" in mcp.names
|
||||
assert "mark_rule_verified" in mcp.names
|
||||
# milestone 323: what a rule used to say
|
||||
assert "rule_history" in mcp.names
|
||||
assert "unsuppress_rule_for_project" in mcp.names
|
||||
assert "suppress_topic_for_project" in mcp.names
|
||||
assert "unsuppress_topic_for_project" in mcp.names
|
||||
@@ -433,3 +436,115 @@ async def test_unrelate_rules_raises_when_the_edge_is_gone():
|
||||
from scribe.mcp.tools.rulebooks import unrelate_rules
|
||||
with pytest.raises(ValueError, match="not found"):
|
||||
await unrelate_rules(relation_id=99)
|
||||
|
||||
|
||||
# ── rule_history (milestone 323 step 3) ────────────────────────────────
|
||||
|
||||
def _fake_version(**over):
|
||||
"""A RuleVersion-shaped stand-in. A real model instance rather than a
|
||||
MagicMock, because the tool calls `to_dict` and a mock would hand back
|
||||
another mock instead of failing."""
|
||||
from scribe.models.rule_version import RuleVersion
|
||||
from datetime import datetime, timezone
|
||||
|
||||
defaults = {
|
||||
"id": 5, "rule_id": 100, "user_id": 1,
|
||||
"title": "The runner has no bash", "statement": "Use sh.",
|
||||
"why": "the image ships no bash", "how_to_apply": None,
|
||||
"when_to_apply": None, "tier": "always_on",
|
||||
"verify_with": "read the workflow's shell setting",
|
||||
"expires_when": None,
|
||||
"created_at": datetime(2026, 8, 29, tzinfo=timezone.utc),
|
||||
}
|
||||
return RuleVersion(**{**defaults, **over})
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rule_history_lists_without_the_heavy_text():
|
||||
"""The listing answers "when, and by whom". A rule's statement runs to
|
||||
thousands of characters, so a history carrying every field would cost
|
||||
more to read than the answer is worth."""
|
||||
with patch(
|
||||
"scribe.mcp.tools.rulebooks.rulebooks_svc.list_rule_versions",
|
||||
AsyncMock(return_value=[_fake_version(), _fake_version(id=4)]),
|
||||
):
|
||||
from scribe.mcp.tools.rulebooks import rule_history
|
||||
out = await rule_history(rule_id=100)
|
||||
|
||||
assert out["total"] == 2
|
||||
assert out["versions"][0]["title"] == "The runner has no bash"
|
||||
assert "statement" not in out["versions"][0]
|
||||
assert "why" not in out["versions"][0]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rule_history_opens_one_version_in_full():
|
||||
"""Passing a version id switches from the index to the text — which is
|
||||
the whole reason the listing can afford to omit it."""
|
||||
with patch(
|
||||
"scribe.mcp.tools.rulebooks.rulebooks_svc.get_rule_version",
|
||||
AsyncMock(return_value=_fake_version()),
|
||||
):
|
||||
from scribe.mcp.tools.rulebooks import rule_history
|
||||
out = await rule_history(rule_id=100, version_id=5)
|
||||
|
||||
assert out["statement"] == "Use sh."
|
||||
assert out["verify_with"] == "read the workflow's shell setting"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rule_history_says_an_empty_history_is_ordinary():
|
||||
"""Most rules have never been reworded, and nothing was written before
|
||||
milestone 323. Without this line an empty list reads as a lost history or
|
||||
a broken tool."""
|
||||
with patch(
|
||||
"scribe.mcp.tools.rulebooks.rulebooks_svc.list_rule_versions",
|
||||
AsyncMock(return_value=[]),
|
||||
):
|
||||
from scribe.mcp.tools.rulebooks import rule_history
|
||||
out = await rule_history(rule_id=100)
|
||||
|
||||
assert out["total"] == 0
|
||||
assert "never been reworded" in out["note"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rule_history_raises_when_the_rule_is_not_yours():
|
||||
"""None from the service means "not readable", and the tool must not
|
||||
turn that into an empty history — which would read as "this rule has no
|
||||
past" rather than "this is not your rule"."""
|
||||
with patch(
|
||||
"scribe.mcp.tools.rulebooks.rulebooks_svc.list_rule_versions",
|
||||
AsyncMock(return_value=None),
|
||||
):
|
||||
from scribe.mcp.tools.rulebooks import rule_history
|
||||
with pytest.raises(ValueError, match="rule 100 not found"):
|
||||
await rule_history(rule_id=100)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rule_history_raises_for_a_version_on_another_rule():
|
||||
with patch(
|
||||
"scribe.mcp.tools.rulebooks.rulebooks_svc.get_rule_version",
|
||||
AsyncMock(return_value=None),
|
||||
):
|
||||
from scribe.mcp.tools.rulebooks import rule_history
|
||||
with pytest.raises(ValueError, match="version 5 not found"):
|
||||
await rule_history(rule_id=100, version_id=5)
|
||||
|
||||
|
||||
def test_rule_history_docstring_says_what_a_version_HOLDS():
|
||||
"""The one thing a reader gets wrong unaided: an entry is the text the
|
||||
edit REPLACED, not the text it introduced. Read the other way, every
|
||||
diff comes out backwards — so the docstring has to say it, and this is
|
||||
the guard against a later tidy-up dropping the line."""
|
||||
from scribe.mcp.tools.rulebooks import rule_history
|
||||
|
||||
doc = rule_history.__doc__ or ""
|
||||
assert "REPLACED" in doc
|
||||
assert "no restore" in doc.lower(), (
|
||||
"the docstring no longer explains that a rule version cannot be "
|
||||
"restored. A caller who assumes a revert exists will look for one "
|
||||
"and, not finding it, is likely to hand-copy the old text back with "
|
||||
"no record of why."
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user