Service Resilience and Credential Topology Implementation Plan

flowchart TB
    subgraph Lifecycle
      A1[Task 1 readiness 90s, exit 75, breaker] --> A2[Task 2 systemd limits]
      A1 --> A3[Task 3 release_ownership before NSApp quit]
      A3 --> A4[Task 4 one stale rule, dead-pid handoff, derived budgets]
    end
    subgraph Credentials
      B5[Task 5 topology registry] --> B6[Task 6 records, build_stores, guards]
      B6 --> B7[Task 7 security tool backend]
      B6 --> B8[Task 8 statusline conflict rendering]
    end
    A4 --> D[Task 9 docs + full gate]
    B7 --> D
    B8 --> D
    D --> E[Task 10 evidence on the real Mac]
  

Tasks 1-4 lifecycle, Tasks 5-8 credentials, Task 9 docs and full test gate, Task 10 evidence.

Task 1: Cold-start readiness budget, fail-fast, retryable exit with a give-up breaker

Traces to: AC 1, 2, 3, 3b

Files

• Modify: jacked/service/__init__.py — shared timing constants
• Create: jacked/service/start_failures.py — start-failure breaker (record / clear)
• Modify: jacked/service/tray.py — _wait_for_ready (~line 482), _abort_unready_start (~line 500), both ready log lines (~1307, ~1473)
• Test: tests/unit/service/test_start_failures.py — new
• Test: tests/unit/service/test_tray.py — new TestReadinessBudget class

Interfaces

• Produces in jacked/service/__init__.py: COLD_START_READY_TIMEOUT = 90.0, HANDOFF_EXIT_TIMEOUT = 30.0, REPLACEMENT_READY_TIMEOUT = COLD_START_READY_TIMEOUT + 15.0, EX_TEMPFAIL = 75, START_FAILURE_LIMIT = 5, START_FAILURE_WINDOW_SECONDS = 600.0.
• Produces in jacked/service/start_failures.py: record_start_failure(path: Path, now: float, *, window: float = START_FAILURE_WINDOW_SECONDS) -> int (appends, prunes, returns the count inside the window) and clear_start_failures(path: Path) -> None.
• Produces: ServiceRunner._wait_for_ready(self, timeout: float | None = None) -> bool; sets self._ready_elapsed: float | None. ServiceRunner._note_ready(elapsed: float) -> None; ServiceRunner._start_failure_path() -> Path.
• Consumes: SupervisorKind from jacked/service/spec.py; default_service_paths().root for the breaker file.

Why: at cold boot the app import plus lifespan startup took over 10 s and the tray gave up (07:05:18 in ~/.claude/jacked-tray.log), then exited 0, which launchd's KeepAlive.SuccessfulExit=false treats as final. A non-zero exit makes every supervisor retry, but launchd has no burst limit, so a permanently failing start (port stolen, corrupt DB) would relaunch every 10 s forever. The breaker turns that into five attempts per ten minutes, then a loud clean exit.

Step 1.1 Add the constants to jacked/service/__init__.py:

# A cold boot on a slow disk needs well over 10 s to import the app and run
# lifespan startup (observed 2026-09-04: >10.5 s). The wait returns as soon
# as the port answers, so a long budget costs nothing on a warm start.
COLD_START_READY_TIMEOUT = 90.0
# A restart handoff waits this long for the old owner to exit, then this long
# for the replacement to report ready. The replacement is a cold start.
HANDOFF_EXIT_TIMEOUT = 30.0
REPLACEMENT_READY_TIMEOUT = COLD_START_READY_TIMEOUT + 15.0
# sysexits.h EX_TEMPFAIL: launchd (SuccessfulExit=false), systemd
# (Restart=on-failure) and Task Scheduler (RestartOnFailure) all relaunch a
# non-zero exit. A clean exit would be treated as final.
EX_TEMPFAIL = 75
# After this many failed starts inside the window the service exits cleanly
# so a permanently broken environment does not relaunch forever.
START_FAILURE_LIMIT = 5
START_FAILURE_WINDOW_SECONDS = 600.0

Step 1.2 Write the failing breaker tests in tests/unit/service/test_start_failures.py:

from __future__ import annotations

import json

from jacked.service.start_failures import clear_start_failures, record_start_failure


def test_record_counts_failures_inside_window_and_prunes_old_ones(tmp_path):
    path = tmp_path / "start-failures.json"

    assert record_start_failure(path, 1000.0, window=600.0) == 1
    assert record_start_failure(path, 1100.0, window=600.0) == 2
    assert record_start_failure(path, 1700.0, window=600.0) == 2  # 1000.0 pruned, 1100.0 on the edge kept
    assert json.loads(path.read_text()) == [1100.0, 1700.0]


def test_record_drops_future_stamps(tmp_path):
    path = tmp_path / "start-failures.json"

    assert record_start_failure(path, 1000.0, window=600.0) == 1
    assert record_start_failure(path, 500.0, window=600.0) == 1  # 1000.0 is in the future


def test_record_tolerates_corrupt_file(tmp_path):
    path = tmp_path / "start-failures.json"
    path.write_text("not json")

    assert record_start_failure(path, 5.0, window=600.0) == 1


def test_clear_removes_file_and_is_idempotent(tmp_path):
    path = tmp_path / "start-failures.json"
    record_start_failure(path, 1.0)

    clear_start_failures(path)
    clear_start_failures(path)

    assert not path.exists()

Step 1.3 Run uv run python -m pytest tests/unit/service/test_start_failures.py -v. Expected: ImportError. Step 1.4 Create jacked/service/start_failures.py:

"""Bounded retry memory for supervised service starts."""

from __future__ import annotations

import json
from pathlib import Path

from jacked.service import START_FAILURE_WINDOW_SECONDS


def record_start_failure(
    path: Path, now: float, *, window: float = START_FAILURE_WINDOW_SECONDS
) -> int:
    """Append one failure timestamp and return how many fall inside the window."""
    stamps: list[float] = []
    try:
        loaded = json.loads(path.read_text(encoding="utf-8"))
        stamps = [float(item) for item in loaded if isinstance(item, (int, float))]
    except (OSError, ValueError, TypeError):
        stamps = []
    # A stamp in the future (clock stepped back after a reboot) must not
    # count forever, so both bounds are enforced.
    stamps = [stamp for stamp in stamps if 0 <= now - stamp <= window]
    stamps.append(now)
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    path.write_text(json.dumps(stamps), encoding="utf-8")
    return len(stamps)


def clear_start_failures(path: Path) -> None:
    try:
        path.unlink()
    except FileNotFoundError:
        return

Step 1.5 Run the breaker tests again. Expected: pass. Step 1.6 Write the failing tray tests. Append to tests/unit/service/test_tray.py (add import socket, import time, from types import SimpleNamespace at the top if missing; the file already imports MagicMock, patch, pytest):

class TestReadinessBudget:
    """Cold-start readiness: long budget, fail fast on a dead server, retryable exit."""

    def test_timing_constants(self):
        from jacked import service as service_pkg

        assert service_pkg.COLD_START_READY_TIMEOUT == 90.0
        assert service_pkg.REPLACEMENT_READY_TIMEOUT == 105.0
        assert service_pkg.EX_TEMPFAIL == 75

    def test_wait_for_ready_returns_true_once_port_answers_and_records_elapsed(self):
        from jacked.service.tray import ServiceRunner

        listener = socket.socket()
        listener.bind(("127.0.0.1", 0))
        listener.listen(1)
        port = listener.getsockname()[1]
        try:
            runner = ServiceRunner(port=port)
            runner._uvicorn_thread = SimpleNamespace(is_alive=lambda: True)
            assert runner._wait_for_ready(timeout=5) is True
            assert runner._ready_elapsed is not None and runner._ready_elapsed < 5
        finally:
            listener.close()

    def test_wait_for_ready_fails_fast_when_server_thread_died(self):
        from jacked.service.tray import ServiceRunner

        probe = socket.socket()
        probe.bind(("127.0.0.1", 0))
        port = probe.getsockname()[1]
        probe.close()  # nothing listens here now
        runner = ServiceRunner(port=port)
        runner._uvicorn_thread = SimpleNamespace(is_alive=lambda: False)
        started = time.monotonic()
        assert runner._wait_for_ready(timeout=30) is False
        assert time.monotonic() - started < 3

    @pytest.mark.parametrize("kind", ["LAUNCHD", "SYSTEMD_USER", "TASK_SCHEDULER"])
    def test_unready_start_under_native_supervisor_exits_tempfail(self, kind, tmp_path):
        from jacked.service.spec import SupervisorKind
        from jacked.service.tray import ServiceRunner

        runner = ServiceRunner()
        runner._service_spec = SimpleNamespace(supervisor=SupervisorKind[kind])
        runner._uvicorn_thread = None
        with (
            patch.object(runner, "_request_stop"),
            patch.object(runner, "_start_failure_path", return_value=tmp_path / "f.json"),
            pytest.raises(SystemExit) as exc_info,
        ):
            runner._abort_unready_start()
        assert exc_info.value.code == 75
        assert runner._service_state == "degraded"

    def test_unready_start_gives_up_cleanly_after_limit(self, tmp_path, caplog):
        from jacked.service import START_FAILURE_LIMIT
        from jacked.service.spec import SupervisorKind
        from jacked.service.start_failures import record_start_failure
        from jacked.service.tray import ServiceRunner

        path = tmp_path / "f.json"
        for _ in range(START_FAILURE_LIMIT - 1):
            record_start_failure(path, time.time())
        runner = ServiceRunner()
        runner._service_spec = SimpleNamespace(supervisor=SupervisorKind.LAUNCHD)
        runner._uvicorn_thread = None
        with (
            patch.object(runner, "_request_stop"),
            patch.object(runner, "_start_failure_path", return_value=path),
            pytest.raises(SystemExit) as exc_info,
        ):
            runner._abort_unready_start()
        assert exc_info.value.code == 0
        assert "giving up" in caplog.text

    def test_unready_start_under_manual_supervisor_keeps_message(self):
        from jacked.service.spec import SupervisorKind
        from jacked.service.tray import ServiceRunner

        runner = ServiceRunner()
        runner._service_spec = SimpleNamespace(supervisor=SupervisorKind.MANUAL)
        runner._uvicorn_thread = None
        with patch.object(runner, "_request_stop"), pytest.raises(SystemExit) as exc_info:
            runner._abort_unready_start()
        assert "did not become ready" in str(exc_info.value.code)

    def test_both_ready_sites_record_success_and_elapsed(self):
        """_note_ready is the only thing that clears the breaker; both platform
        ready sites (macOS menu bar and pystray) must call it and log timing."""
        import inspect

        from jacked.service import tray as tray_module

        source = inspect.getsource(tray_module)
        assert source.count("self._note_ready(self._ready_elapsed or 0.0)") == 2
        assert source.count("ready_in=%.1fs") == 2

    def test_successful_start_clears_failure_memory(self, tmp_path):
        from jacked.service.start_failures import record_start_failure
        from jacked.service.tray import ServiceRunner

        path = tmp_path / "f.json"
        record_start_failure(path, time.time())
        runner = ServiceRunner()
        with patch.object(runner, "_start_failure_path", return_value=path):
            runner._note_ready(0.5)
        assert not path.exists()

Step 1.7 Run uv run python -m pytest tests/unit/service/test_tray.py::TestReadinessBudget -v. Expected: failures (missing constants, exit code 0 instead of 75, no _note_ready). Step 1.8 In jacked/service/tray.py extend the import from jacked.service import DEFAULT_HOST, DEFAULT_PORT, PID_FILE with COLD_START_READY_TIMEOUT, EX_TEMPFAIL, START_FAILURE_LIMIT. Initialise self._ready_elapsed: float | None = None in __init__ next to self._server_ready = False. Replace _wait_for_ready:

    def _wait_for_ready(self, timeout: float | None = None) -> bool:
        """Poll until the server accepts connections.

        Returns False early when the uvicorn thread has already died: there is
        nothing left to wait for. ``timeout=None`` uses the cold-start budget.
        """
        import socket
        import time

        if timeout is None:
            timeout = COLD_START_READY_TIMEOUT
        probe_host = (
            self.bind_plan.probe_host if self.bind_plan is not None else "127.0.0.1"
        )
        started = time.monotonic()
        deadline = started + timeout
        while time.monotonic() < deadline:
            try:
                sock = socket.create_connection((probe_host, self.port), timeout=0.5)
                sock.close()
                self._ready_elapsed = time.monotonic() - started
                return True
            except OSError:
                thread = self._uvicorn_thread
                if thread is not None and not thread.is_alive():
                    logger.error(
                        "Server thread exited before becoming ready (%.1fs)",
                        time.monotonic() - started,
                    )
                    return False
                time.sleep(0.3)
        logger.error("Server did not become ready within %.0fs", timeout)
        return False

Step 1.9 Add two helpers next to _abort_unready_start and rewrite its exit branch:

    def _start_failure_path(self):
        from jacked.service.lifecycle import default_service_paths

        return default_service_paths().root / "start-failures.json"

    def _note_ready(self, elapsed: float) -> None:
        """Record a successful start: reset the retry breaker, remember timing."""
        from jacked.service.start_failures import clear_start_failures

        self._ready_elapsed = elapsed
        try:
            clear_start_failures(self._start_failure_path())
        except OSError:
            logger.exception("Could not clear start-failure memory")

    def _abort_unready_start(self) -> None:
        """Stop before any tray UI exists when startup never became ready."""
        import time

        logger.error("Service did not become ready; aborting before UI startup")
        self._service_state = "degraded"
        self._lifecycle_failure = "initial service start did not become ready"
        self._request_stop()
        if self._uvicorn_thread is not None:
            self._uvicorn_thread.join(timeout=5)
        from jacked.service.spec import SupervisorKind
        from jacked.service.start_failures import record_start_failure

        if (
            self._service_spec is None
            or self._service_spec.supervisor is SupervisorKind.MANUAL
        ):
            raise SystemExit(
                "Jacked did not become ready. Check ~/.claude/jacked-tray.log."
            )
        try:
            failures = record_start_failure(self._start_failure_path(), time.time())
        except OSError:
            logger.exception("Could not record start failure")
            failures = 1
        if failures >= START_FAILURE_LIMIT:
            # A clean exit stops every supervisor from relaunching. The log
            # line is the operator's signal; `jacked service restart` resets.
            logger.error(
                "Service failed to start %d times in a row; giving up until "
                "`jacked service restart` is run",
                failures,
            )
            raise SystemExit(0)
        # A non-zero exit is what makes launchd, systemd and Task Scheduler
        # retry; a clean exit would be treated as final.
        raise SystemExit(EX_TEMPFAIL)

Note the existing behaviour for a bare ServiceRunner() (no spec, used by the pre-existing abort test at ~line 355) is preserved: it takes the message branch.

Step 1.10 At both ready sites, and only those two (macOS _run_mac_menubar ~line 1307 and pystray _run ~line 1473; the in-process restart site at ~line 695 is deliberately excluded), insert the exact line self._note_ready(self._ready_elapsed or 0.0) right after self._started_at = time.time(), and append elapsed time to the log lines: macOS becomes "Service ready — macOS menu-bar agent (pid=%d, port=%d, ready_in=%.1fs)"; pystray becomes "Service ready (pid=%d, port=%d, autostart=%s, ready_in=%.1fs)", passing self._ready_elapsed or 0.0. Grep tests/ for "Service ready" and relax any exact-string assertion to a substring check. Step 1.11 Run uv run python -m pytest tests/unit/service/test_tray.py tests/unit/service/test_start_failures.py -q. Expected: pass. Step 1.12 Commit.

git add jacked/service/__init__.py jacked/service/start_failures.py jacked/service/tray.py tests/unit/service/test_start_failures.py tests/unit/service/test_tray.py
git commit -m "fix(service): 90s cold-start budget, retryable exit, bounded start breaker"

Task 2: Bounded systemd restart limits

Traces to: AC 4

Files

• Modify: jacked/service/supervisors/__init__.py — render_systemd_user unit lines (~lines 85-105)
• Test: tests/unit/service/test_supervisor_core.py — new tests

Interfaces

• Consumes: tests/unit/service/test_supervisor_core.py already imports the spec builder as from tests.unit.service.supervisor_test_support import make_spec as _spec (line 24); use _spec(...). Import render_launchd and render_systemd_user from jacked.service.supervisors if not already imported.

Why: systemd's default start-limit is 5 starts per 10 s. With RestartSec=3 a slow boot burns the whole burst in fifteen seconds and systemd stops retrying. launchd and Task Scheduler already retry non-zero exits; launchd is bounded by Task 1's breaker.

Step 2.1 Write the failing tests in tests/unit/service/test_supervisor_core.py:

def test_systemd_unit_retries_failed_starts_with_bounded_limits():
    spec = _spec(SupervisorKind.SYSTEMD_USER)
    rendered = render_systemd_user(spec, environment={"HOME": "/tmp/user"})
    text = rendered.content.decode("utf-8")
    unit_section, service_section = text.split("[Service]", 1)

    assert "StartLimitIntervalSec=300" in unit_section
    assert "StartLimitBurst=5" in unit_section
    assert "Restart=on-failure" in service_section
    assert "RestartSec=5" in service_section


def test_launchd_plist_keeps_keepalive_on_failure_only():
    import plistlib

    spec = _spec(SupervisorKind.LAUNCHD)
    rendered = render_launchd(spec, environment={"HOME": "/tmp/user"})
    payload = plistlib.loads(rendered.content)

    assert payload["KeepAlive"] == {"SuccessfulExit": False}
    assert payload["RunAtLoad"] is True

Step 2.2 Run: uv run python -m pytest tests/unit/service/test_supervisor_core.py -k "bounded_limits or keepalive" -v. Expected: the systemd test FAILS on StartLimitIntervalSec; the launchd test passes (it pins current behaviour). Step 2.3 Edit the lines list in render_systemd_user: after "After=network.target" insert "StartLimitIntervalSec=300" and "StartLimitBurst=5"; change "RestartSec=3" to "RestartSec=5". Step 2.4 Run the supervisor suites: uv run python -m pytest tests/unit/service/test_supervisor_core.py tests/unit/service/test_supervisor_launchd.py tests/unit/service/test_supervisor_task_scheduler.py -q. Expected: pass. Step 2.5 Commit.

git add jacked/service/supervisors/__init__.py tests/unit/service/test_supervisor_core.py
git commit -m "fix(service): bound systemd restart limits for slow boots"

Task 3: Deterministic ownership teardown on macOS quit

Traces to: AC 5

Files

• Modify: jacked/service/tray.py — new release_ownership; run() finally (~line 1379)
• Modify: jacked/service/menubar_mac.py — _shutdown (~line 817)
• Test: tests/unit/service/test_tray.py — new TestReleaseOwnership
• Test: tests/unit/service/test_menubar_platform.py — new test

Interfaces

• Produces: ServiceRunner.release_ownership(self) -> None, idempotent; closes the control server, then the ownership manifest, then clears both attributes (in that order, so the control server's manifest_provider lambda stays valid while its handler thread joins).
• Consumes: ServiceOwnership.close() (removes the manifest if this instance still owns it) and the control server's close().

Why: rumps.quit_application() calls NSApplication.terminate_, which exits the process without unwinding Python. The finally in ServiceRunner.run() that closes the ownership manifest never executes on macOS, so the manifest keeps naming a dead pid and a restart handoff waits on it in vain (07:22 today).

Step 3.1 Write the failing tests. In tests/unit/service/test_tray.py (add call to the unittest.mock import if missing):

class TestReleaseOwnership:
    def test_release_closes_control_then_ownership_once(self):
        from jacked.service.tray import ServiceRunner

        runner = ServiceRunner()
        parent = MagicMock()
        runner._control_server = parent.control
        runner._ownership = parent.ownership

        runner.release_ownership()
        runner.release_ownership()  # idempotent

        assert parent.mock_calls == [call.control.close(), call.ownership.close()]
        assert runner._control_server is None
        assert runner._ownership is None

    def test_release_still_retires_manifest_when_control_close_raises(self):
        from jacked.service.tray import ServiceRunner

        runner = ServiceRunner()
        runner._control_server = MagicMock(close=MagicMock(side_effect=OSError("boom")))
        ownership = MagicMock()
        runner._ownership = ownership

        runner.release_ownership()

        ownership.close.assert_called_once()
        assert runner._ownership is None

In tests/unit/service/test_menubar_platform.py (add from types import SimpleNamespace and from unittest.mock import MagicMock, call if missing):

def test_mac_shutdown_releases_ownership_before_quit():
    pytest.importorskip("rumps")
    from jacked.service import menubar_mac

    runner = MagicMock()
    app = SimpleNamespace(_runner=runner, _screen_observer=None)

    menubar_mac.MacMenuBarApp._shutdown(app)

    assert runner.mock_calls == [call._shutdown_uvicorn(), call.release_ownership()]

Step 3.2 Run: uv run python -m pytest tests/unit/service/test_tray.py::TestReleaseOwnership tests/unit/service/test_menubar_platform.py -k release -v. Expected: FAIL with AttributeError: release_ownership. Step 3.3 Add the method to ServiceRunner (right before _handle_control_action):

    def release_ownership(self) -> None:
        """Close the control channel and retire this instance's manifest.

        Idempotent. ``run()`` calls it in its ``finally`` on every platform.
        macOS also calls it explicitly before ``NSApp.terminate_`` because
        that call ends the process without unwinding Python, so the
        ``finally`` never runs there and the manifest would outlive the pid.
        The attributes are cleared only after both closes so the control
        server's manifest provider stays valid while its handler joins.
        """
        control, ownership = self._control_server, self._ownership
        if control is not None:
            try:
                control.close()
            except OSError:
                logger.exception("Control server close failed during shutdown")
        if ownership is not None:
            try:
                ownership.close()
            except OSError:
                logger.exception("Ownership release failed during shutdown")
        self._control_server = None
        self._ownership = None

Step 3.4 In run()'s outer finally replace the two inline close blocks (control server, ownership) with a single self.release_ownership() after self._uninstall_windows_console_handler(). Step 3.5 In menubar_mac.py _shutdown, after the uvicorn shutdown try/except and before the screen-observer removal, add:

            # NSApp.terminate_ exits without unwinding Python, so the runner's
            # finally never runs on this path. Retire the manifest now so a
            # restart handoff sees a real exit instead of a dead pid.
            self._runner.release_ownership()

Step 3.6 Run uv run python -m pytest tests/unit/service/test_tray.py tests/unit/service/test_menubar_platform.py -q. Expected: pass. Step 3.7 Commit.

git add jacked/service/tray.py jacked/service/menubar_mac.py tests/unit/service/test_tray.py tests/unit/service/test_menubar_platform.py
git commit -m "fix(service): retire the instance manifest before macOS quits"

Task 4: One stale-process rule; handoff tolerates a dead owner; budgets derived from the cold-start constant

Traces to: AC 6, 7, 7b

Files

• Modify: jacked/service/instance_storage.py — new process_is_stale, manifest_is_proven_stale
• Modify: jacked/service/instance.py — re-exports
• Modify: jacked/service/instance_ownership.py — _clear_proven_stale_manifest (~199) uses the shared rule
• Modify: jacked/cli.py — _manifest_is_proven_stale (~413) delegates; restart failure message (~557); _wait_owned_service_ready call (~5644)
• Modify: jacked/service/handoff.py — _wait_for_handoff_exit, timeouts
• Test: tests/unit/service/test_instance.py — new tests
• Test: tests/unit/service/test_lifecycle.py — new tests

Interfaces

• Produces: process_is_stale(process: ProcessIdentity | None) -> bool in jacked.service.instance_storage: True when the pid is gone or its identity differs; False when it matches, when process is None, or when the probe itself timed out (subprocess.SubprocessError: fail closed, not proven stale).
• Produces: manifest_is_proven_stale(path: Path) -> bool = process_is_stale(read_manifest(path).process); raises OSError/ValueError when the manifest is unreadable. Both re-exported from jacked.service.instance.
• Produces: handoff_owned_service(spec, *, environment, paths=None, timeout: float = HANDOFF_EXIT_TIMEOUT, ready_timeout: float = REPLACEMENT_READY_TIMEOUT).
• Consumes: HANDOFF_EXIT_TIMEOUT, REPLACEMENT_READY_TIMEOUT from jacked.service (Task 1).

Why: _wait_for_handoff_exit only notices a removed manifest. Task 3 fixes the common case, but a crash, a kill, or an older service still leaves a dead pid behind; a proven-dead owner is an exit. The proven-stale rule currently exists twice (cli.py, instance_ownership.py); it becomes one primitive. On macOS the identity probe shells out to ps with a 2 s timeout, and subprocess.TimeoutExpired is not an OSError, so the probe must treat a timeout as "not proven" instead of crashing the restart. Finally the handoff's readiness wait must cover a cold start (Task 1 allows 90 s); today the same 10 s serves both phases, and the CLI's own wait is 15 s.

Step 4.1 Write the failing tests. In tests/unit/service/test_instance.py:

def test_process_is_stale_rules(monkeypatch):
    import subprocess
    from types import SimpleNamespace

    from jacked.service import instance_storage
    from jacked.service.instance import process_is_stale

    process = SimpleNamespace(pid=4242, creation_id="c1", executable="/x")

    assert process_is_stale(None) is False

    monkeypatch.setattr(instance_storage, "process_identity", lambda pid: process)
    assert process_is_stale(process) is False

    other = SimpleNamespace(pid=4242, creation_id="c2", executable="/x")
    monkeypatch.setattr(instance_storage, "process_identity", lambda pid: other)
    assert process_is_stale(process) is True

    def dead(pid):
        raise ProcessLookupError(pid)

    monkeypatch.setattr(instance_storage, "process_identity", dead)
    assert process_is_stale(process) is True

    def slow(pid):
        raise subprocess.TimeoutExpired(["ps"], 2)

    monkeypatch.setattr(instance_storage, "process_identity", slow)
    assert process_is_stale(process) is False  # not proven; fail closed


def test_manifest_is_proven_stale_reads_then_applies_rule(tmp_path, monkeypatch):
    from types import SimpleNamespace

    from jacked.service import instance_storage
    from jacked.service.instance import manifest_is_proven_stale

    process = SimpleNamespace(pid=1, creation_id="c", executable="/x")
    monkeypatch.setattr(
        instance_storage, "read_manifest", lambda _path: SimpleNamespace(process=process)
    )
    monkeypatch.setattr(instance_storage, "process_is_stale", lambda p: p is process)

    assert manifest_is_proven_stale(tmp_path / "m") is True

In tests/unit/service/test_lifecycle.py:

def test_handoff_treats_dead_owner_pid_as_exit(monkeypatch, tmp_path):
    process = SimpleNamespace(pid=999_999, creation_id="gone", executable="/x")
    old = SimpleNamespace(
        instance_id="old",
        generation="old-generation",
        supervisor=SupervisorKind.LAUNCHD.value,
        process=process,
    )
    spec = MagicMock(generation="a" * 64, supervisor=SupervisorKind.LAUNCHD)
    paths = SimpleNamespace(manifest=tmp_path / "manifest", root=tmp_path)
    control = MagicMock(
        side_effect=[
            {"ok": True, "result": {"accepted": True}},
            {
                "ok": True,
                "result": {
                    "state": "running",
                    "generation": spec.generation,
                    "instance_id": "new",
                },
            },
        ]
    )
    install = MagicMock(return_value=SupervisorAction(True, "install", "exact"))
    monkeypatch.setattr("jacked.service.instance.read_manifest", lambda _path: old)
    monkeypatch.setattr("jacked.service.instance.process_is_stale", lambda p: True)
    monkeypatch.setattr("jacked.service.ipc.send_native_control", control)
    monkeypatch.setattr("jacked.service.lifecycle.install_owned_supervisor", install)
    monkeypatch.setattr("jacked.service.lifecycle.spawn_exact_service", MagicMock())

    result = handoff_owned_service(spec, environment={}, paths=paths, timeout=2)

    assert result.ok is True
    install.assert_called_once()


def test_handoff_keeps_waiting_while_owner_pid_is_alive(monkeypatch, tmp_path):
    process = SimpleNamespace(pid=4242, creation_id="live", executable="/x")
    old = SimpleNamespace(instance_id="old", generation="old-generation", process=process)
    spec = MagicMock(generation="a" * 64, supervisor=SupervisorKind.MANUAL)
    paths = SimpleNamespace(manifest=tmp_path / "manifest", root=tmp_path)
    monkeypatch.setattr("jacked.service.instance.read_manifest", lambda _path: old)
    monkeypatch.setattr("jacked.service.instance.process_is_stale", lambda p: False)
    monkeypatch.setattr(
        "jacked.service.ipc.send_native_control",
        lambda *_args, **_kwargs: {"ok": True, "result": {"accepted": True}},
    )
    spawn = MagicMock()
    monkeypatch.setattr("jacked.service.lifecycle.spawn_exact_service", spawn)

    result = handoff_owned_service(spec, environment={}, paths=paths, timeout=0.3)

    assert result.ok is False
    assert "did not exit" in result.reason
    spawn.assert_not_called()


def test_handoff_budgets_come_from_service_constants():
    import inspect

    from jacked.service import HANDOFF_EXIT_TIMEOUT, REPLACEMENT_READY_TIMEOUT

    params = inspect.signature(handoff_owned_service).parameters
    assert params["timeout"].default == HANDOFF_EXIT_TIMEOUT
    assert params["ready_timeout"].default == REPLACEMENT_READY_TIMEOUT
    assert REPLACEMENT_READY_TIMEOUT > 90

Step 4.2 Run: uv run python -m pytest tests/unit/service/test_instance.py -k stale tests/unit/service/test_lifecycle.py -v. Expected: ImportErrors / failures. Step 4.3 Add to jacked/service/instance_storage.py (after process_identity; add import subprocess if missing):

def process_is_stale(process: ProcessIdentity | None) -> bool:
    """True only when a recorded process is proven dead or replaced.

    A probe that times out (macOS shells out to ``ps``) proves nothing and
    returns False so callers never treat a busy machine as a dead owner.
    """
    if process is None:
        return False
    try:
        observed = process_identity(process.pid)
    except subprocess.SubprocessError:
        return False
    except (OSError, ProcessLookupError, ValueError):
        return True
    return observed != process


def manifest_is_proven_stale(path: Path) -> bool:
    """True only when a valid manifest names a dead or identity-mismatched PID.

    Unreadable or invalid manifests raise (OSError/ValueError) so callers can
    route them to explicit recovery instead of guessing.
    """
    return process_is_stale(read_manifest(path).process)

Add both names to the instance_storage import block and __all__ in jacked/service/instance.py.

Step 4.4 In jacked/service/instance_ownership.py _clear_proven_stale_manifest replace the try: observed = process_identity(...) block and the if observed != stale.process block with one branch:

    if process_is_stale(stale.process):
        _remove_owned_stale_control(paths.control)
        if not remove_manifest_if_current(paths.manifest, stale.instance_id):
            raise ServiceOwnershipInvalid("stale manifest changed during recovery")
        return
    raise ServiceOwnershipInvalid(
        "an existing instance manifest still names a live process"
    )

Import process_is_stale from jacked.service.instance_storage there (and drop process_identity if it becomes unused).

Step 4.5 Replace the body of _manifest_is_proven_stale in jacked/cli.py (keep the name; tests patch jacked.cli._manifest_is_proven_stale):

def _manifest_is_proven_stale(path: Path) -> bool:
    """True only when a valid manifest names a dead or identity-mismatched PID."""
    from jacked.service.instance import manifest_is_proven_stale

    return manifest_is_proven_stale(path)

Step 4.6 Rewrite _wait_for_handoff_exit and the public entry point in jacked/service/handoff.py:

from jacked.service import HANDOFF_EXIT_TIMEOUT, REPLACEMENT_READY_TIMEOUT

_LIVENESS_PROBE_INTERVAL = 1.0


def _wait_for_handoff_exit(
    paths: ServicePaths,
    old: InstanceManifest,
    spec: ServiceSpec,
    timeout: float,
    ready_timeout: float,
) -> SupervisorAction | None:
    from jacked.service.instance import process_is_stale, read_manifest

    deadline = time.monotonic() + timeout
    next_liveness_probe = time.monotonic()
    while time.monotonic() < deadline:
        try:
            current = read_manifest(paths.manifest)
        except FileNotFoundError:
            return None
        except (OSError, ValueError):
            return SupervisorAction(
                False, "handoff", "ownership became indeterminate"
            )
        if current.instance_id != old.instance_id:
            if current.generation == spec.generation:
                return _await_ready_generation(
                    paths,
                    spec.generation,
                    previous_instance=old.instance_id,
                    timeout=ready_timeout,
                )
            return SupervisorAction(False, "handoff", "supervisor started stale build")
        # macOS terminates through NSApp without unwinding Python and a
        # crashed service never removes its manifest: a proven-dead owner is
        # an exit. Probe once a second; the macOS probe shells out to ps.
        if time.monotonic() >= next_liveness_probe:
            if process_is_stale(getattr(current, "process", None)):
                return None
            next_liveness_probe = time.monotonic() + _LIVENESS_PROBE_INTERVAL
        time.sleep(0.05)
    return SupervisorAction(False, "handoff", "old ownership did not exit")


def handoff_owned_service(
    spec: ServiceSpec,
    *,
    environment: dict[str, str],
    paths: ServicePaths | None = None,
    timeout: float = HANDOFF_EXIT_TIMEOUT,
    ready_timeout: float = REPLACEMENT_READY_TIMEOUT,
) -> SupervisorAction:
    """Authenticate shutdown, await lease release, then start the new build.

    ``timeout`` bounds the old owner's exit (a graceful shutdown alone can
    take ~9 s). ``ready_timeout`` bounds the replacement's cold start and
    must cover the tray's own cold-start budget.
    """
    from jacked.service.instance import read_manifest
    from jacked.service.ipc import ControlAction, send_native_control
    from jacked.service.lifecycle import default_service_paths

    selected_paths = paths or default_service_paths()
    try:
        old = read_manifest(selected_paths.manifest)
        response = send_native_control(
            selected_paths.manifest, ControlAction.RESTART_HANDOFF
        )
    except (OSError, ValueError) as exc:
        return SupervisorAction(False, "handoff", type(exc).__name__)
    if not response.get("ok"):
        return SupervisorAction(False, "handoff", "service rejected shutdown")
    waiting = _wait_for_handoff_exit(selected_paths, old, spec, timeout, ready_timeout)
    if waiting is not None:
        return waiting
    previous = _HandoffPrevious(old.instance_id, old.supervisor, ready_timeout)
    return _activate_handoff(spec, environment, selected_paths, previous)

_HandoffPrevious.timeout now carries the ready budget; _activate_handoff is unchanged.

Step 4.7 In jacked/cli.py: (a) change the definition default of _wait_owned_service_ready (~line 371) from timeout: float = 15.0 to timeout: float = REPLACEMENT_READY_TIMEOUT. A def default is evaluated at import time, so this import must be module-level: add from jacked.service import REPLACEMENT_READY_TIMEOUT next to the existing eager jacked.dcr_settings imports (~line 21) and extend that block's comment: "jacked.service is stdlib-only at import time too". Then delete the explicit timeout=15.0 override at every call site (~lines 652, 3589, 5642, 5786) so all four cold-start waits share the budget; (b) change the message at ~line 663 from "didn't answer within 15s" to f"didn't answer within {int(REPLACEMENT_READY_TIMEOUT)}s"; (c) replace the restart failure message at ~line 557 so it stops claiming nothing was signalled:

                console.print(
                    "[red]Owned service restart did not complete.[/red] "
                    f"{handoff.reason}."
                )
                console.print(
                    "[dim]The shutdown request may already have been accepted. "
                    "Check `jacked service status` in a minute; if the service "
                    "is still down, run `jacked service restart`, then "
                    "`jacked service recover` to inspect ownership.[/dim]"
                )
                sys.exit(1)

Grep tests/unit/service/test_cli.py for "No process was signalled" and "within 15s" and update any assertion to the new wording. Add this test to tests/unit/service/test_cli.py:

def test_cli_ready_waits_share_the_replacement_budget():
    import inspect

    from jacked import cli
    from jacked.service import REPLACEMENT_READY_TIMEOUT

    assert inspect.signature(cli._wait_owned_service_ready).parameters["timeout"].default == REPLACEMENT_READY_TIMEOUT
    assert "timeout=15.0" not in inspect.getsource(cli)
    assert "within 15s" not in inspect.getsource(cli)

The in-process tray restart at jacked/service/tray.py ~line 691 keeps its 15 s wait: it retries three times and never exits the process.

Step 4.8 Run: uv run python -m pytest tests/unit/service/test_instance.py tests/unit/service/test_lifecycle.py tests/unit/service/test_cli.py tests/unit/service/test_ensure_native_lifecycle.py -q. Expected: pass. Step 4.9 Commit.

git add jacked/service/instance_storage.py jacked/service/instance.py jacked/service/instance_ownership.py jacked/cli.py jacked/service/handoff.py tests/unit/service/test_instance.py tests/unit/service/test_lifecycle.py tests/unit/service/test_cli.py
git commit -m "fix(service): one stale-process rule; handoff survives a dead owner and a cold replacement"

Task 5: Topology-keyed capability registry

Traces to: AC 8 (registry half), 9

Files

• Modify: jacked/credentials/capabilities.py — CapabilityRecord, parse_build, CapabilityRegistry
• Modify: jacked/credentials/models.py — CapabilityResolution.evidence
• Test: tests/unit/test_credential_capabilities.py — rewrite registry tests

Interfaces

• Produces: CapabilityRecord(platform_system: str, config_mode: str, min_build: str, inspected_through: str, capability: CredentialCapability) frozen dataclass.
• Produces: parse_build(version: str) -> tuple[int, ...]; raises ValueError on unparseable input.
• Produces: CapabilityRegistry(records: tuple[CapabilityRecord, ...] = ()); resolve(identity) -> CapabilityResolution; CapabilityResolution gains evidence: tuple[str, ...] = (). Evidence strings: build:<version>, inspected-through:<version>, and the constant NEWER_THAN_INSPECTED = "build-newer-than-inspected" when applicable.
• Consumes: ExecutableIdentity, CredentialCapability, CapabilityMode, StoreDeclaration, StoreRole from jacked/credentials/models.py.

Why: the registry certifies exact executable bytes, so every Claude Code release (2.1.259 to 2.1.260 last night) turns the whole credential path off. What jacked depends on is the store topology per platform and config mode, which is stable across builds. The build and hash still travel as evidence, and Task 6 uses the newer-than-inspected marker to fail closed on the one operation that could paper over a moved store: creating a missing authority.

Step 5.1 Replace the registry tests in tests/unit/test_credential_capabilities.py (keep test_executable_resolution_follows_symlink_and_hashes_target unchanged):

from __future__ import annotations

import hashlib
from pathlib import Path

import pytest

from jacked.credentials.capabilities import (
    CapabilityRecord,
    CapabilityRegistry,
    parse_build,
    resolve_executable,
)
from jacked.credentials.models import (
    CapabilityMode,
    CredentialCapability,
    ExecutableIdentity,
    StoreDeclaration,
    StoreRole,
)


def _template() -> CredentialCapability:
    return CredentialCapability(
        executable=ExecutableIdentity("<template>", "<template>", "1.0.0", "global", "linux", ""),
        mode=CapabilityMode.GLOBAL_UNCOOPERATIVE,
        authority=StoreDeclaration("file", "global-credential-file", StoreRole.AUTHORITY),
        consumers=("claude",),
        capability_epoch=2,
        writer_protocol_epoch=2,
        provenance="shipped:test",
        registry_version=2,
    )


def _record(**changes) -> CapabilityRecord:
    values = {
        "platform_system": "linux",
        "config_mode": "global",
        "min_build": "1.0.0",
        "inspected_through": "1.2.0",
        "capability": _template(),
    }
    values.update(changes)
    return CapabilityRecord(**values)


def _identity(**changes) -> ExecutableIdentity:
    values = {
        "resolved_path": "/opt/claude",
        "sha256": hashlib.sha256(b"any-build").hexdigest(),
        "build_version": "1.1.0",
        "config_mode": "global",
        "platform_system": "linux",
        "platform_machine": "x86_64",
    }
    values.update(changes)
    return ExecutableIdentity(**values)


def test_parse_build_reads_leading_dotted_integers() -> None:
    assert parse_build("2.1.260") == (2, 1, 260)
    assert parse_build("2.1.260-beta (Claude Code)") == (2, 1, 260)
    with pytest.raises(ValueError):
        parse_build("nightly")


def test_empty_registry_disables_mutation(tmp_path: Path) -> None:
    executable = tmp_path / "claude"
    executable.write_bytes(b"unknown-build")
    identity = resolve_executable(executable, build_version="1.0.0", config_mode="global")

    resolution = CapabilityRegistry().resolve(identity)

    assert resolution.capability.mode is CapabilityMode.UNSUPPORTED
    assert resolution.can_mutate is False
    assert "topology" in resolution.reason


def test_topology_match_ignores_hash_and_machine_but_keeps_them_as_provenance() -> None:
    registry = CapabilityRegistry((_record(),))

    for identity in (
        _identity(),
        _identity(sha256="0" * 64),
        _identity(platform_machine="aarch64"),
        _identity(resolved_path="/elsewhere/claude"),
    ):
        resolution = registry.resolve(identity)
        assert resolution.can_mutate is True
        assert resolution.capability.executable == identity
        assert resolution.capability.mode is CapabilityMode.GLOBAL_UNCOOPERATIVE
        assert "build:1.1.0" in resolution.evidence
        assert "inspected-through:1.2.0" in resolution.evidence
        assert "build-newer-than-inspected" not in resolution.evidence


def test_topology_requires_platform_and_config_mode() -> None:
    registry = CapabilityRegistry((_record(),))

    assert registry.resolve(_identity(platform_system="darwin")).can_mutate is False
    assert registry.resolve(_identity(config_mode="scoped")).can_mutate is False


def test_build_floor_and_newer_than_inspected_marker() -> None:
    registry = CapabilityRegistry((_record(),))

    below = registry.resolve(_identity(build_version="0.9.9"))
    assert below.can_mutate is False
    assert "predates" in below.reason

    newer = registry.resolve(_identity(build_version="1.3.0"))
    assert newer.can_mutate is True
    assert "build-newer-than-inspected" in newer.evidence

    assert registry.resolve(_identity(build_version="nightly")).can_mutate is False


def test_executable_resolution_follows_symlink_and_hashes_target(
    tmp_path: Path,
) -> None:
    target = tmp_path / "real-claude"
    target.write_bytes(b"binary")
    wrapper = tmp_path / "claude"
    wrapper.symlink_to(target)

    identity = resolve_executable(wrapper, build_version="1.2.3", config_mode="scoped")

    assert identity.resolved_path == str(target.resolve())
    assert identity.sha256 == hashlib.sha256(b"binary").hexdigest()


def test_kill_switch_needs_fresh_resolution_to_reenable() -> None:
    registry = CapabilityRegistry((_record(),))

    registry.disable_mutation("consumer contract drift")
    assert registry.resolve(_identity()).can_mutate is False
    registry.begin_fresh_resolution()
    assert registry.resolve(_identity()).can_mutate is True

Step 5.2 Run: uv run python -m pytest tests/unit/test_credential_capabilities.py -v. Expected: ImportError for CapabilityRecord. Step 5.3 In jacked/credentials/models.py add the evidence field:

@dataclass(frozen=True)
class CapabilityResolution:
    capability: CredentialCapability
    can_mutate: bool
    reason: str
    evidence: tuple[str, ...] = ()

Step 5.4 Rewrite the registry section of jacked/credentials/capabilities.py (keep resolve_executable and _unsupported; add import re and from dataclasses import dataclass):

_BUILD_RE = re.compile(r"^(\d+(?:\.\d+)*)")


def parse_build(version: str) -> tuple[int, ...]:
    """Parse the leading dotted integers of a build version ("2.1.260" -> (2, 1, 260))."""
    match = _BUILD_RE.match(version or "")
    if match is None:
        raise ValueError(f"unparseable build version: {version!r}")
    return tuple(int(part) for part in match.group(1).split("."))


@dataclass(frozen=True)
class CapabilityRecord:
    """One certified credential-store topology for a platform and config mode.

    Certification is keyed by where Claude Code keeps its credentials, which
    is stable across builds, rather than by executable bytes, which change on
    every release. ``min_build`` is the oldest build the topology was verified
    against and ``inspected_through`` the newest; a newer build still resolves
    and is flagged in the evidence so mutation can be more conservative.
    """

    platform_system: str
    config_mode: str
    min_build: str
    inspected_through: str
    capability: CredentialCapability


NEWER_THAN_INSPECTED = "build-newer-than-inspected"


class CapabilityRegistry:
    """Conservative registry keyed by platform and config mode."""

    def __init__(self, records: tuple[CapabilityRecord, ...] = ()) -> None:
        self._records = {
            (record.platform_system, record.config_mode): record for record in records
        }
        self._disabled_reason: str | None = None
        self._fresh_generation = 0
        self._disabled_generation: int | None = None

    def disable_mutation(self, reason: str) -> None:
        """Latch mutation off until a caller starts a fresh resolution."""
        self._disabled_reason = reason
        self._disabled_generation = self._fresh_generation

    def begin_fresh_resolution(self) -> None:
        """Start a new explicit capability-resolution generation."""
        self._fresh_generation += 1
        self._disabled_reason = None
        self._disabled_generation = None

    def resolve(self, identity: ExecutableIdentity) -> CapabilityResolution:
        if self._disabled_reason is not None:
            return _unsupported(
                identity, f"mutation kill switch: {self._disabled_reason}"
            )
        record = self._records.get((identity.platform_system, identity.config_mode))
        if record is None:
            return _unsupported(
                identity,
                "no certified credential-store topology for "
                f"{identity.platform_system or 'unknown'}/{identity.config_mode}",
            )
        try:
            build = parse_build(identity.build_version)
        except ValueError:
            return _unsupported(identity, "Claude build version is not parseable")
        if build < parse_build(record.min_build):
            return _unsupported(
                identity,
                f"Claude build {identity.build_version} predates the certified "
                f"floor {record.min_build}",
            )
        evidence = [
            f"build:{identity.build_version}",
            f"inspected-through:{record.inspected_through}",
        ]
        if build > parse_build(record.inspected_through):
            evidence.append(NEWER_THAN_INSPECTED)
        capability = CredentialCapability(
            **{**record.capability.__dict__, "executable": identity}
        )
        return CapabilityResolution(
            capability, True, "certified topology matched", tuple(evidence)
        )


DEFAULT_REGISTRY = CapabilityRegistry()

Step 5.5 Run uv run python -m pytest tests/unit/test_credential_capabilities.py -q. Expected: pass. tests/unit/test_credential_runtime.py will fail until Task 6; that is expected at this commit. Step 5.6 Commit.

git add jacked/credentials/capabilities.py jacked/credentials/models.py tests/unit/test_credential_capabilities.py
git commit -m "feat(credentials): certify store topology per platform instead of exact build bytes"

Task 6: Shipped topology records, store wiring, identity cache, fail-closed rules for uninspected builds

Traces to: AC 8, 9, 10, 10b, 10c, 11, 11b

Files

• Modify: jacked/credentials/runtime.py — records, build_stores, detect_claude_identity, _engine_for, activate_account, resolve_active_identity, scoped_launch_needs_global_activation
• Modify: jacked/credentials/transaction.py — TransactionDependencies.allow_missing_authority; schema-drift warning in _activate_unfenced
• Modify: jacked/credentials/resolver.py — identity:stamp-absent evidence
• Modify: jacked/credentials/file_store.py — compare-and-swap between read() and write()
• Modify: jacked/api/session_observer.py — declaration locators use the canonical constants
• Modify: jacked/launch.py — skip global activation when the authority is the scoped-capable file
• Modify: tests/conftest.py — autouse identity-cache reset
• Test: tests/unit/test_credential_runtime.py — rewrite registry/platform tests; new tests
• Test: tests/unit/test_credential_transactions.py — missing-authority rule; schema-drift warning
• Test: tests/unit/test_credential_resolver.py — stamp-absent evidence
• Test: tests/unit/test_credential_stores.py — file CAS
• Test: tests/unit/test_launch.py — scoped launch on a file-authority platform

Interfaces

• Produces: MIN_CLAUDE_BUILD = "2.1.0", INSPECTED_CLAUDE_BUILD = "2.1.260", KEYCHAIN_LOCATOR = "macos-keychain", GLOBAL_FILE_LOCATOR = "global-credential-file", SHIPPED_REGISTRY.
• Produces: build_stores(capability, home) -> dict[str, CredentialStore] keyed by declaration locator; raises ValueError for an unknown locator.
• Produces: clear_identity_cache() -> None; scoped_launch_needs_global_activation() -> bool (True unless the certified authority is the global credential file).
• Produces: TransactionDependencies.allow_missing_authority: bool = True.
• Produces: FileCredentialStore returns StoreStatus.CONCURRENT_WRITE from write() when the file changed since this adapter's last read().
• Consumes: CapabilityRecord, CapabilityRegistry, NEWER_THAN_INSPECTED from Task 5; MacOSCredentialStore, FileCredentialStore; CredentialStore protocol from jacked/credentials/store.py.

Why: the shipped registry has one macOS arm64 record and two hard sys.platform != "darwin" gates, so Linux and Windows have never resolved since 0.99.0 and jacked launch refuses on those platforms. Claude Code's documented storage is the Keychain item on macOS and ~/.claude/.credentials.json on Linux and Windows. The stores dict is also keyed by the adapter's own locator while the resolver looks up the declaration locator, so today's darwin read path is UNUSABLE regardless of the hash; build_stores keys by declaration. Four guards come with the widened certification: an uninspected newer build may not create a missing authority (a moved store looks exactly like a missing one); a schema change in Claude's claudeAiOauth object is logged by name; a lost _jackedAccountId stamp is named in evidence rather than hidden inside UNUSABLE; and the file store refuses to overwrite a file that changed after it was read.

Step 6.1 Replace the registry and platform tests in tests/unit/test_credential_runtime.py. Keep _account(), test_detection_uses_known_install_locations_when_path_is_sanitized, test_activation_maps_unknown_build_to_unsupported_without_mutation (change its identity to _identity(build_version="2.0.0")), and test_active_identity_callers_do_not_reintroduce_file_precedence. Replace the rest with:

import pytest

from jacked.credentials.file_store import FileCredentialStore
from jacked.credentials.models import CredentialCapability, StoreDeclaration, StoreRole
from jacked.credentials.runtime import (
    GLOBAL_FILE_LOCATOR,
    INSPECTED_CLAUDE_BUILD,
    KEYCHAIN_LOCATOR,
    SHIPPED_REGISTRY,
    activate_account,
    build_stores,
    clear_identity_cache,
    detect_claude_identity,
    resolve_active_identity,
    scoped_launch_needs_global_activation,
)


def _identity(**changes) -> ExecutableIdentity:
    values = {
        "resolved_path": "/different/install/location/claude",
        "sha256": "0" * 64,
        "build_version": INSPECTED_CLAUDE_BUILD,
        "config_mode": "global",
        "platform_system": "darwin",
        "platform_machine": "arm64",
    }
    values.update(changes)
    return ExecutableIdentity(**values)


def test_shipped_registry_resolves_any_build_hash_on_each_platform() -> None:
    darwin = SHIPPED_REGISTRY.resolve(_identity(sha256="f" * 64))
    assert darwin.can_mutate is True
    assert darwin.capability.authority.locator == KEYCHAIN_LOCATOR
    assert [m.locator for m in darwin.capability.required_mirrors] == [GLOBAL_FILE_LOCATOR]

    for system in ("linux", "windows"):
        resolution = SHIPPED_REGISTRY.resolve(
            _identity(platform_system=system, platform_machine="x86_64")
        )
        assert resolution.can_mutate is True
        assert resolution.capability.authority.locator == GLOBAL_FILE_LOCATOR
        assert resolution.capability.required_mirrors == ()


def test_shipped_registry_flags_uninspected_newer_builds_and_rejects_old_ones() -> None:
    newer = SHIPPED_REGISTRY.resolve(_identity(build_version="2.1.999"))
    assert newer.can_mutate is True
    assert "build-newer-than-inspected" in newer.evidence

    assert SHIPPED_REGISTRY.resolve(_identity(build_version="2.0.9")).can_mutate is False
    assert SHIPPED_REGISTRY.resolve(_identity(config_mode="scoped")).can_mutate is False


def test_build_stores_keys_adapters_by_declaration_locator(tmp_path: Path) -> None:
    linux = SHIPPED_REGISTRY.resolve(_identity(platform_system="linux")).capability
    stores = build_stores(linux, tmp_path)
    assert set(stores) == {GLOBAL_FILE_LOCATOR}
    assert isinstance(stores[GLOBAL_FILE_LOCATOR], FileCredentialStore)
    assert stores[GLOBAL_FILE_LOCATOR].path == tmp_path / ".claude" / ".credentials.json"

    darwin = SHIPPED_REGISTRY.resolve(_identity()).capability
    with mock.patch("jacked.credentials.runtime.MacOSCredentialStore") as keychain:
        stores = build_stores(darwin, tmp_path)
    assert set(stores) == {KEYCHAIN_LOCATOR, GLOBAL_FILE_LOCATOR}
    assert stores[KEYCHAIN_LOCATOR] is keychain.return_value


def test_build_stores_rejects_unknown_locator(tmp_path: Path) -> None:
    capability = SHIPPED_REGISTRY.resolve(_identity(platform_system="linux")).capability
    unknown = CredentialCapability(
        **{
            **capability.__dict__,
            "authority": StoreDeclaration("x", "nowhere", StoreRole.AUTHORITY),
        }
    )
    with pytest.raises(ValueError):
        build_stores(unknown, tmp_path)


def test_linux_activation_writes_file_authority_end_to_end(tmp_path: Path) -> None:
    from jacked.credentials.repository import InMemoryCredentialSwitchRepository

    home = tmp_path
    (home / ".claude").mkdir()
    with (
        mock.patch(
            "jacked.credentials.runtime.detect_claude_identity",
            return_value=_identity(platform_system="linux", platform_machine="x86_64"),
        ),
        mock.patch("jacked.credentials.runtime.Path.home", return_value=home),
        mock.patch(
            "jacked.credentials.runtime.DatabaseCredentialSwitchRepository",
            lambda _db: InMemoryCredentialSwitchRepository(),
        ),
    ):
        result = activate_account(object(), _account(), SwitchContext.MANUAL, "op-linux")

    assert result.outcome is SwitchOutcome.OBSERVED_TARGET_UNFENCED
    written = home / ".claude" / ".credentials.json"
    assert written.exists()
    assert (written.stat().st_mode & 0o777) == 0o600
    assert '"_jackedAccountId":7' in written.read_text(encoding="utf-8").replace(" ", "")


def test_resolve_active_identity_on_linux_reads_credential_file(tmp_path: Path) -> None:
    home = tmp_path
    (home / ".claude").mkdir()
    (home / ".claude" / ".credentials.json").write_text(
        '{"_jackedAccountId": 4, "claudeAiOauth": {"accessToken": "a"}}', encoding="utf-8"
    )
    with (
        mock.patch(
            "jacked.credentials.runtime.detect_claude_identity",
            return_value=_identity(platform_system="linux", platform_machine="x86_64"),
        ),
        mock.patch("jacked.credentials.runtime.Path.home", return_value=home),
    ):
        observation = resolve_active_identity()

    assert observation.state.value == "resolved"
    assert observation.identity.account_id == 4
    assert f"build:{INSPECTED_CLAUDE_BUILD}" in observation.evidence


def test_unstamped_credential_file_is_unusable_with_named_evidence(tmp_path: Path) -> None:
    """A first-run Linux install has a Claude-written file with no jacked stamp."""
    home = tmp_path
    (home / ".claude").mkdir()
    (home / ".claude" / ".credentials.json").write_text(
        '{"claudeAiOauth": {"accessToken": "a", "refreshToken": "r"}}', encoding="utf-8"
    )
    with (
        mock.patch(
            "jacked.credentials.runtime.detect_claude_identity",
            return_value=_identity(platform_system="linux", platform_machine="x86_64"),
        ),
        mock.patch("jacked.credentials.runtime.Path.home", return_value=home),
    ):
        observation = resolve_active_identity()

    assert observation.state.value == "unusable"
    assert "identity:stamp-absent" in observation.evidence


def test_scoped_launch_needs_global_activation_only_for_keychain_authority() -> None:
    with mock.patch(
        "jacked.credentials.runtime.detect_claude_identity",
        return_value=_identity(platform_system="linux", platform_machine="x86_64"),
    ):
        assert scoped_launch_needs_global_activation() is False
    with mock.patch(
        "jacked.credentials.runtime.detect_claude_identity", return_value=_identity()
    ):
        assert scoped_launch_needs_global_activation() is True
    with mock.patch(
        "jacked.credentials.runtime.detect_claude_identity",
        side_effect=OSError("no claude"),
    ):
        assert scoped_launch_needs_global_activation() is True  # fail closed


def test_detection_caches_identity_until_the_binary_changes(tmp_path: Path) -> None:
    executable = tmp_path / "claude"
    executable.write_bytes(b"build-one")
    executable.chmod(0o755)
    completed = SimpleNamespace(returncode=0, stdout="2.1.260 (Claude Code)\n")
    run = mock.MagicMock(return_value=completed)

    with (
        mock.patch("jacked.credentials.runtime.find_bin", return_value=str(executable)),
        mock.patch("jacked.credentials.runtime.subprocess.run", run),
    ):
        first = detect_claude_identity(tmp_path)
        second = detect_claude_identity(tmp_path)
        executable.write_bytes(b"build-two")  # same size, new mtime
        os.utime(executable, ns=(1, 1))
        third = detect_claude_identity(tmp_path)

    assert first == second
    assert third.sha256 != first.sha256
    assert run.call_count == 2


def test_detection_turns_version_probe_timeout_into_oserror(tmp_path: Path) -> None:
    import subprocess

    executable = tmp_path / "claude"
    executable.write_bytes(b"x")
    executable.chmod(0o755)
    with (
        mock.patch("jacked.credentials.runtime.find_bin", return_value=str(executable)),
        mock.patch(
            "jacked.credentials.runtime.subprocess.run",
            side_effect=subprocess.TimeoutExpired(["claude"], 5),
        ),
        pytest.raises(OSError),
    ):
        detect_claude_identity(tmp_path)

Add to tests/conftest.py an autouse fixture so no test is served another test's cached identity:

@pytest.fixture(autouse=True)
def _reset_claude_identity_cache():
    from jacked.credentials.runtime import clear_identity_cache

    clear_identity_cache()
    yield
    clear_identity_cache()

Step 6.2 Run: uv run python -m pytest tests/unit/test_credential_runtime.py -v. Expected: ImportErrors and failures. Step 6.3 Rewrite the top of jacked/credentials/runtime.py (records replace _SHIPPED_CAPABILITIES; add import threading, import CapabilityRecord and NEWER_THAN_INSPECTED from .capabilities, CredentialStore from .store):

# Claude Code keeps its credentials in a per-platform store topology that is
# stable across builds: the macOS Keychain item mirrored to the global file,
# or the global file alone on Linux and Windows. Certification is keyed by
# that topology; the observed build and hash travel as evidence.
MIN_CLAUDE_BUILD = "2.1.0"
INSPECTED_CLAUDE_BUILD = "2.1.260"
KEYCHAIN_LOCATOR = "macos-keychain"
GLOBAL_FILE_LOCATOR = "global-credential-file"

_KEYCHAIN_AUTHORITY = StoreDeclaration("macOS Keychain", KEYCHAIN_LOCATOR, StoreRole.AUTHORITY)
_GLOBAL_FILE_AUTHORITY = StoreDeclaration(
    "global credential file", GLOBAL_FILE_LOCATOR, StoreRole.AUTHORITY
)
_GLOBAL_FILE_MIRROR = StoreDeclaration(
    "global credential file", GLOBAL_FILE_LOCATOR, StoreRole.REQUIRED_MIRROR
)


def _shipped_record(
    platform_system: str,
    authority: StoreDeclaration,
    mirrors: tuple[StoreDeclaration, ...] = (),
) -> CapabilityRecord:
    return CapabilityRecord(
        platform_system=platform_system,
        config_mode="global",
        min_build=MIN_CLAUDE_BUILD,
        inspected_through=INSPECTED_CLAUDE_BUILD,
        capability=CredentialCapability(
            executable=ExecutableIdentity(
                "<resolved-at-runtime>",
                "<observed-at-runtime>",
                INSPECTED_CLAUDE_BUILD,
                "global",
                platform_system,
                "",
            ),
            mode=CapabilityMode.GLOBAL_UNCOOPERATIVE,
            authority=authority,
            required_mirrors=mirrors,
            consumers=("claude-code",),
            capability_epoch=1,
            writer_protocol_epoch=2,
            provenance=f"shipped:claude-{platform_system}-global-topology",
            registry_version=2,
        ),
    )


_SHIPPED_RECORDS = (
    _shipped_record("darwin", _KEYCHAIN_AUTHORITY, (_GLOBAL_FILE_MIRROR,)),
    _shipped_record("linux", _GLOBAL_FILE_AUTHORITY),
    _shipped_record("windows", _GLOBAL_FILE_AUTHORITY),
)

SHIPPED_REGISTRY = CapabilityRegistry(_SHIPPED_RECORDS)
_PROCESS_SWITCH_LEASE = ProcessSwitchLease()

_identity_cache_lock = threading.Lock()
_identity_cache: dict[tuple[str, str], tuple[tuple[int, int], ExecutableIdentity]] = {}


def clear_identity_cache() -> None:
    with _identity_cache_lock:
        _identity_cache.clear()


def build_stores(capability: CredentialCapability, home: Path) -> dict[str, CredentialStore]:
    """Instantiate one adapter per declared store, keyed by the declaration locator.

    The resolver and the transaction engine look stores up by the locator in
    the capability declaration, never by the adapter's own locator string.
    """
    stores: dict[str, CredentialStore] = {}
    for declaration in (
        capability.authority,
        *capability.required_mirrors,
        *capability.optional_metadata,
    ):
        if declaration.locator == KEYCHAIN_LOCATOR:
            stores[declaration.locator] = MacOSCredentialStore()
        elif declaration.locator == GLOBAL_FILE_LOCATOR:
            stores[declaration.locator] = FileCredentialStore(
                home / ".claude" / ".credentials.json", trusted_root=home
            )
        else:
            raise ValueError(f"no adapter for credential store {declaration.locator!r}")
    return stores

Step 6.4 Replace detect_claude_identity:

def detect_claude_identity(home: Path) -> ExecutableIdentity:
    """Identify the exact Claude executable, hashing it once per build.

    Every poll loop asks for the identity; hashing a ~200 MB binary and
    spawning ``claude --version`` each time is wasteful, so the result is
    cached until the resolved file's size or mtime changes.
    """
    executable_name = find_bin("claude")
    if not executable_name:
        raise OSError("Claude executable was not found")
    executable = Path(executable_name).resolve(strict=True)
    config_mode = _config_mode(home)
    status = executable.stat()
    stamp = (status.st_size, status.st_mtime_ns)
    key = (str(executable), config_mode)
    with _identity_cache_lock:
        cached = _identity_cache.get(key)
        if cached is not None and cached[0] == stamp:
            return cached[1]
    try:
        result = subprocess.run(
            [str(executable), "--version"],
            capture_output=True,
            text=True,
            timeout=5,
            check=False,
        )
    except subprocess.SubprocessError as exc:
        raise OSError("Claude executable version probe failed") from exc
    match = _VERSION_PATTERN.match(result.stdout.strip())
    if result.returncode != 0 or not match:
        raise OSError("Claude executable version could not be identified")
    identity = resolve_executable(
        executable, build_version=match.group(1), config_mode=config_mode
    )
    with _identity_cache_lock:
        _identity_cache[key] = (stamp, identity)
    return identity

Step 6.5 Restructure activate_account so it stays under 50 lines: delete the sys.platform block and the hand-built stores, and extract the engine construction:

def _engine_for(
    db, resolution: CapabilityResolution, home: Path
) -> CredentialTransactionEngine | str:
    """Build the transaction engine for a resolved capability, or a refusal reason."""
    capability = resolution.capability
    if capability.mode is CapabilityMode.GLOBAL_UNCOOPERATIVE:
        key_provider = StaticInstallKeyProvider(None)
        install_id = "unfenced-local"
    else:
        key_provider = FileInstallKeyProvider(home / ".claude" / "credential-recovery.key")
        key = key_provider.get_key()
        if key is None:
            return "private recovery key unavailable"
        install_id = machine_install_id(key)
    try:
        stores = build_stores(capability, home)
    except (ValueError, RuntimeError) as exc:
        return str(exc)
    authority = stores.pop(capability.authority.locator)
    return CredentialTransactionEngine(
        TransactionDependencies(
            capability=capability,
            repository=DatabaseCredentialSwitchRepository(db),
            authority=authority,
            mirrors=stores,
            writer_fence=WriterFence(StaticWriterInspector((), is_complete=False)),
            install_key=key_provider,
            machine_install_id=install_id,
            snapshot_sink=FileResolverSnapshotSink(
                home / ".claude" / "jacked-resolver-snapshot.json"
            ),
            switch_lease=_PROCESS_SWITCH_LEASE,
            # An uninspected newer build may have moved its store; a missing
            # authority then looks identical to "never logged in". Refuse to
            # create it so jacked never writes where Claude no longer reads.
            allow_missing_authority=NEWER_THAN_INSPECTED not in resolution.evidence,
        )
    )


def activate_account(db, account: dict, context: SwitchContext, operation_id: str) -> SwitchResult:
    """Resolve the exact runtime contract and activate one local account."""
    home = Path.home()
    try:
        identity = detect_claude_identity(home)
    except OSError as exc:
        return _unsupported(operation_id, account, str(exc))
    resolution = SHIPPED_REGISTRY.resolve(identity)
    if not resolution.can_mutate:
        return _unsupported(operation_id, account, resolution.reason)
    engine = _engine_for(db, resolution, home)
    if isinstance(engine, str):
        return _unsupported(operation_id, account, engine)
    payload = CredentialPayload.from_mapping(
        {"_jackedAccountId": int(account["id"]), "claudeAiOauth": build_oauth_data(account)}
    )
    request = SwitchRequest(
        operation_id=operation_id,
        account_id=int(account["id"]),
        email=account.get("email") or "",
        organization_id=account.get("organization_uuid") or None,
        payload=payload,
        context=context,
        interaction=InteractionMode.FOREGROUND,
    )
    return engine.activate(request)


def scoped_launch_needs_global_activation() -> bool:
    """True unless the certified authority is the global credential file.

    On macOS the Keychain outranks ``CLAUDE_CONFIG_DIR``, so a scoped launch
    must also switch the global authority. Where the file is the authority,
    Claude reads the scoped file and touching the global one would change
    every other session's account. Unknown state fails closed (True).
    """
    try:
        resolution = SHIPPED_REGISTRY.resolve(detect_claude_identity(Path.home()))
    except OSError:
        return True
    if not resolution.can_mutate:
        return True
    return resolution.capability.authority.locator != GLOBAL_FILE_LOCATOR

Step 6.6 Replace the tail of resolve_active_identity (everything after the registry check) and drop the sys import if unused:

    try:
        stores = build_stores(resolution.capability, Path.home())
    except (ValueError, RuntimeError) as exc:
        return ResolverObservation(
            ResolverState.UNSUPPORTED, CredentialIdentity(), (str(exc),)
        )
    observation = CanonicalCredentialResolver(resolution.capability, stores).resolve()
    return ResolverObservation(
        observation.state,
        observation.identity,
        (*observation.evidence, *resolution.evidence),
    )

Step 6.7 Transaction engine. In jacked/credentials/transaction.py add allow_missing_authority: bool = True to TransactionDependencies (after switch_lease), and in _activate_unfenced change the call to allow_missing_authority=self._deps.allow_missing_authority. After before is known and before the authority write, add _warn_on_schema_drift(before, request) with this module-level helper:

def _warn_on_schema_drift(before: StoreReadResult, request: SwitchRequest) -> None:
    """Name any claudeAiOauth keys Claude wrote that jacked's payload does not carry.

    jacked replaces the whole object; a key it does not know is dropped. That
    is logged loudly so a Claude Code schema change is diagnosable from the
    tray log instead of surfacing as a mysterious re-login.
    """
    if before.payload is None:
        return
    current = before.payload.to_mapping().get("claudeAiOauth")
    target = request.payload.to_mapping().get("claudeAiOauth")
    if not isinstance(current, dict) or not isinstance(target, dict):
        return
    dropped = sorted(set(current) - set(target))
    if dropped:
        logger.warning(
            "Credential schema drift: authority carries claudeAiOauth keys jacked "
            "does not write and will drop: %s",
            ", ".join(dropped),
        )

CredentialPayload.to_mapping() already exists in canonical.py; do not add a property. Add logger = logging.getLogger(__name__) to transaction.py if absent.

Contract: StoreStatus.CONCURRENT_WRITE from write() now means exactly "refused before any mutation because the store changed since this adapter read it". Document that on the CredentialStore protocol in jacked/credentials/store.py. The macOS store's post-write readback mismatch (macos_store.py _readback, ~line 262) is a different situation and must stop using that status: change it to StoreStatus.ERROR with the existing reason text, so the engine's re-read path classifies it. Then give _classify_after_failure an observation-based branch right after the INTERACTIVE_REQUIRED branch:

        if status is StoreStatus.CONCURRENT_WRITE:
            # The adapter refused before writing because the authority changed
            # since it was read (Claude Code refreshed or re-logged in). Nothing
            # of ours landed, so report what is there now as preserved.
            observed = self._deps.authority.read()
            if observed.payload is not None:
                return self._record(
                    request, SwitchOutcome.FAILED_PRESERVED, observed.payload.identity, reason
                )
            return self._record(request, SwitchOutcome.INDETERMINATE, message=reason)

The same race exists on the darwin mirror: _prepare_preserving_target reads the required mirror (arming its stamp), the Keychain is written, then _publish_mirrors writes the mirror. If Claude refreshed the file in between, the mirror now refuses and today's code would return INDETERMINATE and leave a pending row. A required mirror is by definition a copy of the authority, so the right move is to re-read and write once more. In _publish_mirrors, replace write = mirror.write(request.payload, request.interaction) with:

            write = mirror.write(request.payload, request.interaction)
            if write.status is StoreStatus.CONCURRENT_WRITE:
                # The mirror changed since it was read; re-arm and try once more.
                mirror.read()
                write = mirror.write(request.payload, request.interaction)

Tests in tests/unit/test_credential_transactions.py. The module already has _engine, _payload(account_id, token), _capability(mode), and _request(payload) (one argument; it stamps account_id=2, so requests must carry _payload(2, ...)). Import InMemoryCredentialSwitchRepository from jacked.credentials.repository, MemoryCredentialStore from jacked.credentials.store, MemoryResolverSnapshotSink from jacked.credentials.resolver, and StoreWriteResult, StoreDeclaration, StoreRole from jacked.credentials.models. Also update tests/unit/test_credential_stores.py: any test expecting CONCURRENT_WRITE from the macOS readback mismatch now expects ERROR. Change the allow_missing_authority=False refusal reason in _prepare_preserving_target to "credential authority is missing; jacked will not create it for a Claude build newer than the inspected one; run `claude` and log in once" (the existing "authority is missing" assertion still matches):

def test_uninspected_build_may_not_create_a_missing_authority():
    repository = InMemoryCredentialSwitchRepository()
    store = MemoryCredentialStore("auth", None)
    deps = TransactionDependencies(
        capability=_capability(CapabilityMode.GLOBAL_UNCOOPERATIVE),
        repository=repository,
        authority=store,
        mirrors={},
        writer_fence=WriterFence(StaticWriterInspector(())),
        install_key=StaticInstallKeyProvider(None),
        machine_install_id="unfenced-local",
        snapshot_sink=MemoryResolverSnapshotSink(),
        allow_missing_authority=False,
    )
    engine = CredentialTransactionEngine(deps)

    result = engine.activate(_request(_payload(2, "token")))

    assert result.outcome is SwitchOutcome.UNUSABLE
    assert "authority is missing" in result.message
    assert store.read().status is StoreStatus.MISSING


def test_inspected_build_keeps_authority_creation_enabled(tmp_path):
    from jacked.credentials.runtime import SHIPPED_REGISTRY, _engine_for
    from tests.unit.test_credential_runtime import _identity

    inspected = _engine_for(object(), SHIPPED_REGISTRY.resolve(_identity(platform_system="linux")), tmp_path)
    newer = _engine_for(
        object(),
        SHIPPED_REGISTRY.resolve(_identity(platform_system="linux", build_version="2.1.261")),
        tmp_path,
    )
    assert inspected._deps.allow_missing_authority is True
    assert newer._deps.allow_missing_authority is False


def test_concurrent_write_on_authority_reports_the_refreshed_contents_as_preserved():
    refreshed = _payload(7, "refreshed-by-claude")

    class RefreshedUnderneath(MemoryCredentialStore):
        def write(self, payload, interaction):
            self._payload = refreshed  # what the other writer left behind
            return StoreWriteResult(StoreStatus.CONCURRENT_WRITE, "changed since read")

    before = _payload(1, "old")
    repository = InMemoryCredentialSwitchRepository()
    store = RefreshedUnderneath("auth", before)
    deps = TransactionDependencies(
        capability=_capability(CapabilityMode.GLOBAL_UNCOOPERATIVE),
        repository=repository,
        authority=store,
        mirrors={},
        writer_fence=WriterFence(StaticWriterInspector(())),
        install_key=StaticInstallKeyProvider(None),
        machine_install_id="unfenced-local",
        snapshot_sink=MemoryResolverSnapshotSink(),
    )

    result = CredentialTransactionEngine(deps).activate(_request(_payload(2, "new")))

    assert result.outcome is SwitchOutcome.FAILED_PRESERVED
    assert result.observed_identity.account_id == 7
    assert "changed since read" in result.message


def test_required_mirror_concurrent_write_is_retried_once_after_reread():
    class RefreshedMirror(MemoryCredentialStore):
        def __init__(self, *args, **kwargs):
            super().__init__(*args, **kwargs)
            self.refusals_left = 1
            self.reads = 0

        def read(self):
            self.reads += 1
            return super().read()

        def write(self, payload, interaction):
            if self.refusals_left:
                self.refusals_left -= 1
                return StoreWriteResult(StoreStatus.CONCURRENT_WRITE, "changed since read")
            return super().write(payload, interaction)

    before = _payload(1, "old")
    authority = MemoryCredentialStore("auth", before)
    mirror = RefreshedMirror("mirror", before)
    capability = CredentialCapability(
        **{
            **_capability(CapabilityMode.GLOBAL_UNCOOPERATIVE).__dict__,
            "required_mirrors": (StoreDeclaration("mirror", "mirror", StoreRole.REQUIRED_MIRROR),),
        }
    )
    deps = TransactionDependencies(
        capability=capability,
        repository=InMemoryCredentialSwitchRepository(),
        authority=authority,
        mirrors={"mirror": mirror},
        writer_fence=WriterFence(StaticWriterInspector(())),
        install_key=StaticInstallKeyProvider(None),
        machine_install_id="unfenced-local",
        snapshot_sink=MemoryResolverSnapshotSink(),
    )

    result = CredentialTransactionEngine(deps).activate(_request(_payload(2, "new")))

    assert result.outcome is SwitchOutcome.OBSERVED_TARGET_UNFENCED
    assert mirror.read().payload.digest == _payload(2, "new").digest
    assert mirror.reads >= 2  # re-armed before the retry


def test_schema_drift_is_logged_by_key_name(caplog):
    before = CredentialPayload.from_mapping(
        {
            "_jackedAccountId": 1,
            "claudeAiOauth": {"accessToken": "a", "refreshToken": "r", "newField": 1},
        }
    )
    engine, _repository, _store = _engine(CapabilityMode.GLOBAL_UNCOOPERATIVE, before)

    with caplog.at_level("WARNING"):
        engine.activate(_request(_payload(2, "token")))

    assert "newField" in caplog.text
    assert "refresh-" not in caplog.text  # key names only, never values

Step 6.8 Resolver evidence. In jacked/credentials/resolver.py CanonicalCredentialResolver.resolve, change the identity.account_id is None branch to return evidence (*evidence, "identity:stamp-absent"). Test in tests/unit/test_credential_resolver.py (adapt the capability helper to whatever that module already uses for a single-authority capability with locator "auth"):

def test_unstamped_payload_is_unusable_with_named_evidence():
    payload = CredentialPayload.from_mapping({"claudeAiOauth": {"accessToken": "a"}})
    store = MemoryCredentialStore("auth", payload)
    resolver = CanonicalCredentialResolver(_capability(), {"auth": store})

    observation = resolver.resolve()

    assert observation.state is ResolverState.UNUSABLE
    assert "identity:stamp-absent" in observation.evidence

Step 6.9 File store compare-and-swap. In jacked/credentials/file_store.py add a module constant _SEEN_MISSING = ("missing",) and give __init__ self._seen: tuple | None = None. read() takes the stat() before read_bytes() (stat afterwards would pair new metadata with old bytes and hide a change), and on an OK read sets self._seen = (st_size, st_mtime_ns, st_ino, hashlib.sha256(raw).digest()); the digest closes the coarse-mtime blind spot (tokens are fixed length, so an in-place refresh inside one mtime tick keeps size and inode). It sets _SEEN_MISSING when the file is MISSING and None on any other result, including exceptions. Add a helper and call it twice in write(): once as a cheap pre-flight right after the first _validate_existing, and once immediately before _durable_replace (next to the second _validate_existing), because staging and fsync take real time on a slow disk:

    def _changed_since_read(self) -> bool:
        """True when this adapter read the file and it changed afterwards.

        A never-read adapter never refuses. A file that appeared after a
        MISSING read counts as a change: Claude Code may have just logged in.
        """
        if self._seen is None:
            return False
        exists = self.path.exists()
        if self._seen == _SEEN_MISSING:
            return exists
        if not exists:
            return True
        status = self.path.stat()
        if (status.st_size, status.st_mtime_ns, status.st_ino) != self._seen[:3]:
            return True
        return hashlib.sha256(self.path.read_bytes()).digest() != self._seen[3]

At both call sites: if self._changed_since_read(): return StoreWriteResult(StoreStatus.CONCURRENT_WRITE, "credential file changed since it was read"). After a successful replace, refresh self._seen from the new file (stat plus the digest of the bytes just written); do that after the try/except OSError that returns UNUSABLE, wrapped in its own try/except OSError that only logs, so a stat failure after a committed replace cannot report the write as failed. Tests in tests/unit/test_credential_stores.py (the third uses the fsync hook, which runs during staging and therefore before the late check; the fourth pins the coarse-mtime case):

def test_file_store_refuses_to_overwrite_a_file_changed_since_read(tmp_path: Path) -> None:
    path = tmp_path / ".credentials.json"
    path.write_bytes(_payload(1).to_bytes())
    store = FileCredentialStore(path, trusted_root=tmp_path)
    assert store.read().status is StoreStatus.OK

    path.write_bytes(_payload(2).to_bytes())  # Claude Code refreshed a token
    os.utime(path, ns=(10**12, 10**12))

    result = store.write(_payload(3), InteractionMode.FOREGROUND)

    assert result.status is StoreStatus.CONCURRENT_WRITE
    assert CredentialPayload.from_json(path.read_bytes()).identity.account_id == 2


def test_file_store_refuses_when_a_file_appears_after_a_missing_read(tmp_path: Path) -> None:
    path = tmp_path / ".credentials.json"
    store = FileCredentialStore(path, trusted_root=tmp_path)
    assert store.read().status is StoreStatus.MISSING

    path.write_bytes(_payload(9).to_bytes())  # Claude Code logged in meanwhile

    result = store.write(_payload(3), InteractionMode.FOREGROUND)

    assert result.status is StoreStatus.CONCURRENT_WRITE
    assert CredentialPayload.from_json(path.read_bytes()).identity.account_id == 9


def test_file_store_checks_again_right_before_replace(tmp_path: Path, monkeypatch) -> None:
    from jacked.credentials import file_store as file_store_module

    path = tmp_path / ".credentials.json"
    path.write_bytes(_payload(1).to_bytes())
    store = FileCredentialStore(path, trusted_root=tmp_path)
    assert store.read().status is StoreStatus.OK

    def rewrite_during_staging(fd):
        path.write_bytes(_payload(2).to_bytes())
        os.utime(path, ns=(10**12, 10**12))

    monkeypatch.setattr(file_store_module.os, "fsync", rewrite_during_staging)

    result = store.write(_payload(3), InteractionMode.FOREGROUND)

    assert result.status is StoreStatus.CONCURRENT_WRITE
    assert CredentialPayload.from_json(path.read_bytes()).identity.account_id == 2


def test_file_store_detects_same_size_rewrite_inside_one_mtime_tick(tmp_path: Path) -> None:
    path = tmp_path / ".credentials.json"
    path.write_bytes(_payload(1).to_bytes())
    store = FileCredentialStore(path, trusted_root=tmp_path)
    assert store.read().status is StoreStatus.OK
    original = path.stat()

    path.write_bytes(_payload(2).to_bytes())  # same length as _payload(1)
    os.utime(path, ns=(original.st_atime_ns, original.st_mtime_ns))

    result = store.write(_payload(3), InteractionMode.FOREGROUND)

    assert result.status is StoreStatus.CONCURRENT_WRITE

Step 6.10 Session observer. In jacked/api/session_observer.py _build_resolver, use the canonical constants for the declaration locators and key the stores dict by the declaration locator: StoreDeclaration("macOS Keychain", KEYCHAIN_LOCATOR, StoreRole.AUTHORITY) with stores = {KEYCHAIN_LOCATOR: authority}; and for the file case StoreDeclaration("configured credential file", GLOBAL_FILE_LOCATOR, StoreRole.AUTHORITY) with stores = {GLOBAL_FILE_LOCATOR: authority} (the file path still comes from config_root so scoped mode keeps working). Import the constants from jacked.credentials.runtime. Run uv run python -m pytest tests/unit -k session_observer -q. Step 6.11 Launch. In jacked/launch.py import at module scope, next to the other top-level imports, from jacked.credentials.runtime import scoped_launch_needs_global_activation (module scope so tests can patch jacked.launch.scoped_launch_needs_global_activation). The snapshot block that follows the activation (~lines 556-588) reads activation.outcome, activation.operation_id and activation.observed_identity, so the skip path must define what it publishes. Restructure the region from the "The exact certified Claude build reads the global macOS Keychain" comment through the snapshot publish as:

    from jacked.credentials.models import CredentialIdentity, SwitchOutcome

    identity = CredentialIdentity(
        account_id=account_id,
        email=account.get("email"),
        organization_id=account.get("organization_uuid") or None,
    )
    if scoped_launch_needs_global_activation():
        # On macOS the Keychain outranks CLAUDE_CONFIG_DIR, so the global
        # authority must switch too. Unknown builds fail closed here.
        activation = _activate_launch_credentials(account, db)
        truthful_outcomes = {
            SwitchOutcome.COMMITTED,
            SwitchOutcome.COMMITTED_DEGRADED,
            SwitchOutcome.OBSERVED_TARGET_UNFENCED,
        }
        if (
            activation.outcome not in truthful_outcomes
            or activation.observed_identity.account_id != account_id
        ):
            reason = activation.message or activation.outcome.value.replace("_", " ")
            raise click.ClickException(
                f"Could not establish Claude credentials for account {account_id}: {reason}"
            )
        click.echo(
            "Note: this certified Claude build uses the global credential authority. "
            "Launching this account also changes the default for future Claude "
            "sessions; existing sessions keep their current credentials.",
            err=True,
        )
        observed = activation.observed_identity
        evidence = (
            "launch:scoped-file-readback",
            f"launch:global-authority:{activation.outcome.value}",
        )
        revision = f"switch:{activation.operation_id}"
    else:
        click.echo(
            "Note: this platform reads credentials from the launch directory; "
            "the default account for other Claude sessions is unchanged.",
            err=True,
        )
        # "observed" must mean read back. The scoped file was just written;
        # read it through the strict store and compare the OAuth object to
        # what build_oauth_data produced before claiming anything.
        from jacked.credentials.file_store import FileCredentialStore

        readback = FileCredentialStore(
            config_dir / ".credentials.json", trusted_root=config_dir
        ).read()
        written = readback.payload.to_mapping() if readback.payload is not None else None
        if written is None or written.get("claudeAiOauth") != build_oauth_data(account):
            raise click.ClickException(
                f"Credentials for account {account_id} did not read back from the "
                f"launch directory: {readback.reason or 'content mismatch'}"
            )
        observed = CredentialIdentity(
            account_id=account_id,
            email=account.get("email"),
            organization_id=readback.payload.identity.organization_id,
        )
        evidence = ("launch:scoped-file-readback", "launch:global-authority:skipped")
        revision = f"launch:{uuid.uuid4()}"

    # Publish the token-free observation beside the launch directory for
    # session hooks. Labelled global on purpose: the scoped file is a launch
    # input, not the certified credential authority.
    try:
        from jacked.credentials.resolver import (
            FileResolverSnapshotSink,
            ResolverState,
            SnapshotUpdate,
        )

        FileResolverSnapshotSink(config_dir / "jacked-resolver-snapshot.json").publish(
            SnapshotUpdate(
                scope="global",
                state=ResolverState.RESOLVED,
                evidence=evidence,
                credential_revision=revision,
                desired=identity,
                observed=observed,
            )
        )
    except OSError as exc:
        logger.warning(
            "Failed to publish launch credential evidence for account %d: %s",
            account_id,
            exc,
        )

Import uuid at module scope in launch.py. The scoped file is unstamped by design (no _jackedAccountId), so the account id comes from the account row and only the organization comes from the readback. A real revision string is required: launch_claude exports JACKED_CREDENTIAL_SCOPE and the revision to the child only when the revision is a non-empty string.

prepare_account_dir is exercised by ~40 existing tests (test_launch.py, test_active_account_cc_skip.py, test_dual_token.py) and the new call would reach the real Claude detector on every one of them, with a host-dependent result. So the autouse fixture _block_keychain_writes in tests/conftest.py must also patch jacked.launch.scoped_launch_needs_global_activation to return True (add it to the same with patch(...), patch(...): block); the new test below overrides it locally. Add the test to class TestPrepareAccountDir in tests/unit/test_launch.py, modelled on test_creates_cred_file just above it:

    def test_scoped_launch_skips_global_activation_on_file_authority_platforms(self, tmp_path):
        """Where the credential file is the authority, the launch dir is what Claude
        reads; the global file must stay untouched and the snapshot must say so."""
        db = _make_db(tmp_path)
        account = db.get_account(1)

        with (
            mock.patch("jacked.launch.ACCOUNTS_DIR", tmp_path / "accounts"),
            mock.patch("jacked.launch.should_refresh", return_value=False),
            mock.patch("jacked.launch.scoped_launch_needs_global_activation", return_value=False),
            mock.patch("jacked.launch._activate_launch_credentials") as activate,
        ):
            from jacked.launch import prepare_account_dir

            result = prepare_account_dir(account, db)

        activate.assert_not_called()
        snapshot = json.loads((result / "jacked-resolver-snapshot.json").read_text())
        assert snapshot["state"] == "resolved"
        assert "launch:global-authority:skipped" in snapshot["evidence"]
        assert "launch:scoped-file-readback" in snapshot["evidence"]
        assert snapshot["observed"]["account_id"] == 1
        assert snapshot["observed"]["organization_id"] == snapshot["desired"]["organization_id"]
        assert snapshot["credential_revision"].startswith("launch:")
        assert "alice_cc_access" not in json.dumps(snapshot)

    def test_scoped_launch_fails_when_the_written_file_does_not_read_back(self, tmp_path):
        db = _make_db(tmp_path)
        account = db.get_account(1)

        def corrupt_replace(src, dst, **kwargs):
            Path(dst).write_text("{not json", encoding="utf-8")

        with (
            mock.patch("jacked.launch.ACCOUNTS_DIR", tmp_path / "accounts"),
            mock.patch("jacked.launch.should_refresh", return_value=False),
            mock.patch("jacked.launch.scoped_launch_needs_global_activation", return_value=False),
            mock.patch("jacked.launch._safe_replace", corrupt_replace),
            pytest.raises(click.ClickException, match="read back"),
        ):
            from jacked.launch import prepare_account_dir

            prepare_account_dir(account, db)

(_safe_replace is the helper prepare_account_dir uses to land the scoped file; confirm its name and keyword shape at ~line 495 before writing the corrupting stand-in. Import click, pytest, and Path in the test module if missing.)

Step 6.12 Run: uv run python -m pytest tests/unit/test_credential_runtime.py tests/unit/test_credential_capabilities.py tests/unit/test_credential_transactions.py tests/unit/test_credential_resolver.py tests/unit/test_credential_stores.py tests/unit/test_launch.py tests/unit -k "session_observer" -q. Expected: pass. Step 6.13 Commit.

git add jacked/credentials/runtime.py jacked/credentials/transaction.py jacked/credentials/resolver.py jacked/credentials/file_store.py jacked/credentials/canonical.py jacked/api/session_observer.py jacked/launch.py tests/conftest.py tests/unit/test_credential_runtime.py tests/unit/test_credential_transactions.py tests/unit/test_credential_resolver.py tests/unit/test_credential_stores.py tests/unit/test_launch.py
git commit -m "feat(credentials): per-platform topology records, declaration-keyed stores, fail-closed rules for uninspected builds"

Task 7: Keychain access through the signed security tool, prompt-free by construction

Traces to: AC 13, 13b, 13c, 14

Files

• Modify: jacked/credentials/macos_store.py — new SecurityCliBackend, keychain_is_locked, cooling latch; delete PyObjCSecurityBackend
• Test: tests/unit/test_credential_stores.py — replace the PyObjC test with CLI backend and latch tests

Interfaces

• Produces: SecurityCliBackend(*, run=subprocess.run, tool="/usr/bin/security", interactive_timeout=60.0, noninteractive_timeout=2.0) implementing NativeSecurityBackend. Reads use argv; writes use security -i with the command on stdin so the secret never appears in process arguments.
• Produces: keychain_is_locked(*, security_module=None) -> bool: a prompt-free status query; returns False when the framework is unavailable.
• Produces: MacOSCredentialStore(account=None, *, backend=None, noninteractive_timeout=3.0, latch_cooldown=600.0, lock_probe=keychain_is_locked, clock=time.monotonic); the timed-out latch is process-wide, keyed by (service, account) in a module dict (stores are rebuilt on every resolution), expires after latch_cooldown seconds, applies to background reads and background writes, and is cleared by a successful foreground call. clear_keychain_latches() exists for tests. SECURITY_STDIN_MAX_LINE = 4095: an upsert is sent as hex on stdin when it fits, as escaped ASCII JSON (-w) on stdin when that fits, and otherwise fails closed with UNUSABLE unless JACKED_KEYCHAIN_ARGV_FALLBACK=1 allows an argv write (warned once per process). The macOS readback mismatch reports ERROR, not CONCURRENT_WRITE.
• Consumes: NativeReadResult, StoreStatus unchanged.

Why: Claude Code touches the Keychain only through /usr/bin/security, so that Apple-signed tool is on the item's access list from day one and reads it without any prompt (verified on this Mac: 20 ms, no dialog). 0.99.0 moved to in-process Security.framework calls, which macOS attributes to the python3.14 binary; it is not on the list, so interactive operations ask for the login password and every uv Python upgrade starts the cycle again, while background reads fail closed and blank the pill. The tool has no "never prompt" flag (verified: reading an item whose access list lacks the tool blocks until killed), so this task replaces the framework's structural guarantee with three cheap ones: a prompt-free lock-status query before any background call (verified: 3 ms via SecKeychainGetStatus, no item access, no ACL), a subprocess timeout strictly shorter than the store's thread timeout with the child killed on expiry, and a cooling latch that covers background reads and writes and heals itself. Writes go through security -i on stdin (verified: it executes quoted commands and propagates the exit status), so the token is never in process arguments.

Step 7.1 Replace test_pyobjc_noninteractive_query_uses_ui_fail_without_auth_context in tests/unit/test_credential_stores.py with these tests (import SecurityCliBackend, keychain_is_locked instead of PyObjCSecurityBackend; add import json, import re, import subprocess, and from unittest import mock):

def _completed(returncode=0, stdout=b"", stderr=b""):
    return SimpleNamespace(returncode=returncode, stdout=stdout, stderr=stderr)


def test_security_cli_read_uses_argv_and_returns_payload_bytes() -> None:
    calls = []

    def run(args, **kwargs):
        calls.append((args, kwargs))
        return _completed(0, b'{"x":1}\n')

    backend = SecurityCliBackend(run=run)
    result = backend.read(service="Claude Code-credentials", account="alice", is_interactive=False)

    assert result.status is StoreStatus.OK
    assert result.data == b'{"x":1}'
    assert calls[0][0] == [
        "/usr/bin/security", "find-generic-password",
        "-a", "alice", "-s", "Claude Code-credentials", "-w",
    ]
    assert calls[0][1]["timeout"] == 2.0
    assert calls[0][1]["text"] is False


@pytest.mark.parametrize(
    "returncode,stderr,status",
    [
        (44, b"security: SecKeychainSearchCopyNext: The specified item could not be found in the keychain.", StoreStatus.MISSING),
        (36, b"security: SecKeychainItemCopyContent: User interaction is not allowed.", StoreStatus.INTERACTIVE_REQUIRED),
        (128, b"security: SecKeychainItemCopyContent: User canceled the operation.", StoreStatus.DENIED),
        (1, b"security: something else entirely", StoreStatus.ERROR),
    ],
)
def test_security_cli_read_maps_failures_by_exit_code(returncode, stderr, status) -> None:
    backend = SecurityCliBackend(run=lambda *a, **k: _completed(returncode, b"", stderr))

    result = backend.read(service="s", account="a", is_interactive=False)

    assert result.status is status
    assert result.data is None
    assert "entirely" not in result.reason  # raw stderr never leaves the backend


def test_security_cli_noninteractive_timeout_is_reported_not_raised() -> None:
    def run(args, **kwargs):
        raise subprocess.TimeoutExpired(args, kwargs["timeout"])

    backend = SecurityCliBackend(run=run)
    result = backend.read(service="s", account="a", is_interactive=False)

    assert result.status is StoreStatus.ERROR
    assert "timed out" in result.reason


def test_security_cli_write_sends_command_on_stdin_never_in_argv() -> None:
    calls = []

    def run(args, **kwargs):
        calls.append((args, kwargs))
        return _completed(0)

    backend = SecurityCliBackend(run=run)
    secret = b'{"a":"sk-ant-oat01-secret"}'
    status, _ = backend.update(
        service="Claude Code-credentials", account="alice", data=secret, is_interactive=True
    )

    assert status is StoreStatus.OK
    args, kwargs = calls[0]
    assert args == ["/usr/bin/security", "-i"]
    assert kwargs["timeout"] == 60.0
    assert kwargs["input"] == (
        b'add-generic-password -U -a "alice" -s "Claude Code-credentials" -X '
        + secret.hex().encode("ascii") + b"\n"
    )
    assert all(secret.hex() not in item and "sk-ant" not in item for item in args)

    status, _ = backend.add(service="s", account="a", data=b"{}", is_interactive=True)
    assert status is StoreStatus.OK
    assert calls[1][0] == ["/usr/bin/security", "-i"]


def test_security_cli_rejects_unquotable_locator_parts() -> None:
    backend = SecurityCliBackend(run=lambda *a, **k: _completed(0))

    status, reason = backend.update(
        service='bad"name', account="alice", data=b"{}", is_interactive=True
    )

    assert status is StoreStatus.ERROR
    assert "locator" in reason


class _ObjcLikeError(Exception):
    """Stands in for objc.error, which is an Exception but not an OSError."""


def test_keychain_probe_swallows_framework_errors() -> None:
    def explode(_none):
        raise _ObjcLikeError("bridge failure")

    with mock.patch("jacked.credentials.macos_store._PROBE_ERRORS", (_ObjcLikeError,)):
        fake = SimpleNamespace(SecKeychainCopyDefault=explode)
        assert keychain_is_locked(security_module=fake) is False


def test_security_cli_medium_payload_uses_escaped_json_on_stdin() -> None:
    calls = []

    def run(args, **kwargs):
        calls.append((args, kwargs))
        return _completed(0)

    backend = SecurityCliBackend(run=run)
    medium = b'{"a":"' + b"x" * 3000 + b'","q":"say \\"hi\\""}'  # JSON-escaped quotes; hex would exceed 4095
    status, _ = backend.update(service="s", account="a", data=medium, is_interactive=True)

    assert status is StoreStatus.OK
    args, kwargs = calls[0]
    assert args == ["/usr/bin/security", "-i"]
    line = kwargs["input"]
    assert line.startswith(b'add-generic-password -U -a "a" -s "s" -w "')
    assert len(line) <= 4095
    prefix = b'add-generic-password -U -a "a" -s "s" -w "'
    quoted = line[len(prefix):-2]  # drop the closing quote and newline
    unescaped = re.sub(r"\\(.)", r"\1", quoted.decode("ascii"))  # the -i lexer's escape rule
    assert json.loads(unescaped) == json.loads(medium)
    assert medium.hex().encode() not in line


def test_security_cli_oversized_payload_fails_closed_unless_argv_opt_in(monkeypatch, caplog) -> None:
    calls = []

    def run(args, **kwargs):
        calls.append((args, kwargs))
        return _completed(0)

    backend = SecurityCliBackend(run=run)
    huge = b'{"a":"' + b"x" * 5000 + b'"}'

    monkeypatch.delenv("JACKED_KEYCHAIN_ARGV_FALLBACK", raising=False)
    status, reason = backend.update(service="s", account="a", data=huge, is_interactive=True)
    assert status is StoreStatus.UNUSABLE
    assert "stdin line limit" in reason and "5008 bytes" in reason
    assert calls == []

    monkeypatch.setenv("JACKED_KEYCHAIN_ARGV_FALLBACK", "1")
    with caplog.at_level("WARNING"):
        status, _ = backend.update(service="s", account="a", data=huge, is_interactive=True)
        backend.update(service="s", account="a", data=huge, is_interactive=True)
    assert status is StoreStatus.OK
    assert calls[0][0][:3] == ["/usr/bin/security", "add-generic-password", "-U"]
    assert "input" not in calls[0][1]
    assert caplog.text.count("process argument") == 1  # warned once per process


def test_security_cli_timeout_on_argv_path_never_exposes_the_command(monkeypatch, caplog) -> None:
    def run(args, **kwargs):
        raise subprocess.TimeoutExpired(args, kwargs["timeout"])

    monkeypatch.setenv("JACKED_KEYCHAIN_ARGV_FALLBACK", "1")
    huge = b'{"a":"' + b"x" * 5000 + b'"}'
    with caplog.at_level("DEBUG"):
        status, reason = SecurityCliBackend(run=run).update(
            service="s", account="a", data=huge, is_interactive=True
        )
    assert status is StoreStatus.ERROR
    assert huge.hex() not in caplog.text and huge.hex() not in reason


def test_security_cli_small_payload_uses_hex_on_stdin() -> None:
    calls = []

    def run(args, **kwargs):
        calls.append((args, kwargs))
        return _completed(0)

    backend = SecurityCliBackend(run=run)
    backend.update(service="s", account="a", data=b'{"a":1}', is_interactive=True)
    assert calls[0][0] == ["/usr/bin/security", "-i"]
    assert calls[0][1]["input"].endswith(b"\n")
    assert b" -X " in calls[0][1]["input"]
    assert len(calls[0][1]["input"]) <= 4095


def test_security_cli_write_failure_never_logs_stderr(caplog) -> None:
    secret_hex = b'{"a":"sk-ant-oat01-secret"}'.hex().encode()
    backend = SecurityCliBackend(
        run=lambda *a, **k: _completed(1, b"", b'security: unknown command "' + secret_hex + b'"')
    )
    with caplog.at_level("DEBUG"):
        status, reason = backend.update(service="s", account="a", data=b"{}", is_interactive=True)

    assert status is StoreStatus.ERROR
    assert secret_hex.decode() not in caplog.text
    assert secret_hex.decode() not in reason


def test_security_cli_read_decodes_hex_output_for_non_ascii_payloads() -> None:
    backend = SecurityCliBackend(run=lambda *a, **k: _completed(0, b"7b22c3a9223a317d\n"))
    assert backend.read(service="s", account="a", is_interactive=False).data == '{"é":1}'.encode()

    backend = SecurityCliBackend(run=lambda *a, **k: _completed(0, b'{"x":1}\n'))
    assert backend.read(service="s", account="a", is_interactive=False).data == b'{"x":1}'


def test_keychain_is_locked_reads_status_bit_without_touching_items() -> None:
    fake = SimpleNamespace(
        SecKeychainCopyDefault=lambda _none: (0, "kc"),
        SecKeychainGetStatus=lambda kc, _none: (0, 0b110),  # unlock bit clear
        kSecUnlockStateStatus=1,
    )
    assert keychain_is_locked(security_module=fake) is True
    fake.SecKeychainGetStatus = lambda kc, _none: (0, 0b111)
    assert keychain_is_locked(security_module=fake) is False


def test_locked_keychain_short_circuits_background_read_without_spawning() -> None:
    backend = FakeSecurityBackend(NativeReadResult(StoreStatus.OK, _payload().to_bytes()))
    store = MacOSCredentialStore("alice", backend=backend, lock_probe=lambda: True)

    result = store.read()

    assert result.status is StoreStatus.INTERACTIVE_REQUIRED


def test_timed_out_latch_is_shared_across_store_instances_and_expires() -> None:
    """Stores are rebuilt on every resolution, so the latch must outlive them."""
    backend = BlockingSecurityBackend()
    clock = [1000.0]

    def make():
        return MacOSCredentialStore(
            "alice",
            backend=backend,
            noninteractive_timeout=0.05,
            latch_cooldown=60.0,
            lock_probe=lambda: False,
            clock=lambda: clock[0],
        )

    assert make().read().status is StoreStatus.ERROR  # times out, latches
    backend.release.set()

    second = make()
    assert second.read().status is StoreStatus.ERROR  # latched, no backend call
    assert second.write(_payload(), InteractionMode.BACKGROUND).status is StoreStatus.INTERACTIVE_REQUIRED
    assert backend.calls == 1

    clock[0] += 61.0
    backend.read_result = NativeReadResult(StoreStatus.OK, _payload().to_bytes())
    assert make().read().status is StoreStatus.OK  # expired, backend consulted again
    assert backend.calls == 2


def test_successful_interactive_call_clears_the_latch() -> None:
    backend = BlockingSecurityBackend()
    store = MacOSCredentialStore(
        "alice", backend=backend, noninteractive_timeout=0.05, lock_probe=lambda: False
    )
    assert store.read().status is StoreStatus.ERROR
    backend.release.set()
    backend.read_result = NativeReadResult(StoreStatus.OK, _payload().to_bytes())

    assert store.write(_payload(2), InteractionMode.FOREGROUND).status is StoreStatus.OK
    assert store.read().status is StoreStatus.OK


def test_macos_store_defaults_to_security_cli_backend() -> None:
    store = MacOSCredentialStore("alice")
    assert isinstance(store._backend, SecurityCliBackend)


def test_credential_package_never_uses_security_framework_for_item_access() -> None:
    root = Path(__file__).resolve().parents[2] / "jacked" / "credentials"
    for source in root.glob("*.py"):
        text = source.read_text(encoding="utf-8")
        assert "SecItemCopyMatching" not in text, source
        assert "SecItemAdd" not in text, source
        assert "SecItemUpdate" not in text, source
        assert "LocalAuthentication" not in text, source

The existing test_macos_noninteractive_read_is_bounded_and_circuit_breaks, test_macos_background_write_bounds_its_preflight_read and test_macos_missing_item_requires_foreground_before_add tests must pass lock_probe=lambda: False so they do not depend on the real Keychain state of the test machine, and any test that inspects _timed_out_reads now inspects _latches. Add from unittest import mock if the file lacks it.

Step 7.2 Run: uv run python -m pytest tests/unit/test_credential_stores.py -v. Expected: ImportError for SecurityCliBackend. Step 7.3 In jacked/credentials/macos_store.py delete PyObjCSecurityBackend, add import json, import logging, import os, import subprocess, import time, set the module docstring to "macOS Keychain credential store adapter driven by the signed security tool.", and add:

logger = logging.getLogger(__name__)

SECURITY_TOOL = "/usr/bin/security"
DEFAULT_INTERACTIVE_TIMEOUT_SECONDS = 60.0
DEFAULT_SUBPROCESS_TIMEOUT_SECONDS = 2.0  # strictly below the store's thread timeout
DEFAULT_LATCH_COOLDOWN_SECONDS = 600.0

# security(1) exit statuses are the OSStatus truncated to a byte:
# errSecItemNotFound (-25300) -> 44, errSecInteractionNotAllowed (-25308) -> 36,
# errSecUserCanceled (-128) -> 128. Text is a fallback only; it is versioned
# by macOS and never surfaces in a reason string.
_EXIT_STATUS = {
    44: (StoreStatus.MISSING, "Keychain item not found"),
    36: (StoreStatus.INTERACTIVE_REQUIRED, "Keychain interaction required"),
    128: (StoreStatus.DENIED, "Keychain authorization canceled"),
}
_STDERR_HINTS = (
    (b"could not be found", StoreStatus.MISSING, "Keychain item not found"),
    (b"User interaction is not allowed", StoreStatus.INTERACTIVE_REQUIRED, "Keychain interaction required"),
    (b"User canceled", StoreStatus.DENIED, "Keychain authorization canceled"),
)


SECURITY_STDIN_MAX_LINE = 4095  # security -i splits longer lines
_HEX_ALPHABET = frozenset(b"0123456789abcdef")


def _classify_failure(
    returncode: int, stderr: bytes, *, log_stderr: bool = True
) -> tuple[StoreStatus, str]:
    known = _EXIT_STATUS.get(returncode)
    if known is not None:
        return known
    for needle, status, reason in _STDERR_HINTS:
        if needle in stderr:
            return status, reason
    if log_stderr:
        logger.debug("security exit %d: %r", returncode, stderr[:300])
    else:
        logger.debug("security exit %d (write path; stderr withheld)", returncode)
    return StoreStatus.ERROR, f"security exit {returncode}"


def _unhex_if_needed(data: bytes) -> bytes:
    """``find-generic-password -w`` prints hex for any non-ASCII payload."""
    if data and len(data) % 2 == 0 and set(data) <= _HEX_ALPHABET:
        return bytes.fromhex(data.decode("ascii"))
    return data


def _quoted(value: str) -> str:
    """Quote one security -i argument; reject values its lexer cannot carry."""
    if any(ch in value for ch in '"\\\n\r\0'):
        raise ValueError("Keychain locator part contains unquotable characters")
    return f'"{value}"'


def _ascii_json(data: bytes) -> str:
    """Re-serialise a canonical payload as escaped ASCII for a quoted -w value.

    The lexer honours backslash escapes inside double quotes, so quotes and
    backslashes are escaped and non-ASCII becomes a JSON unicode escape. The parsed
    mapping, and therefore the readback digest, is unchanged.
    """
    text = json.dumps(json.loads(data.decode("utf-8")), ensure_ascii=True, separators=(",", ":"))
    return text.replace("\\", "\\\\").replace('"', '\\"')


_argv_fallback_warned = False


def _warn_argv_fallback_once() -> None:
    global _argv_fallback_warned
    if not _argv_fallback_warned:
        _argv_fallback_warned = True
        logger.warning(
            "Keychain payload exceeds the security stdin line limit; "
            "passing it as a process argument (JACKED_KEYCHAIN_ARGV_FALLBACK=1)"
        )


try:  # PyObjC's own exception class is not an OSError
    import objc as _objc

    _PROBE_ERRORS: tuple[type[BaseException], ...] = (
        AttributeError, TypeError, ValueError, OSError, _objc.error
    )
except ImportError:  # pragma: no cover - non-macOS
    _PROBE_ERRORS = (AttributeError, TypeError, ValueError, OSError)

# One latch per Keychain locator for the whole process: stores are rebuilt on
# every resolution, so instance state would never survive a poll.
_latches: dict[tuple[str, str], float] = {}
_latches_lock = threading.Lock()


def clear_keychain_latches() -> None:
    with _latches_lock:
        _latches.clear()


def keychain_is_locked(*, security_module=None) -> bool:
    """Prompt-free: reads the default keychain's status bits, touches no item.

    Returns False when the framework bridge is unavailable so the caller
    falls through to the bounded tool call instead of failing closed twice.
    """
    module = security_module
    if module is None:
        try:
            import Security as module  # type: ignore[import-not-found]
        except ImportError:
            return False
    try:
        status, keychain = module.SecKeychainCopyDefault(None)
        if status != 0 or keychain is None:
            return False
        status, flags = module.SecKeychainGetStatus(keychain, None)
    except _PROBE_ERRORS:
        logger.debug("keychain status probe failed", exc_info=True)
        return False
    if status != 0:
        return False
    return not bool(flags & module.kSecUnlockStateStatus)


class SecurityCliBackend:
    """Drive Apple's signed ``security`` tool, the same client Claude Code uses.

    A Keychain item carries an access list of the applications allowed to read
    it. Claude Code creates its item through ``security``, so ``security`` is
    trusted from the start, while an in-process Security.framework caller is
    identified as the Python binary and prompts for the login password on
    every new Python build. Using the same tool removes that prompt class.

    Reads pass only the locator on argv. Writes run ``security -i`` and send
    the command on stdin so the secret is never a process argument.
    ``subprocess.run`` kills the child when a timeout expires.
    """

    def __init__(
        self,
        *,
        run=subprocess.run,
        tool: str = SECURITY_TOOL,
        interactive_timeout: float = DEFAULT_INTERACTIVE_TIMEOUT_SECONDS,
        noninteractive_timeout: float = DEFAULT_SUBPROCESS_TIMEOUT_SECONDS,
    ) -> None:
        self._run = run
        self._tool = tool
        self._interactive_timeout = interactive_timeout
        self._noninteractive_timeout = noninteractive_timeout

    def _invoke(self, args: list[str], *, is_interactive: bool, stdin: bytes | None = None):
        timeout = (
            self._interactive_timeout if is_interactive else self._noninteractive_timeout
        )
        kwargs = {"capture_output": True, "text": False, "timeout": timeout, "check": False}
        if stdin is not None:
            kwargs["input"] = stdin
        return self._run([self._tool, *args], **kwargs)

    def read(
        self, *, service: str, account: str, is_interactive: bool
    ) -> NativeReadResult:
        try:
            completed = self._invoke(
                ["find-generic-password", "-a", account, "-s", service, "-w"],
                is_interactive=is_interactive,
            )
        except subprocess.TimeoutExpired:
            return NativeReadResult(StoreStatus.ERROR, reason="security tool timed out")
        except OSError as exc:
            return NativeReadResult(StoreStatus.ERROR, reason=f"security tool unavailable: {exc}")
        if completed.returncode != 0:
            status, reason = _classify_failure(completed.returncode, completed.stderr or b"")
            return NativeReadResult(status, reason=reason)
        return NativeReadResult(StoreStatus.OK, _unhex_if_needed((completed.stdout or b"").strip()))

    def _upsert(
        self, *, service: str, account: str, data: bytes, is_interactive: bool
    ) -> tuple[StoreStatus, str]:
        try:
            command = (
                f"add-generic-password -U -a {_quoted(account)} -s {_quoted(service)} "
                f"-X {data.hex()}\n"
            ).encode("utf-8")
        except ValueError:
            return StoreStatus.ERROR, "Keychain locator part contains unquotable characters"
        if len(command) > SECURITY_STDIN_MAX_LINE:
            # security -i splits lines at 4096 bytes, which would store a
            # truncated secret. Hex doubles the payload; the escaped JSON form
            # (-w) fits roughly twice as much. Beyond that, fail closed: argv
            # is readable by other local users through setuid ps and by
            # endpoint agents, so it is opt-in only.
            command = (
                f"add-generic-password -U -a {_quoted(account)} -s {_quoted(service)} "
                f"-w {_quoted(_ascii_json(data))}\n"
            ).encode("utf-8")
        if len(command) <= SECURITY_STDIN_MAX_LINE:
            args, stdin = ["-i"], command
        elif os.environ.get("JACKED_KEYCHAIN_ARGV_FALLBACK") == "1":
            _warn_argv_fallback_once()
            args = ["add-generic-password", "-U", "-a", account, "-s", service, "-X", data.hex()]
            stdin = None
        else:
            return (
                StoreStatus.UNUSABLE,
                f"Keychain payload of {len(data)} bytes exceeds the security tool's "
                f"{SECURITY_STDIN_MAX_LINE}-byte stdin line limit; set "
                "JACKED_KEYCHAIN_ARGV_FALLBACK=1 to allow a process-argument write",
            )
        try:
            completed = self._invoke(args, is_interactive=is_interactive, stdin=stdin)
        except subprocess.TimeoutExpired:
            # Never stringify the exception: its .cmd carries the full argv.
            return StoreStatus.ERROR, "security tool timed out"
        except OSError as exc:
            return StoreStatus.ERROR, f"security tool unavailable: {exc}"
        if completed.returncode != 0:
            # Never log write-path stderr: a split or rejected line can echo
            # fragments of the hex payload.
            return _classify_failure(completed.returncode, completed.stderr or b"", log_stderr=False)
        return StoreStatus.OK, ""

    def update(
        self, *, service: str, account: str, data: bytes, is_interactive: bool
    ) -> tuple[StoreStatus, str]:
        return self._upsert(service=service, account=account, data=data, is_interactive=is_interactive)

    def add(
        self, *, service: str, account: str, data: bytes, is_interactive: bool
    ) -> tuple[StoreStatus, str]:
        return self._upsert(service=service, account=account, data=data, is_interactive=is_interactive)

Step 7.4 Rework MacOSCredentialStore: constructor (account=None, *, backend=None, noninteractive_timeout=DEFAULT_NONINTERACTIVE_TIMEOUT_SECONDS, latch_cooldown=DEFAULT_LATCH_COOLDOWN_SECONDS, lock_probe=keychain_is_locked, clock=time.monotonic); default backend SecurityCliBackend(); delete the module-global _timed_out_reads set and its lock (replaced by _latches above). Then:

    @property
    def _latch_key(self) -> tuple[str, str]:
        return (SERVICE_NAME, self.account)

    def _latched(self) -> bool:
        with _latches_lock:
            return _latches.get(self._latch_key, 0.0) > self._clock()

    def _latch(self) -> None:
        with _latches_lock:
            _latches[self._latch_key] = self._clock() + self._latch_cooldown

    def _note_interactive_success(self) -> None:
        with _latches_lock:
            _latches.pop(self._latch_key, None)

    def _latched_refusal(self) -> NativeReadResult | None:
        if self._latched():
            return NativeReadResult(
                StoreStatus.ERROR, reason="Keychain access paused after a prior timeout"
            )
        return None

_bounded_noninteractive_read starts with refusal = self._latched_refusal(); if refusal is not None: return refusal. Inside its worker function execute(), before self._backend.read(...), add the lock probe so it runs on the bounded worker thread rather than the caller: if self._lock_probe(): result = NativeReadResult(StoreStatus.INTERACTIVE_REQUIRED, reason="login keychain is locked") (else call the backend as today). On queue.Empty call self._latch() instead of adding to the global set.

In write(), add as the very first statement: if interaction is InteractionMode.BACKGROUND and (refusal := self._latched_refusal()) is not None: return StoreWriteResult(StoreStatus.INTERACTIVE_REQUIRED, refusal.reason). The BACKGROUND preflight read then proceeds through _bounded_noninteractive_read as today. Latch clearing has exactly two spots, both on the FOREGROUND path of write(): (1) right after current = self._backend.read(..., is_interactive=True) when current.status in {StoreStatus.OK, StoreStatus.MISSING}; (2) right after self._backend.update(...) or self._backend.add(...) returns StoreStatus.OK. Both call self._note_interactive_success(). _readback is unchanged.

Add to tests/conftest.py an autouse fixture that calls jacked.credentials.macos_store.clear_keychain_latches() before and after each test (same shape as the identity-cache fixture).

Grep the repo for PyObjCSecurityBackend and _timed_out_reads (including jacked/credentials/__init__.py, docs) and remove every reference.

Step 7.5 Run: uv run python -m pytest tests/unit/test_credential_stores.py tests/unit/test_credential_runtime.py -q. Expected: pass. Step 7.6 Commit.

git add jacked/credentials/macos_store.py jacked/credentials/store.py tests/unit/test_credential_stores.py tests/conftest.py
git commit -m "fix(credentials): drive the Keychain through the signed security tool with prompt-free guards"

Task 8: Statusline shows the observed account on a desired-default conflict

Traces to: AC 12

Files

• Modify: jacked/statusline_account.py — account_facts
• Test: tests/unit/test_statusline.py — _write_account gains observed and evidence overrides; new tests

Interfaces

• Consumes: snapshot dict with state, evidence (list), desired, observed (both {account_id, email, organization_id}), written by jacked/api/session_observer.py. The observer emits desired-default:conflict when the stores agree on an identity that is not the desired default, and account-metadata:organization-conflict when the credential's organization contradicts the DB row.

Why: when the DB's active account differs from what the Keychain holds, the snapshot carries both identities but the statusline hides the one the runtime is using. Showing "observed · desired X" tells the user exactly what will happen on the next request. The organization-conflict case must keep the current rendering: there the observed identity's organization is known to be wrong, and publishing it would key the usage lookup to the wrong org.

Step 8.1 In tests/unit/test_statusline.py change _write_account so observed and evidence can be overridden: the evidence entry becomes account.get("evidence", ["store_consensus"]) (accept a list; update any caller that passes a string) and the observed entry becomes account.get("observed", identity if account.get("state", "resolved") == "resolved" else None). Add the tests:

def test_desired_default_conflict_renders_observed_and_desired(home):
    observed = {"account_id": 4, "email": "runtime@co.com", "organization_id": "org-4"}
    _write_account(
        home,
        emailAddress="target@co.com",
        state="conflict",
        observed=observed,
        evidence=["authority:macos-keychain:ok", "desired-default:conflict"],
    )

    assert _render({}, home) == f"runtime@co.com {MIDDOT} desired target@co.com"

    facts = statusline_account.account_facts(str(home), NOW)
    assert facts["email"] == "runtime@co.com"
    assert facts["org_uuid"] == "org-4"
    assert facts["state"] == "credential conflict"


def test_organization_conflict_keeps_runtime_unknown_rendering(home):
    observed = {"account_id": 4, "email": "target@co.com", "organization_id": "org-db"}
    _write_account(
        home,
        emailAddress="target@co.com",
        state="conflict",
        observed=observed,
        evidence=["account-metadata:organization-conflict"],
    )

    assert _render({}, home) == (
        f"desired target@co.com {MIDDOT} runtime unknown (credential conflict)"
    )
    assert statusline_account.account_facts(str(home), NOW)["org_uuid"] == ""

Import jacked.statusline_account as statusline_account at the top if not present.

Step 8.2 Run: uv run python -m pytest tests/unit/test_statusline.py -k conflict -v. Expected: the first new test fails (renders "runtime unknown"); the second passes. Step 8.3 account_facts is already 48 lines, so the new branch lives in a helper. Add above account_facts:

def _desired_default_conflict(snapshot: dict, observed: dict | None) -> bool:
    """True when the stores agree on an identity that is not the desired default.

    An organization conflict also reports ``conflict`` but its observed
    identity is known to be wrong, so it must keep the unknown rendering.
    """
    evidence = snapshot.get("evidence")
    evidence = evidence if isinstance(evidence, list) else []
    return (
        snapshot.get("state") == "conflict"
        and observed is not None
        and "desired-default:conflict" in evidence
        and "account-metadata:organization-conflict" not in evidence
    )

plus a second helper that builds the facts, so account_facts gains only two lines:

def _observed_with_desired(observed: dict, desired_label: str) -> dict:
    """Facts naming what the runtime will actually use; the conflict is not hidden."""
    return {
        "segment": f"{observed['email']} {MIDDOT} desired {desired_label}",
        "email": observed["email"],
        "org_uuid": observed["organization_id"],
        "state": "credential conflict",
    }

and in account_facts, after desired_label is computed and before the reason ladder:

    if valid_clock and not scoped_unverified and _desired_default_conflict(snapshot, observed):
        return {**facts, **_observed_with_desired(observed, desired_label)}

Step 8.4 Run: uv run python -m pytest tests/unit/test_statusline.py -q. Expected: pass. Step 8.5 Commit.

git add jacked/statusline_account.py tests/unit/test_statusline.py
git commit -m "feat(statusline): show the observed runtime account on a desired-default conflict"

Task 9: Documentation and full gate

Traces to: spec sections B and C, all

Files

• Modify: docs/architecture/oauth-and-credential-flows.md — "Shipped capability" table (~lines 96-114); Keychain access paragraph (~131-135)
• Modify: docs/architecture/auto-swap-system.md — section 6 paragraph on the shipped record (~line 124)

Interfaces

• None beyond the files above.

Step 9.1 Replace the "Shipped capability" table in docs/architecture/oauth-and-credential-flows.md with:

### Shipped capability records

Certification is keyed by credential-store topology per platform and config
mode, not by executable bytes. The observed build and SHA-256 are recorded as
evidence on every resolution.

| Platform | Config mode | Build floor | Inspected through | Authority | Required mirror | Mode |
| --- | --- | --- | --- | --- | --- | --- |
| `darwin` | `global` | `2.1.0` | `2.1.260` | macOS Keychain (`Claude Code-credentials`) | `~/.claude/.credentials.json` | `global_uncooperative` |
| `linux` | `global` | `2.1.0` | `2.1.260` | `~/.claude/.credentials.json` | none | `global_uncooperative` |
| `windows` | `global` | `2.1.0` | `2.1.260` | `%USERPROFILE%\.claude\.credentials.json` | none | `global_uncooperative` |

A build newer than "inspected through" still resolves and carries the
`build-newer-than-inspected` evidence marker; on such a build jacked refuses
to create a missing authority (a moved store looks identical to a missing
one) and logs any `claudeAiOauth` keys it would drop. Scoped config mode
(`CLAUDE_CONFIG_DIR`) has no shipped record. On Linux and Windows a scoped
launch does not touch the global file. `~/.claude` must be a real directory
(a symlinked dotfiles setup is refused with a clear reason); on Windows the
file's privacy is the profile directory's ACL, the same as Claude Code's own.

### Keychain access

All Keychain reads and writes go through `/usr/bin/security`, the same
Apple-signed tool Claude Code uses, so its access-list entry is shared and no
password prompt ever names a Python binary. Writes run `security -i` with the
command on stdin, so tokens never appear in process arguments. Background
calls are guarded by a prompt-free lock-status probe, a 2 s subprocess
timeout (the child is killed on expiry), and a 10 minute cooling latch that
also blocks background writes; a successful foreground call clears the latch.

Step 9.2 In docs/architecture/auto-swap-system.md section 6 replace both sentences at ~lines 122-124: "Credential capabilities are keyed to exact executable bytes, version, config mode, platform, and architecture. The only shipped production record is Claude 2.1.259 with a specific SHA-256 on macOS arm64 in global mode." becomes "Credential capabilities are keyed to the credential-store topology per platform and config mode, with a certified build floor; the executable hash and build travel as evidence. The shipped production records certify the topology on macOS (Keychain plus file mirror), Linux and Windows (file authority) for Claude builds from 2.1.0, inspected through 2.1.260." Rewrite the single sentence at ~135-137 "Linux, Windows, Intel macOS, other Claude builds, and scoped modes also have no shipped mutation capability. The portable file-store code does not itself certify Claude's consumption behavior on those platforms." as "Scoped config modes have no shipped mutation capability. Linux and Windows are certified for the file topology; a switch there is unfenced and foreground-only, exactly as on macOS." Also update README.md ~line 206 (the sentence beginning "The exact Claude Code 2.1.259 macOS arm64 build can report observed_target_unfenced") to describe topology certification with a build floor, and the file-map row at oauth-and-credential-flows.md ~line 382 ("Security.framework Keychain authority adapter" becomes "Keychain authority adapter driven by the signed security tool"). In oauth-and-credential-flows.md also fix ~lines 92-95 ("An unknown digest, version, config mode, platform, or architecture resolves to unsupported" becomes "An unknown platform or config mode, or a build below the certified floor, resolves to unsupported; digest and architecture are evidence, not gates") and ~lines 133-138 (the sentences "It uses Security.framework through PyObjC, not the security subprocess" and "After a read timeout, further noninteractive reads for that locator are disabled for the process" are replaced by the Keychain access section from Step 9.1). Step 9.3 Run the full suite once as the gate (about seven minutes; do not run it while iterating):

uv run python -m pytest -q 2>&1 | tail -20

Expected: all pass. Fix anything the earlier task subsets missed before continuing.

Step 9.4 Commit.

git add README.md docs/architecture/oauth-and-credential-flows.md docs/architecture/auto-swap-system.md
git commit -m "docs: describe topology-keyed certification and security-tool Keychain access"

Task 10: Prove it on the real Mac (Phase 6 evidence)

Traces to: AC 15

Files

• Read: ~/.claude/jacked-tray.log — evidence source

Interfaces

• Consumes: an installed build of this branch. Install it into the tool environment from the repo without touching PyPI: uv tool install --force --editable /Users/jackneil/Github/claude-jacked, then confirm cd /tmp && ~/.local/share/uv/tools/claude-jacked/bin/python -c 'import jacked; print(jacked.__file__)' points at the repo.

Why: green tests prove the units; only the assembled service on the machine that failed this morning proves the fix. Every step below records its command and output; paste them into the PR.

Step 10.1 Restart through the real handoff path and time it:

cd /tmp && time ~/.local/bin/jacked service restart
launchctl print gui/$(id -u)/ai.hank.jacked | grep -E "state|pid ="
python3 -c "import json;d=json.load(open('$HOME/.claude/jacked-service-v2/api-v2.instance.json'));print(d['process']['pid'], d['supervisor'])"
grep -E "Service ready|giving up|did not become ready" ~/.claude/jacked-tray.log | tail -3

Expected: "Owned service handoff: new generation is ready", launchd state running, manifest pid equals launchd's pid, and a "Service ready ... ready_in=N.Ns" line.

Step 10.2 Simulate the boot-time failure class: kill the service without a handoff and confirm launchd brings it back on its own, and that the breaker file is cleared on success:

kill -9 $(launchctl print gui/$(id -u)/ai.hank.jacked | awk '/pid =/{print $3}')
sleep 20; launchctl print gui/$(id -u)/ai.hank.jacked | grep -E "state|pid =|last exit"
ls ~/.claude/jacked-service-v2/start-failures.json 2>&1

Expected: state running with a new pid; the breaker file does not exist.

Step 10.3 Credential path with no prompt. Run these and watch the screen: no dialog may appear.

curl -s http://127.0.0.1:8321/api/auth/active-credential | python3 -m json.tool
curl -s http://127.0.0.1:8321/api/menubar-summary | python3 -c "import json,sys;d=json.load(sys.stdin);print(d['active_account_id'], (d['active'] or {}).get('email'))"
cat ~/.claude/jacked-resolver-snapshot.json | python3 -m json.tool | head -30

Expected: state resolved with evidence containing build:2.1.260 and the Keychain authority marked ok; active_account_id is the account the Keychain holds (4, jackusc@gmail.com, unless the user has switched); the snapshot state is resolved or conflict with an observed identity.

Step 10.4 Menu bar and statusline evidence:

W=$(~/.local/share/uv/tools/claude-jacked/bin/python -c "import Quartz;b=Quartz.CGDisplayBounds(Quartz.CGMainDisplayID());print(int(b.size.width))")
screencapture -x -R $((W-900)),0,900,40 /tmp/menubar_after_fix.png
echo '{"model":{"display_name":"Fable 5.1"}}' | ~/.local/share/uv/tools/claude-jacked/bin/python -m jacked.statusline

Expected: the pill shows percentages, not "—"; the statusline segment names an email (either plain or "email · desired other"), not "runtime unknown".

Step 10.5 Interactive switch through the tool: in the dashboard switch the active account to the one the Keychain already holds, then back. Expected: no password dialog names python3.14; if macOS asks once for security, click Always Allow and confirm a second switch is silent. Then run claude --version and a one-line claude -p "say ok" to prove Claude Code still reads the item jacked wrote. Step 10.6 Record the outputs of 10.1 through 10.5 in the PR description under "Evidence".

Task 11: Dashboard keeps polling the chained Claude Code token flow after a re-auth

Traces to: AC 16 (appended)

Files

• Modify: jacked/data/web/js/components/oauth-flows.js — runOAuthFlow: follow cc_flow_id after the primary flow completes
• Test: tests/unit/test_web_js_oauth_chained_cc.py — node-driven, reuses the guard test harness

Interfaces

• Consumes: GET /api/auth/flow/{flow_id} exposes cc_flow_id on a completed primary flow; the chained flow polls like any other.
• Consumes: HARNESS / _run from tests/unit/test_web_js_oauth_flow_guard.py.

Why: today `runOAuthFlow` polls only the flow it started. A re-auth is a *primary* flow; when it completes, the server stores the account, marks the flow completed with `cc_flow_id`, and opens a second browser window for the Claude Code token. The dashboard sees `completed`, refreshes once, and stops. The Claude Code token then lands server-side (`_store_cc_tokens` writes the token and `validation_status="valid"`), but nothing refreshes the accounts view, so the account's status button keeps showing the pre-re-auth state until the user clicks something else. Reported 2026-09-04: "it doesn't say the cc token is valid, the icon button doesn't change on the account unless I click it again and reauth that one by itself." This is a dashboard polling gap, not a storage bug.

Step 11.1 Write the failing tests in tests/unit/test_web_js_oauth_chained_cc.py:

"""A re-auth is two server flows: the primary sign-in, then the chained Claude
Code token flow the server auto-starts. The dashboard must keep polling the
second one so the account card updates when the token lands, not on the next
manual click.

Node runs the real component source; skipped when node is not installed.
"""

import shutil

import pytest

from tests.unit.test_web_js_oauth_flow_guard import _run

pytestmark = pytest.mark.skipif(shutil.which("node") is None, reason="node not installed")


def test_completed_primary_with_cc_flow_id_keeps_polling_the_chained_flow(tmp_path):
    result = _run(tmp_path, r"""
(async () => {
    const paths = [];
    const answers = [
        { status: 'completed', cc_flow_id: 'cc1', account_id: 7, email: 'a@b.com' },
        { status: 'pending' },
        { status: 'completed', account_id: 7, email: 'a@b.com' },
    ];
    global.api.get = async (p) => { paths.push(p); calls.get++; return answers.shift() || { status: 'completed' }; };
    releaseRefresh();  // refreshes resolve immediately in this test

    startReauthFlow(7, 'a@b.com');
    await tick();
    fireDoc('visibilitychange');            // primary: completed + cc_flow_id
    await tick(); await tick();
    const guardAfterPrimary = window.jackedState._accountActionInFlight;
    const refreshesAfterPrimary = refreshCalls;
    const bannerWhileChained = textAll(statusEl);
    const hooksWhileChained = (listeners.doc['visibilitychange'] || []).length;
    fireDoc('visibilitychange');            // chained: pending
    await tick();
    fireDoc('visibilitychange');            // chained: completed
    await tick(); await tick();
    out({ paths, guardAfterPrimary, refreshesAfterPrimary, bannerWhileChained, hooksWhileChained,
          refreshCalls, banner: textAll(statusEl),
          hooksAtEnd: (listeners.doc['visibilitychange'] || []).length });
    process.exit(0);
})().catch(e => { console.error(e); process.exit(1); });
""")
    assert result["paths"] == ["/api/auth/flow/f1", "/api/auth/flow/cc1", "/api/auth/flow/cc1"]
    assert result["guardAfterPrimary"] is False, "the primary verdict releases the guard"
    assert result["refreshesAfterPrimary"] == 1
    assert "Claude Code token" in result["bannerWhileChained"]
    assert result["hooksWhileChained"] == 1, "still polling while the chained flow runs"
    assert result["refreshCalls"] == 2, "the chained completion refreshes the accounts view again"
    assert "Claude Code token" in result["banner"] and "authorized" in result["banner"].lower()
    assert result["hooksAtEnd"] == 0


def test_chained_flow_failure_names_the_failure_but_keeps_the_reauth(tmp_path):
    result = _run(tmp_path, r"""
(async () => {
    const answers = [
        { status: 'completed', cc_flow_id: 'cc1' },
        { status: 'error', error: 'CC auth email mismatch' },
    ];
    global.api.get = async () => { calls.get++; return answers.shift() || { status: 'error' }; };
    releaseRefresh();

    startReauthFlow(7, 'a@b.com');
    await tick();
    fireDoc('visibilitychange');
    await tick(); await tick();
    fireDoc('visibilitychange');
    await tick(); await tick();
    out({ refreshCalls, banner: textAll(statusEl), guard: window.jackedState._accountActionInFlight,
          hooks: (listeners.doc['visibilitychange'] || []).length });
    process.exit(0);
})().catch(e => { console.error(e); process.exit(1); });
""")
    assert result["refreshCalls"] == 2, "the card must show the stored state after a chained failure too"
    assert "re-authenticated" in result["banner"].lower()
    assert "CC auth email mismatch" in result["banner"]
    assert result["guard"] is False
    assert result["hooks"] == 0


def test_completed_flow_without_cc_flow_id_ends_as_before(tmp_path):
    result = _run(tmp_path, r"""
(async () => {
    global.api.get = async () => { calls.get++; return { status: 'completed' }; };
    releaseRefresh();
    startCcAuthFlow(7, 'a@b.com');
    await tick();
    fireDoc('visibilitychange');
    await tick(); await tick();
    out({ refreshCalls, gets: calls.get, banner: textAll(statusEl),
          hooks: (listeners.doc['visibilitychange'] || []).length });
    process.exit(0);
})().catch(e => { console.error(e); process.exit(1); });
""")
    assert result == {"refreshCalls": 1, "gets": 1, "banner": result["banner"], "hooks": 0}
    assert "authorized successfully" in result["banner"]

Step 11.2 Run `uv run python -m pytest tests/unit/test_web_js_oauth_chained_cc.py -v`. Expected: the first two tests fail (only one refresh; the second GET path never switches to `cc1`); the third passes (it pins current behaviour).

Step 11.3 In `runOAuthFlow` (jacked/data/web/js/components/oauth-flows.js):

- Change `const flowId = start.flow_id;` to `let flowId = start.flow_id;` and add `let chainedFlowId = null;` beside `let terminal = false;`. - Add a helper next to `renderBanner`:

    // A primary sign-in that completes may hand back a second flow: the server
    // opened another browser window for the Claude Code token. Follow it so
    // the account card updates when that token lands, not on the next click.
    function chainTo(ccFlowId) {
        chainedFlowId = ccFlowId;
        flowId = ccFlowId;
        elapsed = 0;
        const slot = document.getElementById('oauth-flow-status') || statusEl;
        slot.textContent = '';
        buildOAuthBanner(slot, OAUTH_ACCENT_ORANGE,
            'Account re-authenticated. Authorizing the Claude Code token in the browser...');
    }

- Declare `let elapsed = 0;` before `handleFlowResult` (move the existing declaration up; the interval callback keeps using it). - In `handleFlowResult`, replace the `completed` branch with:

        if (poll.status === 'completed') {
            // The server's verdict is what the guard was waiting on, so drop
            // it now, not after the refresh: refreshAndRender fetches every
            // account (and, on macOS, reconciles the active one through the
            // Keychain), and a Use Account click during that window used to
            // be refused as "another action in progress".
            window.jackedState._accountActionInFlight = false;
            const ccFlowId = typeof poll.cc_flow_id === 'string' ? poll.cc_flow_id : '';
            if (ccFlowId && !chainedFlowId) {
                // Refresh so the re-authenticated account shows now, then keep
                // polling the chained flow; this call is not terminal.
                await refreshAndRender();
                chainTo(ccFlowId);
                return false;
            }
            terminal = true;
            stopPolling();
            const success = chainedFlowId
                ? { text: 'Account re-authenticated and Claude Code token authorized!', duration: 3000 }
                : msgs.success(poll);
            // Refresh FIRST: refreshAndRender re-renders the route wholesale,
            // which would wipe a banner drawn before it. Render the success
            // message into the fresh slot afterwards.
            await refreshAndRender();
            renderBanner(OAUTH_SUCCESS_CLASS, success.text, success.duration);
        } else if (poll.status === 'error') {
            if (chainedFlowId) {
                // The account itself is stored; only the token step failed.
                // Refresh so the card shows exactly what the server holds.
                terminal = true;
                stopPolling();
                await refreshAndRender();
                renderBanner(OAUTH_ERROR_CLASS,
                    'Account re-authenticated, but the Claude Code token authorization failed: '
                    + (poll.error || 'Unknown error')
                    + '. Use the account menu to authorize the Claude Code token again.');
                window.jackedState._accountActionInFlight = false;
            } else {
                endWith(OAUTH_ERROR_CLASS, msgs.failPrefix + (poll.error || 'Unknown error'));
            }
        } else {
            endWith(OAUTH_WARN_CLASS, msgs.notFound);
        }
        return true;

- `pollOnce` already reads `flowId` at call time (`api.get(`/api/auth/flow/${flowId}`)`), so switching the binding is enough; confirm nothing else captured the old value. - Keep the `submitCode` path working: it posts to `/api/auth/flow/${flowId}/code`, which is right for a chained manual-mode flow too.

Step 11.4 Run `uv run python -m pytest tests/unit/test_web_js_oauth_chained_cc.py tests/unit/test_web_js_oauth_flow_guard.py tests/unit/test_web_js_accounts_reauth.py -q`. Expected: pass, including the three pre-existing guard tests.

Step 11.5 Commit.

git add jacked/data/web/js/components/oauth-flows.js tests/unit/test_web_js_oauth_chained_cc.py
git commit -m "fix(web): keep polling the chained Claude Code token flow after a re-auth"

Task 12: Observation resolves from the authority; mirror divergence is evidence, not a conflict

Traces to: AC 17 (appended from Task 10 evidence)

Files

• Modify: jacked/credentials/resolver.py — require_mirror_consensus keyword; _resolve_from_authority, _mirror_evidence
• Modify: jacked/credentials/runtime.py — resolve_active_identity observes with consensus off
• Test: tests/unit/test_credential_resolver.py — five new tests
• Test: tests/unit/test_credential_runtime.py — divergent darwin mirror end to end

Interfaces

• Produces: CanonicalCredentialResolver(capability, stores, *, require_mirror_consensus=True); mirror evidence required_mirror:<name>:ok|divergent|missing|<status>.
• Consumes: the transaction engine's own consensus read is untouched.

Why: on macOS the certified topology is Keychain authority plus the global credential file as a required mirror. Claude Code refreshes tokens in the Keychain only, and a jacked switch that failed at the Keychain step can leave the file holding another account, so the two legitimately drift between jacked switches. Today `resolve()` demands consensus across both for a plain observation, so any drift yields CONFLICT with no identity: the menu-bar pill goes blank and every consumer of `/api/auth/active-credential` sees nothing, even though the Keychain, which is what Claude Code reads, is unambiguous. Observed live on 2026-09-04: Keychain = account 4, file = account 5 (written 02:25), pill blank with the build certified. Consensus is the right rule when verifying that a write landed everywhere; it is the wrong rule for reporting what the runtime will use. The engine keeps its own consensus check; this task only changes observation.

Step 12.1 Write the failing tests. In tests/unit/test_credential_resolver.py (helpers `_payload(account_id)` and `_capability()` with authority locator "keychain" and required mirror locator "file" already exist):

def test_observation_resolves_from_the_authority_when_the_mirror_diverges() -> None:
    resolver = CanonicalCredentialResolver(
        _capability(),
        {
            "keychain": MemoryCredentialStore("keychain", _payload(1)),
            "file": MemoryCredentialStore("file", _payload(2)),
        },
        require_mirror_consensus=False,
    )

    observation = resolver.resolve()

    assert observation.state is ResolverState.RESOLVED
    assert observation.identity.account_id == 1
    assert "authority:keychain:ok" in observation.evidence
    assert "required_mirror:file:divergent" in observation.evidence


def test_observation_names_a_missing_mirror_but_still_resolves() -> None:
    resolver = CanonicalCredentialResolver(
        _capability(),
        {
            "keychain": MemoryCredentialStore("keychain", _payload(1)),
            "file": MemoryCredentialStore("file", None),
        },
        require_mirror_consensus=False,
    )

    observation = resolver.resolve()

    assert observation.state is ResolverState.RESOLVED
    assert observation.identity.account_id == 1
    assert "required_mirror:file:missing" in observation.evidence


def test_observation_agreeing_mirror_is_marked_ok() -> None:
    resolver = CanonicalCredentialResolver(
        _capability(),
        {
            "keychain": MemoryCredentialStore("keychain", _payload(1)),
            "file": MemoryCredentialStore("file", _payload(1)),
        },
        require_mirror_consensus=False,
    )

    observation = resolver.resolve()

    assert observation.state is ResolverState.RESOLVED
    assert "required_mirror:file:ok" in observation.evidence


def test_observation_still_fails_closed_on_the_authority() -> None:
    missing = CanonicalCredentialResolver(
        _capability(),
        {
            "keychain": MemoryCredentialStore("keychain", None),
            "file": MemoryCredentialStore("file", _payload(2)),
        },
        require_mirror_consensus=False,
    ).resolve()
    assert missing.state is ResolverState.MISSING

    unstamped = CredentialPayload.from_mapping({"claudeAiOauth": {"accessToken": "secret"}})
    unusable = CanonicalCredentialResolver(
        _capability(),
        {
            "keychain": MemoryCredentialStore("keychain", unstamped),
            "file": MemoryCredentialStore("file", _payload(2)),
        },
        require_mirror_consensus=False,
    ).resolve()
    assert unusable.state is ResolverState.UNUSABLE
    assert "identity:stamp-absent" in unusable.evidence


def test_consensus_default_is_unchanged() -> None:
    resolver = CanonicalCredentialResolver(
        _capability(),
        {
            "keychain": MemoryCredentialStore("keychain", _payload(1)),
            "file": MemoryCredentialStore("file", _payload(2)),
        },
    )

    assert resolver.resolve().state is ResolverState.CONFLICT

In tests/unit/test_credential_runtime.py (next to `test_resolve_active_identity_on_linux_reads_credential_file`; reuse its patching pattern and `_identity`):

def test_resolve_active_identity_reports_a_divergent_darwin_mirror_as_evidence(tmp_path: Path) -> None:
    """Keychain says account 4, the file mirror says 5: the runtime uses the Keychain."""
    from jacked.credentials.canonical import CredentialPayload
    from jacked.credentials.store import MemoryCredentialStore

    home = tmp_path
    (home / ".claude").mkdir()
    (home / ".claude" / ".credentials.json").write_text(
        '{"_jackedAccountId": 5, "claudeAiOauth": {"accessToken": "five"}}', encoding="utf-8"
    )
    keychain = MemoryCredentialStore(
        "keychain",
        CredentialPayload.from_mapping({"_jackedAccountId": 4, "claudeAiOauth": {"accessToken": "four"}}),
    )
    with (
        mock.patch("jacked.credentials.runtime.detect_claude_identity", return_value=_identity()),
        mock.patch("jacked.credentials.runtime.Path.home", return_value=home),
        mock.patch("jacked.credentials.runtime.MacOSCredentialStore", return_value=keychain),
    ):
        observation = resolve_active_identity()

    assert observation.state.value == "resolved"
    assert observation.identity.account_id == 4
    assert "required_mirror:global credential file:divergent" in observation.evidence

Step 12.2 Run `uv run python -m pytest tests/unit/test_credential_resolver.py tests/unit/test_credential_runtime.py -k "observation or consensus or divergent" -v`. Expected: TypeError on the unexpected keyword for the new resolver tests; the runtime test fails with state "conflict".

Step 12.3 In jacked/credentials/resolver.py change the constructor and split `resolve()` into two small methods so each stays under 50 lines:

    def __init__(
        self,
        capability: CredentialCapability,
        stores: Mapping[str, CredentialStore],
        *,
        require_mirror_consensus: bool = True,
    ) -> None:
        self._capability = capability
        self._stores = stores
        self._require_mirror_consensus = require_mirror_consensus

    def resolve(self) -> ResolverObservation:
        if self._require_mirror_consensus:
            return self._resolve_by_consensus()
        return self._resolve_from_authority()

`_resolve_by_consensus` is today's `resolve()` body, unchanged. Add:

    def _resolve_from_authority(self) -> ResolverObservation:
        """Report what the runtime will use: the authority decides, mirrors are evidence.

        Claude Code refreshes the authority alone, so a required mirror can lag
        it between jacked writes. Hiding the identity behind CONFLICT on every
        lag blanked the menu bar. Write verification keeps demanding consensus
        in the transaction engine; only observation takes this path.
        """
        authority = self._capability.authority
        store = self._stores.get(authority.locator)
        if store is None:
            return ResolverObservation(
                ResolverState.UNUSABLE, CredentialIdentity(), (f"missing-adapter:{authority.locator}",)
            )
        result = store.read()
        evidence = [f"{authority.role.value}:{authority.name}:{result.status.value}"]
        if result.status is StoreStatus.MISSING:
            return ResolverObservation(ResolverState.MISSING, CredentialIdentity(), tuple(evidence))
        if result.status is not StoreStatus.OK or result.payload is None:
            return ResolverObservation(ResolverState.UNUSABLE, CredentialIdentity(), tuple(evidence))
        evidence.extend(self._mirror_evidence(result.payload))
        identity = result.payload.identity
        if identity.account_id is None:
            return ResolverObservation(
                ResolverState.UNUSABLE, CredentialIdentity(), (*evidence, "identity:stamp-absent")
            )
        return ResolverObservation(ResolverState.RESOLVED, identity, tuple(evidence))

    def _mirror_evidence(self, authority_payload: CredentialPayload) -> list[str]:
        evidence = []
        for declaration in self._capability.required_mirrors:
            store = self._stores.get(declaration.locator)
            if store is None:
                evidence.append(f"{declaration.role.value}:{declaration.name}:missing-adapter")
                continue
            result = store.read()
            if result.status is StoreStatus.OK and result.payload is not None:
                same = (
                    result.payload.digest == authority_payload.digest
                    and result.payload.identity == authority_payload.identity
                )
                verdict = "ok" if same else "divergent"
            else:
                verdict = result.status.value
            evidence.append(f"{declaration.role.value}:{declaration.name}:{verdict}")
        return evidence

Import `CredentialPayload` in resolver.py if it is not already imported there (it is referenced in a type annotation; a `TYPE_CHECKING` import is fine).

Step 12.4 In jacked/credentials/runtime.py `resolve_active_identity`, construct the resolver with `require_mirror_consensus=False`:

    observation = CanonicalCredentialResolver(
        resolution.capability, stores, require_mirror_consensus=False
    ).resolve()

Step 12.5 Run `uv run python -m pytest tests/unit/test_credential_resolver.py tests/unit/test_credential_runtime.py tests/unit/test_credential_transactions.py tests/unit -k "session_observer or credential" -q`. Expected: pass (the engine's tests are unaffected; the existing `test_resolver_reports_conflict_instead_of_precedence_guess` still passes because the default is consensus).

Step 12.6 Commit.

git add jacked/credentials/resolver.py jacked/credentials/runtime.py tests/unit/test_credential_resolver.py tests/unit/test_credential_runtime.py
git commit -m "fix(credentials): observe the active identity from the authority; mirror drift is evidence"