bvandeusenandClaude Opus 5.5
dfcf4df2e9
CI & Build / Python lint (push) Successful in 3s
CI & Build / Plugin hooks (push) Successful in 19s
CI & Build / integration (push) Failing after 43s
CI & Build / Python tests (push) Failing after 46s
CI & Build / TypeScript typecheck (push) Successful in 55s
CI & Build / Build & push image (push) Skipped
A rule written before moments existed is mounted on nothing. Step 7 records, per (rule, moment), whether it belongs there and who said so: - rule_moment_judgments (migration 0118, backup v23): suggested / confirmed / rejected, from a pass, the signal, or an edit. Moment "" is "no moment fits". - The pass: rules_to_mount lists unjudged rules; propose_rule_moments records suggestions that mount nothing; rule_moment_proposals and judge_rule_moments put them to the operator. A confirm mounts, a reject is kept so the pair is never proposed again. Same service behind REST and a "Waiting on you" panel in Settings > Moments. - Edits are judgments: set_rule_moments, the one mount write path, confirms what was added and rejects what was removed in the same transaction. - The signal: scribe_moment.sh keeps a per-session acts ledger; when a rule is opened, scribe_record_opened.sh sends the last three minutes of it to /api/plugin/rule-opened. The acts resolve through the install's mappings; work.run and work.change are not evidence. Counted per distinct session with lesson_rules' evidence model, and once due the open returns one line asking the reader to offer the mount. - scribe_session_end.sh removes the session's scribe-moment files. Plugin 2026.10.05.2003. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
279 lines
13 KiB
Python
279 lines
13 KiB
Python
"""Real-Postgres tests for rule mounts (milestone 458 step 3).
|
|
|
|
A mount is the claim "this rule arrives at this moment". These pin what that
|
|
claim depends on:
|
|
- the set given is the set stored, and a refused name stores nothing;
|
|
- only the owner can mount;
|
|
- the rule's detail reads its mounts back;
|
|
- a backup carries them, attached to the RESTORED rule's new id rather than
|
|
the old number;
|
|
- the delivery lookup (step 4) answers by home: a project's mounted rule
|
|
reaches only work bound to that project, and nobody else's mount arrives.
|
|
"""
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import select
|
|
|
|
from scribe.models import async_session
|
|
from scribe.models.project import Project
|
|
from scribe.models.rule_moment_judgment import RuleMomentJudgment
|
|
from scribe.models.rulebook import Rule, Rulebook, RulebookTopic
|
|
from scribe.models.user import User
|
|
from scribe.services import backup
|
|
from scribe.services import projects as projects_svc
|
|
from scribe.services import rulebooks as rulebooks_svc
|
|
from tests.helpers import ensure_user
|
|
|
|
pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
|
|
|
|
OWNER_USERNAME = "rule_moments_owner"
|
|
STRANGER_USERNAME = "rule_moments_stranger"
|
|
RESTORED_USERNAME = "rule_moments_restored"
|
|
TRIGGER = "about to tell the operator a piece of work is finished"
|
|
|
|
|
|
async def _purge_books(username: str) -> None:
|
|
"""At SETUP: rule writes fire a detached embedding refresh, and a teardown
|
|
delete races it (test_integration_rule_versions records why)."""
|
|
async with async_session() as s:
|
|
for user in (await s.execute(
|
|
select(User).where(User.username == username)
|
|
)).scalars().all():
|
|
for book in (await s.execute(
|
|
select(Rulebook).where(Rulebook.owner_user_id == user.id)
|
|
)).scalars().all():
|
|
await s.delete(book)
|
|
await s.commit()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def world():
|
|
for name in (OWNER_USERNAME, STRANGER_USERNAME, RESTORED_USERNAME):
|
|
await _purge_books(name)
|
|
# The restore below creates a user under a unique username; a previous
|
|
# run's must be gone first, as in the sibling roundtrip files.
|
|
async with async_session() as s:
|
|
for user in (await s.execute(
|
|
select(User).where(User.username == RESTORED_USERNAME)
|
|
)).scalars().all():
|
|
await s.delete(user)
|
|
await s.commit()
|
|
async with async_session() as s:
|
|
owner = await ensure_user(s, OWNER_USERNAME)
|
|
stranger = await ensure_user(s, STRANGER_USERNAME)
|
|
await s.commit()
|
|
uid, sid = owner.id, stranger.id
|
|
book = await rulebooks_svc.create_rulebook(uid, "Moment fixtures")
|
|
topic = await rulebooks_svc.create_topic(book.id, uid, "verification")
|
|
rule = await rulebooks_svc.create_rule(
|
|
topic.id, uid, "Done means delivered and checked",
|
|
"Work is done when it is delivered and its automated checks pass.",
|
|
when_to_apply=TRIGGER,
|
|
)
|
|
return {"uid": uid, "sid": sid, "book_id": book.id, "rule": rule}
|
|
|
|
|
|
async def test_the_set_given_is_the_set_stored(world):
|
|
uid, rule = world["uid"], world["rule"]
|
|
assert await rulebooks_svc.set_rule_moments(
|
|
rule.id, uid, ["reply.report", " Work.Finish"],
|
|
) == ["reply.report", "work.finish"]
|
|
# Read back in CATALOG order, whatever order they were given in.
|
|
assert (await rulebooks_svc.list_rule_moments([rule.id]))[rule.id] == [
|
|
"work.finish", "reply.report",
|
|
]
|
|
|
|
await rulebooks_svc.set_rule_moments(rule.id, uid, ["work.deliver"])
|
|
assert (await rulebooks_svc.list_rule_moments([rule.id]))[rule.id] == ["work.deliver"]
|
|
|
|
await rulebooks_svc.set_rule_moments(rule.id, uid, [])
|
|
assert rule.id not in await rulebooks_svc.list_rule_moments([rule.id])
|
|
|
|
|
|
async def test_a_refused_name_stores_nothing(world):
|
|
uid, rule = world["uid"], world["rule"]
|
|
await rulebooks_svc.set_rule_moments(rule.id, uid, ["work.finish"])
|
|
with pytest.raises(ValueError, match="unknown moment"):
|
|
await rulebooks_svc.set_rule_moments(rule.id, uid, ["work.deliver", "work.shipped"])
|
|
assert (await rulebooks_svc.list_rule_moments([rule.id]))[rule.id] == ["work.finish"]
|
|
|
|
|
|
async def test_only_the_owner_can_mount(world):
|
|
rule = world["rule"]
|
|
assert await rulebooks_svc.set_rule_moments(rule.id, world["sid"], ["work.finish"]) is None
|
|
assert rule.id not in await rulebooks_svc.list_rule_moments([rule.id])
|
|
|
|
|
|
async def test_the_detail_carries_the_mounts_and_omits_an_empty_set(world):
|
|
uid, rule = world["uid"], world["rule"]
|
|
bare = await rulebooks_svc.rule_detail(uid, rule)
|
|
assert "moments" not in bare
|
|
mounted = await rulebooks_svc.rule_detail(uid, rule, moments=["work.finish"])
|
|
assert mounted["moments"] == ["work.finish"]
|
|
# None leaves them alone: a later read without the argument still has them.
|
|
assert (await rulebooks_svc.rule_detail(uid, rule))["moments"] == ["work.finish"]
|
|
|
|
|
|
async def test_a_backup_carries_the_mounts_to_the_restored_rule(world):
|
|
uid, rule = world["uid"], world["rule"]
|
|
await rulebooks_svc.set_rule_moments(rule.id, uid, ["work.finish", "reply.report"])
|
|
|
|
async with async_session() as s:
|
|
payload = {
|
|
"version": backup.BACKUP_VERSION,
|
|
"users": backup._user_rows(
|
|
[(await s.execute(select(User).where(User.id == uid))).scalars().one()]
|
|
),
|
|
"rulebooks": backup._rulebook_rows(
|
|
[(await s.execute(select(Rulebook).where(
|
|
Rulebook.id == world["book_id"]))).scalars().one()]
|
|
),
|
|
"rulebook_topics": backup._topic_rows((await s.execute(
|
|
select(RulebookTopic).where(RulebookTopic.rulebook_id == world["book_id"])
|
|
)).scalars().all()),
|
|
"rules": backup._rule_rows(
|
|
[(await s.execute(select(Rule).where(Rule.id == rule.id))).scalars().one()]
|
|
),
|
|
}
|
|
exported = await backup.export_user_backup(uid)
|
|
payload["rule_moments"] = [
|
|
row for row in exported["rule_moments"] if row["rule_id"] == rule.id
|
|
]
|
|
assert sorted(r["moment"] for r in payload["rule_moments"]) == [
|
|
"reply.report", "work.finish",
|
|
]
|
|
# The mount was an edit, so it was also recorded as two judgments (step 7)
|
|
# — and those travel too, onto the restored rule.
|
|
payload["rule_moment_judgments"] = [
|
|
row for row in exported["rule_moment_judgments"] if row["rule_id"] == rule.id
|
|
]
|
|
assert sorted((r["moment"], r["state"], r["source"])
|
|
for r in payload["rule_moment_judgments"]) == [
|
|
("reply.report", "confirmed", "edit"), ("work.finish", "confirmed", "edit"),
|
|
]
|
|
payload["users"][0]["username"] = RESTORED_USERNAME
|
|
|
|
stats = await backup.restore_full_backup(payload)
|
|
assert stats["rule_moments"] == 2
|
|
assert stats["rule_moment_judgments"] == 2
|
|
|
|
async with async_session() as s:
|
|
restored_user = (await s.execute(
|
|
select(User).where(User.username == RESTORED_USERNAME)
|
|
)).scalars().one()
|
|
restored_rule = (await s.execute(
|
|
select(Rule).join(RulebookTopic, RulebookTopic.id == Rule.topic_id)
|
|
.join(Rulebook, Rulebook.id == RulebookTopic.rulebook_id)
|
|
.where(Rulebook.owner_user_id == restored_user.id)
|
|
)).scalars().one()
|
|
assert restored_rule.id != rule.id
|
|
assert (await rulebooks_svc.list_rule_moments([restored_rule.id]))[restored_rule.id] == [
|
|
"work.finish", "reply.report",
|
|
]
|
|
async with async_session() as s:
|
|
judged = (await s.execute(
|
|
select(RuleMomentJudgment.moment).where(RuleMomentJudgment.rule_id == restored_rule.id)
|
|
)).scalars().all()
|
|
assert sorted(judged) == ["reply.report", "work.finish"]
|
|
|
|
|
|
async def _purge_projects(username: str) -> None:
|
|
async with async_session() as s:
|
|
for user in (await s.execute(
|
|
select(User).where(User.username == username)
|
|
)).scalars().all():
|
|
for project in (await s.execute(
|
|
select(Project).where(Project.user_id == user.id)
|
|
)).scalars().all():
|
|
await s.delete(project)
|
|
await s.commit()
|
|
|
|
|
|
async def test_the_delivery_lookup_answers_by_home(world):
|
|
uid, sid, rule = world["uid"], world["sid"], world["rule"]
|
|
await _purge_projects(OWNER_USERNAME)
|
|
project = await projects_svc.create_project(uid, "Moment delivery fixture")
|
|
local = await rulebooks_svc.create_project_rule(
|
|
project.id, uid, "Ship notes with the release",
|
|
"A release carries its notes.", when_to_apply="publishing a release",
|
|
)
|
|
await rulebooks_svc.set_rule_moments(rule.id, uid, ["work.finish", "reply.report"])
|
|
await rulebooks_svc.set_rule_moments(local.id, uid, ["work.deliver"])
|
|
|
|
# Unbound: the global rule, under the FIRST of the given moments it is on.
|
|
got = await rulebooks_svc.rules_on_moments(uid, ["reply.report", "work.finish"])
|
|
assert [(r.id, m) for r, m in got] == [(rule.id, "reply.report")]
|
|
assert await rulebooks_svc.rules_on_moments(uid, ["work.deliver"]) == []
|
|
|
|
# Bound to the project: its own rule arrives too.
|
|
got = await rulebooks_svc.rules_on_moments(uid, ["work.deliver"], project.id)
|
|
assert [(r.id, m) for r, m in got] == [(local.id, "work.deliver")]
|
|
|
|
# Somebody else's mount never arrives — not even naming their project.
|
|
assert await rulebooks_svc.rules_on_moments(sid, ["work.finish", "work.deliver"], project.id) == []
|
|
|
|
# The plugin's "can anything arrive" answer spans every home.
|
|
assert await rulebooks_svc.mounted_moments(uid) >= {"work.finish", "reply.report", "work.deliver"}
|
|
assert not (await rulebooks_svc.mounted_moments(sid)) & {"work.finish", "work.deliver"}
|
|
|
|
# Counted the same way for the Settings view: a rule mounted twice counts
|
|
# once per moment, and another user's mounts are not in the count.
|
|
counts = await rulebooks_svc.mount_counts(uid)
|
|
assert counts.get("work.deliver", 0) >= 1 and counts.get("reply.report", 0) >= 1
|
|
assert set(counts) == await rulebooks_svc.mounted_moments(uid)
|
|
assert not set(await rulebooks_svc.mount_counts(sid)) & {"work.finish", "work.deliver"}
|
|
|
|
|
|
async def test_moment_usage_counts_deliveries_and_the_agent_opens_that_followed(world):
|
|
"""The per-moment readout (step 6), on real SQL: an EXISTS inside a
|
|
COUNT(CASE …) is a shape no mock can vouch for, and the block's own guard
|
|
would turn a database refusal into a quiet `moment_usage_failed`."""
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import delete
|
|
|
|
from scribe.models.rule_usage import PULLED, SURFACED, RuleUsageEvent
|
|
from scribe.services.retrieval_pipeline import MOMENT_RULE_SOURCE
|
|
from scribe.services.retrieval_telemetry import moment_usage
|
|
|
|
uid, sid, rule = world["uid"], world["sid"], world["rule"]
|
|
other = await rulebooks_svc.create_rule(
|
|
rule.topic_id, uid, "Close the task with the push",
|
|
"A task closes in the turn its CI goes green.",
|
|
when_to_apply="the CI run on a pushed commit has just gone green",
|
|
)
|
|
now = datetime.now(timezone.utc)
|
|
|
|
def ev(user, rule_id, event, source, ago, detail=None):
|
|
return RuleUsageEvent(user_id=user, rule_id=rule_id, event=event, source=source,
|
|
detail=detail, created_at=now - ago)
|
|
|
|
async with async_session() as s:
|
|
await s.execute(delete(RuleUsageEvent).where(RuleUsageEvent.user_id.in_([uid, sid])))
|
|
s.add_all([
|
|
ev(uid, rule.id, SURFACED, MOMENT_RULE_SOURCE, timedelta(hours=3), "work.finish"),
|
|
ev(uid, rule.id, SURFACED, MOMENT_RULE_SOURCE, timedelta(hours=2), "work.finish"),
|
|
ev(uid, other.id, SURFACED, MOMENT_RULE_SOURCE, timedelta(hours=1), "work.finish"),
|
|
ev(uid, rule.id, SURFACED, MOMENT_RULE_SOURCE, timedelta(hours=1), "reply.report"),
|
|
# An agent opened `rule` after its first work.finish delivery, and
|
|
# BEFORE its reply.report one — so it counts at the first only.
|
|
ev(uid, rule.id, PULLED, "mcp_get_rule", timedelta(minutes=150)),
|
|
# A person opening `other` in the browser is not an agent open.
|
|
ev(uid, other.id, PULLED, "rest_rule", timedelta(minutes=5)),
|
|
# A ranked surfacing is not a delivery at a moment.
|
|
ev(uid, other.id, SURFACED, "pre_tool_rule", timedelta(minutes=5)),
|
|
# Nor is a delivery older than the window, or someone else's.
|
|
ev(uid, other.id, SURFACED, MOMENT_RULE_SOURCE, timedelta(days=3), "work.deliver"),
|
|
ev(sid, rule.id, SURFACED, MOMENT_RULE_SOURCE, timedelta(minutes=5), "work.finish"),
|
|
])
|
|
await s.commit()
|
|
|
|
block = await moment_usage(uid, now - timedelta(days=1))
|
|
assert "moment_usage_failed" not in block
|
|
assert set(block["by_moment"]) == {"work.finish", "reply.report"}
|
|
finish = block["by_moment"]["work.finish"]
|
|
assert (finish["delivered"], finish["rules"], finish["opened"]) == (3, 2, 1)
|
|
report = block["by_moment"]["reply.report"]
|
|
assert (report["delivered"], report["rules"], report["opened"]) == (1, 1, 0)
|
|
assert report["last_delivered_at"]
|