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>
194 lines
8.8 KiB
Python
194 lines
8.8 KiB
Python
"""Real-Postgres tests for moment proposals and judgments (milestone 458 step 7).
|
|
|
|
What the step promises, against the real tables:
|
|
- the pass reads only rules nobody has answered for, and any answer — a
|
|
proposal, a mount, "no moment fits" — takes a rule off that list;
|
|
- a proposal mounts nothing; a confirm mounts, a reject is kept, and a judged
|
|
pair is never proposed again by either source;
|
|
- an edit that takes a moment off a rule is a rejection the signal respects;
|
|
- the signal counts SESSIONS, not opens, and asks once the bar is crossed;
|
|
- only the owner can propose, judge, or feed the signal.
|
|
"""
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy import select
|
|
|
|
from scribe.models import async_session
|
|
from scribe.models.rule_moment_judgment import RuleMomentJudgment
|
|
from scribe.models.rulebook import Rulebook
|
|
from scribe.models.user import User
|
|
from scribe.services import rule_moment_judgments as judgments_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 = "moment_judgments_owner"
|
|
STRANGER_USERNAME = "moment_judgments_stranger"
|
|
|
|
|
|
async def _purge_books(username: str) -> None:
|
|
"""At SETUP, for the reason test_integration_rule_moments gives."""
|
|
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):
|
|
await _purge_books(name)
|
|
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, "Judgment fixtures")
|
|
topic = await rulebooks_svc.create_topic(book.id, uid, "verification")
|
|
done = 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="about to tell the operator a piece of work is finished",
|
|
)
|
|
style = await rulebooks_svc.create_rule(
|
|
topic.id, uid, "Dates are written day first",
|
|
"Every date shown to a person is written day, month, year.",
|
|
when_to_apply="formatting a date for display",
|
|
)
|
|
return {"uid": uid, "sid": sid, "done": done, "style": style}
|
|
|
|
|
|
async def _row(rule_id: int, moment: str) -> RuleMomentJudgment | None:
|
|
async with async_session() as s:
|
|
return (await s.execute(
|
|
select(RuleMomentJudgment).where(
|
|
RuleMomentJudgment.rule_id == rule_id, RuleMomentJudgment.moment == moment,
|
|
)
|
|
)).scalar_one_or_none()
|
|
|
|
|
|
async def _unjudged_ids(uid: int) -> set[int]:
|
|
return {r["id"] for r in (await judgments_svc.unjudged_rules(uid, limit=100))["rules"]}
|
|
|
|
|
|
async def test_any_answer_takes_a_rule_off_the_pass(world):
|
|
uid, done, style = world["uid"], world["done"], world["style"]
|
|
assert {done.id, style.id} <= await _unjudged_ids(uid)
|
|
|
|
out = await judgments_svc.propose(uid, [
|
|
{"rule_id": done.id, "moments": ["work.finish", "reply.report"], "why": "about finishing"},
|
|
{"rule_id": style.id, "none": "about how a date looks, whenever one is shown"},
|
|
])
|
|
assert out["proposed"] == 2 and out["no_moment"] == 1 and not out["refused"]
|
|
assert not ({done.id, style.id} & await _unjudged_ids(uid))
|
|
# A proposal mounted nothing.
|
|
assert (await rulebooks_svc.list_rule_moments([done.id])).get(done.id) is None
|
|
|
|
|
|
async def test_a_confirm_mounts_and_a_reject_is_never_proposed_again(world):
|
|
uid, done = world["uid"], world["done"]
|
|
await judgments_svc.propose(uid, [
|
|
{"rule_id": done.id, "moments": ["work.finish", "work.plan"], "why": "w"},
|
|
])
|
|
pending = await judgments_svc.pending(uid, rule_id=done.id)
|
|
assert [p["moment"] for p in pending["rules"][0]["proposals"]] == ["work.finish", "work.plan"]
|
|
|
|
out = await judgments_svc.judge(uid, [
|
|
{"rule_id": done.id, "moment": "work.finish", "verdict": "confirm", "note": "yes"},
|
|
{"rule_id": done.id, "moment": "work.plan", "verdict": "reject", "note": "too early"},
|
|
])
|
|
assert not out["refused"]
|
|
assert (await rulebooks_svc.list_rule_moments([done.id]))[done.id] == ["work.finish"]
|
|
confirmed = await _row(done.id, "work.finish")
|
|
assert (confirmed.state, confirmed.source, confirmed.note) == ("confirmed", "pass", "yes")
|
|
assert (await _row(done.id, "work.plan")).state == "rejected"
|
|
assert (await judgments_svc.pending(uid))["total"] == 0
|
|
|
|
again = await judgments_svc.propose(uid, [
|
|
{"rule_id": done.id, "moments": ["work.plan", "work.finish"], "why": "w"},
|
|
])
|
|
assert again["proposed"] == 0
|
|
assert {s["moment"] for s in again["skipped"]} == {"work.plan", "work.finish"}
|
|
|
|
|
|
async def test_an_edit_that_unmounts_is_a_rejection_the_signal_respects(world):
|
|
uid, done = world["uid"], world["done"]
|
|
await rulebooks_svc.set_rule_moments(done.id, uid, ["work.deliver"])
|
|
await rulebooks_svc.set_rule_moments(done.id, uid, [])
|
|
row = await _row(done.id, "work.deliver")
|
|
assert (row.state, row.source) == ("rejected", "edit")
|
|
|
|
for session in ("a", "b", "c", "d"):
|
|
assert await judgments_svc.co_occurred(
|
|
uid, done.id, ["work.deliver"], situation=session,
|
|
) == ""
|
|
assert (await _row(done.id, "work.deliver")).evidence is None
|
|
|
|
|
|
async def test_the_signal_counts_sessions_and_asks_once_the_bar_is_crossed(world):
|
|
uid, done = world["uid"], world["done"]
|
|
# Ubiquitous moments are not evidence; a repeat in one session counts once.
|
|
assert await judgments_svc.co_occurred(uid, done.id, ["work.run", "work.change"],
|
|
situation="s1") == ""
|
|
assert await _row(done.id, "work.run") is None
|
|
for situation in ("s1", "s1", "s2"):
|
|
assert await judgments_svc.co_occurred(uid, done.id, ["work.verify"],
|
|
situation=situation) == ""
|
|
line = await judgments_svc.co_occurred(uid, done.id, ["work.verify"], situation="s3")
|
|
assert f"#{done.id}" in line and "`work.verify`" in line and "3 distinct sessions" in line
|
|
assert "judge_rule_moments" in line
|
|
row = await _row(done.id, "work.verify")
|
|
assert (row.state, row.source) == ("suggested", "signal")
|
|
# Asked once; the cooldown keeps the next open quiet.
|
|
assert await judgments_svc.co_occurred(uid, done.id, ["work.verify"], situation="s4") == ""
|
|
# And it waits for the operator like any proposal.
|
|
pending = await judgments_svc.pending(uid, rule_id=done.id)
|
|
[proposal] = pending["rules"][0]["proposals"]
|
|
assert proposal["source"] == "signal" and proposal["evidence"]["situations"] == 4
|
|
|
|
|
|
async def test_a_mount_overturns_no_moment_fits(world):
|
|
uid, style = world["uid"], world["style"]
|
|
await judgments_svc.propose(uid, [{"rule_id": style.id, "none": "about how a date looks"}])
|
|
assert (await _row(style.id, "")).state == "confirmed"
|
|
await judgments_svc.judge(uid, [
|
|
{"rule_id": style.id, "moment": "reply.report", "verdict": "confirm", "note": "dates in reports"},
|
|
])
|
|
assert (await _row(style.id, "")).state == "rejected"
|
|
|
|
|
|
async def test_only_the_owner_can_propose_judge_or_feed_the_signal(world):
|
|
sid, done = world["sid"], world["done"]
|
|
out = await judgments_svc.propose(sid, [{"rule_id": done.id, "moments": ["work.finish"]}])
|
|
assert out["proposed"] == 0 and out["refused"]
|
|
judged = await judgments_svc.judge(sid, [
|
|
{"rule_id": done.id, "moment": "work.finish", "verdict": "confirm"},
|
|
])
|
|
assert judged["refused"] and not judged["judged"]
|
|
for situation in ("a", "b", "c"):
|
|
assert await judgments_svc.co_occurred(sid, done.id, ["work.verify"],
|
|
situation=situation) == ""
|
|
assert await _row(done.id, "work.verify") is None
|
|
assert done.id not in await _unjudged_ids(sid)
|
|
assert (await judgments_svc.pending(sid))["total"] == 0
|
|
|
|
|
|
async def test_the_open_resolves_acts_through_the_installs_mappings(world):
|
|
"""`opened_after` reads the acts as delivery does: a push is a deliver."""
|
|
uid, done = world["uid"], world["done"]
|
|
acts = [
|
|
{"tool_name": "Bash", "tool_input": {"command": "git push origin dev"}},
|
|
{"tool_name": "Bash", "tool_input": {"command": "ls"}},
|
|
"not an act",
|
|
]
|
|
out = await judgments_svc.opened_after(uid, done.id, acts, session_id="s1")
|
|
assert "work.deliver" in out["moments"]
|
|
assert "work.run" not in out["moments"]
|
|
assert (await _row(done.id, "work.deliver")).source == "signal"
|