"""Durable fleet events, exact acknowledgements, crash recovery, and Stop scope."""

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 argparse
import copy
import json
import os
import subprocess
import shlex
import shutil
import tempfile
import unittest
from datetime import timedelta
from pathlib import Path
from unittest.mock import patch
from typing import Callable

from foreman import supervision as store
from foreman import supervision_runtime as runtime
from foreman import supervision_hook as hook
from foreman import foreman_reset
from foreman.errors import HerdrError, StateError, UsageError

AT = "2026-09-01T12:00:00+00:00"
PROCESS = {"pid": 101, "identity": "live-process-identity"}
NATIVE = {"source": "herdr:codex", "agent": "codex", "kind": "id", "value": "native-worker"}


class Clock:
    def __init__(self):
        self.value = store.timestamp(AT)
        self.sleeps = []
        self.on_sleep: Callable[[], object] | None = None

    def now(self):
        return self.value.isoformat()

    def sleep(self, seconds):
        self.sleeps.append(seconds)
        self.value += timedelta(seconds=seconds)
        if self.on_sleep:
            self.on_sleep()


class Client:
    def __init__(self):
        self.agents = {}
        self.reads = []
        self.visible = {}
        self.panes = {}
        self.failures = set()

    def agent_get(self, name):
        self.reads.append(name)
        if name in self.failures:
            raise HerdrError("Read unavailable; inspect the worker connection.", {})
        return copy.deepcopy(self.agents[name])

    def agent_read(self, name, **_kwargs):
        return self.visible.get(name, "working")

    def pane_get(self, pane):
        return copy.deepcopy(self.panes[pane])


class SupervisionTest(unittest.TestCase):
    def setUp(self):
        temporary = tempfile.TemporaryDirectory(prefix="foreman-supervision-")
        self.addCleanup(temporary.cleanup)
        self.root = Path(temporary.name).resolve()
        self.path = self.root / "owner.json"
        self.bindings = self.root / "bindings"
        self.evidence = self.root / "TASK-LEDGER.md"
        self.evidence.write_text("Foreman verified this bounded outcome; task release still pending.\n")
        self.who = store.identity("lead-session", str(self.root), "fixture", pane_id="lead-pane")
        store.bind(self.path, self.who, AT, root=self.bindings)
        self.client = Client()
        self.clock = Clock()
        self.probe = lambda _pid: PROCESS
        self.payload = {"cwd": str(self.root), "session_id": "lead-session", "stop_hook_active": False}
        self.environ = {"HERDR_ENV": "fixture", "HERDR_PANE_ID": "lead-pane"}

    def member(self, name="a", status="working", native=NATIVE):
        record = {"id": "dispatch-" + name, "agent": name, "task": "task-" + name,
                  "report": str(self.root / (name + ".md")), "pane_id": "pane-" + name,
                  "native_session": native}
        store.enroll(self.path, record, AT)
        self.client.agents[name] = {"pane_id": record["pane_id"], "agent_status": status, "agent_session": native}
        return record

    def watch(self, **kwargs):
        return runtime.watch(self.path, self.client, self.clock.now(), clock=self.clock.now,
                             sleeper=self.clock.sleep, probe=self.probe, pid=101, duration=4, interval=1, **kwargs)

    def emit(self, member="dispatch-a", kind="test"):
        return store.transaction(self.path, lambda data: store.append_event(data, AT, member, kind, {"observation_only": True}))

    def ack(self, result=None):
        result = result or store.drain(self.path)
        return store.acknowledge(self.path, {"through": result["through"], "outcomes": [
            {"event": row["id"], "outcome": "Reconciled source; no acceptance inferred.", "evidence": [str(self.evidence)]}
            for row in result["events"]]}, AT)

    def stop(self, payload=None, environ=None, probe=None):
        return hook.check(payload or self.payload, self.environ if environ is None else environ,
                          self.clock.now(), root=self.bindings, probe=probe or self.probe)

    def hold(self, members, *, kind="waiting_for_user", hold_id="hold-1"):
        condition = "User supplies the named decision." if kind == "waiting_for_user" else "The scheduled reset resumes this saved stow."
        return store.hold(self.path, {"id": hold_id, "kind": kind, "resume_condition": condition,
            "evidence": [str(self.evidence)], "dispositions": [
                {"member": member, "outcome": "The assignment remains covered by the saved boundary.", "evidence": [str(self.evidence)]}
                for member in members]}, AT)

    def schedule_reset(self, *, stow="hold-1", pane="lead-pane", native=None, claimed=True):
        plan = {"pane_id": pane, "stow": stow}
        result = foreman_reset.schedule(
            self.path, plan, AT, lambda: PROCESS["pid"],
            native_session=native or {"kind": "id", "value": "lead-session"}, probe=self.probe)
        if claimed:
            foreman_reset.claim(self.path, plan, PROCESS)
        return result

    def test_worker_b_event_is_returned_while_a_is_still_working(self):
        self.member("a")
        b = self.member("b")
        self.clock.on_sleep = lambda: Path(b["report"]).write_text("Candidate report; review still required")
        result = self.watch()
        self.assertEqual(result["reason"], "events")
        self.assertIn("a", self.client.reads)
        self.assertIn("b", self.client.reads)
        self.assertTrue(any(row["member"] == "dispatch-b" and row["kind"] == "report_observed" for row in result["events"]))
        self.assertTrue(all(row["active"] for row in store.load(self.path)["members"]))

    def test_done_idle_are_only_observations(self):
        self.member(status="done")
        result = self.watch()
        self.assertEqual(result["events"][0]["kind"], "lifecycle_observed")
        self.ack(result)
        self.assertTrue(store.load(self.path)["members"][0]["active"])
        self.assertIsNotNone(self.stop())

    def test_blocked_is_observed_without_answering_or_confirming_a_dialog(self):
        self.member(status="blocked")
        self.client.visible["a"] = "Approval needed; use downstream two-read dialog verification"
        result = self.watch()
        self.assertIn("lifecycle_observed", [row["kind"] for row in result["events"]])
        self.assertNotIn("Approval needed", json.dumps(result))

    def test_identical_observations_deduplicate_after_ack(self):
        record = self.member(status="idle")
        Path(record["report"]).write_text("Ready for report-delivery reconciliation")
        first = self.watch()
        self.ack(first)
        second = self.watch()
        self.assertEqual(second["reason"], "deadline")
        self.assertEqual(second["events"], [])
        self.assertEqual(second["through"], first["through"])

    def test_new_visible_marker_changes_wake_even_if_report_bytes_do_not(self):
        record = self.member(status="idle")
        Path(record["report"]).write_text("Body")
        self.ack(self.watch())
        self.client.visible["a"] = "REPORT: " + record["report"]
        result = self.watch()
        self.assertEqual([row["kind"] for row in result["events"]], ["visible_observed"])

    def test_pending_ack_schedules_durable_recheck_without_changed_bytes(self):
        self.member()
        self.emit()
        request = {"through": 1, "outcomes": [{"event": "event-1", "outcome": "The exact report checkpoint is still pending.",
                    "evidence": [str(self.evidence)], "pending": True}]}
        store.acknowledge(self.path, request, AT)
        saved = store.load(self.path)["acknowledgements"][0]
        store.acknowledge(self.path, request, "2026-09-01T12:00:01+00:00")
        self.assertEqual(store.load(self.path)["acknowledgements"][0], saved)
        self.clock.value = store.timestamp(saved["recheck_at"])
        result = self.watch()
        rechecks = [row for row in result["events"] if row["kind"] == "recheck_due"]
        self.assertEqual(len(rechecks), 1)
        self.assertEqual(rechecks[0]["data"]["event"], "event-1")
        self.ack(result)
        self.assertEqual(self.watch()["events"], [])

    def test_recheck_before_ack_is_rejected_without_mutation(self):
        self.member()
        self.emit()
        before = store.store_path(self.path).read_bytes()
        with self.assertRaises(UsageError):
            store.acknowledge(self.path, {"through": 1, "outcomes": [{"event": "event-1", "outcome": "pending", "evidence": [str(self.evidence)], "recheck_at": "2026-09-01T11:00:00+00:00"}]}, AT)
        self.assertEqual(store.store_path(self.path).read_bytes(), before)

    def test_pending_replay_does_not_touch_workers(self):
        self.member()
        self.emit()
        result = self.watch()
        self.assertEqual(result["reason"], "pending_events")
        self.assertEqual(self.client.reads, [])

    def test_drain_is_read_only_and_does_not_acknowledge(self):
        self.member()
        self.emit()
        before = store.store_path(self.path).read_bytes()
        first = store.drain(self.path)
        self.assertEqual(first, store.drain(self.path))
        self.assertEqual(store.store_path(self.path).read_bytes(), before)
        self.assertEqual(len(first["events"]), 1)

    def test_ack_bound_does_not_lose_events_arriving_after_drain(self):
        self.member()
        self.emit()
        drained = store.drain(self.path)
        later = self.emit(kind="later")
        result = self.ack(drained)
        self.assertEqual([row["id"] for row in result["remaining"]], [later["id"]])
        with self.assertRaises(UsageError):
            store.acknowledge(self.path, {"through": drained["through"], "outcomes": [
                {"event": later["id"], "outcome": "Not observed in that drain", "evidence": [str(self.evidence)]}]}, AT)

    def test_ack_requires_evidence_and_preserves_original_outcome(self):
        self.member()
        self.emit()
        request = {"through": 1, "outcomes": [{"event": "event-1", "outcome": "Handled", "evidence": []}]}
        with self.assertRaises(UsageError):
            store.acknowledge(self.path, request, AT)
        request["outcomes"][0]["evidence"] = [str(self.evidence)]
        store.acknowledge(self.path, request, AT)
        store.acknowledge(self.path, request, "2026-09-01T13:00:00+00:00")
        request["outcomes"][0]["outcome"] = "Different"
        with self.assertRaises(UsageError):
            store.acknowledge(self.path, request, AT)
        self.assertEqual(len(store.load(self.path)["acknowledgements"]), 1)

    def test_unavailable_worker_does_not_prevent_other_worker_observation(self):
        self.member("a")
        self.member("b", status="blocked")
        self.client.failures.add("a")
        result = self.watch()
        self.assertEqual({row["member"] for row in result["events"]}, {"dispatch-a", "dispatch-b"})

    def test_worker_relaunch_is_an_observation_and_never_reassigns(self):
        self.member()
        self.client.agents["a"]["agent_session"] = {**NATIVE, "value": "replacement"}
        result = self.watch()
        self.assertIn("identity_changed_observed", [row["kind"] for row in result["events"]])
        self.assertEqual(store.load(self.path)["members"][0]["assignment"]["native_session"], NATIVE)

    def test_native_context_session_pane_field_does_not_false_positive(self):
        self.member(native={"pane_id": "pane-a", **NATIVE})
        self.client.agents["a"]["agent_session"] = NATIVE
        self.assertEqual(self.watch()["events"], [])

    def test_first_watch_records_missing_native_identity(self):
        self.member(native=None)
        self.assertIn("identity_unverified_observed", [row["kind"] for row in self.watch()["events"]])

    def test_refine_only_fills_missing_identity_and_preserves_original(self):
        self.member(native=None)
        store.refine(self.path, "dispatch-a", "pane-a", NATIVE, AT)
        row = store.load(self.path)["members"][0]
        self.assertIsNone(row["assignment"]["native_session"])
        self.assertEqual(store.expected_assignment(row)["native_session"], NATIVE)
        with self.assertRaises(UsageError):
            store.refine(self.path, "dispatch-a", "pane-a", {**NATIVE, "value": "changed"}, AT)

    def test_dead_process_receipt_emits_loss_before_restart(self):
        self.member()
        self.watch()
        def corrupt_to_crashed(data):
            data["watchers"][-1].update(status="running", process={"pid": 99, "identity": "old"})
        store.transaction(self.path, corrupt_to_crashed)
        result = self.watch()
        self.assertEqual(result["reason"], "pending_events")
        self.assertEqual(result["events"][0]["kind"], "watcher_lost")
        self.ack(result)
        self.assertEqual(self.watch()["reason"], "deadline")

    def test_live_process_blocks_duplicate_even_with_old_heartbeat(self):
        self.member()
        self.watch()
        store.transaction(self.path, lambda data: data["watchers"][-1].update(status="running"))
        self.clock.value += timedelta(seconds=100)
        self.assertEqual(runtime.status(self.path, self.clock.now(), self.probe)["watcher"]["state"], "unresponsive")
        with self.assertRaisesRegex(UsageError, "still exists"):
            self.watch()

    def test_pid_reuse_is_not_live_identity(self):
        self.member()
        self.watch()
        store.transaction(self.path, lambda data: data["watchers"][-1].update(status="running"))
        changed = lambda _pid: {**PROCESS, "identity": "different-start-or-argv"}
        self.assertEqual(runtime.status(self.path, self.clock.now(), changed)["watcher"]["state"], "dead")

    def test_interrupt_records_stopped_watch_without_resolving_assignment(self):
        self.member()
        def interrupt():
            raise KeyboardInterrupt()
        self.clock.on_sleep = interrupt
        with self.assertRaises(KeyboardInterrupt):
            self.watch()
        data = store.load(self.path)
        self.assertEqual(data["watchers"][-1]["status"], "stopped")
        self.assertEqual(data["watchers"][-1]["reason"], "interrupted")
        self.assertTrue(data["members"][0]["active"])

    def test_only_explicit_resolution_ends_observation(self):
        self.member()
        self.emit()
        request = {"id": "dispatch-a", "outcome": "Foreman reconciled assignment; release tracked separately", "evidence": [str(self.evidence)]}
        with self.assertRaises(UsageError):
            store.resolve(self.path, request, AT)
        self.ack()
        store.resolve(self.path, request, AT)
        self.assertIsNone(self.stop())
        self.assertFalse(self.path.exists(), "Supervision must not rewrite dispatch/task state")

    def test_active_worker_always_blocks_stop_even_with_loop_guard(self):
        self.member()
        self.assertEqual(self.stop()["decision"], "block")
        self.assertEqual(self.stop({**self.payload, "stop_hook_active": True})["decision"], "block")

    def test_worker_and_unrelated_session_do_not_inherit_lead_gate(self):
        self.member()
        self.assertIsNone(self.stop({**self.payload, "session_id": "worker-session"}))
        self.assertIsNone(self.stop(environ={**self.environ, "HERDR_PANE_ID": "worker-pane"}))
        self.assertIsNone(self.stop(environ={}))
        self.assertIsNone(self.stop({**self.payload, "cwd": str(self.root / "other")}))

    def test_user_hold_requires_disposition_for_every_active_member(self):
        self.member("a")
        self.member("b")
        with self.assertRaisesRegex(UsageError, "every active"):
            self.hold(["dispatch-a"])
        self.hold(["dispatch-a", "dispatch-b"])
        self.assertIsNone(self.stop())
        store.resume(self.path, AT)
        self.assertEqual(self.stop()["decision"], "block")

    def test_nominal_handoff_does_not_release_stop_after_a_quiet_watch_deadline(self):
        self.member()
        result = self.watch()
        self.assertEqual(result["reason"], "deadline")
        self.hold(["dispatch-a"], kind="handoff")
        blocked = self.stop()
        self.assertEqual(blocked["decision"], "block")
        self.assertIn("prepares a reset but does not transfer supervision", blocked["reason"])
        self.assertIn("quiet watch deadline", blocked["reason"])
        self.assertIn("Schedule its exact live continuation", blocked["reason"])

    def test_live_unclaimed_reset_never_releases_stop_or_changes_its_record(self):
        self.member()
        self.hold(["dispatch-a"], kind="handoff")
        self.schedule_reset(claimed=False)
        path = foreman_reset.record_path(self.path)
        before = path.read_bytes()
        blocked = self.stop()
        self.assertEqual(blocked["decision"], "block")
        self.assertIn("reset_claim_pending", blocked["reason"])
        self.assertEqual(path.read_bytes(), before)

    def test_matching_live_claimed_reset_releases_the_handoff_boundary(self):
        self.member()
        self.hold(["dispatch-a"], kind="handoff")
        self.schedule_reset()
        self.assertIsNone(self.stop())

    def test_multiple_current_handoffs_refuse_an_older_live_reset(self):
        self.member()
        self.hold(["dispatch-a"], kind="handoff", hold_id="older-stow")
        self.schedule_reset(stow="older-stow")
        self.hold(["dispatch-a"], kind="handoff", hold_id="newer-stow")
        blocked = self.stop()
        self.assertEqual(blocked["decision"], "block")
        self.assertIn("handoff_ambiguous", blocked["reason"])

    def test_schema_1_reset_cannot_release_stop_or_rewrite_the_record(self):
        self.member()
        self.hold(["dispatch-a"], kind="handoff")
        path = foreman_reset.record_path(self.path)
        row = {"schema_version": 1, "pane_id": "lead-pane", "stow": "hold-1",
               "status": "scheduled", "scheduled_at": AT, "options": {},
               "process": PROCESS, "result": None}
        original = json.dumps({"schema_version": 1, "resets": [row]})
        path.write_text(original, encoding="utf-8")
        blocked = self.stop()
        self.assertEqual(blocked["decision"], "block")
        self.assertIn("no usable Stop proof", blocked["reason"])
        self.assertIn("catch-up", blocked["reason"])
        self.assertEqual(path.read_text(encoding="utf-8"), original)

    def test_malformed_legacy_reset_requires_restoration_without_rewriting(self):
        self.member()
        self.hold(["dispatch-a"], kind="handoff")
        path = foreman_reset.record_path(self.path)
        row = {"schema_version": 1, "pane_id": "lead-pane", "stow": "hold-1",
               "status": "scheduled", "scheduled_at": AT, "options": {},
               "process": PROCESS, "result": None}
        cases = ({"schema_version": 1}, {"schema_version": 1, "resets": [None]},
                 {"schema_version": 1, "resets": [dict(row, status="unknown")]},
                 {"schema_version": 1, "resets": [row, row]})
        for document in cases:
            with self.subTest(document=document):
                original = json.dumps(document)
                path.write_text(original, encoding="utf-8")
                blocked = self.stop()
                self.assertEqual(blocked["decision"], "block")
                self.assertIn("restores a valid file", blocked["reason"])
                self.assertNotIn("catch-up", blocked["reason"])
                self.assertEqual(path.read_text(encoding="utf-8"), original)

    def test_dead_reset_deliverer_routes_recovery_through_catch_up(self):
        self.member()
        self.hold(["dispatch-a"], kind="handoff")
        self.schedule_reset()
        blocked = self.stop(probe=lambda _pid: None)
        self.assertEqual(blocked["decision"], "block")
        self.assertIn("reset_deliverer_not_live", blocked["reason"])
        self.assertIn("catch-up", blocked["reason"])
        self.assertIn("obtain its reconciliation command", blocked["reason"])
        self.assertIn("do not schedule another attempt", blocked["reason"])

    def test_handoff_rejects_wrong_reset_identity_and_dead_or_reused_deliverer(self):
        cases = (
            ("wrong stow", {"stow": "another-stow"}, self.probe),
            ("wrong pane", {"pane": "another-pane"}, self.probe),
            ("wrong native", {"native": {"kind": "id", "value": "another-session"}}, self.probe),
            ("dead deliverer", {}, lambda _pid: None),
            ("reused pid", {}, lambda pid: {"pid": pid, "identity": "replacement-process"}),
        )
        for label, reset, probe in cases:
            with self.subTest(label=label):
                with tempfile.TemporaryDirectory(prefix="handoff-case-") as raw:
                    root = Path(raw).resolve()
                    state, bindings = root / "owner.json", root / "bindings"
                    evidence = root / "evidence.md"
                    evidence.write_text("authorized work remains active\n", encoding="utf-8")
                    who = store.identity("lead-session", str(root), "fixture", pane_id="lead-pane")
                    store.bind(state, who, AT, root=bindings)
                    store.enroll(state, {"id": "dispatch-a", "agent": "a", "task": "task-a",
                                        "report": str(root / "a.md"), "pane_id": "pane-a",
                                        "native_session": NATIVE}, AT)
                    store.hold(state, {"id": "hold-1", "kind": "handoff",
                                      "resume_condition": "The scheduled reset resumes this saved stow.",
                                      "evidence": [str(evidence)], "dispositions": [{"member": "dispatch-a",
                                      "outcome": "The assignment remains covered by the saved boundary.",
                                      "evidence": [str(evidence)]}]}, AT)
                    foreman_reset.schedule(
                        state, {"pane_id": reset.get("pane", "lead-pane"),
                                "stow": reset.get("stow", "hold-1")}, AT, lambda: PROCESS["pid"],
                        native_session=reset.get("native", {"kind": "id", "value": "lead-session"}),
                        probe=self.probe)
                    stopped = hook.check(
                        {"cwd": str(root), "session_id": "lead-session"},
                        {"HERDR_ENV": "fixture", "HERDR_PANE_ID": "lead-pane"}, AT,
                        root=bindings, probe=probe)
                    self.assertEqual(stopped["decision"], "block")

    def test_failed_reset_does_not_release_handoff_or_retry_the_stow(self):
        self.member()
        self.hold(["dispatch-a"], kind="handoff")
        def fail_launch():
            raise StateError("launch failed", {})
        with self.assertRaises(foreman_reset.ResetEnded):
            foreman_reset.schedule(
                self.path, {"pane_id": "lead-pane", "stow": "hold-1"}, AT,
                fail_launch,
                native_session={"kind": "id", "value": "lead-session"})
        blocked = self.stop()
        self.assertEqual(blocked["decision"], "block")
        self.assertIn("reset_failed", blocked["reason"])
        self.assertIn("do not retry this stow", blocked["reason"])
        self.assertNotIn("Schedule its exact live continuation", blocked["reason"])
        with self.assertRaises(foreman_reset.ResetEnded):
            self.schedule_reset()

    def test_new_event_or_assignment_invalidates_held_boundary(self):
        self.member()
        self.hold(["dispatch-a"])
        self.emit()
        self.assertIsNotNone(self.stop())
        self.ack()
        self.assertIsNotNone(self.stop(), "A prior hold does not cover a later event merely because it was acked")

    def test_pending_events_cannot_be_hidden_by_hold(self):
        self.member()
        self.emit()
        with self.assertRaisesRegex(UsageError, "pending"):
            self.hold(["dispatch-a"])

    def test_canonical_alias_uses_one_store_and_binding(self):
        alias = self.root / "alias"
        alias.symlink_to(self.root, target_is_directory=True)
        self.member()
        self.assertEqual(store.drain(alias / self.path.name), store.drain(self.path))
        who = store.identity("lead-session", str(alias), "fixture", pane_id="lead-pane")
        self.assertEqual(store.binding_path(who, self.bindings), store.binding_path(self.who, self.bindings))

    def test_corrupt_newer_state_is_preserved_and_blocks_bound_stop(self):
        for body in ('{"schema_version":900}', '{"broken"', '[1,2]', 'null'):
            store.store_path(self.path).write_text(body)
            with self.assertRaises(StateError):
                store.enroll(self.path, {"id": "d", "agent": "a", "task": "t", "report": str(self.evidence), "pane_id": None, "native_session": None}, AT)
            self.assertEqual(store.store_path(self.path).read_text(), body)
            self.assertEqual(self.stop()["decision"], "block")

    def test_missing_owner_state_is_not_proof_of_no_active_work(self):
        store.store_path(self.path).unlink()
        self.assertEqual(self.stop()["decision"], "block")

    def test_conflicting_native_binding_is_preserved(self):
        original = store.binding_path(self.who, self.bindings).read_bytes()
        with self.assertRaises(StateError):
            store.bind(self.root / "different-state.json", self.who, AT, root=self.bindings)
        self.assertEqual(store.binding_path(self.who, self.bindings).read_bytes(), original)

    def test_interrupted_new_lead_binding_blocks_new_session_until_retry(self):
        self.member()
        new_who = store.identity("new-lead", str(self.root), "fixture", pane_id="lead-pane")
        original_save = store.save_state
        def fail_owner(path, data):
            if Path(path) == store.store_path(self.path):
                raise StateError("Simulated interrupted owner commit; retry binding.", {})
            return original_save(path, data)
        with patch.object(store, "save_state", side_effect=fail_owner):
            with self.assertRaises(StateError):
                store.bind(self.path, new_who, AT, root=self.bindings)
        self.assertEqual(self.stop({**self.payload, "session_id": "new-lead"})["decision"], "block")
        store.bind(self.path, new_who, AT, root=self.bindings)
        self.assertIsNone(self.stop(), "Old native foreman no longer owns the committed binding")
        self.assertEqual(self.stop({**self.payload, "session_id": "new-lead"})["decision"], "block")

    def test_bind_current_uses_read_only_native_pane_evidence(self):
        self.client.panes["lead-pane"] = {"pane_id": "lead-pane", "agent_session": {**NATIVE, "value": "lead-session"}}
        result = runtime.bind_current(self.path, self.client, AT, environ=self.environ, cwd=str(self.root), root=self.bindings)
        self.assertEqual(result["identity"], self.who)

    def test_path_identity_matches_native_hook_transcript_path(self):
        transcript = str(self.root / "native-session.jsonl")
        self.client.panes["lead-pane"] = {"pane_id": "lead-pane", "agent_session": {**NATIVE, "kind": "path", "value": transcript}}
        runtime.bind_current(self.path, self.client, AT, environ=self.environ, cwd=str(self.root), root=self.bindings)
        self.member()
        self.assertIsNotNone(self.stop({**self.payload, "session_id": "unrelated-native-id", "transcript_path": transcript}))

    def test_drain_and_stop_do_not_require_original_evidence_files(self):
        self.member()
        self.emit()
        self.ack()
        self.hold(["dispatch-a"])
        self.evidence.unlink()
        self.assertEqual(store.drain(self.path)["events"], [])
        self.assertIsNone(self.stop())

    def test_live_process_probe_invokes_ps_with_start_and_command(self):
        with patch.object(runtime.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, "Tue Sep 1 12:00:00 2026 python watch", "")) as run:
            self.assertIsNotNone(runtime.process_identity(123))
            self.assertEqual(run.call_args.args[0], ["ps", "-p", "123", "-o", "lstart=", "-o", "command="])
        with patch.object(runtime.subprocess, "run", return_value=subprocess.CompletedProcess([], 1, "", "")):
            self.assertIsNone(runtime.process_identity(123))

    def test_registered_commands_route_readback_without_worker_or_config_access(self):
        parser = argparse.ArgumentParser()
        common = argparse.ArgumentParser(add_help=False)
        common.add_argument("--state")
        sub = parser.add_subparsers(dest="command", required=True)
        runtime.register_commands(sub, common)
        self.member()
        self.emit()
        args = parser.parse_args(["supervision-drain", "--state", str(self.path), "--now", AT])
        result = runtime.run_command(args, self.path, AT)
        self.assertEqual(result["through"], 1)
        args = parser.parse_args(["supervision-status", "--now", AT])
        result = runtime.run_command(args, self.path, AT, process_probe=self.probe)
        self.assertEqual(result["active"][0]["id"], "dispatch-a")
        missing = self.root / "missing.json"
        args = parser.parse_args(["supervision-bind", "--record", str(missing), "--now", AT])
        with self.assertRaisesRegex(UsageError, "missing"):
            runtime.run_command(args, self.path, AT)

    def test_mutating_command_inputs_produce_ack_and_resolution_receipts(self):
        parser = argparse.ArgumentParser()
        runtime.register_commands(parser.add_subparsers(dest="command", required=True), argparse.ArgumentParser(add_help=False))
        self.member()
        self.emit()
        request = self.root / "ack.json"
        request.write_text(json.dumps({"through": 1, "outcomes": [{"event": "event-1", "outcome": "Reconciled report candidate", "evidence": [str(self.evidence)]}]}))
        args = parser.parse_args(["supervision-ack", "--record", str(request), "--now", AT])
        self.assertEqual(runtime.run_command(args, self.path, AT)["acknowledged"], ["event-1"])
        request.write_text(json.dumps({"id": "dispatch-a", "outcome": "Accepted assignment in ledger; task release remains open", "evidence": [str(self.evidence)]}))
        args = parser.parse_args(["supervision-resolve", "--record", str(request), "--now", AT])
        self.assertFalse(runtime.run_command(args, self.path, AT)["active"])

    def test_hook_wrapper_subprocess_exact_scope_and_json(self):
        self.member()
        native_root = self.root / "xdg" / "foreman" / "supervision-bindings"
        store.bind(self.path, self.who, AT, root=native_root)
        script = Path(_ROOT).parent.parent / "hooks" / "herdr-supervision-stop.sh"
        environment = {**os.environ, **self.environ, "XDG_STATE_HOME": str(self.root / "xdg")}
        result = subprocess.run(["bash", str(script)], input=json.dumps(self.payload), env=environment, capture_output=True, text=True, check=False)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(json.loads(result.stdout)["decision"], "block")
        unrelated = subprocess.run(["bash", str(script)], input=json.dumps({**self.payload, "session_id": "worker"}), env=environment, capture_output=True, text=True, check=False)
        self.assertEqual(unrelated.returncode, 0, unrelated.stderr)
        self.assertEqual(unrelated.stdout, "")

    def test_wrapper_reports_python_startup_failure_without_native_exit_gate(self):
        binaries = self.root / "binaries"
        binaries.mkdir()
        python = binaries / "python3"
        python.write_text("#!/usr/bin/env bash\necho 'fixture: Python import failed' >&2\nexit 7\n")
        python.chmod(0o755)
        script = Path(_ROOT).parent.parent / "hooks/herdr-supervision-stop.sh"
        environment = {**os.environ, **self.environ, "PATH": str(binaries) + os.pathsep + os.environ["PATH"]}
        result = subprocess.run(["bash", str(script)], input=json.dumps(self.payload), env=environment,
                                capture_output=True, text=True, check=False)
        self.assertEqual(result.returncode, 0)
        self.assertEqual(result.stdout, "")
        self.assertIn("Python hook failed (exit 7)", result.stderr)
        self.assertIn("restore the hook installation", result.stderr)

    def test_published_manifest_hooks_run_from_paths_with_spaces_at_mode_0644(self):
        self.member()
        native_root = self.root / "xdg" / "foreman" / "supervision-bindings"
        store.bind(self.path, self.who, AT, root=native_root)
        repo = Path(_ROOT).parent.parent
        plugin = self.root / "installed plugin"
        shutil.copytree(Path(_ROOT) / "foreman", plugin / "skills/herdr-foreman/foreman",
                        ignore=shutil.ignore_patterns("__pycache__"))
        (plugin / "hooks").mkdir()
        script = plugin / "hooks/herdr-supervision-stop.sh"
        shutil.copyfile(repo / "hooks/herdr-supervision-stop.sh", script)
        script.chmod(0o644)
        manifest = json.loads((repo / ".tessl-plugin/plugin.json").read_text())
        environment = {**os.environ, **self.environ, "XDG_STATE_HOME": str(self.root / "xdg")}
        for agent in ("claude-code", "codex", "grok"):
            with self.subTest(agent=agent):
                configuration = "claude-code" if agent == "grok" else agent
                entries = [item for group in manifest["nativeHooks"][configuration]["Stop"]
                           for item in group["hooks"] if "herdr-supervision-stop.sh" in json.dumps(item)]
                self.assertEqual(len(entries), 1)
                entry = entries[0]
                expand = lambda value: value.replace("${TESSL_PLUGIN_DIR}", str(plugin))
                if agent == "grok":
                    # Grok imports Claude settings but ignores args. A bare
                    # command is relative to the settings directory, not PATH.
                    command = expand(entry["command"])
                    argv = ["sh", "-c", command] if " " in command else [str(plugin / ".claude" / command)]
                else:
                    argv = shlex.split(expand(entry["command"])) + [expand(arg) for arg in entry.get("args", [])]
                result = subprocess.run(argv, input=json.dumps(self.payload), env=environment,
                                        capture_output=True, text=True, check=False)
                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertEqual(json.loads(result.stdout)["decision"], "block")


    def test_wrapper_resolves_directory_names_ending_in_a_newline(self):
        """Command substitution strips trailing newlines from either name (#487)."""
        self.member()
        native_root = self.root / "xdg" / "foreman" / "supervision-bindings"
        store.bind(self.path, self.who, AT, root=native_root)
        repo = Path(_ROOT).parent.parent
        environment = {**os.environ, **self.environ, "XDG_STATE_HOME": str(self.root / "xdg")}
        for plugin_name, hooks_name in (("plain plugin", "hooks\n"), ("newline plugin\n", "hooks")):
            with self.subTest(plugin=plugin_name, hooks=hooks_name):
                plugin = self.root / plugin_name
                shutil.copytree(Path(_ROOT) / "foreman", plugin / "skills/herdr-foreman/foreman",
                                ignore=shutil.ignore_patterns("__pycache__"))
                (plugin / hooks_name).mkdir()
                script = plugin / hooks_name / "herdr-supervision-stop.sh"
                shutil.copyfile(repo / "hooks/herdr-supervision-stop.sh", script)
                result = subprocess.run(["bash", str(script)], input=json.dumps(self.payload), env=environment,
                                        capture_output=True, text=True, check=False)
                self.assertEqual(result.returncode, 0, result.stderr)
                self.assertEqual(json.loads(result.stdout)["decision"], "block")

    def test_wrapper_leaves_no_bytecode_cache_in_the_installed_tree(self):
        """The evaluator's read-only contract covers incidental caches too (#385)."""
        self.member()
        native_root = self.root / "xdg" / "foreman" / "supervision-bindings"
        store.bind(self.path, self.who, AT, root=native_root)
        repo = Path(_ROOT).parent.parent
        plugin = self.root / "fresh plugin"
        shutil.copytree(Path(_ROOT) / "foreman", plugin / "skills/herdr-foreman/foreman",
                        ignore=shutil.ignore_patterns("__pycache__"))
        (plugin / "hooks").mkdir()
        script = plugin / "hooks/herdr-supervision-stop.sh"
        shutil.copyfile(repo / "hooks/herdr-supervision-stop.sh", script)
        self.assertEqual(list(plugin.rglob("__pycache__")), [])
        environment = {**os.environ, **self.environ, "XDG_STATE_HOME": str(self.root / "xdg")}
        environment.pop("PYTHONDONTWRITEBYTECODE", None)
        result = subprocess.run(["bash", str(script)], input=json.dumps(self.payload), env=environment,
                                capture_output=True, text=True, check=False)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(json.loads(result.stdout)["decision"], "block")
        self.assertEqual([str(path.relative_to(plugin)) for path in plugin.rglob("__pycache__")], [])
        self.assertEqual([str(path.relative_to(plugin)) for path in plugin.rglob("*.pyc")], [])


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