"""Public CLI dispatch enrollment and offline supervision with fake workers."""

import os as _os
import sys as _sys
_ROOT = _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__)))
if _ROOT not in _sys.path:
    _sys.path.insert(0, _ROOT)

import io
import json
import os
import unittest
from types import SimpleNamespace
from unittest.mock import Mock, patch

from teamlead import cli, supervision
from teamlead.state import state_lock
from tests import test_cli as fixture
from tests.fakes import FakeRunner, failing_replace, replaces_into

AT = fixture.AT
TASK = "fleet-task"


class SupervisionCliTest(fixture.CliCase):
    _client = fixture.ApplyCommandTest._client

    def setUp(self):
        super().setUp()
        self.runner = FakeRunner()
        environment = patch.dict(os.environ, {"XDG_STATE_HOME": str(self.tmp / "xdg")})
        environment.start()
        self.addCleanup(environment.stop)
        self.bindings = supervision.default_state_path().parent / "supervision-bindings"
        self.identity = supervision.identity("lead-native", str(self.tmp), "fixture", pane_id="lead-pane")
        supervision.bind(self.state, self.identity, AT, root=self.bindings)
        self.reports = {role: str(self.tmp / (role + "-report.md")) for role in self.briefs}

    def invoke(self, arguments, client=None):
        self.out, self.err = io.StringIO(), io.StringIO()
        return self.run_cli(self.base() + arguments, client=client)

    def apply_arguments(self, assignments=None, reports=None, task: str | None = TASK, **options):
        assignments = assignments or {"developer": "grok"}
        reports = {role: self.reports[role] for role in assignments} if reports is None else reports
        result = ["apply", "--assignments", json.dumps(assignments), "--common", str(self.common),
                  "--composer-settle", "0", "--now", AT, "--dispatch-id", "dispatch-fixture"]
        # A seated judge declares what it is for (#425); these fixtures
        # exercise adjudication, which carries no assessment requirement.
        if "judge" in assignments and "judge_mode" not in options:
            result += ["--judge-mode", "adjudication"]
        if task is not None:
            result += ["--task", task]
        result += self.brief_args(*assignments)
        for role, report in reports.items():
            result += ["--report", role + "=" + report]
        for option, value in options.items():
            result.append("--" + option.replace("_", "-"))
            if value is not True:
                result.append(str(value))
        return result

    def saved(self):
        return supervision.load(self.state)

    def test_all_roles_are_enrolled_before_clear_and_assignment(self):
        assignments = {"developer": "grok", "tester": "claude", "reviewer": "codex"}
        client = self._client({name: "idle" for name in assignments.values()})
        original_runner = client._runner
        writes_checked = []
        def checking_runner(argv):
            if "send-text" in argv or "send-keys" in argv or "prompt" in argv:
                enrolled = {row["assignment"]["agent"] for row in self.saved()["members"] if row["active"]}
                if "grok" in argv or "w4:p1" in argv:
                    self.assertIn("grok", enrolled)
                if "claude" in argv or "w2:p1" in argv:
                    self.assertIn("claude", enrolled)
                if "codex" in argv or "w3:p1" in argv:
                    self.assertIn("codex", enrolled)
                writes_checked.append(argv)
            return original_runner(argv)
        client._runner = checking_runner
        code, out, err = self.invoke(self.apply_arguments(assignments), client)
        self.assertEqual(code, 0, err)
        self.assertTrue(writes_checked)
        self.assertEqual(len(json.loads(out)["applied"]), 3)
        members = self.saved()["members"]
        self.assertEqual({row["id"] for row in members}, {"dispatch-fixture:" + role for role in assignments})
        self.assertEqual({row["assignment"]["report"] for row in members}, set(self.reports.values()))
        self.assertTrue(all(row["active"] for row in members))
        self.assertTrue(all(row["assignment"]["native_session"] is None for row in members), "Outgoing context must not become incoming proof")

    def test_release_assignment_uses_the_same_pre_send_enrollment_contract(self):
        self.briefs["release"] = self.tmp / "release.md"
        self.briefs["release"].write_text("Release the accepted task; make no source changes.")
        self.reports["release"] = str(self.tmp / "release-report.md")
        code, _, err = self.invoke(self.apply_arguments({"release": "grok"}), self._client({"grok": "idle"}))
        self.assertEqual(code, 0, err)
        member = self.saved()["members"][0]
        self.assertEqual(member["id"], "dispatch-fixture:release")
        self.assertEqual(member["assignment"]["report"], self.reports["release"])
        self.assertTrue(member["active"])

    def test_pinned_judge_is_enrolled_with_verified_launch_arguments(self):
        self.briefs["judge"] = self.tmp / "judge.md"
        self.briefs["judge"].write_text("Resolve the recorded disputed finding read-only.")
        self.reports["judge"] = str(self.tmp / "judge-report.md")
        config = json.loads(self.config.read_text())
        config["judge"] = {"agent": "claude", "model": "claude-opus-5", "effort": "high"}
        self.config.write_text(json.dumps(config))
        client = self._client({"claude": "idle"}, sessions={"claude": "judge-native"})
        self.runner.set("pane process-info --pane w2:p1", json.dumps({"result": {"process_info": {
            "pane_id": "w2:p1", "foreground_processes": [{"name": "claude", "pid": 200,
            "argv": ["claude", "--dangerously-skip-permissions", "--model", "claude-opus-5", "--effort", "high"]}]}}}))
        code, _, err = self.invoke(self.apply_arguments({"judge": "claude"}, no_clear=True), client)
        self.assertEqual(code, 0, err)
        member = self.saved()["members"][0]
        self.assertEqual(member["id"], "dispatch-fixture:judge")
        self.assertEqual(supervision.expected_assignment(member)["native_session"]["value"], "judge-native")

    def test_bound_dry_run_remains_read_only(self):
        before = supervision.store_path(self.state).read_bytes()
        client = self._client({})
        code, out, err = self.invoke(self.apply_arguments(dry_run=True), client)
        self.assertEqual(code, 0, err)
        self.assertTrue(json.loads(out)["dry_run"])
        self.assertEqual(supervision.store_path(self.state).read_bytes(), before)
        self.assertFalse(self.state.exists())
        self.assertEqual(self.runner.calls, [])

    def test_missing_task_or_any_report_refuses_before_all_live_calls(self):
        for task, reports in ((None, {}), (TASK, {}), (TASK, {"developer": self.reports["developer"]})):
            with self.subTest(task=task, reports=reports):
                client = self._client({"grok": "idle", "claude": "idle"})
                code, _, err = self.invoke(self.apply_arguments({"developer": "grok", "tester": "claude"}, reports=reports, task=task), client)
                self.assertEqual(code, 1)
                self.assertIn("--report", err)
                self.assertEqual(self.runner.calls, [])
                self.assertEqual(self.saved()["members"], [])

    def test_bad_report_map_is_rejected_before_input(self):
        bad_maps = ({"developer": "relative.md"}, {"tester": self.reports["tester"]}, {"developer": str(self.tmp)})
        for reports in bad_maps:
            with self.subTest(reports=reports):
                client = self._client({"grok": "idle"})
                code, _, err = self.invoke(self.apply_arguments(reports=reports), client)
                self.assertEqual(code, 1)
                self.assertIn("report", err)
                self.assertEqual(self.runner.calls, [])
        client = self._client({"grok": "idle"})
        code, _, _ = self.invoke(self.apply_arguments() + ["--report", "developer=" + self.reports["developer"]], client)
        self.assertEqual(code, 1)
        self.assertEqual(self.runner.calls, [])

    def test_shared_report_destination_refuses_the_entire_parallel_assignment(self):
        client = self._client({"grok": "idle", "claude": "idle"})
        same = {"developer": self.reports["developer"], "tester": self.reports["developer"]}
        code, _, err = self.invoke(self.apply_arguments({"developer": "grok", "tester": "claude"}, reports=same), client)
        self.assertEqual(code, 1)
        self.assertIn("distinct report", err)
        self.assertEqual(self.runner.calls, [])

    def test_enrollment_write_failure_prevents_even_clear_input(self):
        client = self._client({"grok": "idle"})
        owner = supervision.store_path(self.state)
        with failing_replace(replaces_into(owner), "injected enrollment write failure"):
            code, _, err = self.invoke(self.apply_arguments(), client)
        self.assertEqual(code, 1)
        self.assertIn("Cannot write the state file", err)
        self.assertIn("injected enrollment write failure", err)
        self.assertEqual(self.runner.writes(), [])
        self.assertFalse(self.state.exists())

    def test_interrupted_send_remains_active_and_retry_never_resends(self):
        client = self._client({"grok": "idle"})
        # The operator interrupts while the brief is being pasted.
        self.runner.raises["agent prompt grok"] = KeyboardInterrupt()
        with self.assertRaises(KeyboardInterrupt):
            self.invoke(self.apply_arguments(), client)
        member = self.saved()["members"][0]
        self.assertTrue(member["active"])
        dispatch = json.loads(self.state.read_text())["recovery"]["dispatches"][0]
        self.assertEqual(dispatch["status"], "sending")
        client = self._client({"grok": "idle"})
        code, _, err = self.invoke(self.apply_arguments(), client)
        self.assertEqual(code, 1)
        self.assertIn("uncertain outcome", err)
        self.assertEqual(self.runner.calls, [])
        self.assertEqual(self.saved()["members"][0], member)

    def test_confirmed_replay_is_idempotent_and_repairs_missing_enrollment_without_send(self):
        code, out, err = self.invoke(self.apply_arguments(), self._client({"grok": "idle"}))
        self.assertEqual(code, 0, err)
        original_id = json.loads(out)["applied"][0]["dispatch_id"]
        original_state = self.state.read_bytes()
        original_member = self.saved()["members"][0]
        # Simulate a previous installation with a persisted dispatch receipt but
        # no enrollment row; this is explicit owner recovery, not normal cleanup.
        supervision.transaction(self.state, lambda data: data["members"].clear())
        client = self._client({})
        code, out, err = self.invoke(self.apply_arguments(), client)
        self.assertEqual(code, 0, err)
        self.assertTrue(json.loads(out)["applied"][0]["replayed"])
        self.assertEqual(self.saved()["members"][0]["id"], original_id)
        self.assertEqual(self.saved()["members"][0]["assignment"], original_member["assignment"])
        self.assertEqual(self.state.read_bytes(), original_state)
        self.assertEqual(self.runner.calls, [])
        code, _, err = self.invoke(self.apply_arguments(), self._client({}))
        self.assertEqual(code, 0, err)
        self.assertEqual(len(self.saved()["members"]), 1)

    def test_gated_mixed_batch_refuses_before_resaving_a_replayed_enrollment(self):
        # Copilot on coding-policy#401: a batch mixing a replayed role and a
        # new role under an open decision must refuse before the replay's
        # enrollment is re-saved, so a refused apply writes nothing.
        from teamlead import attention
        code, out, err = self.invoke(self.apply_arguments(), self._client({"grok": "idle"}))
        self.assertEqual(code, 0, err)
        supervision.transaction(self.state, lambda data: data["members"].clear())
        sidecar = supervision.store_path(self.state)
        before_sidecar, before_state = sidecar.read_bytes(), self.state.read_bytes()
        attention.write(self.state, "record", {
            "id": "fixture-decision", "kind": "decision", "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"}]}, AT)
        client = self._client({"claude": "idle"})
        code, out, err = self.invoke(self.apply_arguments({"developer": "grok", "tester": "claude"}), client)
        self.assertEqual(code, 1)
        self.assertEqual(out, "")
        self.assertIn("fixture-decision", err)
        self.assertEqual(sidecar.read_bytes(), before_sidecar)
        self.assertEqual(self.state.read_bytes(), before_state)
        self.assertEqual(self.runner.calls, [])
        # A batch of replays alone still returns its receipt and repairs enrollment.
        code, out, err = self.invoke(self.apply_arguments(), self._client({}))
        self.assertEqual(code, 0, err)
        self.assertTrue(json.loads(out)["applied"][0]["replayed"])
        self.assertEqual(len(self.saved()["members"]), 1)

    def refusal_args(self, agent, report, dispatch_id, brief=None):
        # Each replacement carries a fresh report path and, unchanged
        # otherwise, the same brief identity. Timestamps advance per dispatch.
        from datetime import datetime, timedelta
        self.briefs["tester"].write_text(brief or "# tester\nWrite `{0}` covering the plan.\nREPORT: {0}\n".format(report))
        count = len(json.loads(self.state.read_text())["assignments"]) if self.state.exists() else 0
        at = (datetime.fromisoformat(AT) + timedelta(seconds=count)).isoformat()
        return ["apply", "--assignments", json.dumps({"tester": agent}), "--common", str(self.common),
                "--composer-settle", "0", "--now", at, "--dispatch-id", dispatch_id, "--task", TASK,
                "--brief", "tester=" + str(self.briefs["tester"]), "--report", "tester=" + report]

    def refusal_receipt(self, agent, report, name):
        path = self.tmp / name
        path.write_text(json.dumps({"agent": agent, "state": "idle", "report_path": report, "found": False,
                                    "elapsed_seconds": 12, "reason": "terminal_provider_refusal"}))
        return str(path)

    def record_refusal(self, dispatch, receipt):
        record = self.tmp / "refusal-record.json"
        record.write_text(json.dumps({"dispatch": dispatch, "receipt": receipt}))
        return self.invoke(["record-refusal", "--record", str(record), "--now", AT])

    def recovery_rows(self):
        return json.loads(self.state.read_text())["recovery"]["dispatches"]

    def test_a_refused_brief_moves_once_to_another_provider_and_then_stops(self):
        # coding-policy#399, live case: codex refused the tester brief
        # mid-execution. The lead may move the brief unchanged to one other
        # provider; a same-provider resend, a reworded brief, and a third
        # dispatch after two refusals are refused before any keystroke.
        reports = {n: str(self.tmp / "tester-{}.md".format(n)) for n in (1, 2, 3, 4)}
        code, out, err = self.invoke(self.refusal_args("codex", reports[1], "attempt-1"), self._client({"codex": "idle"}))
        self.assertEqual(code, 0, err)
        first = json.loads(out)["applied"][0]["dispatch_id"]
        code, _, err = self.record_refusal(first, self.refusal_receipt("codex", reports[2], "stale.json"))
        self.assertEqual(code, 1)
        self.assertIn("is enrolled for " + reports[1], err)
        code, out, err = self.record_refusal(first, self.refusal_receipt("codex", reports[1], "refusal-1.json"))
        self.assertEqual(code, 0, err)
        self.assertEqual((json.loads(out)["provider"], json.loads(out)["report_path"]), ("codex", reports[1]))
        for extra in ((), ("--dry-run",)):
            code, out, err = self.invoke(self.refusal_args("codex", reports[2], "attempt-2") + list(extra), self._client({"codex": "idle"}))
            self.assertEqual(code, 1)
            self.assertEqual(out, "")
            self.assertIn("same provider", err)
            self.assertEqual(self.runner.writes(), [])
        reworded = "# tester\nWrite `{0}` covering the plan.\nSkip the security checks.\nREPORT: {0}\n".format(reports[2])
        code, out, err = self.invoke(self.refusal_args("claude", reports[2], "attempt-2", brief=reworded), self._client({"claude": "idle"}))
        self.assertEqual(code, 1)
        self.assertIn("reworded brief is not a move", err)
        self.assertEqual(self.runner.writes(), [])
        code, out, err = self.invoke(self.refusal_args("claude", reports[2], "attempt-2"), self._client({"claude": "idle"}))
        self.assertEqual(code, 0, err)
        second = json.loads(out)["applied"][0]["dispatch_id"]
        moved = self.recovery_rows()[-1]
        self.assertEqual((moved["id"], moved["refusal_move"]["from"], moved["refusal_move"]["from_provider"], moved["refusal_move"]["provider"]),
                         (second, first, "codex", "claude"))
        self.assertEqual(moved["brief_identity"], self.recovery_rows()[0]["brief_identity"])
        code, out, err = self.invoke(self.refusal_args("grok", reports[3], "attempt-3"), self._client({"grok": "idle"}))
        self.assertEqual(code, 1)
        self.assertIn("already moved to provider claude", err)
        code, _, err = self.record_refusal(second, self.refusal_receipt("claude", reports[2], "refusal-2.json"))
        self.assertEqual(code, 0, err)
        code, out, err = self.invoke(self.refusal_args("grok", reports[4], "attempt-4"), self._client({"grok": "idle"}))
        self.assertEqual(code, 1)
        self.assertEqual(out, "")
        self.assertIn("refused by 2 providers", err)
        self.assertEqual(self.runner.writes(), [])
        grant = self.tmp / "grant.json"
        grant.write_text(json.dumps({"id": "grant-1", "task": TASK, "role": "tester", "fix_round": None, "provider": "grok", "brief": "unchanged",
                                     "decision": "Run the tester on grok with the same brief.",
                                     "authorization": {"source": "fixture operator message", "quote": "Send it to grok."}}))
        code, _, err = self.invoke(["authorize-refused-dispatch", "--record", str(grant), "--now", AT])
        self.assertEqual(code, 0, err)
        code, _, err = self.invoke(self.refusal_args("codex", reports[4], "attempt-4"), self._client({"codex": "idle"}))
        self.assertEqual(code, 1)
        self.assertIn("approves provider grok", err)
        code, _, err = self.invoke(self.refusal_args("grok", reports[4], "attempt-4", brief=reworded.replace(reports[2], reports[4])), self._client({"grok": "idle"}))
        self.assertEqual(code, 1)
        self.assertIn("approves the refused brief unchanged", err)
        code, out, err = self.invoke(self.refusal_args("grok", reports[4], "attempt-4"), self._client({"grok": "idle"}))
        self.assertEqual(code, 0, err)
        self.assertEqual(self.recovery_rows()[-1]["refusal_move"]["authorization"], "grant-1")
        code, out, err = self.invoke(self.refusal_args("grok", str(self.tmp / "tester-5.md"), "attempt-5"), self._client({"grok": "idle"}))
        self.assertEqual(code, 1)
        self.assertIn("refused by 2 providers", err)
        code, _, err = self.invoke(["status"])
        self.assertEqual(code, 0, err)

    def test_a_receipt_naming_the_refined_pane_id_is_accepted(self):
        # coding-policy#403: supervision can fill in a pane id the enrollment
        # never knew, and wait-report may have been given that one.
        report = str(self.tmp / "tester-1.md")
        code, out, err = self.invoke(self.refusal_args("codex", report, "attempt-1"), self._client({"codex": "idle"}))
        self.assertEqual(code, 0, err)
        first = json.loads(out)["applied"][0]["dispatch_id"]
        # A replayed or legacy enrollment can carry no pane id; supervision
        # fills it in afterwards through a refinement.
        supervision.transaction(self.state, lambda data: data["members"][0]["assignment"].update(pane_id=None))
        supervision.refine(self.state, first, "w9:p9", None, AT)
        self.assertEqual(supervision.expected_assignment(self.saved()["members"][0])["pane_id"], "w9:p9")
        code, out, err = self.record_refusal(first, self.refusal_receipt("w9:p9", report, "refined.json"))
        self.assertEqual(code, 0, err)
        self.assertEqual(json.loads(out)["provider"], "codex")

    def test_record_refusal_needs_the_refused_worker_in_config(self):
        report = str(self.tmp / "tester-1.md")
        code, out, err = self.invoke(self.refusal_args("codex", report, "attempt-1"), self._client({"codex": "idle"}))
        self.assertEqual(code, 0, err)
        first = json.loads(out)["applied"][0]["dispatch_id"]
        config = json.loads(self.config.read_text())
        config["agents"] = [agent for agent in config["agents"] if agent["name"] != "codex"]
        self.config.write_text(json.dumps(config))
        code, _, err = self.record_refusal(first, self.refusal_receipt("codex", report, "refusal-1.json"))
        self.assertEqual(code, 1)
        self.assertIn("not in config.json", err)
        code, _, err = self.record_refusal("no-such-dispatch", self.refusal_receipt("codex", report, "refusal-1.json"))
        self.assertEqual(code, 1)
        self.assertIn("dispatch", err)

    def test_sidecar_failure_after_confirmed_send_does_not_make_it_replayable(self):
        client = self._client({"grok": "idle"})
        sent = self.runner
        owner = supervision.store_path(self.state)
        def after_send(_source, target):
            return target == owner and any(call[1:3] == ["agent", "prompt"] for call in sent.calls)
        with failing_replace(after_send, "Post-send sidecar failure"):
            code, _, err = self.invoke(self.apply_arguments(), client)
        self.assertEqual(code, 1)
        self.assertIn("Post-send sidecar", err)
        self.assertEqual(json.loads(self.state.read_text())["recovery"]["dispatches"][0]["status"], "applied")
        client = self._client({})
        code, out, err = self.invoke(self.apply_arguments(), client)
        self.assertEqual(code, 0, err)
        self.assertTrue(json.loads(out)["applied"][0]["replayed"])
        self.assertEqual(self.runner.calls, [])
        self.assertEqual(len(json.loads(self.state.read_text())["assignments"]), 1)

    def test_report_change_under_same_dispatch_id_refuses_even_if_enrollment_missing(self):
        code, _, err = self.invoke(self.apply_arguments(), self._client({"grok": "idle"}))
        self.assertEqual(code, 0, err)
        supervision.transaction(self.state, lambda data: data["members"].clear())
        client = self._client({})
        code, _, err = self.invoke(self.apply_arguments(reports={"developer": str(self.tmp / "wrong-report.md")}), client)
        self.assertEqual(code, 1)
        self.assertIn("different inputs", err)
        self.assertEqual(self.runner.calls, [])

    def test_legacy_applied_retry_after_binding_repairs_without_resend(self):
        # Model a real old installation: this distinct state has never had a
        # supervision owner or discovery record. Never clear a live binding.
        self.state = self.tmp / "never-bound-legacy-state.json"
        arguments = self.apply_arguments(reports={})
        code, _, err = self.invoke(arguments, self._client({"grok": "idle"}))
        self.assertEqual(code, 0, err)
        binding = self.tmp / "legacy-native-binding.json"
        binding.write_text(json.dumps(supervision.identity("legacy-native", str(self.tmp), "fixture", pane_id="lead-pane")))
        code, _, err = self.invoke(["supervision-bind", "--record", str(binding), "--now", AT])
        self.assertEqual(code, 0, err)
        client = self._client({})
        code, out, err = self.invoke(self.apply_arguments(), client)
        self.assertEqual(code, 0, err)
        self.assertTrue(json.loads(out)["applied"][0]["replayed"])
        self.assertEqual(self.runner.calls, [])
        self.assertEqual(len(self.saved()["members"]), 1)

    def test_missing_bound_owner_refuses_legacy_fallback_before_any_worker_read(self):
        discovery = supervision.binding_path(self.identity)
        original = discovery.read_bytes()
        supervision.store_path(self.state).unlink()
        client = self._client({"grok": "idle"})
        code, _, err = self.invoke(self.apply_arguments(reports={}), client)
        self.assertEqual(code, 1)
        self.assertIn("missing supervision owner", err)
        self.assertEqual(self.runner.calls, [])
        self.assertFalse(supervision.store_path(self.state).exists())
        self.assertFalse(self.state.exists())
        self.assertEqual(discovery.read_bytes(), original)

    def test_new_native_lead_cannot_bypass_missing_earlier_owner(self):
        discovery = supervision.binding_path(self.identity)
        original = discovery.read_bytes()
        supervision.store_path(self.state).unlink()
        client = self._client({"grok": "idle"})
        with patch.dict(os.environ, {"HERDR_ENV": "new-session", "HERDR_PANE_ID": "new-lead-pane", "CODEX_THREAD_ID": "replacement-native"}):
            code, _, err = self.invoke(self.apply_arguments(), client)
        self.assertEqual(code, 1)
        self.assertIn("missing supervision owner", err)
        self.assertEqual(self.runner.calls, [])
        self.assertEqual(discovery.read_bytes(), original)

    def test_interrupted_first_bind_requires_completion_before_dispatch(self):
        self.state = self.tmp / "interrupted-first-bind.json"
        who = supervision.identity("initializing-native", str(self.tmp), "fixture", pane_id="lead-pane")
        record = self.tmp / "initializing-binding.json"
        record.write_text(json.dumps(who))
        def bound_owner(source, target):
            return target == supervision.store_path(self.state) and json.loads(source.read_text())["binding"] is not None
        with failing_replace(bound_owner, "Simulated interrupted binding commit; resume initialization."):
            code, _, err = self.invoke(["supervision-bind", "--record", str(record), "--now", AT])
        self.assertEqual(code, 1, err)
        self.assertIsNone(self.saved()["binding"])
        owner_before = supervision.store_path(self.state).read_bytes()
        discovery = supervision.binding_path(who)
        discovery_before = discovery.read_bytes()
        client = self._client({"grok": "idle"})
        code, _, err = self.invoke(self.apply_arguments(), client)
        self.assertEqual(code, 1)
        self.assertIn("initialization is incomplete", err)
        self.assertEqual(self.runner.calls, [])
        self.assertEqual(supervision.store_path(self.state).read_bytes(), owner_before)
        self.assertEqual(discovery.read_bytes(), discovery_before)
        code, _, err = self.invoke(["supervision-bind", "--record", str(record), "--now", AT])
        self.assertEqual(code, 0, err)
        code, _, err = self.invoke(self.apply_arguments(), self._client({"grok": "idle"}))
        self.assertEqual(code, 0, err)
        self.assertTrue(self.saved()["members"][0]["active"])

    def test_no_clear_retains_verified_native_proof_for_all_roles(self):
        native_sessions = {"grok": "grok-native", "claude": "claude-native"}
        assignments = {"developer": "grok", "tester": "claude"}
        code, _, err = self.invoke(self.apply_arguments(assignments, no_clear=True), self._client({name: "idle" for name in native_sessions}, sessions=native_sessions))
        self.assertEqual(code, 0, err)
        for member in self.saved()["members"]:
            self.assertEqual(supervision.expected_assignment(member)["native_session"]["value"], native_sessions[member["assignment"]["agent"]])

    def test_offline_commands_do_not_lock_or_read_main_state_or_config(self):
        self.state.write_text("corrupt main state must remain untouched")
        self.config.unlink()
        evidence = self.tmp / "outcome.md"
        evidence.write_text("Saved verified outcome")
        assignment = {"id": "offline-dispatch", "agent": "grok", "task": TASK, "report": self.reports["developer"], "pane_id": None, "native_session": None}
        record = self.tmp / "enroll.json"
        record.write_text(json.dumps(assignment))
        with state_lock(self.state):
            code, _, err = self.invoke(["supervision-enroll", "--record", str(record), "--now", AT])
            self.assertEqual(code, 0, err)
            code, out, err = self.invoke(["supervision-status", "--now", AT])
            self.assertEqual(code, 0, err)
            self.assertEqual(json.loads(out)["active"][0]["id"], "offline-dispatch")
            code, out, err = self.invoke(["supervision-drain", "--now", AT])
            self.assertEqual(code, 0, err)
            self.assertEqual(json.loads(out)["events"], [])
        self.assertEqual(self.state.read_text(), "corrupt main state must remain untouched")

    def test_watch_uses_injected_cli_clock_and_returns_durable_observation(self):
        assignment = {"id": "blocked-dispatch", "agent": "grok", "task": TASK, "report": self.reports["developer"], "pane_id": "w4:p1", "native_session": None}
        supervision.enroll(self.state, assignment, AT)
        client = self._client({"grok": "blocked"})
        self.runner.set("agent read grok --source visible --lines 200", "Candidate approval dialog")
        # Isolate the fleet sleeper from Python's shared time module: ps's
        # subprocess timeout machinery may sleep while waiting for its child.
        fleet_sleep = Mock(name="fleet_sleep")
        with patch.object(cli, "now_iso", return_value=AT), patch.object(cli, "time", SimpleNamespace(sleep=fleet_sleep)):
            code, out, err = self.invoke(["supervision-watch", "--now", AT], client)
        self.assertEqual(code, 0, err)
        self.assertEqual(json.loads(out)["reason"], "events")
        fleet_sleep.assert_not_called()
        self.assertEqual(self.runner.writes(), [])
        self.assertTrue(self.saved()["events"])


if __name__ == "__main__":
    unittest.main()