diff --git a/changelog.d/2026-07-29-claude-active-model-observation.md b/changelog.d/2026-07-29-claude-active-model-observation.md new file mode 100644 index 0000000..b0ccec4 --- /dev/null +++ b/changelog.d/2026-07-29-claude-active-model-observation.md @@ -0,0 +1,15 @@ +### Fixed + +- Observe the active model on a Claude Code host from the live session's own transcript. Claude Code does not export the model to the environment, so `active_model` resolved to `unknown`, no Claude session was ever `governance_ready`, and every governance route failed closed with `unknown_family` — the only host family with that gap. Callers worked around it by hand-authoring `primary`, which invited an invented `session_identifier` that conflicted with the strongly observed one and turned every route into a configuration error. + +### Changed + +- Document that `primary` should be sent as `{}` and that a `session_identifier` must never be fabricated by a caller, in both the coordinator request schema and the `intent-check` skill. Complete explicit configuration remains supported for a host that genuinely cannot be observed; what is prohibited is inventing a signal the caller does not have. + +### Security + +- The transcript observation is bounded and fails closed: strict lowercase-UUID session keys before any path join, owner/symlink/hardlink/permission checks reused unchanged from the Codex rollout path, post-open `fstat` re-validation against the pre-open identity, a bounded tail read, and strict JSON decoding. A model that is not anthropic-shaped is rejected rather than allowed to reassign `primary_family`, so a forged record cannot defeat reviewer family-exclusion. Model validation is by shape, not by a pinned model list, so future models resolve without a code change. + +- Identity on a Claude host is now observation-only: every identity field is recorded non-empty, so none of them can be supplied through `AGENT_COLLAB_*` or an explicit `primary`. A supplied value that disagrees with observation registers as a conflict rather than an override. Previously an unobserved field was left empty, and the overlay fills an empty field — so configuration alone could make a wholly unobserved session governance-ready, including by pairing a filled model with a filled session identifier. A nonempty session identifier that fails the UUID contract also now fails closed rather than reading as merely absent. + +- Same-uid integrity is explicitly out of scope. The transcript is a weaker signal than the `CLAUDE_CODE_SESSION_ID` it is keyed by: that value is inherited at process start and a sibling same-uid process cannot rewrite it, whereas the transcript can be modified mid-flight. The identity re-checks detect append races and replacement, not a same-size in-place rewrite or an append-then-truncate. diff --git a/plugins/agent-collab/README.md b/plugins/agent-collab/README.md index 404aa67..dad64f8 100644 --- a/plugins/agent-collab/README.md +++ b/plugins/agent-collab/README.md @@ -10,8 +10,12 @@ reference; the repository README carries the purpose and governance narrative. Current: **4.5.4** It resolves `primary_id`, `primary_family`, `active_model`, `host_runtime`, and -`session_identifier` from the current host or explicit configuration. ZCode -model changes are re-observed before routing; OpenCode is a runtime, while the +`session_identifier` from the current host or explicit configuration. On a +Claude Code host the active model is observed from the live session's own +transcript, because that host does not export it to the environment; the +observation is bounded, fails closed, and cannot reassign the host-derived +family. +ZCode model changes are re-observed before routing; OpenCode is a runtime, while the selected model determines artifact family. The current `opencode-go/glm-5.2` preset therefore records **Zhipu** provenance. Kimi models record **Moonshot** provenance. Exact provider/model segments produce @@ -468,15 +472,35 @@ nonblank content continues with an independence warning. `primary` is an object containing any subset of the string fields `primary_id`, `primary_family`, `active_model`, `host_runtime`, -`session_identifier`, `opencode_model`, and `async_inbox`. Use `{}` for host -observation; use explicit fields only when the host cannot expose strong -current-session signals. Strong observed session identity, model, family, -runtime, and session identifier are authoritative. Explicit values may fill +`session_identifier`, `opencode_model`, and `async_inbox`. **Send `{}` and let +the host be observed.** Supply an explicit field only when the host genuinely +cannot expose that signal, and only for that one field. Strong observed session +identity, model, family, runtime, and session identifier are authoritative. +Explicit values may fill missing signals only; conflicting current-session and explicit identity is a configuration error rather than an override. Complete explicit configuration is governance-eligible only when its id, family, active-model lineage, runtime, and session identifier are mutually consistent. +**Never invent a `session_identifier`.** A caller cannot know a session +identifier the host did not expose, so a constructed value is invented rather +than observed. Where the identifier IS observed, an invented one does not +override it: it registers as an identity conflict and turns every route into a +configuration error. If a governance route reports incomplete identity, do not +assemble an identity to satisfy it -- fix the missing observed signal, or let +the route fail closed. This does not withdraw complete explicit configuration +for a host that genuinely cannot be observed; it forbids fabricating a signal +the caller does not have. + +**On a detected Claude host, identity is observation-only.** Every Claude +identity field is recorded non-empty, so none of `primary_id`, +`primary_family`, `active_model`, `host_runtime`, or `session_identifier` can be +supplied through `AGENT_COLLAB_*` or an explicit `primary`; a supplied value +that disagrees registers as a conflict rather than an override. The active model +is observed from the live session transcript, and only a resolved transcript +yields a governance-ready Claude profile. This scoping is specific to that host +and does not change the fill behavior of the other host runtimes. + Exact row contracts are: | Contract | Exact `row` shape | diff --git a/plugins/agent-collab/host_policy.py b/plugins/agent-collab/host_policy.py index 731497d..eaf453b 100644 --- a/plugins/agent-collab/host_policy.py +++ b/plugins/agent-collab/host_policy.py @@ -125,6 +125,14 @@ ) _CODEX_YEAR_RE = re.compile(r"^[0-9]{4}$") _CODEX_MONTH_DAY_RE = re.compile(r"^[0-9]{2}$") +_CLAUDE_SESSION_ID_RE = re.compile( + r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" +) +_CLAUDE_PROJECT_ENTRY_LIMIT = 100_000 +_CLAUDE_MODEL_LENGTH_LIMIT = 128 +# Host sentinel Claude Code writes for synthesized/error turns. Not a model +# name, so it never goes stale as models ship or are deprecated. +_CLAUDE_SYNTHETIC_MODEL = "" _CODEX_ROLLOUT_ENTRY_LIMIT = 100_000 _CODEX_ROLLOUT_LINE_LIMIT = 4 * 1024 * 1024 _CODEX_ROLLOUT_SCAN_LIMIT = 64 * 1024 * 1024 @@ -568,6 +576,219 @@ def _codex_rollout_model(thread_id: str) -> tuple[str, str]: return "invalid", "" +def _claude_projects_root() -> Path | None: + # The POSIX capability guard governs BOTH root selections, not just the + # passwd-home one. Whichever root is chosen is then checked by an ownership + # predicate that calls os.getuid(), so a host lacking it must fail closed + # here; returning a configured root first would raise AttributeError from + # deeper in, and AttributeError is not in the caught tuple, so it would + # propagate out of resolve_profile instead of failing closed. + if _pwd is None or not hasattr(os, "getuid"): + return None + # Claude Code stores session data beneath CLAUDE_CONFIG_DIR when that is + # set, so a resolver pinned to the passwd home would read those supported + # installations as having no transcript at all -- authoritative unknown, and + # therefore never governance-ready. The directory still has to pass the same + # ownership and permission predicate as the default root, so a hostile value + # redirects to a location that fails closed rather than one that is trusted. + configured = os.environ.get("CLAUDE_CONFIG_DIR", "").strip() + if configured: + if ( + "\0" in configured + or len(configured) > 4096 + or not Path(configured).is_absolute() + ): + return None + return Path(configured) / "projects" + try: + home = _pwd.getpwuid(os.getuid()).pw_dir + except (KeyError, OSError): + return None + if ( + type(home) is not str + or not home + or "\0" in home + or len(home) > 4096 + or not Path(home).is_absolute() + ): + return None + return Path(home) / ".claude" / "projects" + + +def _claude_transcript_candidates( + root: Path, session_id: str +) -> list[tuple[Path, os.stat_result]]: + if not _safe_codex_directory(root): + raise ValueError("Claude projects root identity is unsafe") + entry_count = 0 + candidates: list[tuple[Path, os.stat_result]] = [] + name = f"{session_id}.jsonl" + try: + with os.scandir(root) as iterator: + for entry in iterator: + entry_count += 1 + # Bound incrementally so an oversized tree cannot be fully + # materialized before the limit is enforced. + if entry_count > _CLAUDE_PROJECT_ENTRY_LIMIT: + raise ValueError("Claude projects tree exceeds the entry bound") + project = Path(entry.path) + # A project directory that is a symlink, foreign-owned, or + # group/other-writable is SKIPPED rather than fatal: skipping + # denies an attacker-planted redirect the chance to supply the + # transcript at all, while one stray directory elsewhere in the + # tree cannot deny resolution for every session. A hostile real + # directory holding the same session name still surfaces as a + # second candidate and fails closed on ambiguity below. + if not _safe_codex_directory(project): + continue + path = project / name + try: + identity = path.lstat() + except OSError: + continue + if not _safe_codex_rollout_identity(identity): + raise ValueError("Claude transcript identity is unsafe") + candidates.append((path, identity)) + except OSError as exc: + raise ValueError("Claude projects directory is unreadable") from exc + return candidates + + +def _claude_model_from_transcript( + path: Path, observed: os.stat_result, session_id: str +) -> str: + flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) + fd = os.open(os.fspath(path), flags) + try: + opened = os.fstat(fd) + if ( + not _safe_codex_rollout_identity(opened) + or (opened.st_dev, opened.st_ino) != (observed.st_dev, observed.st_ino) + ): + raise ValueError("Claude transcript changed before open") + captured_size = opened.st_size + + model = "" + for raw in reversed(_codex_rollout_window(fd, captured_size).splitlines()): + record = _decode_codex_rollout_line(raw) + if record.get("type") != "assistant": + continue + sidechain = record.get("isSidechain") + if sidechain is not None and type(sidechain) is not bool: + raise ValueError("Claude transcript sidechain flag is malformed") + if sidechain: + # Subagent turns can run a different model than the session. + continue + if record.get("sessionId") != session_id: + raise ValueError("Claude transcript session identity does not match") + message = record.get("message") + if not isinstance(message, Mapping): + raise ValueError("Claude transcript message is malformed") + if message.get("role") != "assistant": + raise ValueError("Claude transcript message role is malformed") + candidate = message.get("model") + if type(candidate) is not str: + raise ValueError("Claude turn model identity is unproven") + # `` is a host sentinel written for synthesized/error + # turns, not a model claim. Skip it and keep scanning; anything + # else that fails the anthropic family shape IS a claim that + # contradicts the observed host, so it fails closed instead. + if candidate.strip() == _CLAUDE_SYNTHETIC_MODEL: + continue + if len(candidate) > _CLAUDE_MODEL_LENGTH_LIMIT or _model_family_signals( + candidate + ) != {"anthropic"}: + raise ValueError("Claude turn model identity is unproven") + model = candidate.strip() + break + if not model: + # A transcript carrying no qualifying assistant turn yet -- a new + # session, or a tail window holding only user/tool records -- is an + # ABSENT observation, not a contradictory one. Raising here would + # set a false identity-conflict signal on a healthy session. This + # deliberately diverges from the Codex rollout precedent, whose + # first record is always session metadata. + return "" + completed = os.fstat(fd) + if ( + not _safe_codex_rollout_identity(completed) + or ( + completed.st_dev, + completed.st_ino, + completed.st_uid, + stat.S_IFMT(completed.st_mode), + completed.st_mode & 0o777, + completed.st_nlink, + ) + != ( + opened.st_dev, + opened.st_ino, + opened.st_uid, + stat.S_IFMT(opened.st_mode), + opened.st_mode & 0o777, + opened.st_nlink, + ) + or completed.st_size != captured_size + ): + raise ValueError("Claude transcript changed during identity proof") + return model + finally: + os.close(fd) + + +def _claude_transcript_model(session_id: str) -> tuple[str, str]: + """Observe the live Claude session's model from its own transcript. + + Claude Code does not export the active model to the environment, so the + only current-session signal is the transcript it writes. + + Same-uid integrity is explicitly OUT OF SCOPE. This is best-effort + anti-confusion, NOT a forgery-resistant attestation: any process running as + this user can append to, rewrite, or truncate the transcript at any time. + It is a WEAKER signal than the `CLAUDE_CODE_SESSION_ID` it is keyed by -- + that value is inherited at process start and a sibling same-uid process + cannot rewrite it, whereas this file can be modified mid-flight. The + identity re-checks below detect ordinary append races and replacement; they + do NOT detect a same-size in-place rewrite or an append-then-truncate. + + What the observation therefore may NOT do is reassign identity: a model + that is not anthropic-shaped fails closed rather than changing the + host-derived `primary_family`, so a forged record cannot defeat reviewer + family-exclusion. The residual is asserting a different model WITHIN the + host family, which affects provenance accuracy, not independence. + """ + if not session_id: + return "absent", "" + if _CLAUDE_SESSION_ID_RE.fullmatch(session_id) is None: + # A NONEMPTY identifier that fails the UUID contract is not merely an + # unresolvable session: the malformed value still reaches the profile as + # `session_identifier`, where it is non-"unknown" and therefore counts + # toward governance eligibility. Reporting "absent" here would let it + # combine with a filled model to manufacture a governance-ready profile + # for a session that cannot exist. Fail closed instead. + return "invalid", "" + root = _claude_projects_root() + if root is None: + return "absent", "" + try: + root.lstat() + except FileNotFoundError: + return "absent", "" + except OSError: + return "invalid", "" + try: + candidates = _claude_transcript_candidates(root, session_id) + if not candidates: + return "absent", "" + if len(candidates) != 1: + raise ValueError("Claude transcript identity is ambiguous") + path, identity = candidates[0] + model = _claude_model_from_transcript(path, identity, session_id) + return ("ok", model) if model else ("absent", "") + except (OSError, RuntimeError, TypeError, ValueError): + return "invalid", "" + + def _environment_profile() -> dict[str, str]: env = os.environ overrides = { @@ -580,13 +801,60 @@ def _environment_profile() -> dict[str, str]: } detected: list[dict[str, str]] = [] if env.get("CLAUDE_CODE_SESSION_ID") or env.get("CLAUDE_CODE_ENTRYPOINT"): + session_identifier = env.get("CLAUDE_CODE_SESSION_ID", "") + environment_model = env.get("CLAUDE_CODE_MODEL", "").strip() + transcript_state, transcript_model = _claude_transcript_model( + session_identifier + ) + conflict = transcript_state == "invalid" + if transcript_state == "ok": + active_model = transcript_model + if ( + environment_model + and environment_model.casefold() != transcript_model.casefold() + ): + conflict = True + elif transcript_state == "invalid": + active_model = "unknown" + elif session_identifier: + # A transcript was EXPECTED for this session and did not resolve. + # Record an AUTHORITATIVE unknown rather than an empty value: an + # empty model is fillable by `AGENT_COLLAB_ACTIVE_MODEL` or by a + # caller-supplied `primary.active_model`, which would manufacture a + # governance-ready profile for a session whose model was never + # observed. A non-empty "unknown" cannot be filled -- an explicit + # value that disagrees with it registers as a conflict instead. + active_model = "unknown" + else: + # Claude detected by entrypoint alone: there is no session + # identifier, so no transcript can be located and no current-session + # identity can be observed. A genuine environment-exported model is + # still recorded; the invariant below makes the missing identifier + # authoritative, which keeps governance closed. + active_model = environment_model + # CLASS INVARIANT -- identity on a Claude host is OBSERVATION-ONLY. + # + # Every identity field is recorded non-empty, because + # `_overlay_profile_values` fills a field only when the observed value is + # empty (`if not current: continue`). A field that is never empty can + # never be supplied by `AGENT_COLLAB_*` or by an explicit `primary`; a + # supplied value that disagrees registers as a conflict instead of an + # override. Both the environment and explicit paths funnel through that + # same overlay, so this one invariant closes both. + # + # This deliberately closes the FAMILY of "an unobserved field is empty, + # and an empty field is fillable" defects rather than its members: the + # unfillable-model and unfillable-session-identifier fixes were two + # instances of it, and patching instances one at a time left a further + # paired-field bypass open each time. detected.append( { "primary_id": "claude", "primary_family": "anthropic", - "active_model": env.get("CLAUDE_CODE_MODEL", ""), + "active_model": active_model or "unknown", "host_runtime": "claude-code", - "session_identifier": env.get("CLAUDE_CODE_SESSION_ID", ""), + "session_identifier": session_identifier or "unknown", + "_identity_conflict": "1" if conflict else "", } ) if env.get("CODEX_THREAD_ID"): diff --git a/plugins/agent-collab/skills/intent-check/SKILL.md b/plugins/agent-collab/skills/intent-check/SKILL.md index 7c16b72..e4d6948 100644 --- a/plugins/agent-collab/skills/intent-check/SKILL.md +++ b/plugins/agent-collab/skills/intent-check/SKILL.md @@ -36,6 +36,16 @@ high effort to preserve that sealed role's mandatory floor. constraints, success criteria, and stop conditions. 3. Capture artifact author-model provenance. If primary or artifact family is unknown, fail closed; do not accept a caller-provided family assertion. + Send `primary` as `{}` on any host that exposes strong identity signals, so + the host is observed; on a detected Claude host identity is observation-only + and no explicit identity field is accepted at all. Never invent a + `session_identifier` the host did not expose: where the real identifier is + observed, the invented one registers as an identity conflict that fails every + route. On a host that genuinely exposes no identity, a COMPLETE and mutually + consistent explicit configuration remains the supported path -- that is + configuration, not fabrication. What must never happen is assembling a + partial or guessed identity to satisfy a route that reported incomplete + identity; let that fail closed. 4. Ask the dynamically selected independent reviewer to return exactly: ```text diff --git a/skill-specs/intent-check.md b/skill-specs/intent-check.md index 700a8df..fea35d7 100644 --- a/skill-specs/intent-check.md +++ b/skill-specs/intent-check.md @@ -29,6 +29,16 @@ high effort to preserve that sealed role's mandatory floor. constraints, success criteria, and stop conditions. 3. Capture artifact author-model provenance. If primary or artifact family is unknown, fail closed; do not accept a caller-provided family assertion. + Send `primary` as `{}` on any host that exposes strong identity signals, so + the host is observed; on a detected Claude host identity is observation-only + and no explicit identity field is accepted at all. Never invent a + `session_identifier` the host did not expose: where the real identifier is + observed, the invented one registers as an identity conflict that fails every + route. On a host that genuinely exposes no identity, a COMPLETE and mutually + consistent explicit configuration remains the supported path -- that is + configuration, not fabrication. What must never happen is assembling a + partial or guessed identity to satisfy a route that reported incomplete + identity; let that fail closed. 4. Ask the dynamically selected independent reviewer to return exactly: ```text diff --git a/tests/test_claude_transcript_model.py b/tests/test_claude_transcript_model.py new file mode 100644 index 0000000..17cc688 --- /dev/null +++ b/tests/test_claude_transcript_model.py @@ -0,0 +1,611 @@ +"""Bounded observation of the live Claude session model from its transcript. + +Claude Code does not export the active model to the environment, so before this +observation existed a Claude host resolved ``active_model='unknown'`` and every +governance route failed closed with ``unknown_family``. Callers then hand-filled +``primary``, fabricated ``session_identifier`` (which a caller cannot know), and +tripped the identity-conflict guard. + +These tests pin the mechanism AND its fail-closed posture. The observation is +same-uid best-effort anti-confusion, not a forgery-resistant attestation. +""" + +from __future__ import annotations + +import importlib.util +import json +import os +import stat +import sys +import tempfile +import unittest +from pathlib import Path +from unittest import mock + + +ROOT = Path(__file__).resolve().parents[1] +PLUGIN = ROOT / "plugins" / "agent-collab" + +SESSION = "07a2aa37-f15c-4604-a184-d969ef9d01ac" +# A session id that must NOT exist in the real ~/.claude, so relocation tests +# cannot be satisfied by the passwd-home fallback. +RELOCATED_SESSION = "11111111-2222-3333-4444-555555555555" + + +def _load(name: str, path: Path): + spec = importlib.util.spec_from_file_location(name, path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +host_policy = _load("claude_transcript_host_policy", PLUGIN / "host_policy.py") + + +def _assistant(model: str, *, session: str = SESSION, sidechain=False) -> str: + record = { + "type": "assistant", + "sessionId": session, + "isSidechain": sidechain, + "message": {"role": "assistant", "model": model}, + } + return json.dumps(record) + "\n" + + +class ClaudeTranscriptModelTests(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.home = Path(self._tmp.name) + self.projects = self.home / ".claude" / "projects" + self.projects.mkdir(parents=True) + os.chmod(self.projects, 0o700) + self.addCleanup(self._tmp.cleanup) + + def _project(self, name: str = "-Users-x-repo") -> Path: + path = self.projects / name + path.mkdir(exist_ok=True) + os.chmod(path, 0o700) + return path + + def _write(self, body: str, *, project: str = "-Users-x-repo", session: str = SESSION) -> Path: + path = self._project(project) / f"{session}.jsonl" + path.write_text(body, encoding="utf-8") + os.chmod(path, 0o600) + return path + + def _resolve(self, session: str = SESSION): + with mock.patch.object( + host_policy, "_claude_projects_root", return_value=self.projects + ): + return host_policy._claude_transcript_model(session) + + # -- happy path ---------------------------------------------------------- + + def test_resolves_newest_assistant_model(self): + self._write(_assistant("claude-sonnet-5") + _assistant("claude-opus-5")) + self.assertEqual(self._resolve(), ("ok", "claude-opus-5")) + + def test_unreleased_model_name_resolves_without_an_allowlist(self): + """No pinned model list: a future model must work with no code edit.""" + self._write(_assistant("claude-nextgen-9-20990101")) + self.assertEqual(self._resolve(), ("ok", "claude-nextgen-9-20990101")) + + # -- sentinel and non-qualifying records --------------------------------- + + def test_synthetic_sentinel_is_skipped_not_accepted(self): + self._write(_assistant("claude-opus-5") + _assistant("")) + self.assertEqual(self._resolve(), ("ok", "claude-opus-5")) + + def test_sidechain_record_is_skipped(self): + self._write(_assistant("claude-opus-5") + _assistant("claude-haiku-4-5", sidechain=True)) + self.assertEqual(self._resolve(), ("ok", "claude-opus-5")) + + def test_non_bool_sidechain_flag_fails_closed(self): + self._write(_assistant("claude-opus-5").replace('"isSidechain": false', '"isSidechain": "true"')) + self.assertEqual(self._resolve(), ("invalid", "")) + + def test_no_assistant_record_is_absent_not_invalid(self): + self._write(json.dumps({"type": "user", "sessionId": SESSION}) + "\n") + self.assertEqual(self._resolve(), ("absent", "")) + + # -- family binding: the load-bearing bypass defence --------------------- + + def test_cross_family_model_fails_closed_and_cannot_reassign_family(self): + """A forged cross-family record must not become an OpenAI primary.""" + self._write(_assistant("claude-opus-5") + _assistant("gpt-4o")) + self.assertEqual(self._resolve(), ("invalid", "")) + + def test_google_and_xai_model_claims_also_fail_closed(self): + for model in ("gemini-3.1-pro", "grok-4.5", "glm-5.2"): + with self.subTest(model=model): + self._write(_assistant(model)) + self.assertEqual(self._resolve(), ("invalid", "")) + + def test_oversized_model_string_fails_closed(self): + self._write(_assistant("claude-" + "a" * 200)) + self.assertEqual(self._resolve(), ("invalid", "")) + + def test_session_id_mismatch_inside_record_fails_closed(self): + self._write(_assistant("claude-opus-5", session="ffffffff-0000-0000-0000-000000000000")) + self.assertEqual(self._resolve(), ("invalid", "")) + + # -- path and identity hardening ----------------------------------------- + + def test_empty_session_id_is_absent(self): + self.assertEqual(self._resolve(""), ("absent", "")) + + def test_malformed_session_id_fails_closed_rather_than_reading_absent(self): + """A nonempty non-UUID id still reaches the profile, so it must conflict.""" + for bad in ( + "../../etc/passwd", + "*", + "07A2AA37-F15C-4604-A184-D969EF9D01AC", + "a/b", + "not-a-uuid-at-all", + ): + with self.subTest(session=bad): + self.assertEqual(self._resolve(bad), ("invalid", "")) + + def test_missing_transcript_is_absent(self): + self._project() + self.assertEqual(self._resolve(), ("absent", "")) + + def test_ambiguous_duplicate_across_projects_fails_closed(self): + self._write(_assistant("claude-opus-5"), project="-Users-x-repo") + self._write(_assistant("claude-sonnet-5"), project="-Users-x-other") + self.assertEqual(self._resolve(), ("invalid", "")) + + def test_symlinked_transcript_is_rejected(self): + real = self.home / "planted.jsonl" + real.write_text(_assistant("claude-opus-5"), encoding="utf-8") + os.chmod(real, 0o600) + os.symlink(real, self._project() / f"{SESSION}.jsonl") + self.assertEqual(self._resolve(), ("invalid", "")) + + def test_group_writable_transcript_is_rejected(self): + path = self._write(_assistant("claude-opus-5")) + # Deliberately insecure fixture: the assertion IS that this is refused. + path.chmod(0o620) # group-write transcript + self.assertEqual(self._resolve(), ("invalid", "")) + + def test_hardlinked_transcript_is_rejected(self): + path = self._write(_assistant("claude-opus-5")) + os.link(path, self.home / "second-link.jsonl") + self.assertEqual(self._resolve(), ("invalid", "")) + + def test_symlinked_project_directory_is_skipped_not_followed(self): + outside = self.home / "attacker" + outside.mkdir() + os.chmod(outside, 0o700) + planted = outside / f"{SESSION}.jsonl" + planted.write_text(_assistant("gpt-4o"), encoding="utf-8") + os.chmod(planted, 0o600) + os.symlink(outside, self.projects / "-evil") + self.assertEqual(self._resolve(), ("absent", "")) + + def test_group_writable_project_directory_is_skipped(self): + self._write(_assistant("claude-opus-5"), project="-Users-x-repo") + loose = self.projects / "-loose" + loose.mkdir() + # Plant a same-session transcript INSIDE the loose directory, so the + # permission is the only thing under test. If the directory were + # honoured this would be a second candidate and resolution would fail + # closed on ambiguity; resolution succeeding is what proves the + # directory was skipped. Without this the directory contributes no + # candidate either way and the assertion holds for the wrong reason. + planted = loose / f"{SESSION}.jsonl" + planted.write_text(_assistant("claude-sonnet-5"), encoding="utf-8") + planted.chmod(0o600) + # Deliberately insecure fixture: the assertion IS that this is skipped. + loose.chmod(0o777) # world-write project directory + self.assertEqual(self._resolve(), ("ok", "claude-opus-5")) + + def test_secure_duplicate_project_directory_is_not_skipped(self): + """Control for the test above: identical fixture, secure permissions. + + Proves the skip is caused by the permission and nothing else -- with a + secure directory the planted transcript IS a second candidate and + resolution fails closed on ambiguity. + """ + self._write(_assistant("claude-opus-5"), project="-Users-x-repo") + secure = self.projects / "-secure-dupe" + secure.mkdir() + planted = secure / f"{SESSION}.jsonl" + planted.write_text(_assistant("claude-sonnet-5"), encoding="utf-8") + planted.chmod(0o600) + secure.chmod(0o700) + self.assertEqual(self._resolve(), ("invalid", "")) + + def test_malformed_json_fails_closed(self): + self._write("{not json at all\n") + self.assertEqual(self._resolve(), ("invalid", "")) + + def test_incomplete_final_line_fails_closed(self): + self._write(_assistant("claude-opus-5").rstrip("\n")) + self.assertEqual(self._resolve(), ("invalid", "")) + + +class ClaudeProfileWiringTests(unittest.TestCase): + """The observation must reach resolve_profile with the Codex state contract.""" + + def _profile(self, state, model, env=None): + environ = {"CLAUDE_CODE_SESSION_ID": SESSION, "CLAUDE_CODE_ENTRYPOINT": "cli"} + environ.update(env or {}) + with mock.patch.dict(os.environ, environ, clear=True), mock.patch.object( + host_policy, "_claude_transcript_model", return_value=(state, model) + ): + return host_policy.resolve_profile(None) + + def test_observed_model_makes_a_claude_host_governance_ready(self): + profile = self._profile("ok", "claude-opus-5") + self.assertEqual(profile.active_model, "claude-opus-5") + self.assertEqual(profile.primary_family, "anthropic") + self.assertFalse(profile.identity_conflict) + self.assertTrue(profile.governance_ready) + + def test_invalid_observation_is_unknown_and_conflicting(self): + profile = self._profile("invalid", "") + self.assertEqual(profile.active_model, "unknown") + self.assertTrue(profile.identity_conflict) + self.assertFalse(profile.governance_ready) + + def test_absent_observation_preserves_prior_environment_behaviour(self): + profile = self._profile("absent", "") + self.assertEqual(profile.active_model, "unknown") + self.assertFalse(profile.identity_conflict) + self.assertFalse(profile.governance_ready) + + def test_environment_model_agreeing_with_observation_is_not_a_conflict(self): + profile = self._profile("ok", "claude-opus-5", {"CLAUDE_CODE_MODEL": "Claude-Opus-5"}) + self.assertFalse(profile.identity_conflict) + + def test_environment_model_disagreeing_with_observation_conflicts(self): + profile = self._profile("ok", "claude-opus-5", {"CLAUDE_CODE_MODEL": "claude-sonnet-5"}) + self.assertTrue(profile.identity_conflict) + self.assertFalse(profile.governance_ready) + + def test_explicit_caller_model_disagreeing_with_observation_conflicts(self): + """A hand-authored `primary` cannot override strong observation.""" + environ = {"CLAUDE_CODE_SESSION_ID": SESSION, "CLAUDE_CODE_ENTRYPOINT": "cli"} + with mock.patch.dict(os.environ, environ, clear=True), mock.patch.object( + host_policy, "_claude_transcript_model", return_value=("ok", "claude-opus-5") + ): + profile = host_policy.resolve_profile({"active_model": "claude-sonnet-5"}) + self.assertTrue(profile.identity_conflict) + self.assertFalse(profile.governance_ready) + + def test_fabricated_session_identifier_still_conflicts(self): + """The original defect: a caller-invented session id is never an override.""" + environ = {"CLAUDE_CODE_SESSION_ID": SESSION, "CLAUDE_CODE_ENTRYPOINT": "cli"} + with mock.patch.dict(os.environ, environ, clear=True), mock.patch.object( + host_policy, "_claude_transcript_model", return_value=("ok", "claude-opus-5") + ): + profile = host_policy.resolve_profile( + {"session_identifier": "11111111-1111-1111-1111-111111111111"} + ) + self.assertEqual(profile.session_identifier, SESSION) + self.assertTrue(profile.identity_conflict) + + +class ClaudeUnfillableIdentityTests(unittest.TestCase): + """An unobserved Claude session must not be made governance-ready by config. + + These exercise `resolve_profile` end to end against real files, WITHOUT + mocking `_claude_transcript_model`, so they prove the state labels cannot be + manoeuvred into eligibility rather than merely restating the wiring. + """ + + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.home = Path(self._tmp.name) + self.projects = self.home / ".claude" / "projects" + self.projects.mkdir(parents=True) + os.chmod(self.projects, 0o700) + self.addCleanup(self._tmp.cleanup) + + def _write(self, body: str, *, session: str = SESSION) -> None: + project = self.projects / "-Users-x-repo" + project.mkdir(exist_ok=True) + os.chmod(project, 0o700) + path = project / f"{session}.jsonl" + path.write_text(body, encoding="utf-8") + os.chmod(path, 0o600) + + def _profile(self, env, explicit=None): + environ = {"CLAUDE_CODE_ENTRYPOINT": "cli"} + environ.update(env) + with mock.patch.dict(os.environ, environ, clear=True), mock.patch.object( + host_policy, "_claude_projects_root", return_value=self.projects + ): + return host_policy.resolve_profile(explicit) + + def test_end_to_end_real_transcript_yields_governance_ready(self): + self._write(_assistant("claude-opus-5")) + profile = self._profile({"CLAUDE_CODE_SESSION_ID": SESSION}) + self.assertEqual(profile.active_model, "claude-opus-5") + self.assertEqual(profile.primary_family, "anthropic") + self.assertFalse(profile.identity_conflict) + self.assertTrue(profile.governance_ready) + + def test_absent_transcript_cannot_be_filled_by_environment(self): + profile = self._profile( + { + "CLAUDE_CODE_SESSION_ID": SESSION, + "AGENT_COLLAB_ACTIVE_MODEL": "claude-opus-5", + } + ) + self.assertEqual(profile.active_model, "unknown") + self.assertFalse(profile.governance_ready) + + def test_absent_transcript_cannot_be_filled_by_explicit_primary(self): + profile = self._profile( + {"CLAUDE_CODE_SESSION_ID": SESSION}, {"active_model": "claude-opus-5"} + ) + self.assertEqual(profile.active_model, "unknown") + self.assertFalse(profile.governance_ready) + + def test_malformed_session_with_filled_model_is_not_governance_ready(self): + profile = self._profile( + {"CLAUDE_CODE_SESSION_ID": "not-a-uuid-at-all"}, + {"active_model": "claude-opus-5"}, + ) + self.assertTrue(profile.identity_conflict) + self.assertFalse(profile.governance_ready) + + def test_cross_family_transcript_record_cannot_be_rescued_by_config(self): + self._write(_assistant("gpt-4o")) + profile = self._profile( + {"CLAUDE_CODE_SESSION_ID": SESSION}, {"active_model": "claude-opus-5"} + ) + self.assertEqual(profile.primary_family, "anthropic") + self.assertTrue(profile.identity_conflict) + self.assertFalse(profile.governance_ready) + + def test_unreadable_transcript_is_not_filled(self): + """Inject the failure at the open boundary rather than via mode bits. + + A mode-000 fixture does not block `os.open` when the suite runs as root, + so a permission-based fixture would silently stop testing anything in a + root container and the assertion would hold for the wrong reason. + """ + self._write(_assistant("claude-opus-5")) + real_open = os.open + + def denied(path, *args, **kwargs): + if str(path).endswith(f"{SESSION}.jsonl"): + raise PermissionError(13, "Permission denied") + return real_open(path, *args, **kwargs) + + with mock.patch.object(os, "open", denied): + profile = self._profile( + { + "CLAUDE_CODE_SESSION_ID": SESSION, + "AGENT_COLLAB_ACTIVE_MODEL": "claude-opus-5", + } + ) + self.assertFalse(profile.governance_ready) + + def test_configured_config_dir_is_honoured(self): + """CLAUDE_CONFIG_DIR relocates the projects root; it must be followed. + + Uses a session id that exists ONLY in the relocated root. Reusing the + live session id would let the passwd-home fallback resolve it from the + real ~/.claude, so the assertion would hold even with the honouring + removed -- the test would pass for the wrong reason. + """ + relocated = self.home / "custom-config" + projects = relocated / "projects" / "-Users-x-repo" + projects.mkdir(parents=True) + (relocated / "projects").chmod(0o700) + projects.chmod(0o700) + path = projects / f"{RELOCATED_SESSION}.jsonl" + path.write_text( + _assistant("claude-sonnet-5", session=RELOCATED_SESSION), encoding="utf-8" + ) + path.chmod(0o600) + environ = { + "CLAUDE_CODE_ENTRYPOINT": "cli", + "CLAUDE_CODE_SESSION_ID": RELOCATED_SESSION, + "CLAUDE_CONFIG_DIR": str(relocated), + } + with mock.patch.dict(os.environ, environ, clear=True): + profile = host_policy.resolve_profile(None) + self.assertEqual(profile.active_model, "claude-sonnet-5") + self.assertTrue(profile.governance_ready) + + def test_relative_config_dir_fails_closed(self): + """Build a REAL relative tree, so the validation is what fails it. + + A relative path pointing at nothing would fail closed whether or not the + absolute-path check existed, and the assertion would hold for the wrong + reason. Here the relative path genuinely resolves from the working + directory to a well-formed transcript, so dropping the check would make + this resolve and the test fail. + + An EMPTY value means "unset" and correctly falls back to the passwd + home, so it is not a malformed case. A NUL byte cannot be placed in + os.environ at all, so that guard is unreachable through this path and + exists as defence in depth for any other caller of the resolver. + """ + relative = Path("relative-config") + projects = self.home / relative / "projects" / "-Users-x-repo" + projects.mkdir(parents=True) + (self.home / relative / "projects").chmod(0o700) + projects.chmod(0o700) + path = projects / f"{RELOCATED_SESSION}.jsonl" + path.write_text( + _assistant("claude-sonnet-5", session=RELOCATED_SESSION), encoding="utf-8" + ) + path.chmod(0o600) + environ = { + "CLAUDE_CODE_ENTRYPOINT": "cli", + "CLAUDE_CODE_SESSION_ID": RELOCATED_SESSION, + "CLAUDE_CONFIG_DIR": str(relative), + } + cwd = os.getcwd() + os.chdir(self.home) + self.addCleanup(os.chdir, cwd) + with mock.patch.dict(os.environ, environ, clear=True): + profile = host_policy.resolve_profile(None) + self.assertEqual(profile.active_model, "unknown") + self.assertFalse(profile.governance_ready) + + def test_non_posix_host_fails_closed_with_a_configured_dir(self): + """The POSIX guard must gate BOTH root selections, not just the default. + + With a configured root returned before the guard, the ownership + predicate raises AttributeError from deeper in -- which is not in the + caught tuple, so it escapes resolve_profile rather than failing closed. + + Patching `_pwd` alone would NOT exercise this: the configured branch + never consults `_pwd`, so the assertion would hold for the wrong reason. + The hazard is specifically the missing `os.getuid`, so remove that + attribute for the duration and restore it unconditionally. + """ + configured = self.home / "configured" + (configured / "projects").mkdir(parents=True) + (configured / "projects").chmod(0o700) + environ = { + "CLAUDE_CODE_ENTRYPOINT": "cli", + "CLAUDE_CODE_SESSION_ID": SESSION, + "CLAUDE_CONFIG_DIR": str(configured), + } + saved = os.getuid + del os.getuid + self.addCleanup(setattr, os, "getuid", saved) + with mock.patch.dict(os.environ, environ, clear=True): + profile = host_policy.resolve_profile(None) + self.assertEqual(profile.active_model, "unknown") + self.assertFalse(profile.governance_ready) + + def test_assistant_record_evicted_from_the_bounded_window_is_not_filled(self): + """Padding the tail past the scan bound must not open an env fill path.""" + limit = host_policy._CODEX_ROLLOUT_SCAN_LIMIT + filler = json.dumps({"type": "user", "sessionId": SESSION, "pad": "p" * 4096}) + "\n" + body = _assistant("claude-opus-5") + filler * ((limit // len(filler)) + 2) + self._write(body) + profile = self._profile( + { + "CLAUDE_CODE_SESSION_ID": SESSION, + "AGENT_COLLAB_ACTIVE_MODEL": "claude-sonnet-5", + } + ) + self.assertEqual(profile.active_model, "unknown") + self.assertFalse(profile.governance_ready) + + def test_no_session_identifier_leaves_governance_closed(self): + profile = self._profile({"AGENT_COLLAB_ACTIVE_MODEL": "claude-opus-5"}) + self.assertFalse(profile.governance_ready) + + def test_entrypoint_only_paired_environment_fields_cannot_manufacture_identity(self): + """Supplying BOTH model and session id must not make an unobserved + session eligible -- filling only one field is not the whole bypass.""" + profile = self._profile( + { + "AGENT_COLLAB_ACTIVE_MODEL": "claude-opus-5", + "AGENT_COLLAB_SESSION_ID": "invented-session", + } + ) + self.assertEqual(profile.session_identifier, "unknown") + self.assertFalse(profile.governance_ready) + + def test_entrypoint_only_paired_explicit_fields_cannot_manufacture_identity(self): + profile = self._profile( + {}, + {"active_model": "claude-opus-5", "session_identifier": "invented-session"}, + ) + self.assertEqual(profile.session_identifier, "unknown") + self.assertFalse(profile.governance_ready) + + def test_entrypoint_only_paired_fields_with_valid_uuid_also_rejected(self): + """A well-formed but unobserved UUID is still not an observation.""" + profile = self._profile( + {}, + {"active_model": "claude-opus-5", "session_identifier": SESSION}, + ) + self.assertEqual(profile.session_identifier, "unknown") + self.assertFalse(profile.governance_ready) + + # -- the class invariant, not another instance of it --------------------- + + _HOST_SCENARIOS = { + "resolved transcript": ({"CLAUDE_CODE_SESSION_ID": SESSION}, True), + "absent transcript": ({"CLAUDE_CODE_SESSION_ID": "99999999-9999-9999-9999-999999999999"}, False), + "malformed session": ({"CLAUDE_CODE_SESSION_ID": "not-a-uuid"}, False), + "entrypoint only": ({}, False), + } + + _OVERRIDES = { + "primary_id": "codex", + "primary_family": "openai", + "active_model": "claude-opus-5", + "host_runtime": "codex", + "session_identifier": "invented-session", + } + + def _observed(self, env): + """Raw `_environment_profile()` output, BEFORE any normalization. + + `HostProfile` maps empty values to "unknown" downstream, so asserting + the invariant against the profile is vacuous -- it holds whether or not + the record was written non-empty. The overlay reads the raw record, so + the raw record is where the invariant has to be pinned. + """ + environ = {"CLAUDE_CODE_ENTRYPOINT": "cli"} + environ.update(env) + with mock.patch.dict(os.environ, environ, clear=True), mock.patch.object( + host_policy, "_claude_projects_root", return_value=self.projects + ): + return host_policy._environment_profile() + + def test_no_identity_field_is_ever_recorded_empty_on_a_claude_host(self): + """The invariant the fill class depends on: nothing is left fillable.""" + self._write(_assistant("claude-opus-5")) + for name, (env, _) in self._HOST_SCENARIOS.items(): + with self.subTest(scenario=name): + observed = self._observed(env) + for field in ( + "primary_id", + "primary_family", + "active_model", + "host_runtime", + "session_identifier", + ): + self.assertTrue( + str(observed.get(field, "")).strip(), + f"{field} was recorded empty and is therefore fillable", + ) + + def test_no_identity_field_can_be_supplied_on_a_claude_host(self): + """Every field, every scenario, both fill paths -- none may be overridden.""" + self._write(_assistant("claude-opus-5")) + for name, (env, _) in self._HOST_SCENARIOS.items(): + baseline = self._profile(env) + for field, value in self._OVERRIDES.items(): + with self.subTest(scenario=name, field=field, path="explicit"): + profile = self._profile(env, {field: value}) + self.assertEqual( + getattr(profile, field), + getattr(baseline, field), + f"explicit {field} overrode observation", + ) + with self.subTest(scenario=name, field=field, path="environment"): + key = f"AGENT_COLLAB_{'SESSION_ID' if field == 'session_identifier' else field.upper()}" + profile = self._profile({**env, key: value}) + self.assertEqual( + getattr(profile, field), + getattr(baseline, field), + f"{key} overrode observation", + ) + + def test_only_a_resolved_transcript_yields_governance_on_a_claude_host(self): + self._write(_assistant("claude-opus-5")) + for name, (env, expected) in self._HOST_SCENARIOS.items(): + with self.subTest(scenario=name): + self.assertEqual(self._profile(env).governance_ready, expected) + + +if __name__ == "__main__": # pragma: no cover + unittest.main()