from controllers.types import Facts, Nudges, Rules from features.journal_laws.briefing import BEGIN, END, brief from runner.hooks import handle from providers import PROVIDERS from resources.base import AGENT, USER from tests.conftest import fresh, refused from tests.kit import nudges, report def test_a_command_that_touches_a_rules_keyword_is_whispered_the_rule_once_per_session(): claude = PROVIDERS["claude"]() record = fresh() def use(tool, given, session="claude-1"): return handle(claude, record.root, record.env, {"hook_event_name": "PreToolUse", "session_id": session, "tool_name": tool, "tool_input": given}) received = set() def text(session="claude-1"): fresh_nudges = [n for n in Nudges(record, actor=AGENT).rows.every() if n.data.get("session") == session and n.n not in received] received.update(n.n for n in fresh_nudges) return "\n".join(n.agent_line() for n in fresh_nudges) rule = Rules(record, actor=USER).create("Never change the git branch", brief="A branch change belongs to the user", keywords=["git checkout", "git switch"]) pin = Facts(record, actor=USER).create("The viewer is built from web/", brief="web/dist is what the server serves", keywords=["npm run build"]) assert (use("Bash", {"command": "ls -la"}), text()) == ({}, ""), "a command with none of the words says nothing" assert use("Bash", {"command": "git checkout -b spike"}).get("hookSpecificOutput", {}).get("permissionDecision") is None, \ "a command carrying a rule's word is not refused" assert text() == f"rule {rule.n} — Never change the git branch — A branch change belongs to the user", \ "and the rule is whispered, naming it and its reasoning" use("Bash", {"command": "git checkout main"}) assert text() == "", "the same rule is not said again within 100 tool uses" use("Bash", {"command": "git checkout main"}, session="claude-2") assert text("claude-2").startswith(f"rule {rule.n} —") is True, "another session hears it once of its own" use("Write", {"file_path": "notes.md", "content": "then npm run build"}) assert text().startswith(f"fact {pin.n} —") is True, "a fact whose word is in what is being written is whispered" Rules(record, actor=USER).create("Keep the tree clean", keywords="word") use("Bash", {"command": "keep the tree clean"}) assert text() == "", "a row with no keywords is never whispered" assert ("keywords" in type(rule).fields) is True, "keywords are a field of the type, not loose data" Rules(record, actor=USER).set(rule.n, "keywords", '["git checkout", "git rebase"]') assert Rules(record).load(rule.n).data["keywords"] == ["git checkout", "git rebase"], "set reads a list the same way --set does" Rules(record, actor=USER).set(rule.n, "keywords", "git checkout, git rebase") assert Rules(record).load(rule.n).data["keywords"] == ["git checkout", "git rebase"], "plain words are read as the list they plainly are" Rules(record, actor=USER).set(rule.n, "keywords", "git checkout") assert Rules(record).load(rule.n).data["keywords"] == ["git checkout"], "one word is a list of one" assert ("keywords is a list" in refused(lambda: Rules(record, actor=USER).set(rule.n, "keywords", "7"))) is True, \ "something that is no kind of list is refused" def test_standing_rules_are_repeated_at_every_quarter_and_a_struck_rule_drops_out(): record = fresh() rules = Rules(record, actor=USER) rules.create("name the model on every dispatch", keywords="word") rules.create("a title never explains with a colon", keywords="word") for pct in (10, 24, 25, 40, 49.5, 50, 80): report(record, "working", "PostToolUse", context=pct) assert [n for n in nudges(record) if "in force" in n] == ["2 rules in force, read them"] * 3, "said at 25, 50 and 80: the standing rules by number" assert Nudges(record).load(1).brief == "1. name the model on every dispatch; 2. a title never explains with a colon", \ "the words carry the rules" rules.complete(2, "retired") report(record, "working", "PostToolUse", context=100) assert [n for n in nudges(record) if "in force" in n][-1] == "1 rule in force, read them", "a struck rule is not said" def test_inject_and_uninject_change_the_rules_in_the_one_block_and_pin_notices_the_rule(): record = fresh() rules = Rules(record, actor=USER) rules.create("name the model on every dispatch", keywords="word") claude_md = record.root.parent / "CLAUDE.md" agents_md = record.root.parent / "AGENTS.md" claude_md.write_text("# My project\n\nkeep this.\n") agents_md.write_text("keep this too.\n") rules.inject(1) assert (rules.load(1).injected, claude_md.read_text(), agents_md.read_text().startswith(f"{BEGIN}\n"), agents_md.read_text().endswith(f"## Rules\n\n- name the model on every dispatch\n\n{END}\n\nkeep this too.\n")) == (True, "# My project\n\n@AGENTS.md\n\nkeep this.\n", True, True), \ "inject puts the rule in the one block, in AGENTS.md, which CLAUDE.md imports" rules.update(1, title="name the model on every subagent dispatch") assert ("- name the model on every subagent dispatch\n" in agents_md.read_text(), agents_md.read_text().count(BEGIN), claude_md.read_text().count(BEGIN)) == (True, 1, 0), \ "a change rewrites the one block, in AGENTS.md" rules.uninject(1) assert (rules.load(1).injected, "## Rules" in agents_md.read_text(), agents_md.read_text().endswith(f"{END}\n\nkeep this too.\n")) == (False, False, True), \ "uninject takes the rule out and keeps the law and the project's text" notice = rules.pin(1) assert (notice.title, notice.refs, notice.data["link"], notice.data["label"]) == \ ("name the model on every subagent dispatch", ["rule:1"], "#/t/rule/1", "Open rule"), \ "pin puts the rule over chat and links it" assert rules.pin(1).n == notice.n, "pin keeps one standing notice per rule" def test_the_briefing_retires_the_old_blocks_and_leaves_a_file_it_cannot_read_safely(): record = fresh() project = record.root.parent Rules(record, actor=USER).inject(Rules(record, actor=USER).create("only rule", keywords="word").n) old = ("# Notes\n\nfirst part.\n\n\nB1\n\n\n" "\nL1\n\n\nmiddle.\n\n" "\n# Rules\n\n- old rule\n\n\nlast part.\n") (project / "AGENTS.md").write_text(old) (project / "CLAUDE.md").write_text(old) first = brief(project, record) text = (project / "AGENTS.md").read_text() assert (len(first.written), text.startswith(f"# Notes\n\n{BEGIN}\n"), text.endswith(f"- only rule\n\n{END}\n\nfirst part.\n\n\n\n\n\nmiddle.\n\n\n\nlast part.\n"), "B1" in text or "old rule" in text) == (2, True, True, False), "the three old blocks become one block at the head, the project's text kept in order" assert brief(project, record).written == (), "a file already right is not written again" spaced = "# Notes\n\n\n indented, with trailing spaces \n\n\n\nend without a newline" (project / "AGENTS.md").write_text(spaced) brief(project, record) written = (project / "AGENTS.md").read_text() assert (written.endswith(f"{END}\n" + spaced.removeprefix("# Notes\n")), brief(project, record).written) == (True, ()), \ "the project's own text is kept byte for byte, and a second write changes nothing" crlf = ("# Notes\r\n\r\nfirst \r\n\r\nold\r\n\r\n\r\nlast\r\n") (project / "AGENTS.md").write_bytes(crlf.encode()) brief(project, record) kept = (project / "AGENTS.md").read_bytes() assert kept.endswith(b"first \r\n\r\n\r\nlast\r\n") and b"\n" not in kept.replace(b"\r\n", b""), "CRLF and the bytes around a retired block survive" left = {"lone": "a\n\nb\n", "lone current": "a\n\nb\n", "conflicted": "a\n<<<<<<< ours\nb\n=======\nc\n>>>>>>> theirs\n", "newer": "\nx\n\n"} for name, had in left.items(): (project / "AGENTS.md").write_text(had) assert ((project / "AGENTS.md").read_text(), len(brief(project, record).left)) == (had, 1), f"a {name} file is left as it is and named" empty = fresh() Rules(empty, actor=USER).inject(Rules(empty, actor=USER).create("only rule", keywords="word").n) assert ((empty.root.parent / "AGENTS.md").read_text().startswith(f"# {empty.root.parent.name}\n\n{BEGIN}\n"), (empty.root.parent / "CLAUDE.md").read_text()) == (True, "@AGENTS.md\n"), \ "no instruction files yet: AGENTS.md is made with the block under the project's name, and CLAUDE.md imports it" def test_a_rule_the_agent_makes_is_sent_back_to_be_read_as_a_ruling_for_the_whole_project(): record = fresh() report(record, "working", "PreToolUse") made = Rules(record, actor=AGENT).create("Run the dashboard warm-up in this worktree", keywords=["dashboard"]) assert f"Is rule {made.n} a ruling for the whole project?" in nudges(record), "the agent is asked whether its rule binds every environment" Rules(record, actor=USER).create("Never change the git branch", keywords=["git switch"]) assert len([n for n in nudges(record) if n.startswith("Is rule")]) == 1, "a rule the user makes is not sent back" def test_old_rule_injection_flags_are_folded_into_one_flag_by_the_upgrades(): from migrations.m0002_rule_targets import run as targets from migrations.m0003_unify_rule_injection import run as unify record = fresh() rules = Rules(record, actor=AGENT) both, plain, claude = (rules.create(title, keywords="word") for title in ("both", "plain", "claude")) for rule, data in ((both, {"injected": True, "injected_codex": True}), (claude, {"injected": True})): rules.update(rule.n, **data) assert targets(record.root) == [both.ref, claude.ref], "only a rule with an old flag is rewritten" assert rules.load(both.n).data["targets"] == ["claude", "codex"] and rules.load(claude.n).data["targets"] == ["claude"], "each old flag becomes a target" assert unify(record.root) == [both.ref, claude.ref], "the targets are folded into one flag again" assert rules.load(both.n).data["injected"] is True and "targets" not in rules.load(both.n).data, "a rule with any target is injected" assert "injected" not in rules.load(plain.n).data, "a rule without any flag stays as it was"