bvandeusenandClaude Opus 5.5
4e1320120d
CI & Build / Python lint (push) Successful in 2s
CI & Build / Plugin hooks (push) Successful in 16s
CI & Build / TypeScript typecheck (push) Successful in 54s
CI & Build / integration (push) Successful in 1m19s
CI & Build / Python tests (push) Successful in 2m3s
CI & Build / Build & push image (push) Successful in 18s
The open-after-moment signal proposes a mount; nothing proposed taking one off, so a wrong mount was noise at every occurrence until someone happened to notice. rule_misfired(rule_id, moment, why, reached_by) records a report against a MOUNTED pair, counted per distinct day (the MCP door carries no session id) on a new rule_moment_judgments.misfire column (migration 0119, backup v24). At three days the response carries a line asking the agent to offer the operator the fix - reject takes the rule off, unmap_action stops the action reaching the moment, confirm keeps the mount and stops the asking - and Settings > Moments lists it as an unmount proposal with the reasons and the actions that reached it. A re-mount clears the count. Taught in moments.md, missed-retrieval.md and the reply hold's wording. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
105 lines
4.6 KiB
Python
105 lines
4.6 KiB
Python
"""Moment proposals and judgments — what a mock can see (milestone 458 step 7).
|
||
|
||
The behaviour against real tables is tests/test_integration_rule_moment_judgments.py.
|
||
These pin the pieces that need no database: that the migration's CHECKs and
|
||
the model agree (rule 36), which moments count as evidence, and that the line
|
||
the signal hands a session names the tool that answers it.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import importlib.util
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from types import SimpleNamespace
|
||
|
||
from scribe.models.rule_moment_judgment import JUDGMENT_STATES, NO_MOMENT, SOURCES
|
||
from scribe.services import moments as moments_svc
|
||
from scribe.services import rule_moment_judgments as svc
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
|
||
|
||
def _migration():
|
||
path = ROOT / "alembic" / "versions" / "0118_rule_moment_judgments.py"
|
||
spec = importlib.util.spec_from_file_location("m0118", path)
|
||
module = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(module)
|
||
return module
|
||
|
||
|
||
def test_the_migration_checks_and_the_model_agree():
|
||
m = _migration()
|
||
assert tuple(m._STATES) == tuple(JUDGMENT_STATES)
|
||
assert tuple(m._SOURCES) == tuple(SOURCES)
|
||
|
||
|
||
def test_verdicts_name_real_states():
|
||
assert set(svc.VERDICTS.values()) <= set(JUDGMENT_STATES)
|
||
|
||
|
||
def test_no_moment_is_not_a_mountable_name():
|
||
""""" is the "none fits" answer; if it ever became mountable, a pass's
|
||
"no moment" would read as a mount on nothing."""
|
||
assert not moments_svc.is_moment(NO_MOMENT)
|
||
|
||
|
||
def test_the_ubiquitous_moments_are_not_evidence():
|
||
assert svc.signal_moments(["work.run", "work.verify", "work.change", "work.verify",
|
||
"not.a.moment", "skill.brainstorming"]) == [
|
||
"work.verify", "skill.brainstorming",
|
||
]
|
||
# Every skipped name is a real moment — a typo here would skip nothing.
|
||
assert all(moments_svc.is_moment(m) for m in svc.SIGNAL_SKIP)
|
||
|
||
|
||
def test_the_proposal_line_names_the_tool_that_answers_it():
|
||
rule = SimpleNamespace(id=11, title="Definition of done", kind="rule")
|
||
line = svc._proposal_line(rule, "work.verify", {"situations": ["a", "b", "c"]})
|
||
assert line.startswith("> Rule #11")
|
||
assert "`work.verify`" in line and "3 distinct sessions" in line
|
||
assert '"rule_id": 11' in line and '"moment": "work.verify"' in line
|
||
assert "judge_rule_moments" in line and '"reject"' in line
|
||
|
||
|
||
# ── The misfire (step 7b) ────────────────────────────────────────────────────
|
||
|
||
|
||
def test_a_day_is_one_situation_and_days_are_distinct():
|
||
"""The day arm keys verbatim: tokenised, "10" and "05" fall under the
|
||
minimum token length and every day of a year would be one situation."""
|
||
a = datetime(2026, 10, 5, 1, tzinfo=timezone.utc)
|
||
assert svc.misfire_situation(a) == svc.misfire_situation(a + timedelta(hours=20))
|
||
assert svc.misfire_situation(a) != svc.misfire_situation(a + timedelta(days=1))
|
||
assert svc.misfire_situation(a) != svc.misfire_situation(a + timedelta(days=31))
|
||
|
||
|
||
def test_a_misfire_counts_reasons_and_actions():
|
||
now = datetime(2026, 10, 5, tzinfo=timezone.utc)
|
||
mf = None
|
||
for i in range(svc.MISFIRE_REASONS + 2):
|
||
mf = svc.add_misfire(mf, svc.misfire_situation(now + timedelta(days=i)), 7, now,
|
||
why=f"reason {i}", reached_by="git push" if i % 2 else "make ship")
|
||
assert len(mf["situations"]) == svc.MISFIRE_REASONS + 2
|
||
assert [r["why"] for r in mf["reasons"]][-1] == f"reason {svc.MISFIRE_REASONS + 1}"
|
||
assert len(mf["reasons"]) == svc.MISFIRE_REASONS
|
||
assert mf["reached_by"] == {"make ship": 4, "git push": 3}
|
||
assert mf["projects"] == [7]
|
||
|
||
|
||
def test_the_unmount_line_offers_both_fixes():
|
||
rule = SimpleNamespace(id=10, title="CI verifies", kind="rule")
|
||
mf = {"situations": ["a", "b", "c"], "reached_by": {"actions_run_read": 3}}
|
||
line = svc._unmount_line(rule, "work.verify", mf)
|
||
assert line.startswith("> Rule #10") and "3 distinct days" in line
|
||
assert "`actions_run_read` ×3" in line
|
||
assert '"verdict": "reject"' in line and "unmap_action" in line and '"confirm"' in line
|
||
|
||
|
||
def test_a_held_reply_names_the_misfire_call_only_for_a_mount():
|
||
from scribe.services.moment_delivery import reply_hold_reason
|
||
|
||
mounted = {"rule_id": 11, "title": "Definition of done", "moment": "reply.report", "trigger": ""}
|
||
scored = {"rule_id": 12, "title": "Other", "moment": "", "score": 0.81, "trigger": ""}
|
||
assert 'rule_misfired(11, "reply.report", why)' in reply_hold_reason([mounted])
|
||
assert "rule_misfired" not in reply_hold_reason([scored])
|