jbaruch/coding-policy
General-purpose coding policy for Baruch's AI agents
73
Quality
91%
Does it follow best practices?
Run evals on this skill
Adds up to 20 points to the overall score
View guide
Securityby
Low
Low-risk findings worth noting
test_cli.pyskills/herdr-foreman/tests/
"""Tests for foreman.cli.
Every test drives `main()` with an injected herdr client and captured streams,
so nothing here spawns a process or reads the real clock (`--now` is always
passed where a timestamp would otherwise be taken).
"""
import os as _os
import sys as _sys
# Run as a script (`python3 tests/test_x.py`), Python puts tests/ on sys.path
# rather than the repo root, so neither `foreman` nor `tests.fakes` would
# resolve. Under `-m unittest` from the root this is already true and the
# insert is a no-op. The consuming repo's runner executes files as scripts.
_ROOT = _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__)))
if _ROOT not in _sys.path:
_sys.path.insert(0, _ROOT)
import contextlib
import io
import json
import shutil
import tempfile
import unittest
from unittest.mock import patch
from pathlib import Path
from types import SimpleNamespace
from foreman import attention, cli, runnable
from foreman.report_delivery import marker_columns
from foreman.cli import build_parser, main
from foreman.errors import UsageError
from foreman.herdr import HerdrClient
from foreman.partition import slice_digest
from foreman.state import STATE_SCHEMA_VERSION, add_assignment, empty_state, load_state_checked, save_state
from tests.fakes import (
FakeRunner,
ScriptedReads,
agent_json,
composer_reads,
ok_json,
pane_layout,
)
AT = "2026-02-03T10:00:00+00:00"
CONFIG = {
"schema_version": 1,
"agents": [
{
"name": "claude",
"kind": "claude",
"usage_prompt": "/usage",
"usage_marker": "Current week",
"usage_read_source": "visible",
"slash_delivery": "paste",
"composer_glyph": "❯ ",
"recover_keys": ["esc"],
"close_keys": ["esc"],
"clear_prompt": "/clear",
},
{
"name": "codex",
"kind": "codex",
"usage_prompt": "/status",
"usage_marker": "Weekly limit",
"usage_read_source": "recent-unwrapped",
"slash_delivery": "type",
"slash_enter_count": 2,
"composer_glyph": "› ",
"composer_placeholders": ["Ask Codex to do anything"],
"recover_keys": [],
"clear_prompt": "/new",
},
{
"name": "grok",
"kind": "grok",
"usage_prompt": "/usage",
"usage_marker": "Weekly limit",
"usage_read_source": "visible",
"slash_delivery": "type",
"composer_glyph": "│ ❯",
"recover_keys": ["esc"],
"close_keys": ["esc"],
"clear_prompt": "/new",
"dialog_next_tab_keys": ["tab"],
"idle_markers": ["Shift+Tab:mode"],
"working_markers": ["Esc:cancel"],
},
],
}
SNAPSHOT = {
"schema_version": 1,
"measured_at": AT,
"agents": {
"claude": {"kind": "claude", "state": "idle", "headroom_pct": 92.0},
"codex": {"kind": "codex", "state": "idle", "headroom_pct": 87.0},
"grok": {"kind": "grok", "state": "done", "headroom_pct": 100.0},
},
}
CLAUDE_PANE = (
" Current session\n ████ 8% used\n Resets 12:49am (Europe/Oslo)\n"
" Current week (all models)\n █ 2% used\n Resets Sep 5 at 11:59pm (Europe/Oslo)\n"
)
GROK_PANE = " Weekly limit: 0%\n Next reset: September 6, 12:55\n Credits: $16.42\n"
# Verbatim grok footers, as the idle probe reads them.
GROK_WORKING_FOOTER = " │ ❯ go │\n Shift+Tab:mode │ Esc:cancel │ Ctrl+.:shortcuts\n"
GROK_IDLE_FOOTER = " │ ❯ │\n Shift+Tab:mode │ Ctrl+.:shortcuts\n"
class CliCase(unittest.TestCase):
"""Shared temp workspace: config, state, briefs, snapshot."""
def setUp(self):
# These fixtures isolate the existing CLI/dispatch contracts. Real
# retrospective gates and storage are exercised in test_retrospective_runtime.
retro = patch("foreman.cli.retrospective_runtime.Guard")
retro.start()
self.addCleanup(retro.stop)
self.tmp = Path(tempfile.mkdtemp(prefix="foreman-cli-test-"))
self.addCleanup(shutil.rmtree, self.tmp)
self.config = self.tmp / "config.json"
self.config.write_text(json.dumps(CONFIG), encoding="utf-8")
self.state = self.tmp / "state.json"
self.snapshot = self.tmp / "snapshot.json"
self.snapshot.write_text(json.dumps(SNAPSHOT), encoding="utf-8")
self.common = self.tmp / "COMMON.md"
self.common.write_text("# common\n", encoding="utf-8")
self.briefs = {}
for role in ("developer", "tester", "reviewer"):
path = self.tmp / (role + ".md")
path.write_text("# " + role + "\n", encoding="utf-8")
self.briefs[role] = path
# The seat brief carries its partition digest the way the composer
# renders it, so a bound apply can read it back (#453).
# Its own attribute, never `self.briefs`: subclasses derive per-role
# fixtures from that map, and a seat is not one of the round's roles.
self.seat_plan = bound_seat_plan({"reviewer#api": "grok"})
self.seat_brief = self.tmp / "reviewer-api.md"
self.seat_brief.write_text(
seat_brief_text("reviewer#api", self.seat_plan), encoding="utf-8")
# A stand-in herdr that always fails, for the paths that build a real
# client instead of taking an injected one.
self.fake_herdr = self.tmp / "herdr-stub"
self.fake_herdr.write_text(
'#!/bin/sh\necho \'{"error":{"code":"agent_not_found","message":"no"}}\' >&2\nexit 1\n',
encoding="utf-8",
)
self.fake_herdr.chmod(0o755)
self.out = io.StringIO()
self.err = io.StringIO()
def run_cli(self, argv, client=None):
code = main(argv, stdout=self.out, stderr=self.err, client=client)
return code, self.out.getvalue(), self.err.getvalue()
def base(self):
return ["--config", str(self.config), "--state", str(self.state)]
@staticmethod
def assign_report(brief, report):
"""Give a fixture brief the `REPORT:` line a composed brief carries."""
text = Path(brief).read_text(encoding="utf-8")
if "REPORT: " + str(report) not in text.splitlines():
separator = "" if not text or text.endswith("\n") else "\n"
Path(brief).write_text(text + separator + "REPORT: " + str(report) + "\n", encoding="utf-8")
def seat_report(self, name):
"""A seat report path its brief assigns, as a composed brief would."""
report = str(self.tmp / name)
self.assign_report(self.seat_brief, report)
return report
def brief_args(self, *roles):
args = []
for role in roles:
args += ["--brief", "{}={}".format(role, self.briefs[role])]
return args
class ParserTest(unittest.TestCase):
def test_every_subcommand_is_registered(self):
parser = build_parser()
for command in ("measure", "plan", "apply", "state"):
args = parser.parse_args([command, "--assignments", "{}", "--common", "x"] if command == "apply" else [command])
self.assertEqual(args.command, command)
def test_a_missing_subcommand_exits_two(self):
# argparse prints its usage to the real stderr; capture it so a
# deliberate failure does not litter the test run.
with contextlib.redirect_stderr(io.StringIO()):
with self.assertRaises(SystemExit) as caught:
build_parser().parse_args([])
self.assertEqual(caught.exception.code, 2)
def test_apply_requires_assignments_and_common(self):
with contextlib.redirect_stderr(io.StringIO()):
with self.assertRaises(SystemExit):
build_parser().parse_args(["apply"])
class TraceFlagTest(CliCase):
"""--trace has to reach the real client, which tests never inject."""
def test_the_flag_is_accepted_before_and_after_the_subcommand(self):
parser = build_parser()
self.assertTrue(parser.parse_args(["--trace", "state"]).trace)
self.assertTrue(parser.parse_args(["state", "--trace"]).trace)
def test_it_is_absent_by_default(self):
self.assertFalse(getattr(build_parser().parse_args(["state"]), "trace", False))
def test_every_herdr_command_is_printed_to_stderr(self):
# No client injected: the CLI builds one and wires the trace sink to
# the same stderr stream the warnings use.
code, out, err = self.run_cli(
self.base()
+ [
"--trace",
"measure",
"--marker-poll-interval",
"0",
"--agent",
"grok",
"--now",
AT,
"--herdr-bin",
str(self.fake_herdr),
]
)
self.assertEqual(code, 1) # the stub herdr refuses every command
self.assertIn("herdr> {} agent get grok".format(self.fake_herdr), err)
self.assertIn("exit=1", err)
self.assertIn("agent_not_found", err)
# stdout stays the machine-readable snapshot; tracing never pollutes it
self.assertEqual(json.loads(out)["failed_agents"], ["grok"])
def test_without_the_flag_nothing_is_traced(self):
code, _, err = self.run_cli(
self.base()
+ [
"measure",
"--marker-poll-interval",
"0",
"--agent",
"grok",
"--now",
AT,
"--herdr-bin",
str(self.fake_herdr),
]
)
self.assertEqual(code, 1)
self.assertNotIn("herdr>", err)
class StateCommandTest(CliCase):
def test_prints_a_fresh_document_when_no_state_exists(self):
code, out, err = self.run_cli(self.base() + ["state"])
self.assertEqual(code, 0)
self.assertEqual(
json.loads(out), empty_state()
)
self.assertEqual(err, "")
def test_prints_what_was_written(self):
self.state.write_text(
json.dumps(
{
"schema_version": 1,
"snapshots": [SNAPSHOT],
"assignments": [{"at": AT, "role": "developer", "agent": "grok"}],
}
),
encoding="utf-8",
)
code, out, _ = self.run_cli(self.base() + ["state"])
self.assertEqual(code, 0)
self.assertEqual(json.loads(out)["assignments"][0]["agent"], "grok")
def test_routes_state_warnings_to_the_cli_stderr(self):
self.state.write_text('{"schema_version": 2, broken', encoding="utf-8")
code, out, err = self.run_cli(self.base() + ["state"])
self.assertEqual(code, 0)
self.assertEqual(json.loads(out)["snapshots"], [])
self.assertIn("foreman: state file", err)
self.assertIn(str(self.state), err)
class PlanCommandTest(CliCase):
def test_plans_from_a_snapshot_file(self):
code, out, err = self.run_cli(
self.base() + ["plan", "--roles", "developer,tester,reviewer", "--snapshot", str(self.snapshot)]
)
self.assertEqual(code, 0)
self.assertEqual(
json.loads(out)["assignments"],
{"developer": "grok", "tester": "claude", "reviewer": "codex"},
)
self.assertEqual(err, "")
def test_routes_state_warnings_to_the_cli_stderr(self):
self.state.write_text('{"schema_version": 2, broken', encoding="utf-8")
code, out, err = self.run_cli(
self.base() + ["plan", "--snapshot", str(self.snapshot)]
)
self.assertEqual(code, 0)
self.assertIn("assignments", json.loads(out))
self.assertIn("foreman: state file", err)
self.assertIn(str(self.state), err)
def test_roles_default_to_developer_tester_reviewer(self):
code, out, _ = self.run_cli(self.base() + ["plan", "--snapshot", str(self.snapshot)])
self.assertEqual(code, 0)
self.assertEqual(list(json.loads(out)["assignments"]), ["developer", "tester", "reviewer"])
def test_snapshot_ref_names_the_file(self):
_, out, _ = self.run_cli(self.base() + ["plan", "--snapshot", str(self.snapshot)])
self.assertEqual(json.loads(out)["snapshot_ref"]["source"], str(self.snapshot))
def test_falls_back_to_the_newest_snapshot_in_state(self):
self.state.write_text(
json.dumps({"schema_version": 1, "snapshots": [SNAPSHOT], "assignments": []}),
encoding="utf-8",
)
code, out, _ = self.run_cli(self.base() + ["plan"])
self.assertEqual(code, 0)
self.assertEqual(json.loads(out)["assignments"]["developer"], "grok")
def test_previous_assignments_break_a_headroom_tie(self):
tied = json.loads(json.dumps(SNAPSHOT))
for record in tied["agents"].values():
record["headroom_pct"] = 50.0
tied_path = self.tmp / "tied.json"
tied_path.write_text(json.dumps(tied), encoding="utf-8")
self.state.write_text(
json.dumps(
{
"schema_version": 1,
"snapshots": [],
"assignments": [
{"at": AT, "role": "developer", "agent": "claude"},
{"at": AT, "role": "developer", "agent": "codex"},
],
}
),
encoding="utf-8",
)
_, out, _ = self.run_cli(self.base() + ["plan", "--snapshot", str(tied_path)])
self.assertEqual(json.loads(out)["assignments"]["developer"], "grok")
def test_no_snapshot_anywhere_is_an_actionable_error(self):
code, out, err = self.run_cli(self.base() + ["plan"])
self.assertEqual(code, 1)
self.assertEqual(out, "")
self.assertIn("`{}`".format(runnable.command("measure")), json.loads(err)["message"])
def test_missing_snapshot_file_is_an_actionable_error(self):
code, _, err = self.run_cli(self.base() + ["plan", "--snapshot", str(self.tmp / "no.json")])
self.assertEqual(code, 1)
self.assertEqual(json.loads(err)["error"], "plan_error")
def test_plan_never_touches_herdr(self):
runner = FakeRunner()
code, _, _ = self.run_cli(
self.base() + ["plan", "--snapshot", str(self.snapshot)],
client=HerdrClient(runner=runner),
)
self.assertEqual(code, 0)
self.assertEqual(runner.calls, [])
def test_exclude_bars_an_agent_from_one_role(self):
code, out, _ = self.run_cli(
self.base()
+ ["plan", "--snapshot", str(self.snapshot), "--exclude", "developer=grok"]
)
self.assertEqual(code, 0)
assignments = json.loads(out)["assignments"]
self.assertEqual(assignments["developer"], "claude")
self.assertNotEqual(assignments["developer"], "grok")
def test_exclude_repeats_to_bar_the_author_from_two_seats(self):
code, out, _ = self.run_cli(
self.base()
+ [
"plan",
"--snapshot",
str(self.snapshot),
"--exclude",
"reviewer=grok",
"--exclude",
"tester=grok",
]
)
self.assertEqual(code, 0)
assignments = json.loads(out)["assignments"]
self.assertEqual(assignments["developer"], "grok")
self.assertNotIn("grok", [assignments["reviewer"], assignments["tester"]])
def test_exclude_takes_a_comma_separated_list(self):
code, out, _ = self.run_cli(
self.base()
+ [
"plan",
"--roles",
"developer,reviewer",
"--snapshot",
str(self.snapshot),
"--exclude",
"reviewer=grok,claude",
]
)
self.assertEqual(code, 0)
self.assertEqual(json.loads(out)["assignments"]["reviewer"], "codex")
def test_a_malformed_exclude_is_an_actionable_usage_error(self):
code, out, err = self.run_cli(
self.base() + ["plan", "--snapshot", str(self.snapshot), "--exclude", "grok"]
)
self.assertEqual(code, 1)
self.assertEqual(out, "")
self.assertEqual(json.loads(err)["error"], "usage_error")
self.assertIn("ROLE=AGENT", json.loads(err)["message"])
def test_a_snapshot_missing_a_configured_agent_refuses_end_to_end(self):
# The CLI is what hands the planner its roster; a regression dropping
# that argument restores jbaruch/coding-policy#395 while the planner's
# own tests stay green.
partial = json.loads(json.dumps(SNAPSHOT))
partial["agents"] = {"codex": partial["agents"]["codex"]}
self.snapshot.write_text(json.dumps(partial), encoding="utf-8")
code, out, err = self.run_cli(
self.base() + ["plan", "--roles", "reviewer", "--snapshot", str(self.snapshot)]
)
self.assertEqual(code, 1)
self.assertEqual(out, "")
self.assertEqual(json.loads(err)["error"], "plan_error")
self.assertIn("does not cover claude, grok", json.loads(err)["message"])
self.assertEqual(json.loads(err)["details"]["uncovered"], ["claude", "grok"])
def test_role_costs_in_the_config_reweigh_the_seats(self):
config = json.loads(json.dumps(CONFIG))
config["role_costs"] = {"reviewer": 40}
self.config.write_text(json.dumps(config), encoding="utf-8")
code, out, _ = self.run_cli(self.base() + ["plan", "--snapshot", str(self.snapshot)])
self.assertEqual(code, 0)
# reviewer now outweighs developer, so it is filled first and takes
# the agent with the most headroom.
self.assertEqual(json.loads(out)["assignments"]["reviewer"], "grok")
def test_a_broken_role_costs_map_fails_naming_the_config(self):
config = json.loads(json.dumps(CONFIG))
config["role_costs"] = {"reviewer": "cheap"}
self.config.write_text(json.dumps(config), encoding="utf-8")
code, _, err = self.run_cli(self.base() + ["plan", "--snapshot", str(self.snapshot)])
self.assertEqual(code, 1)
self.assertEqual(json.loads(err)["error"], "config_error")
self.assertIn(str(self.config), json.loads(err)["message"])
def test_planning_works_on_a_machine_with_no_config_at_all(self):
code, out, _ = self.run_cli(
[
"--config",
str(self.tmp / "absent.json"),
"--state",
str(self.state),
"plan",
"--snapshot",
str(self.snapshot),
]
)
self.assertEqual(code, 0)
self.assertEqual(json.loads(out)["assignments"]["developer"], "grok")
class MeasureCommandTest(CliCase):
CORRUPT_STATE = '{"schema_version": 2, "snapshots": [broken'
def _client(self, statuses, footers=None):
footers = footers or {}
runner = FakeRunner()
panes = {"claude": "w2:p1", "codex": "w3:p1", "grok": "w4:p1"}
texts = {"claude": CLAUDE_PANE, "grok": GROK_PANE}
for name, status in statuses.items():
runner.set("agent get " + name, agent_json(name, status, panes[name]))
runner.set("agent prompt " + name, ok_json("agent_prompt"))
runner.set("agent send-keys " + name, ok_json("agent_send_keys"))
if name in texts:
runner.set("agent read " + name, texts[name])
if name in footers:
runner.set(
"agent read {} --source visible --lines 40".format(name), footers[name]
)
runner.set("pane wait-output", ok_json("output_matched"))
runner.set("pane send-text", ok_json("pane_send_text"))
runner.set("pane send-keys", ok_json("pane_send_keys"))
runner.set("pane rename", ok_json("pane_rename"))
for name in statuses:
runner.responses[
"agent read {} --source visible --lines 20".format(name)
] = composer_reads(name)
self.runner = runner
return HerdrClient(runner=runner)
def test_measures_the_named_agents_only(self):
client = self._client({"claude": "idle", "grok": "done"})
code, out, err = self.run_cli(
self.base() + ["measure", "--marker-poll-interval", "0", "--composer-settle", "0", "--agent", "claude", "--agent", "grok", "--now", AT],
client=client,
)
self.assertEqual(code, 0)
payload = json.loads(out)
self.assertEqual(sorted(payload["agents"]), ["claude", "grok"])
self.assertEqual(payload["measured_at"], AT)
self.assertEqual(err, "")
def test_snapshot_is_appended_to_the_state_file(self):
client = self._client({"grok": "idle"})
self.run_cli(self.base() + ["measure", "--marker-poll-interval", "0", "--composer-settle", "0", "--agent", "grok", "--now", AT], client=client)
state = json.loads(self.state.read_text(encoding="utf-8"))
self.assertEqual(len(state["snapshots"]), 1)
self.assertEqual(state["snapshots"][0]["agents"]["grok"]["headroom_pct"], 100.0)
def test_a_busy_agent_is_skipped_and_never_written_to(self):
client = self._client({"claude": "idle", "codex": "working", "grok": "done"})
code, out, _ = self.run_cli(self.base() + ["measure", "--marker-poll-interval", "0", "--composer-settle", "0", "--now", AT], client=client)
self.assertEqual(code, 0)
codex = json.loads(out)["agents"]["codex"]
self.assertEqual(codex["state"], "working")
self.assertIsNone(codex["headroom_pct"])
self.assertEqual([c for c in self.runner.writes() if "codex" in c], [])
def test_a_parse_failure_exits_one_but_still_prints_the_snapshot(self):
client = self._client({"claude": "idle"})
self.runner.set("agent read claude", "nothing useful here\n")
code, out, err = self.run_cli(
self.base() + ["measure", "--marker-poll-interval", "0", "--composer-settle", "0", "--agent", "claude", "--now", AT], client=client
)
self.assertEqual(code, 1)
self.assertEqual(json.loads(out)["failed_agents"], ["claude"])
self.assertEqual(json.loads(err)["error"], "measure_incomplete")
def test_records_which_signal_decided_the_state(self):
client = self._client({"grok": "idle"})
_, out, _ = self.run_cli(
self.base() + ["measure", "--marker-poll-interval", "0", "--composer-settle", "0", "--agent", "grok", "--now", AT], client=client
)
record = json.loads(out)["agents"]["grok"]
self.assertEqual(record["state_source"], "herdr")
self.assertEqual(record["herdr_state"], "idle")
def test_a_stale_herdr_state_is_overturned_and_warned_about_on_stderr(self):
client = self._client({"grok": "working"}, footers={"grok": GROK_IDLE_FOOTER})
code, out, err = self.run_cli(
self.base() + ["measure", "--marker-poll-interval", "0", "--composer-settle", "0", "--agent", "grok", "--now", AT], client=client
)
self.assertEqual(code, 0)
record = json.loads(out)["agents"]["grok"]
self.assertEqual(record["state"], "idle")
self.assertEqual(record["herdr_state"], "working")
self.assertEqual(record["state_source"], "probe")
self.assertEqual(record["headroom_pct"], 100.0)
self.assertIn("stale", err)
def test_a_genuinely_working_agent_is_still_skipped(self):
client = self._client({"grok": "working"}, footers={"grok": GROK_WORKING_FOOTER})
code, out, _ = self.run_cli(
self.base() + ["measure", "--marker-poll-interval", "0", "--composer-settle", "0", "--agent", "grok", "--now", AT], client=client
)
self.assertEqual(code, 0)
self.assertTrue(json.loads(out)["agents"]["grok"]["skipped"])
self.assertEqual(self.runner.writes(), [])
def test_each_agent_asks_for_usage_by_its_configured_path_end_to_end(self):
client = self._client({"claude": "idle", "grok": "done"})
code, _, _ = self.run_cli(
self.base()
+ [
"measure",
"--marker-poll-interval",
"0",
"--agent",
"claude",
"--agent",
"grok",
"--now",
AT,
],
client=client,
)
self.assertEqual(code, 0)
commands = self.runner.commands()
self.assertIn("agent prompt claude /usage", commands) # claude pastes
self.assertIn("pane send-text w4:p1 /usage", commands) # grok types
self.assertNotIn("agent prompt grok /usage", commands)
def test_unknown_agent_name_is_an_actionable_error(self):
code, out, err = self.run_cli(
self.base() + ["measure", "--marker-poll-interval", "0", "--composer-settle", "0", "--agent", "gemini", "--now", AT],
client=HerdrClient(runner=FakeRunner()),
)
self.assertEqual(code, 1)
self.assertEqual(out, "")
self.assertIn("gemini", json.loads(err)["message"])
def test_missing_config_names_the_example_file(self):
code, _, err = self.run_cli(
["--config", str(self.tmp / "gone.json"), "--state", str(self.state), "measure"],
client=HerdrClient(runner=FakeRunner()),
)
self.assertEqual(code, 1)
self.assertIn("config.example.json", json.loads(err)["message"])
def test_an_unreadable_state_file_is_never_overwritten(self):
# load_state leaves such a file exactly as found and hands back an
# empty document. Saving that document over it would undo precisely
# that preservation, taking the ledger with it.
self.state.write_text(self.CORRUPT_STATE, encoding="utf-8")
client = self._client({"grok": "done"})
code, out, err = self.run_cli(
self.base()
+ [
"measure",
"--marker-poll-interval", "0",
"--composer-settle", "0",
"--agent", "grok",
"--now", AT,
],
client=client,
)
self.assertNotEqual(code, 0)
self.assertIn("would destroy its contents", err)
self.assertIn(".bak", err)
self.assertEqual(self.state.read_text(encoding="utf-8"), self.CORRUPT_STATE)
self.assertEqual(out, "")
def test_a_missing_state_file_still_writes(self):
# Nothing to lose is not the same as something unreadable.
client = self._client({"grok": "done"})
code, _out, _err = self.run_cli(
self.base()
+ [
"measure",
"--marker-poll-interval", "0",
"--composer-settle", "0",
"--agent", "grok",
"--now", AT,
],
client=client,
)
self.assertEqual(code, 0)
self.assertTrue(self.state.exists())
def bound_seat_plan(assignments, slice_paths=None):
"""An apply document for seats, carrying the boundary `plan` checked.
A seated apply is refused without it: the digests are what tie each brief
and its dispatch to the partition `validate-partition` accepted (#453).
"""
from foreman.partition import seat_digest, slice_digest
paths = slice_paths or {seat: ["src/{}/*".format(seat.split("#", 1)[1])] for seat in assignments}
return {"schema_version": 1, "assignments": dict(assignments),
"slice_paths": paths, "slice_digest": slice_digest(paths),
"seat_digests": {seat: seat_digest(seat, globs) for seat, globs in paths.items()}}
def seat_brief_text(seat, plan):
"""A brief carrying the canonical scope block a seated dispatch checks."""
from foreman.partition import slice_scope
return "# {}\n\n{}\n".format(
seat, slice_scope(seat, plan["slice_paths"][seat], plan["seat_digests"][seat]))
def validated_partition(slices=None, changed=None):
"""What `validate-partition` writes: RESOLVED paths plus its `changed` set.
Not the globs the document it read carried — `validate()` replaces each
slice's patterns with the changed files it assigned them, so a result's
`slices[].paths` are literal names. `plan --partition` seats from that
result, never the document, so a round cannot be seated against a
partition nobody checked (#453).
"""
slices = slices or [{"name": "api", "paths": ["src/api/routes.py"]},
{"name": "core", "paths": ["src/core/db.py"]}]
return {"schema_version": 2, "role": "reviewer", "slices": slices,
"changed": changed or ["src/api/routes.py", "src/core/db.py"],
# What `validate-partition` stamps: the diff the slices were proven over (#460).
"proof": {"repo": "/repo", "base": "b" * 40, "head": "c" * 40}}
class ApplyCommandTest(CliCase):
CORRUPT_STATE = '{"schema_version": 2, "snapshots": [broken'
def _client(self, statuses, footers=None, sessions=None):
footers = footers or {}
runner = FakeRunner()
panes = {"claude": "w2:p1", "codex": "w3:p1", "grok": "w4:p1"}
for name, status in statuses.items():
runner.set("agent get " + name, agent_json(name, status, panes[name], (sessions or {}).get(name)))
runner.set("agent prompt " + name, ok_json("agent_prompt"))
runner.set("agent wait " + name, ok_json("agent_wait"))
runner.set(
"agent read {} --source visible --lines 40".format(name),
footers.get(name, GROK_WORKING_FOOTER),
)
runner.set("pane send-text", ok_json("pane_send_text"))
runner.set("pane send-keys", ok_json("pane_send_keys"))
runner.set("pane rename", ok_json("pane_rename"))
for name in statuses:
runner.responses[
"agent read {} --source visible --lines 20".format(name)
] = composer_reads(name)
self.runner = runner
return HerdrClient(binary="herdr", runner=runner)
def test_dry_run_prints_commands_and_sends_nothing(self):
client = self._client({})
code, out, err = self.run_cli(
self.base()
+ [
"apply",
"--composer-settle",
"0",
"--assignments",
json.dumps({"developer": "grok"}),
"--common",
str(self.common),
"--dry-run",
]
+ self.brief_args("developer"),
client=client,
)
self.assertEqual(code, 0)
payload = json.loads(out)
self.assertTrue(payload["dry_run"])
self.assertEqual(self.runner.calls, [])
shells = [command["shell"] for command in payload["steps"][0]["commands"]]
self.assertEqual(shells[0], "herdr agent get grok")
self.assertIn("herdr pane send-text PANE-ID-RESOLVED-AT-RUN-TIME /new", shells)
self.assertIn("herdr pane send-keys PANE-ID-RESOLVED-AT-RUN-TIME enter", shells)
self.assertTrue(any("DEVELOPER" in shell for shell in shells))
self.assertEqual(shells[-1], "herdr agent wait grok --until working --timeout 15000")
self.assertEqual(err, "")
def test_dry_run_writes_nothing_to_the_state_file(self):
client = self._client({})
self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", json.dumps({"developer": "grok"}), "--common", str(self.common), "--dry-run"]
+ self.brief_args("developer"),
client=client,
)
self.assertFalse(self.state.exists())
def test_assignments_can_be_a_file_of_plan_output(self):
plan_file = self.tmp / "plan.json"
plan_file.write_text(
json.dumps({"schema_version": 1, "assignments": {"developer": "grok"}}), encoding="utf-8"
)
client = self._client({})
code, out, _ = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", str(plan_file), "--common", str(self.common), "--dry-run"]
+ self.brief_args("developer"),
client=client,
)
self.assertEqual(code, 0)
self.assertEqual(json.loads(out)["steps"][0]["agent"], "grok")
def test_a_partitioned_plan_dispatches_each_seat(self):
# coding-policy#434: the seat names the planner emits must survive
# apply — brief templates, requirements, round tiers and the ledger all
# resolve the responsibility a seat fills.
partition = self.tmp / "partition.json"
partition.write_text(json.dumps(validated_partition()))
# With --task, which is the documented invocation and the one that
# exercises the contribution-exclusion path.
out = io.StringIO()
code = main(self.base() + ["plan", "--roles", "reviewer", "--partition", str(partition),
"--task", "t-partition", "--now", AT,
"--snapshot", str(self.snapshot)], stdout=out)
self.assertEqual(code, 0, out.getvalue())
plan = json.loads(out.getvalue())
self.assertEqual(sorted(plan["assignments"]), ["reviewer#api", "reviewer#core"])
self.assertEqual(len(set(plan["assignments"].values())), 2)
# The composer requires each seat's owned paths and reads no partition,
# so the plan carries them out of the validated document (#434).
self.assertEqual(plan["slice_paths"],
{"reviewer#api": ["src/api/routes.py"],
"reviewer#core": ["src/core/db.py"]})
# The plan stamps the digest that binds this boundary to the briefs
# and the dispatch (#453).
self.assertEqual(plan["slice_digest"], slice_digest(plan["slice_paths"], plan["partition_proof"]))
# Each seat takes its role's brief template, carrying the digest, and
# dispatches.
for seat in plan["assignments"]:
brief = self.tmp / (seat.replace("#", "-") + ".md")
brief.write_text(seat_brief_text(seat, plan), encoding="utf-8")
self.briefs[seat] = brief
plan_file = self.tmp / "partitioned-plan.json"
plan_file.write_text(json.dumps(plan), encoding="utf-8")
code, applied, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", str(plan_file),
"--task", "t-partition", "--now", AT, "--common", str(self.common), "--dry-run"]
+ self.brief_args(*plan["assignments"]),
client=self._client({}),
)
self.assertEqual(code, 0, err)
steps = {step["role"]: step["agent"] for step in json.loads(applied)["steps"]}
self.assertEqual(sorted(steps), ["reviewer#api", "reviewer#core"])
self.assertEqual(len(set(steps.values())), 2)
def test_the_review_gate_refuses_a_task_with_no_recorded_base(self):
# coding-policy#460: the gate binds the proof to the task's recorded base.
partition = self.tmp / "gate.json"
partition.write_text(json.dumps(validated_partition()))
out = io.StringIO()
code = main(self.base() + ["plan", "--roles", "reviewer", "--partition", str(partition),
"--now", AT, "--snapshot", str(self.snapshot)], stdout=out)
self.assertEqual(code, 0, out.getvalue())
plan_file = self.tmp / "gate-plan.json"
plan_file.write_text(out.getvalue(), encoding="utf-8")
self.assertEqual(json.loads(out.getvalue())["partition_proof"]["head"], "c" * 40)
err = io.StringIO()
code = main(self.base() + ["verify-partition", "--plan", str(plan_file), "--repo", str(self.tmp),
"--head", "HEAD", "--task", "never-registered"], stdout=io.StringIO(), stderr=err)
self.assertEqual(code, 1)
self.assertIn("no registered base", err.getvalue())
def plan_at(self, name, head):
document = validated_partition()
document["proof"]["head"] = head
partition = self.tmp / (name + ".json")
partition.write_text(json.dumps(document))
out = io.StringIO()
code = main(self.base() + ["plan", "--roles", "reviewer", "--partition", str(partition),
"--now", AT, "--snapshot", str(self.snapshot)], stdout=out)
self.assertEqual(code, 0, out.getvalue())
return json.loads(out.getvalue())
def test_briefs_dispatched_at_an_older_head_fail_a_replan_over_the_same_paths(self):
# coding-policy#460: a push can change a file's content without changing
# which paths changed. The seat digests bind the proof, so briefs sent for
# the old tip do not satisfy a plan proven at the new one.
old, new = self.plan_at("old", "c" * 40), self.plan_at("new", "d" * 40)
self.assertEqual(old["slice_paths"], new["slice_paths"])
briefs = {}
for seat in old["slice_paths"]:
brief = self.tmp / ("old-" + seat.replace("#", "-") + ".md")
brief.write_text(seat_brief_text(seat, old), encoding="utf-8")
briefs[seat] = str(brief)
cli._require_bound_slices(old, sorted(old["slice_paths"]), briefs)
with self.assertRaises(UsageError):
cli._require_bound_slices(new, sorted(new["slice_paths"]), briefs)
def test_a_validated_result_without_a_proof_is_refused_at_planning(self):
document = validated_partition()
del document["proof"]
partition = self.tmp / "unproven.json"
partition.write_text(json.dumps(document))
code, _, err = self.run_cli(self.base() + ["plan", "--roles", "reviewer", "--partition", str(partition),
"--now", AT, "--snapshot", str(self.snapshot)])
self.assertEqual(code, 1)
self.assertIn("carries no usable proof", err)
def test_a_seat_inherits_its_role_bars_alongside_its_own(self):
# An exclusion is a bar, not a setting a seat overrides. Naming one
# seat's own excluded worker must not lift the contributor bar the
# role already carries, or the planner hands independent verification
# of a slice to the worker that wrote the task (#434).
partition = self.tmp / "bars.json"
partition.write_text(json.dumps(validated_partition()))
state = empty_state()
add_assignment(state, AT, "developer", "grok", task="t-bar", status="applied")
save_state(self.state, state)
out = io.StringIO()
code = main(self.base() + ["plan", "--roles", "reviewer", "--partition", str(partition),
"--task", "t-bar", "--now", AT,
"--exclude", "reviewer#api=claude",
"--snapshot", str(self.snapshot)], stdout=out)
self.assertEqual(code, 0, out.getvalue())
assignments = json.loads(out.getvalue())["assignments"]
self.assertEqual(sorted(assignments), ["reviewer#api", "reviewer#core"])
self.assertEqual(assignments["reviewer#api"], "codex")
self.assertNotIn("grok", assignments.values())
def test_an_unpartitioned_plan_carries_no_slice_paths(self):
out = io.StringIO()
code = main(self.base() + ["plan", "--roles", "reviewer", "--now", AT,
"--snapshot", str(self.snapshot)], stdout=out)
self.assertEqual(code, 0, out.getvalue())
self.assertNotIn("slice_paths", json.loads(out.getvalue()))
def test_a_pre_seated_role_input_is_refused(self):
# `--partition` is what proves the slices disjoint and exhaustive, so a
# seat named straight into `--roles` would plan several seats over an
# unchecked surface (#434).
code, _, err = self.run_cli(
self.base() + ["plan", "--roles", "reviewer#api,reviewer#core", "--now", AT,
"--snapshot", str(self.snapshot)])
self.assertEqual(code, 1)
self.assertIn("came pre-seated", err)
def test_a_seat_exclusion_does_not_reach_the_tier_candidates(self):
# Tier candidacy is decided per responsibility, so a seat key would
# reach the planner's exclusion parser as an unknown role (#434).
partition = self.tmp / "tiers.json"
partition.write_text(json.dumps(validated_partition()))
out = io.StringIO()
code = main(self.base() + ["plan", "--roles", "reviewer", "--partition", str(partition),
"--task", "t-tiers", "--now", AT,
"--exclude", "reviewer#api=claude",
"--snapshot", str(self.snapshot)], stdout=out, stderr=self.err)
self.assertEqual(code, 0, self.err.getvalue())
assignments = json.loads(out.getvalue())["assignments"]
self.assertEqual(sorted(assignments), ["reviewer#api", "reviewer#core"])
self.assertNotEqual(assignments["reviewer#api"], "claude")
def test_a_seat_of_an_unseatable_responsibility_is_refused(self):
# A partition document can only seat a reviewer or a tester, but the
# role names reaching plan and apply come straight off the command
# line. `developer#api` would take a reviewer's seat contract while
# `validate_fix_history` keeps counting the literal `developer`, so
# the fix counter would fragment across seats (#434).
for role in ("developer#api", "release#core", "judge#api"):
with self.subTest(role=role, command="plan"):
code, _, err = self.run_cli(
self.base() + ["plan", "--roles", role, "--now", AT,
"--snapshot", str(self.snapshot)])
self.assertEqual(code, 1)
self.assertIn("names a seat of", err)
with self.subTest(role=role, command="apply"):
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0",
"--assignments", json.dumps({role: "grok"}),
"--common", str(self.common), "--now", AT, "--dry-run"]
+ ["--brief", role + "=" + str(self.briefs["developer"])],
client=self._client({}),
)
self.assertEqual(code, 1)
self.assertIn("names a seat of", err)
def test_a_seat_records_its_responsibility_in_the_ledger(self):
# The per-role history must not fragment across seat names (#434).
client = self._client({"grok": "idle"})
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0",
"--assignments", json.dumps(self.seat_plan),
"--task", "t-ledger-seat", "--common", str(self.common), "--now", AT]
+ ["--brief", "reviewer#api=" + str(self.seat_brief),
"--report", "reviewer#api=" + self.seat_report("ledger-seat.md")],
client=client,
)
self.assertEqual(code, 0, err)
rows = json.loads(self.state.read_text())["assignments"]
self.assertEqual([row["role"] for row in rows], ["reviewer"])
def test_planning_from_an_unvalidated_partition_is_refused(self):
# `plan` has no repo, base or head, so it cannot check ownership
# itself. Seating from the checked RESULT is what makes the round's
# slices provably disjoint and exhaustive (#453).
raw = self.tmp / "raw-partition.json"
raw.write_text(json.dumps({"schema_version": 1, "slices": [
{"name": "api", "paths": ["src/api/*"]}, {"name": "core", "paths": ["src/core/*"]}]}))
code, _, err = self.run_cli(
self.base() + ["plan", "--roles", "reviewer", "--partition", str(raw),
"--task", "t-unchecked", "--now", AT,
"--snapshot", str(self.snapshot)])
self.assertEqual(code, 1)
self.assertIn("carries no `changed` set", err)
self.assertIn("validate-partition", err)
def test_a_resolved_path_with_glob_characters_still_seats(self):
# A result's paths are resolved FILENAMES, so re-matching them as
# patterns reads `[x]` as a character class and reports the file it
# names unowned — rejecting a valid result (#453).
tricky = self.tmp / "tricky-result.json"
tricky.write_text(json.dumps(validated_partition(
slices=[{"name": "api", "paths": ["src/api/[x].py"]},
{"name": "core", "paths": ["src/core/a?b.py"]}],
changed=["src/api/[x].py", "src/core/a?b.py"])), encoding="utf-8")
code, out, err = self.run_cli(
self.base() + ["plan", "--roles", "reviewer", "--partition", str(tricky),
"--task", "t-tricky", "--now", AT,
"--snapshot", str(self.snapshot)])
self.assertEqual(code, 0, err)
self.assertEqual(json.loads(out)["slice_paths"],
{"reviewer#api": ["src/api/[x].py"],
"reviewer#core": ["src/core/a?b.py"]})
def test_planning_from_an_edited_validated_result_is_refused(self):
# The result's verdict is not taken on faith: overlapping slices and a
# changed file no slice owns both pass a shape check, so ownership is
# re-derived against the `changed` set the result carries (#453).
edited = self.tmp / "edited-result.json"
edited.write_text(json.dumps(validated_partition(
slices=[{"name": "api", "paths": ["src/**"]},
{"name": "core", "paths": ["src/core/*"]}],
changed=["src/core/db.py", "docs/guide.md"])), encoding="utf-8")
code, _, err = self.run_cli(
self.base() + ["plan", "--roles", "reviewer", "--partition", str(edited),
"--task", "t-edited-result", "--now", AT,
"--snapshot", str(self.snapshot)])
self.assertEqual(code, 1)
self.assertIn("docs/guide.md", err)
def test_a_boundary_edited_after_planning_is_refused(self):
# The digest is stamped over the map the validated partition carried;
# editing the plan's paths afterwards no longer matches it.
edited = {**self.seat_plan, "slice_paths": {"reviewer#api": ["src/**"]}}
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", json.dumps(edited),
"--task", "t-edited", "--common", str(self.common), "--now", AT, "--dry-run"]
+ ["--brief", "reviewer#api=" + str(self.seat_brief)],
client=self._client({}),
)
self.assertEqual(code, 1)
self.assertIn("no longer match its slice_digest", err)
def test_an_unreadable_seat_brief_names_its_repair(self):
# coding-policy#461: the handler reported the OSError and stopped.
# rules/error-handling.md Actionable Messages wants the next step.
unreadable = self.tmp / "undecodable-brief.md"
unreadable.write_bytes(b"# reviewer\n\xff\xfe not utf-8\n")
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", json.dumps(self.seat_plan),
"--task", "t-unreadable", "--common", str(self.common), "--now", AT, "--dry-run"]
+ ["--brief", "reviewer#api=" + str(unreadable)],
client=self._client({}),
)
self.assertEqual(code, 1)
self.assertIn("Cannot read the brief for seat", err)
self.assertIn("compose-briefs.sh", err)
def test_a_malformed_slice_paths_map_is_refused_not_crashed(self):
# `slice_paths` rides in an editable document, and a seat mapped to a
# non-list reached the digest as an unhashable value — a traceback
# where `apply` promises a refusal (#453).
for broken in ({"reviewer#api": None}, {"reviewer#api": []},
{"reviewer#api": ["src/api/*", 7]}):
with self.subTest(broken=broken):
edited = {**self.seat_plan, "slice_paths": broken}
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", json.dumps(edited),
"--task", "t-malformed", "--common", str(self.common), "--now", AT,
"--dry-run"]
+ ["--brief", "reviewer#api=" + str(self.seat_brief)],
client=self._client({}),
)
self.assertEqual(code, 1)
self.assertIn("non-empty list of globs", err)
self.assertNotIn("Traceback", err)
def test_a_seat_brief_missing_its_boundary_is_refused(self):
# A hand-written or differently-composed brief carries no digest, so a
# full-surface brief cannot be dispatched as a slice verdict.
bare = self.tmp / "bare-seat.md"
bare.write_text("# reviewer#api\n", encoding="utf-8")
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", json.dumps(self.seat_plan),
"--task", "t-bare", "--common", str(self.common), "--now", AT, "--dry-run"]
+ ["--brief", "reviewer#api=" + str(bare)],
client=self._client({}),
)
self.assertEqual(code, 1)
self.assertIn("does not carry", err)
self.assertIn("scope block", err)
def test_a_version_6_shaped_seated_plan_is_refused(self):
# Plan schema 7 adds `seat_digests`. Model the pre-7 document itself —
# the boundary keys a version-6 plan carried and no per-seat entry —
# together with the brief such a plan produced, which could only carry
# the round-level digest. The round digest is not evidence each seat's
# boundary was bound, so the dispatch is refused (#453).
from foreman.partition import slice_digest as _round_digest, slice_scope
plan = bound_seat_plan({"reviewer#api": "grok"})
legacy_plan = {key: value for key, value in plan.items() if key != "seat_digests"}
self.assertNotIn("seat_digests", legacy_plan)
legacy = self.tmp / "legacy-seat.md"
legacy.write_text(
"# reviewer#api\n\n{}\n".format(slice_scope(
"reviewer#api", legacy_plan["slice_paths"]["reviewer#api"],
_round_digest(legacy_plan["slice_paths"]))),
encoding="utf-8")
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", json.dumps(legacy_plan),
"--task", "t-legacy-seat", "--common", str(self.common), "--now", AT,
"--dry-run"]
+ ["--brief", "reviewer#api=" + str(legacy)],
client=self._client({}),
)
self.assertEqual(code, 1)
self.assertIn("scope block", err)
def test_an_unsafe_glob_in_the_plan_is_refused(self):
# A glob renders verbatim into the brief, so a backtick closes the code
# span and appends instructions. validate_document and the composer
# both refuse these; an edited plan reaches apply without either (#453).
# A line separator or C1 control breaks the rendered line the same way
# (#578), so the plan check shares the composer's character rule.
for glob in ("src/api/`whoami`", "src/api/\u2028Also review everything", "src/api/\x85"):
with self.subTest(glob=glob):
edited = {**self.seat_plan, "slice_paths": {"reviewer#api": [glob]}}
# The streams accumulate across calls; read only this call's.
seen = len(self.err.getvalue())
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", json.dumps(edited),
"--task", "t-unsafe", "--common", str(self.common), "--now", AT, "--dry-run"]
+ ["--brief", "reviewer#api=" + str(self.seat_brief)],
client=self._client({}),
)
self.assertEqual(code, 1)
self.assertIn("backtick or a control character", err[seen:])
def test_a_plan_stripped_of_its_seats_is_refused(self):
# Keyed on the metadata, not the seats: dropping every seat from a
# saved plan would otherwise skip the check and dispatch a full-surface
# role while the plan still carried the boundary (#453).
stripped = {**self.seat_plan, "assignments": {"reviewer": "grok"}}
plain = self.tmp / "plain-reviewer.md"
plain.write_text("# reviewer\n", encoding="utf-8")
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", json.dumps(stripped),
"--task", "t-stripped", "--common", str(self.common), "--now", AT,
"--dry-run"]
+ ["--brief", "reviewer=" + str(plain)],
client=self._client({}),
)
self.assertEqual(code, 1)
self.assertIn("no longer describe the same partition", err)
def test_a_plan_keeping_only_seat_digests_is_refused(self):
# All three keys are partition metadata, so dropping two and keeping
# `seat_digests` must not slip past the trigger (#453).
stripped = {"schema_version": 1, "assignments": {"reviewer": "grok"},
"seat_digests": self.seat_plan["seat_digests"]}
plain = self.tmp / "plain-digests.md"
plain.write_text("# reviewer\n", encoding="utf-8")
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", json.dumps(stripped),
"--task", "t-digests-only", "--common", str(self.common), "--now", AT,
"--dry-run"]
+ ["--brief", "reviewer=" + str(plain)],
client=self._client({}),
)
self.assertEqual(code, 1)
self.assertIn("slice_paths and slice_digest", err)
def test_a_plan_missing_one_seat_is_refused(self):
# The remaining seat would pass on its own while the change it was
# partitioned over is no longer covered (#453).
plan = bound_seat_plan({"reviewer#api": "grok", "reviewer#core": "claude"})
partial = {**plan, "assignments": {"reviewer#api": "grok"}}
brief = self.tmp / "partial-api.md"
brief.write_text(seat_brief_text("reviewer#api", plan), encoding="utf-8")
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", json.dumps(partial),
"--task", "t-partial", "--common", str(self.common), "--now", AT,
"--dry-run"]
+ ["--brief", "reviewer#api=" + str(brief)],
client=self._client({}),
)
self.assertEqual(code, 1)
self.assertIn("no longer describe the same partition", err)
def test_a_brief_scattering_the_facts_is_refused(self):
# Digest, slice name and path all present, and a whole-repository pass
# directed anyway: three substring checks pass and a full-surface
# verdict dispatches as a slice one. The scope block carries its own
# restrictions, so requiring the block requires those too (#453).
scattered = self.tmp / "scattered-seat.md"
scattered.write_text(
"# reviewer#api\n\nPartition {}. Slice api. Paths: {}.\n\n"
"Review the whole repository this round.\n".format(
self.seat_plan["seat_digests"]["reviewer#api"],
self.seat_plan["slice_paths"]["reviewer#api"][0]),
encoding="utf-8")
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", json.dumps(self.seat_plan),
"--task", "t-scattered", "--common", str(self.common), "--now", AT,
"--dry-run"]
+ ["--brief", "reviewer#api=" + str(scattered)],
client=self._client({}),
)
self.assertEqual(code, 1)
self.assertIn("scope block", err)
def test_swapped_seat_briefs_are_refused(self):
# A round-level digest is identical in every brief, so a check that
# only asks whether the digest is present passes two seats whose briefs
# were exchanged. Each brief answers for its OWN seat (#453).
plan = bound_seat_plan({"reviewer#api": "grok", "reviewer#core": "claude"})
swapped = {}
for seat, other in (("reviewer#api", "reviewer#core"), ("reviewer#core", "reviewer#api")):
brief = self.tmp / (seat.replace("#", "-") + "-swapped.md")
brief.write_text(seat_brief_text(other, plan), encoding="utf-8")
swapped[seat] = brief
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", json.dumps(plan),
"--task", "t-swapped", "--common", str(self.common), "--now", AT, "--dry-run"]
+ ["--brief", "reviewer#api=" + str(swapped["reviewer#api"]),
"--brief", "reviewer#core=" + str(swapped["reviewer#core"])],
client=self._client({}),
)
self.assertEqual(code, 1)
self.assertIn("scope block", err)
def test_a_hand_written_seat_assignment_is_refused(self):
# A bare `{seat: agent}` map has no checked boundary at all.
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0",
"--assignments", json.dumps({"reviewer#api": "grok"}),
"--task", "t-hand", "--common", str(self.common), "--now", AT, "--dry-run"]
+ ["--brief", "reviewer#api=" + str(self.seat_brief)],
client=self._client({}),
)
self.assertEqual(code, 1)
self.assertIn("need the plan's slice_paths and slice_digest", err)
def test_an_unseated_round_needs_no_boundary(self):
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0",
"--assignments", json.dumps({"developer": "grok"}),
"--common", str(self.common), "--now", AT, "--dry-run"]
+ self.brief_args("developer"),
client=self._client({}),
)
self.assertEqual(code, 0, err)
def test_a_seated_apply_without_a_task_is_refused(self):
# The ledger row records the responsibility and the DISPATCH records
# the seat, and a dispatch exists only under a task. A task-less seated
# apply would leave nothing naming the slice, so its verdict could
# never be read back (#434).
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0",
"--assignments", json.dumps(self.seat_plan),
"--common", str(self.common), "--now", AT, "--dry-run"]
+ ["--brief", "reviewer#api=" + str(self.seat_brief)],
client=self._client({}),
)
self.assertEqual(code, 1)
self.assertIn("need --task", err)
def test_a_live_seat_dispatch_reserves_and_records(self):
# Not a dry run: reserving a live `reviewer#api` dispatch exercises the
# recovery store's metadata validation, which read the literal role
# (#434).
state = empty_state()
save_state(self.state, state)
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0",
"--assignments", json.dumps(self.seat_plan),
"--task", "t-live-seat", "--common", str(self.common), "--now", AT]
+ ["--brief", "reviewer#api=" + str(self.seat_brief),
"--report", "reviewer#api=" + self.seat_report("seat-report.md")],
client=self._client({"grok": "idle"}),
)
self.assertEqual(code, 0, err)
saved = json.loads(self.state.read_text())
self.assertEqual([row["role"] for row in saved["assignments"]], ["reviewer"])
dispatch = saved["recovery"]["dispatches"][0]
self.assertEqual(dispatch["role"], "reviewer#api")
self.assertEqual(dispatch["reviewer_scope"], "verification")
def _seat_dispatch_into(self, width_delta):
# The report path's length depends on the temp dir, so the pane is
# sized from the marker itself: exactly fitting, or one column short.
report = self.seat_report("wrap-report.md")
client = self._client({"grok": "idle"})
self.runner.set("pane layout --pane w4:p1", pane_layout("w4:p1", marker_columns("grok", report) + width_delta))
save_state(self.state, empty_state())
return self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0",
"--assignments", json.dumps(self.seat_plan),
"--task", "t-wrap", "--common", str(self.common), "--now", AT]
+ ["--brief", "reviewer#api=" + str(self.seat_brief),
"--report", "reviewer#api=" + report],
client=client,
)
def test_a_report_marker_the_pane_would_wrap_is_refused_before_input(self):
code, _, err = self._seat_dispatch_into(-1)
self.assertNotEqual(code, 0)
self.assertIn("REPORT marker would wrap", err)
self.assertEqual(self.runner.writes(), [])
def test_a_report_the_brief_does_not_assign_is_refused_before_input(self):
# A short --report must not stand in for the brief's own longer marker.
self.seat_report("long-" + "x" * 200 + ".md")
client = self._client({"grok": "idle"})
save_state(self.state, empty_state())
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0",
"--assignments", json.dumps(self.seat_plan),
"--task", "t-mismatch", "--common", str(self.common), "--now", AT]
+ ["--brief", "reviewer#api=" + str(self.seat_brief),
"--report", "reviewer#api=" + str(self.tmp / "r.md")],
client=client,
)
self.assertNotEqual(code, 0)
self.assertIn("is not the `REPORT: <path>` line its brief assigns", err)
self.assertEqual(self.runner.writes(), [])
def test_a_report_marker_that_exactly_fits_the_pane_dispatches(self):
code, _, err = self._seat_dispatch_into(0)
self.assertEqual(code, 0, err)
self.assertIn("pane layout --pane w4:p1", self.runner.commands())
def test_a_saved_seat_dispatch_reloads_with_its_history(self):
# The ledger row carries the responsibility and the dispatch carries
# the seat, so the NEXT load must still read them as one confirmed
# outcome. A store that rejects the pair returns an empty ledger and
# loses the contribution and recovery history it was keeping (#434).
save_state(self.state, empty_state())
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0",
"--assignments", json.dumps(self.seat_plan),
"--task", "t-reload-seat", "--common", str(self.common), "--now", AT]
+ ["--brief", "reviewer#api=" + str(self.seat_brief),
"--report", "reviewer#api=" + self.seat_report("reload-report.md")],
client=self._client({"grok": "idle"}),
)
self.assertEqual(code, 0, err)
warnings = []
reloaded, usable = load_state_checked(self.state, warn=warnings.append)
self.assertTrue(usable, warnings)
self.assertEqual(warnings, [])
self.assertEqual([row["role"] for row in reloaded["assignments"]], ["reviewer"])
self.assertEqual(reloaded["assignments"][0]["reviewer_scope"], "verification")
self.assertEqual([row["role"] for row in reloaded["recovery"]["dispatches"]], ["reviewer#api"])
def test_a_contributor_cannot_take_a_review_seat(self):
# The responsibility decides independence: a worker the ledger records
# as a contributor on this task is barred from every seat of the
# reviewer role, not only from the literal name `reviewer` (#434).
state = empty_state()
add_assignment(state, AT, "developer", "grok", task="t-seat", fix_round=1)
save_state(self.state, state)
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0",
"--assignments", json.dumps(self.seat_plan),
"--task", "t-seat", "--now", AT, "--common", str(self.common), "--dry-run"]
+ ["--brief", "reviewer#api=" + str(self.seat_brief)],
client=self._client({}),
)
self.assertEqual(code, 1)
self.assertIn("ineligible", err)
def test_live_apply_clears_then_assigns_and_records_the_ledger(self):
client = self._client({"grok": "idle", "claude": "done"})
code, out, _ = self.run_cli(
self.base()
+ [
"apply",
"--composer-settle",
"0",
"--assignments",
json.dumps({"developer": "grok", "tester": "claude"}),
"--common",
str(self.common),
"--now",
AT,
]
+ self.brief_args("developer", "tester"),
client=client,
)
self.assertEqual(code, 0)
self.assertEqual(len(json.loads(out)["applied"]), 2)
ledger = json.loads(self.state.read_text(encoding="utf-8"))["assignments"]
self.assertEqual(
ledger,
[
{
"schema_version": STATE_SCHEMA_VERSION,
"at": AT,
"role": "developer",
"agent": "grok",
"status": "applied",
"cleared": True,
"clear_reason": "automatic",
"task": None,
"fix_round": None,
"context_session": None,
"tier": None,
"requirements": None,
"reviewer_scope": None,
"judge_mode": None,
},
{
"schema_version": STATE_SCHEMA_VERSION,
"at": AT,
"role": "tester",
"agent": "claude",
"status": "applied",
"cleared": True,
"clear_reason": "automatic",
"task": None,
"fix_round": None,
"context_session": None,
"tier": None,
"requirements": None,
"reviewer_scope": None,
"judge_mode": None,
},
],
)
def test_each_agent_clears_by_its_configured_path_end_to_end(self):
client = self._client({"grok": "idle", "claude": "done"})
code, _, _ = self.run_cli(
self.base()
+ [
"apply",
"--composer-settle",
"0",
"--assignments",
json.dumps({"developer": "grok", "tester": "claude"}),
"--common",
str(self.common),
"--now",
AT,
]
+ self.brief_args("developer", "tester"),
client=client,
)
self.assertEqual(code, 0)
commands = self.runner.commands()
self.assertIn("pane send-text w4:p1 /new", commands) # grok types
self.assertIn("agent prompt claude /clear", commands) # claude pastes
self.assertNotIn("agent prompt grok /new", commands)
def test_a_not_started_round_is_recorded_but_never_counted(self):
# End to end: apply writes the row, plan does not count it as
# experience of the role.
client = self._client({"grok": "idle"}, footers={"grok": GROK_IDLE_FOOTER})
# The clear must still change the screen (that gate is real); it is
# the assignment that never lands.
self.runner.responses[
"agent read grok --source visible --lines 20"
] = ScriptedReads(
[
" old transcript\n │ ❯ │\n",
" fresh session\n │ ❯ │\n",
" fresh session\n │ ❯ │\n",
" fresh session\n │ ❯ │\n",
]
)
self.runner.set(
"agent wait grok --until working",
stdout="",
returncode=1,
stderr='{"error":{"code":"timeout","message":"never left idle"}}',
)
code, out, err = self.run_cli(
self.base()
+ [
"apply",
"--composer-settle",
"0",
"--assignments",
json.dumps({"developer": "grok"}),
"--common",
str(self.common),
"--now",
AT,
]
+ self.brief_args("developer"),
client=client,
)
self.assertEqual(code, 1)
self.assertEqual(json.loads(out)["applied"][0]["status"], "sent_but_not_started")
self.assertIn("sent_but_not_started", err)
ledger = json.loads(self.state.read_text(encoding="utf-8"))["assignments"]
self.assertEqual(len(ledger), 1)
self.assertEqual(ledger[0]["status"], "sent_but_not_started")
self.out, self.err = io.StringIO(), io.StringIO()
code, out, _ = self.run_cli(
self.base() + ["plan", "--roles", "developer", "--snapshot", str(self.snapshot)]
)
self.assertEqual(code, 0)
self.assertIn("held this role 0x before", json.loads(out)["rationale"][0])
def test_busy_agent_is_refused_with_a_json_error_and_no_writes(self):
client = self._client({"grok": "working"})
code, out, err = self.run_cli(
self.base()
+ [
"apply",
"--composer-settle",
"0",
"--assignments",
json.dumps({"developer": "grok"}),
"--common",
str(self.common),
"--now",
AT,
]
+ self.brief_args("developer"),
client=client,
)
self.assertEqual(code, 1)
self.assertEqual(out, "")
payload = json.loads(err)
self.assertEqual(payload["error"], "agent_busy")
self.assertEqual(payload["details"]["busy"], {"grok": "working"})
self.assertEqual(self.runner.writes(), [])
def test_a_refused_round_leaves_the_ledger_untouched(self):
client = self._client({"grok": "blocked"})
self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", json.dumps({"developer": "grok"}), "--common", str(self.common), "--now", AT]
+ self.brief_args("developer"),
client=client,
)
self.assertFalse(self.state.exists())
def test_apply_proceeds_when_the_probe_overturns_a_stale_herdr_state(self):
client = self._client({"grok": "working"}, footers={"grok": GROK_IDLE_FOOTER})
code, out, err = self.run_cli(
self.base()
+ [
"apply",
"--composer-settle",
"0",
"--assignments",
json.dumps({"developer": "grok"}),
"--common",
str(self.common),
"--now",
AT,
]
+ self.brief_args("developer"),
client=client,
)
self.assertEqual(code, 0)
applied = json.loads(out)["applied"][0]
self.assertEqual(applied["state_source"], "probe")
self.assertEqual(applied["herdr_state_before"], "working")
self.assertIn("stale", err)
def _rejects(self, argv):
# argparse prints its usage to stderr on a bad flag; swallow it so the
# suite's own output stays readable.
with contextlib.redirect_stderr(io.StringIO()):
with self.assertRaises(SystemExit):
build_parser().parse_args(argv)
def test_apply_has_no_force_flag(self):
# The override is gone from the surface, not merely discouraged.
self._rejects(["apply", "--assignments", "{}", "--common", "x", "--force"])
def test_measure_has_no_force_flag(self):
self._rejects(["measure", "--force"])
def test_a_busy_agent_refuses_the_whole_round(self):
client = self._client({"grok": "working"})
code, out, err = self.run_cli(
self.base()
+ [
"apply",
"--composer-settle",
"0",
"--assignments",
json.dumps({"developer": "grok"}),
"--common",
str(self.common),
"--now",
AT,
]
+ self.brief_args("developer"),
client=client,
)
self.assertNotEqual(code, 0)
self.assertEqual(out, "")
self.assertIn("Refusing to interrupt", err)
def test_no_clear_skips_the_clear_prompt(self):
client = self._client({"grok": "idle"})
self.run_cli(
self.base()
+ [
"apply",
"--composer-settle",
"0",
"--assignments",
json.dumps({"developer": "grok"}),
"--common",
str(self.common),
"--now",
AT,
"--no-clear",
]
+ self.brief_args("developer"),
client=client,
)
self.assertEqual(len(self.runner.writes()), 1)
row = json.loads(self.state.read_text(encoding="utf-8"))["assignments"][-1]
self.assertFalse(row["cleared"])
self.assertEqual(row["clear_reason"], "hand")
def _seed_context(self, *, task="repo#322", role="developer", status="applied", fix_round=None, session="task-session"):
state = empty_state()
add_assignment(
state, AT, role, "grok", status=status, task=task, fix_round=fix_round,
cleared=True, clear_reason="automatic",
context_session={"pane_id": "w4:p1", "source": "herdr:grok", "agent": "grok",
"kind": "id", "value": session},
)
save_state(self.state, state)
def _foreground(self, pane_id, argv):
return json.dumps({"result": {"process_info": {"pane_id": pane_id,
"foreground_processes": [{"name": argv[0], "pid": 200, "argv": argv}]}}})
def _fix_args(self, round_number=1, *extra):
return self.base() + [
"apply", "--composer-settle", "0", "--assignments",
json.dumps({"developer": "grok"}), "--common", str(self.common),
"--now", AT, "--task", "repo#322", "--fix-round", str(round_number),
*extra,
] + self.brief_args("developer")
def _obligation(self, name, kind="decision", task: "str | None" = "repo#322"):
return {"id": name, "kind": kind, "task": task, "title": "Choose the replacement tester",
"context": "The tester's provider refused the brief.", "consequence": "No tester report exists.",
"resolution_condition": "Record the user's choice of replacement tester.", "priority": 99,
"sources": [{"schema_version": 1, "kind": "user_message", "ref": "conversation/1/message/3"}]}
def _answer(self, name, event_id="answer-1", revision=1):
return {"event_id": event_id, "id": name, "expected_revision": revision, "action": "resolve",
"reason": "The user answered.", "evidence": {"schema_version": 1, "kind": "user_answer",
"ref": "conversation/1/message/5", "summary": "Use the other provider."}}
def _retained(self):
return self._client({"grok": "idle"}, sessions={"grok": "task-session"})
def test_apply_refuses_while_a_decision_on_the_task_is_unanswered(self):
# coding-policy#399: the foreman withheld a tester on an unanswered
# priority-99 decision and kept dispatching fix rounds on the same
# task. A foreman that can keep dispatching has not been blocked.
self._seed_context()
attention.write(self.state, "record", self._obligation("acr14-tester"), AT)
for extra in ((), ("--dry-run",)):
self.out, self.err = io.StringIO(), io.StringIO()
code, out, err = self.run_cli(self._fix_args(1, "--retain-context", *extra), client=self._retained())
self.assertEqual(code, 1)
self.assertEqual(out, "")
failure = json.loads(err)
self.assertEqual(failure["error"], "usage_error")
self.assertIn("acr14-tester", failure["message"])
self.assertIn("Record the user's choice of replacement tester.", failure["message"])
self.assertEqual(failure["details"]["gating"][0]["id"], "acr14-tester")
self.assertEqual(self.runner.writes(), [])
self.assertEqual(json.loads(self.state.read_text(encoding="utf-8"))["assignments"][-1]["fix_round"], None)
# A presentation is not an answer; a decision on another task gates nothing.
attention.write(self.state, "update", {"event_id": "shown-1", "id": "acr14-tester", "expected_revision": 1,
"action": "present", "reason": "Shown in commentary.", "evidence": {"schema_version": 1, "kind": "delivery",
"ref": "conversation/1/message/4", "summary": "Mentioned in a long message."}}, AT)
attention.write(self.state, "record", self._obligation("elsewhere", task="repo#999"), AT)
self.out, self.err = io.StringIO(), io.StringIO()
code, _out, err = self.run_cli(self._fix_args(1, "--retain-context"), client=self._retained())
self.assertEqual(code, 1)
self.assertIn("acr14-tester", err)
attention.write(self.state, "update", self._answer("acr14-tester", revision=2), AT)
self.out, self.err = io.StringIO(), io.StringIO()
code, out, err = self.run_cli(self._fix_args(1, "--retain-context"), client=self._retained())
self.assertEqual(code, 0, err)
self.assertEqual(json.loads(out)["applied"][0]["agent"], "grok")
self.assertEqual(len(self.runner.writes()), 1)
def test_apply_refuses_a_malformed_attention_history(self):
self._seed_context()
attention.storage_path(self.state).write_text("{", encoding="utf-8")
code, out, err = self.run_cli(self._fix_args(1, "--retain-context"), client=self._retained())
self.assertEqual(code, 1)
self.assertEqual(out, "")
self.assertEqual(json.loads(err)["error"], "state_error")
self.assertEqual(self.runner.writes(), [])
def test_apply_without_a_task_ignores_the_attention_queue(self):
attention.write(self.state, "record", self._obligation("untasked", task=None), AT)
client = self._client({"grok": "idle"})
code, out, err = self.run_cli(
self.base() + ["apply", "--composer-settle", "0", "--assignments", json.dumps({"developer": "grok"}),
"--common", str(self.common), "--now", AT] + self.brief_args("developer"),
client=client)
self.assertEqual(code, 0, err)
self.assertEqual(len(json.loads(out)["applied"]), 1)
def test_retained_context_sends_one_prompt_and_persists_its_reason(self):
self._seed_context()
code, out, err = self.run_cli(
self._fix_args(1, "--retain-context"), client=self._client({"grok": "idle"}, sessions={"grok": "task-session"})
)
self.assertEqual(code, 0, err)
self.assertEqual(len(self.runner.writes()), 1)
row = json.loads(self.state.read_text(encoding="utf-8"))["assignments"][-1]
self.assertFalse(row["cleared"])
self.assertEqual(row["clear_reason"], "retained")
self.assertEqual((row["task"], row["fix_round"]), ("repo#322", 1))
self.assertEqual(json.loads(out)["applied"][0]["clear_reason"], "retained")
def test_retained_fix_accepts_a_resumed_yolo_developer_and_keeps_its_identity(self):
# coding-policy#382: the developer was restored with the documented resume form
# and its explicit YOLO flag. The retained fix verifies that process instead of
# demanding a fresh worker; task, native session and the fix count are unchanged.
session = "3b1a2c4d-5e6f-4a7b-8c9d-0e1f2a3b4c5d"
self._seed_context(session=session)
client = self._client({"grok": "idle"}, sessions={"grok": session})
self.runner.set("pane process-info --pane w4:p1",
self._foreground("w4:p1", ["grok", "--resume", session, "--always-approve", "--no-subagents"]))
code, out, err = self.run_cli(self._fix_args(1, "--retain-context"), client=client)
self.assertEqual(code, 0, err)
record = json.loads(out)["applied"][0]
self.assertEqual((record["task"], record["fix_round"], record["cleared"], record["clear_reason"]),
("repo#322", 1, False, "retained"))
self.assertEqual(record["context_session"]["value"], session)
self.assertEqual(len(self.runner.writes()), 1)
self.assertFalse(any(command.startswith(("agent start", "-TERM")) or "/new" in command
for command in self.runner.commands()))
rows = json.loads(self.state.read_text(encoding="utf-8"))["assignments"]
self.assertEqual([(row["task"], row["fix_round"]) for row in rows], [("repo#322", None), ("repo#322", 1)])
self.assertEqual(rows[-1]["context_session"]["value"], session)
def test_retained_fix_refuses_an_ambiguous_resume_before_any_input(self):
session = "3b1a2c4d-5e6f-4a7b-8c9d-0e1f2a3b4c5d"
for argv in (["grok", "--continue", "--always-approve"],
["grok", "--resume", "--always-approve"],
["grok", "--resume", session]):
with self.subTest(argv=argv):
self.out, self.err = io.StringIO(), io.StringIO()
self._seed_context(session=session)
before = self.state.read_bytes()
client = self._client({"grok": "idle"}, sessions={"grok": session})
self.runner.set("pane process-info --pane w4:p1", self._foreground("w4:p1", argv))
code, out, err = self.run_cli(self._fix_args(1, "--retain-context"), client=client)
self.assertEqual(code, 1)
self.assertEqual(out, "")
self.assertIn("before dispatch", err)
self.assertEqual(self.runner.writes(), [])
self.assertEqual(self.state.read_bytes(), before)
def test_retention_rejects_wrong_task_role_status_or_round_before_herdr(self):
for task, role, status, fix_round in (
("another", "developer", "applied", None),
("repo#322", "tester", "applied", None),
("repo#322", "developer", "sent_but_not_started", None),
("repo#322", "developer", "applied", 2),
):
with self.subTest(task=task, role=role, status=status, fix_round=fix_round):
self.out = io.StringIO()
self.err = io.StringIO()
self._seed_context(task=task, role=role, status=status, fix_round=fix_round)
code, _, err = self.run_cli(
self._fix_args(1, "--retain-context"), client=self._client({"grok": "idle"})
)
self.assertEqual(code, 1)
self.assertEqual(json.loads(err)["error"], "usage_error")
self.assertEqual(self.runner.calls, [])
def test_retention_requires_confirmed_history(self):
code, _, err = self.run_cli(
self._fix_args(1, "--retain-context"), client=self._client({"grok": "idle"})
)
self.assertEqual(code, 1)
self.assertIn("preceding confirmed", err)
self.assertEqual(self.runner.calls, [])
def test_fix_round_four_clears_context(self):
self._seed_context(fix_round=3)
code, out, err = self.run_cli(
self._fix_args(4), client=self._client({"grok": "idle"})
)
self.assertEqual(code, 0, err)
row = json.loads(out)["applied"][0]
self.assertTrue(row["cleared"])
self.assertEqual((row["clear_reason"], row["fix_round"]), ("automatic", 4))
self.assertGreater(len(self.runner.writes()), 1)
def test_retention_at_four_and_fix_round_six_are_refused(self):
for number, extra in ((4, ["--retain-context"]), (6, [])):
with self.subTest(number=number):
self.out = io.StringIO()
self.err = io.StringIO()
code, _, _ = self.run_cli(
self._fix_args(number, *extra), client=self._client({"grok": "idle"})
)
self.assertEqual(code, 1)
self.assertEqual(self.runner.calls, [])
def test_retained_dry_run_has_no_clear_and_no_state_write(self):
code, out, err = self.run_cli(
self._fix_args(1, "--retain-context", "--dry-run"), client=self._client({})
)
self.assertEqual(code, 0, err)
payload = json.loads(out)
self.assertEqual(payload["clear_reason"], "retained")
commands = [command["argv"] for command in payload["steps"][0]["commands"]]
self.assertFalse(any("/new" in command for command in commands))
self.assertEqual(self.runner.calls, [])
self.assertFalse(self.state.exists())
def test_context_flags_are_mutually_exclusive(self):
self._rejects(self._fix_args(1, "--retain-context", "--no-clear"))
def test_task_padding_is_rejected_without_rewriting_existing_history(self):
self._seed_context(task="repo#322 ")
before = self.state.read_bytes()
for task in (" repo#322", "repo#322 ", "\trepo#322", "repo#322\n", "repo#322\u00a0"):
for dry_run in (False, True):
with self.subTest(task=task, dry_run=dry_run):
self.out, self.err = io.StringIO(), io.StringIO()
args = self._fix_args(1, "--retain-context")
args[args.index("--task") + 1] = task
if dry_run:
args.append("--dry-run")
code, _, err = self.run_cli(args, client=self._client({}))
self.assertEqual(code, 1)
self.assertIn("leading or trailing whitespace", err)
self.assertEqual(self.runner.calls, [])
self.assertEqual(self.state.read_bytes(), before)
def test_existing_task_with_internal_spaces_is_not_renamed(self):
task = "repo task 322"
self._seed_context(task=task)
args = self._fix_args(1, "--retain-context")
args[args.index("--task") + 1] = task
code, out, err = self.run_cli(
args, client=self._client({"grok": "idle"}, sessions={"grok": "task-session"})
)
self.assertEqual(code, 0, err)
self.assertEqual(json.loads(out)["applied"][0]["task"], task)
def test_retention_round_three_succeeds_after_confirmed_round_two(self):
self._seed_context(fix_round=2)
code, out, err = self.run_cli(
self._fix_args(3, "--retain-context"), client=self._client({"grok": "idle"}, sessions={"grok": "task-session"})
)
self.assertEqual(code, 0, err)
self.assertEqual(len(self.runner.writes()), 1)
self.assertEqual(json.loads(out)["applied"][0]["fix_round"], 3)
def test_retention_still_refuses_a_live_busy_worker(self):
self._seed_context()
before = self.state.read_bytes()
code, _, err = self.run_cli(
self._fix_args(1, "--retain-context"), client=self._client({"grok": "blocked"})
)
self.assertEqual(code, 1)
self.assertEqual(json.loads(err)["error"], "agent_busy")
self.assertEqual(self.runner.writes(), [])
self.assertEqual(self.state.read_bytes(), before)
def test_fresh_rounds_refuse_no_clear(self):
for number in (4, 5):
with self.subTest(number=number):
self.out, self.err = io.StringIO(), io.StringIO()
code, _, err = self.run_cli(
self._fix_args(number, "--no-clear"), client=self._client({})
)
self.assertEqual(code, 1)
self.assertIn("automatic clear", err)
self.assertEqual(self.runner.calls, [])
def test_retention_cannot_use_migrated_history(self):
self.state.write_text(json.dumps({"schema_version": 2, "snapshots": [],
"assignments": [{"schema_version": 2, "at": AT, "agent": "grok",
"role": "developer", "status": "applied"}]}), encoding="utf-8")
code, _, err = self.run_cli(
self._fix_args(1, "--retain-context"), client=self._client({})
)
self.assertEqual(code, 1)
self.assertIn("preceding confirmed", err)
self.assertEqual(self.runner.calls, [])
row = json.loads(self.state.read_text(encoding="utf-8"))["assignments"][0]
self.assertEqual(row["clear_reason"], "unknown")
def test_completed_fixes_cannot_restart_as_initial_development(self):
self._seed_context(fix_round=5)
args = self._fix_args(5)
position = args.index("--fix-round")
del args[position:position + 2]
code, _, err = self.run_cli(args, client=self._client({}))
self.assertEqual(code, 1)
self.assertIn("do not reset", err)
self.assertEqual(self.runner.calls, [])
def test_fresh_fix_cannot_skip_the_task_history(self):
self._seed_context(fix_round=1)
code, _, err = self.run_cli(self._fix_args(4), client=self._client({}))
self.assertEqual(code, 1)
self.assertIn("next fix number 2", err)
self.assertEqual(self.runner.calls, [])
def test_early_developer_fix_requires_explicit_retention(self):
self._seed_context()
code, _, err = self.run_cli(self._fix_args(1), client=self._client({}))
self.assertEqual(code, 1)
self.assertIn("require --retain-context", err)
self.assertEqual(self.runner.calls, [])
def test_fresh_worker_continues_another_workers_fix_count(self):
state = empty_state()
add_assignment(state, AT, "developer", "claude", task="repo#322",
fix_round=3, cleared=False, clear_reason="retained")
save_state(self.state, state)
code, out, err = self.run_cli(self._fix_args(4), client=self._client({"grok": "idle"}))
self.assertEqual(code, 0, err)
row = json.loads(out)["applied"][0]
self.assertEqual((row["agent"], row["fix_round"]), ("grok", 4))
self.assertTrue(row["cleared"])
def test_intervening_role_prevents_retention_even_with_matching_task_history(self):
state = empty_state()
add_assignment(state, AT, "developer", "grok", task="repo#322",
cleared=True, clear_reason="automatic")
add_assignment(state, "2026-02-03T10:00:01+00:00", "tester", "grok", task="another-task",
cleared=True, clear_reason="automatic")
save_state(self.state, state)
code, _, err = self.run_cli(self._fix_args(1, "--retain-context"), client=self._client({}))
self.assertEqual(code, 1)
self.assertIn("Cannot retain", err)
self.assertEqual(self.runner.calls, [])
def test_retention_refuses_a_missing_or_changed_live_native_session(self):
for session in (None, "after-manual-clear", "restarted-worker"):
with self.subTest(session=session):
self.out, self.err = io.StringIO(), io.StringIO()
self._seed_context()
before = self.state.read_bytes()
code, _, err = self.run_cli(
self._fix_args(1, "--retain-context"),
client=self._client({"grok": "idle"}, sessions={"grok": session}),
)
self.assertEqual(code, 1)
self.assertIn("native session continuity", err)
self.assertEqual(self.runner.writes(), [])
self.assertEqual(self.state.read_bytes(), before)
def test_native_session_change_between_readiness_and_send_refuses_retention(self):
self._seed_context()
before = self.state.read_bytes()
client = self._client({"grok": "idle"})
self.runner.responses["agent get grok"] = ScriptedReads([
agent_json("grok", "idle", "w4:p1", "task-session"),
agent_json("grok", "idle", "w4:p1", "new-session"),
])
code, _, err = self.run_cli(self._fix_args(1, "--retain-context"), client=client)
self.assertEqual(code, 1)
self.assertIn("native session continuity", err)
self.assertEqual(self.runner.writes(), [])
after = json.loads(self.state.read_text())
self.assertEqual(after["assignments"], json.loads(before)["assignments"])
self.assertEqual(after["recovery"]["dispatches"][-1]["status"], "not_sent")
def test_initial_dispatch_captures_post_clear_identity_for_a_real_retained_fix(self):
client = self._client({"grok": "idle"})
self.runner.responses["agent get grok"] = ScriptedReads([
agent_json("grok", "idle", "w4:p1", "old-task"),
agent_json("grok", "idle", "w4:p1", "new-task"),
])
args = self._fix_args(1)
position = args.index("--fix-round")
del args[position:position + 2]
code, out, err = self.run_cli(args, client=client)
self.assertEqual(code, 0, err)
self.assertEqual(json.loads(out)["applied"][0]["context_session"]["value"], "new-task")
self.out, self.err = io.StringIO(), io.StringIO()
code, out, err = self.run_cli(
self._fix_args(1, "--retain-context"),
client=self._client({"grok": "idle"}, sessions={"grok": "new-task"}),
)
self.assertEqual(code, 0, err)
self.assertEqual(len(self.runner.writes()), 1)
row = json.loads(self.state.read_text(encoding="utf-8"))["assignments"][-1]
self.assertEqual((row["fix_round"], row["clear_reason"]), (1, "retained"))
self.assertEqual(row["context_session"]["value"], "new-task")
def test_unchanged_pre_clear_session_is_not_saved_as_fresh_context(self):
args = self._fix_args(1)
position = args.index("--fix-round")
del args[position:position + 2]
code, out, err = self.run_cli(
args, client=self._client({"grok": "idle"}, sessions={"grok": "stale-reference"}),
)
self.assertEqual(code, 0)
self.assertIsNone(json.loads(out)["applied"][0]["context_session"])
self.assertIn("no verified post-clear", err)
def test_identity_appearing_after_first_prompt_supports_the_next_retained_fix(self):
client = self._client({"grok": "idle"})
self.runner.responses["agent get grok"] = ScriptedReads([
agent_json("grok", "idle", "w4:p1", "previous-task"),
agent_json("grok", "idle", "w4:p1"),
agent_json("grok", "working", "w4:p1"),
agent_json("grok", "working", "w4:p1", "delayed-native-id"),
])
args = self._fix_args(1)
position = args.index("--fix-round")
del args[position:position + 2]
code, out, err = self.run_cli(args, client=client)
self.assertEqual(code, 0, err)
row = json.loads(self.state.read_text())["assignments"][-1]
self.assertEqual(row["context_session"]["value"], "delayed-native-id")
self.assertEqual(row["status"], "applied")
self.assertEqual(sum(command.startswith("agent prompt grok 'New assignment")
for command in self.runner.commands()), 1)
self.out, self.err = io.StringIO(), io.StringIO()
code, out, err = self.run_cli(self._fix_args(1, "--retain-context"),
client=self._client({"grok": "idle"}, sessions={"grok": "delayed-native-id"}))
self.assertEqual(code, 0, err)
self.assertEqual(json.loads(out)["applied"][0]["fix_round"], 1)
self.assertEqual(len(self.runner.writes()), 1)
def test_unprovable_post_dispatch_identity_preserves_the_sent_assignment(self):
malformed = json.loads(agent_json("grok", "working", "w4:p1", "new-session"))
malformed["result"]["agent"]["agent_session"]["source"] = "pane-label"
for before, after in (
(None, agent_json("grok", "working", "w4:p1")),
(None, agent_json("grok", "working", "w4:p1", "old-session")),
(None, json.dumps(malformed)),
(None, agent_json("grok", "working", "different-pane", "new-session")),
("first-session", agent_json("grok", "working", "w4:p1", "replacement-session")),
):
with self.subTest(before=before, after=after):
save_state(self.state, empty_state())
self.out, self.err = io.StringIO(), io.StringIO()
client = self._client({"grok": "idle"})
self.runner.responses["agent get grok"] = ScriptedReads([
agent_json("grok", "idle", "w4:p1", "old-session"),
agent_json("grok", "idle", "w4:p1", before), after,
])
args = self._fix_args(1)
position = args.index("--fix-round")
del args[position:position + 2]
code, out, err = self.run_cli(args, client=client)
self.assertEqual(code, 0, err)
records = json.loads(self.state.read_text())["assignments"]
self.assertEqual(len(records), 1)
self.assertEqual(records[0]["status"], "applied")
self.assertIsNone(records[0]["context_session"])
self.assertTrue(err)
self.assertEqual(sum(command.startswith("agent prompt grok 'New assignment")
for command in self.runner.commands()), 1)
def test_missing_brief_flag_is_an_actionable_error(self):
client = self._client({})
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", json.dumps({"developer": "grok"}), "--common", str(self.common), "--dry-run"],
client=client,
)
self.assertEqual(code, 1)
self.assertIn("--brief developer=", json.loads(err)["message"])
def test_malformed_brief_pair_is_rejected(self):
client = self._client({})
code, _, err = self.run_cli(
self.base()
+ [
"apply",
"--composer-settle",
"0",
"--assignments",
json.dumps({"developer": "grok"}),
"--common",
str(self.common),
"--brief",
"developer",
"--dry-run",
],
client=client,
)
self.assertEqual(code, 1)
self.assertIn("ROLE=PATH", json.loads(err)["message"])
def test_nonexistent_brief_file_is_rejected_before_anything_is_sent(self):
client = self._client({"grok": "idle"})
code, _, err = self.run_cli(
self.base()
+ [
"apply",
"--composer-settle",
"0",
"--assignments",
json.dumps({"developer": "grok"}),
"--common",
str(self.common),
"--brief",
"developer={}".format(self.tmp / "missing.md"),
"--now",
AT,
],
client=client,
)
self.assertEqual(code, 1)
self.assertIn("missing.md", json.loads(err)["message"])
self.assertEqual(self.runner.calls, [])
def test_malformed_assignments_json_is_rejected(self):
client = self._client({})
code, _, err = self.run_cli(
self.base()
+ ["apply", "--composer-settle", "0", "--assignments", "{not json", "--common", str(self.common), "--dry-run"],
client=client,
)
self.assertEqual(code, 1)
self.assertEqual(json.loads(err)["error"], "usage_error")
def test_an_unreadable_state_file_is_never_overwritten(self):
self.state.write_text(self.CORRUPT_STATE, encoding="utf-8")
client = self._client({"grok": "idle"})
code, _out, err = self.run_cli(
self.base()
+ [
"apply",
"--assignments",
json.dumps({"developer": "grok"}),
"--common",
str(self.common),
"--now",
AT,
]
+ self.brief_args("developer"),
client=client,
)
self.assertNotEqual(code, 0)
self.assertIn("would destroy its contents", err)
self.assertEqual(self.state.read_text(encoding="utf-8"), self.CORRUPT_STATE)
class JudgeModeTest(unittest.TestCase):
"""A seat's mode is chosen once, at plan, and read everywhere after (#425)."""
def setUp(self):
self.args = SimpleNamespace(judge_mode=None)
def plan_doc(self, mode=None):
judge = {"agent": "claude", "model": "opus-5", "effort": "high"}
if mode is not None:
judge["mode"] = mode
return {"schema_version": 6, "assignments": {"judge": "claude"}, "judge": judge}
def test_the_plan_supplies_the_mode(self):
self.assertEqual(cli._judge_mode_for(self.args, self.plan_doc("diagnosis")), "diagnosis")
def test_a_matching_flag_agrees(self):
args = SimpleNamespace(judge_mode="diagnosis")
self.assertEqual(cli._judge_mode_for(args, self.plan_doc("diagnosis")), "diagnosis")
def test_a_differing_flag_refuses(self):
args = SimpleNamespace(judge_mode="adjudication")
with self.assertRaisesRegex(UsageError, "the plan's mode is the one its brief was composed for"):
cli._judge_mode_for(args, self.plan_doc("diagnosis"))
def test_a_plan_seating_a_judge_without_a_mode_refuses(self):
# A flag cannot supply what the plan's brief was never composed for.
for args in (self.args, SimpleNamespace(judge_mode="adjudication")):
with self.assertRaisesRegex(UsageError, "re-plan with --judge-mode"):
cli._judge_mode_for(args, self.plan_doc())
def test_a_bare_assignments_map_takes_the_flag(self):
# Not a plan document: there is no seat to have recorded a mode.
args = SimpleNamespace(judge_mode="adjudication")
self.assertEqual(cli._judge_mode_for(args, {"judge": "claude"}), "adjudication")
self.assertIsNone(cli._judge_mode_for(self.args, {"judge": "claude"}))
class ReportPathTests(unittest.TestCase):
"""`apply --report` shares the report-marker character rule (#578)."""
def test_a_report_path_that_splits_the_marker_is_refused(self):
for char in ("\n", "\x7f", "\x85", "\u2028", "\u2029", "\u202e"):
with self.subTest(char=hex(ord(char))):
with self.assertRaisesRegex(UsageError, "ROLE=ABS_PATH"):
cli._parse_reports(["developer=/r/a" + char + "b.md"], {"developer": {}})
def test_an_ordinary_report_path_is_accepted(self):
self.assertEqual(cli._parse_reports(["developer=/r/報告 1.md"], {"developer": {}}),
{"developer": "/r/報告 1.md"})
if __name__ == "__main__":
unittest.main().tessl-plugin
hooks
rules
agent-team-operation.mdagent-worktree-isolation.mdboy-scout.mdci-safety.mdcode-formatting.mdcommit-conventions.mdcontext-artifacts.mdcontext-writing-style.mddependency-management.mderror-handling.mdexternal-repo-contributions.mdfile-hygiene.mdhook-action-reporting.mdlanguage-diagnostics.mdno-secrets.mdresponse-clarity.mdreview-severity.mdreviewer-feedback-reading.mdrule-frontmatter.mdscript-as-black-box.mdscript-delegation.mdship-on-green.mdskill-authoring.mdstateful-artifacts.mdsync-before-work.mdtesting-standards.md
skills
adopt-fork-pr
herdr-foreman
classify
foreman
__init__.py__main__.pyassign.pyattention_view.pyattention.pybilling.pycapabilities.pychronology.pychurn.pyclaude_native.pycli.pycomposer.pycomposition.pyconfig.pycost_report.pydiagnostics.pyengagement.pyerrors.pyforeman_queue.pyforeman_reset.pyherdr.pyhistorical.pyhome.pylaunch.pyload_set.pymeasure.pymembers.pymemory.pyoracle.pyparsers.pypartition.pyplanner.pyprobe.pyprune_result.pyrecovery.pyrenderable.pyreport_contract.pyreport_delivery.pyreport_gates.pyrestoration.pyretrospective_runtime.pyretrospective.pyrole_clear.pyrunnable.pyselection.pystate.pysupervision_gate.pysupervision_hook.pysupervision_runtime.pysupervision.pytiers.pytriggers.py
references
templates
tests
__init__.pyfakes.pytest_assign.pytest_attention.pytest_billing.pytest_bounded_run.shtest_capabilities.pytest_capability_routing.pytest_chronology.pytest_churn.pytest_classify.shtest_claude_native.pytest_cli.pytest_compose_briefs.shtest_composer.pytest_composition.pytest_config.pytest_continuity_cli.pytest_cost_report.pytest_diagnostics.pytest_engagement.pytest_entrypoints.pytest_foreman_launcher.shtest_foreman_queue.pytest_foreman_reset.pytest_foreman_seat.pytest_foreman_tier_check.pytest_freeze.pytest_herdr.pytest_historical.pytest_home.pytest_label_workspaces.shtest_launch.pytest_legacy_recovery.pytest_load_set.pytest_measure.pytest_members.pytest_memory.pytest_oracle.pytest_parsers.pytest_partition.pytest_planner.pytest_probe.pytest_provision_worktree.shtest_prune_remote_branches.shtest_prune_report_caches.pytest_prune_result.pytest_prune_worktrees.shtest_recovery_cli.pytest_recovery.pytest_renderable.pytest_report_contract.pytest_report_delivery.pytest_report_gates.pytest_report_verdict.pytest_resolve_gates.shtest_resolve_policy_paths.pytest_restoration.pytest_retrospective_runtime.pytest_retrospective.pytest_review_package.pytest_role_clear.pytest_roster.shtest_round_preflight.shtest_runnable.pytest_scoring.pytest_script_dir_newline.shtest_seat_holds.pytest_selection.pytest_skill_invocations.shtest_slice_scope_parity.pytest_specialist_cli.pytest_specialist_delivery.pytest_specialist_recovery.pytest_specialist_retention.pytest_stale_grok_delivery.pytest_start_judge_worker.pytest_state.pytest_supervision_cli.pytest_supervision_diagnostics.pytest_supervision_gate.pytest_supervision_replay.pytest_supervision.pytest_sweep_worktrees.shtest_tier_integration.pytest_tiers.pytest_triggers.pytest_typesafe_client.pytest_verdict_gates.pytest_verify_authority.shtest_wait_report.shtier_fixture.py
herdr-standup
migrate-to-plugin
onboard-repo
release
references
tests
test_capture_registry_baseline.shtest_check_changelog_placement.pytest_check_closing_issues.pytest_check_leftovers.shtest_commit_stamp.shtest_confirm_publish_landed.shtest_copilot_run.shtest_dismiss_ruled_review.shtest_dismiss_stale_reviews.shtest_origin_repo.pytest_poll_pr_reviews.shtest_pr_create_repo_binding.pytest_registry_has_version.shtest_registry_version.shtest_release_helpers.shtest_request_copilot_review.shtest_resolve_publish_run.shtest_script_dir_newline.shtest_smart_publish.shtest_stamp_changelog.pytest_verify_github_release.shtest_verify_moderation_cleared.shtest_verify_publish_landed.shtest_version_compare.shtest_watch_pr_reviews.sh