diff --git a/packages/mcp-server-akf/mcp_server_akf/server.py b/packages/mcp-server-akf/mcp_server_akf/server.py index a091068..0ed63b4 100644 --- a/packages/mcp-server-akf/mcp_server_akf/server.py +++ b/packages/mcp-server-akf/mcp_server_akf/server.py @@ -1,6 +1,7 @@ """MCP server implementation for AKF — Agent Knowledge Format. -Exposes 10 tools via Model Context Protocol: +Exposes 11 tools via Model Context Protocol: + - replay_file: Replay-verify a stamp's falsifiable probe recipe - check_file: One-line trust check — can an agent build on this file? - create_claim: Create AKF trust metadata - validate_file: Validate an .akf file @@ -29,6 +30,17 @@ # Tool implementations # --------------------------------------------------------------------------- +def replay_file(path: str, run: bool = False) -> dict: + """Replay-verify a stamp's probe recipe. run=False inspects only; + run=True executes the recorded command — only for trusted files.""" + from akf.verify import verify_file as _verify + + result = _verify(path, run=run) + payload = result.to_dict() + payload["summary"] = result.summary_line() + return payload + + def check_file(path: str, threshold: float = 0.6) -> dict: """One-line trust check: can an agent build on this file without re-verifying?""" from akf.check import check_file as _check @@ -169,6 +181,18 @@ def detect_threats(path: str) -> dict: # --------------------------------------------------------------------------- TOOLS = [ + Tool( + name="replay_file", + description="Replay-verify a stamp's falsifiable probe recipe. Default inspects only: shows the recorded command and whether the claim's inputs drifted since issuance. With run=true, EXECUTES the recorded command (arbitrary code — only for files you trust) and returns CONFIRMED / CONFIRMED_DRIFTED (succeeded against diverged inputs) / REFUTED / UNREPLAYABLE.", + inputSchema={ + "type": "object", + "required": ["path"], + "properties": { + "path": {"type": "string", "description": "Path to the stamped file"}, + "run": {"type": "boolean", "default": False, "description": "Execute the recorded command (review it first)"}, + }, + }, + ), Tool( name="check_file", description="One-line trust check before building on a file. Returns OK (fresh stamp, trust above threshold — skip re-verification), LOW (trust below threshold), STALE (modified after stamping or claims expired — re-verify), or UNSTAMPED (no metadata). Use this before re-reading, re-testing, or re-deriving work another agent already verified.", @@ -298,6 +322,7 @@ def detect_threats(path: str) -> dict: # Map tool names to handler functions HANDLERS = { + "replay_file": replay_file, "check_file": check_file, "create_claim": create_claim, "validate_file": validate_file, diff --git a/python/akf/__init__.py b/python/akf/__init__.py index 6a3a3bc..7a165f9 100644 --- a/python/akf/__init__.py +++ b/python/akf/__init__.py @@ -53,6 +53,7 @@ from .knowledge_base import KnowledgeBase from .stamp import stamp, stamp_file from .check import check_file, CheckResult +from .verify import verify_file, VerifyResult from .certify import ( certify_file, certify_directory, certify_team, CertifyResult, CertifyReport, AgentCertifyReport, TeamCertifyReport, @@ -417,6 +418,8 @@ def read(filepath): "stamp_file", "check_file", "CheckResult", + "verify_file", + "VerifyResult", "stamp_commit", "read_commit", "trust_log", diff --git a/python/akf/cli.py b/python/akf/cli.py index bcb1ea5..e5571ab 100644 --- a/python/akf/cli.py +++ b/python/akf/cli.py @@ -225,6 +225,34 @@ def check_cmd(file, threshold, as_json) -> None: raise SystemExit(result.exit_code) +@main.command("replay") +@click.argument("file", type=click.Path(exists=True)) +@click.option("--run", "do_run", is_flag=True, + help="Execute the recorded replay command (review it first — it runs with your privileges)") +@click.option("--json", "as_json", is_flag=True, help="Output as JSON") +def replay_cmd(file, do_run, as_json) -> None: + """Re-verify a stamp's replay recipe — claimed evidence, checked. + + Without --run: shows the recipe and whether the claim's inputs have + drifted since issuance. With --run: executes the probe. + Verdicts: CONFIRMED (0), CONFIRMED_DRIFTED (1), REFUTED (2), + UNREPLAYABLE (3). + """ + from .verify import verify_file + + result = verify_file(file, run=do_run) + + if as_json: + click.echo(json.dumps(result.to_dict(), indent=2, ensure_ascii=False)) + else: + color = {"CONFIRMED": "green", "REPLAY_AVAILABLE": "cyan", + "CONFIRMED_DRIFTED": "yellow"}.get(result.verdict, "red") + click.secho(result.summary_line(), fg=color) + + if result.exit_code: + raise SystemExit(result.exit_code) + + @main.command("create") @click.argument("file", type=click.Path(), required=False) @click.option("--claim", "-c", multiple=True, help="Claim content") @@ -1186,8 +1214,10 @@ def kb_prune_cmd(directory, max_age, min_trust) -> None: @click.option("--label", default=None, help="Classification label") @click.option("--preset", type=click.Choice(["memory", "skill"]), default=None, help="Context preset: memory (30-day trust decay) or skill (public, supply-chain trust)") +@click.option("--replay", default=None, + help='Falsifiable probe recipe, e.g. --replay "pytest -q" — akf verify can re-run it') @click.option("--format", "fmt", default="auto", help="Output format: auto, embed, sidecar") -def stamp_cmd(file, agent, evidence, confidence, claims, model, label, preset, fmt): +def stamp_cmd(file, agent, evidence, confidence, claims, model, label, preset, replay, fmt): """Add AKF trust metadata to any file. Stamps the file with trust scores, provenance, and classification. @@ -1230,6 +1260,7 @@ def stamp_cmd(file, agent, evidence, confidence, claims, model, label, preset, f trust_score=trust_score, classification=classification, evidence=evidence_list, + replay=replay, **preset_claim_kwargs, ) diff --git a/python/akf/deps.py b/python/akf/deps.py index 7491c91..8353639 100644 --- a/python/akf/deps.py +++ b/python/akf/deps.py @@ -110,6 +110,21 @@ def hash_source(source: Optional[str], base_dir: str) -> Optional[str]: return None +def input_fingerprint(dep_hashes: Dict[str, str], src_hashes: list) -> str: + """Digest over a claim's input closure — recorded deps + pinned sources. + + Deliberately excludes the stamped file's own content: its drift is + already covered by mtime/dependency staleness, and frontmatter stamping + rewrites the file itself. Order-independent. + """ + h = hashlib.sha256() + for key in sorted(dep_hashes): + h.update(f"{key}={dep_hashes[key]}".encode()) + for s in sorted(x for x in src_hashes if x): + h.update(s.encode()) + return "sha256:" + h.hexdigest()[:16] + + def changed_deps(filepath: str, recorded: Dict[str, str]) -> list: """Return the recorded dependencies whose content no longer matches.""" base_dir = os.path.dirname(os.path.abspath(filepath)) diff --git a/python/akf/models.py b/python/akf/models.py index 6031bb7..1856b79 100644 --- a/python/akf/models.py +++ b/python/akf/models.py @@ -177,6 +177,24 @@ def to_dict(self, compact: bool = False) -> dict: return d +class Replay(BaseModel): + """A falsifiable probe recipe carried inside evidence (#128). + + A signature proves who said it; a replay proves it could have been + true. ``input_hash`` pins the claim's input closure (dependencies + + cited sources) at issuance so a successful replay against drifted + inputs is distinguishable from one against the original world. + """ + + model_config = ConfigDict(extra="allow", populate_by_name=True) + + command: str + cwd: Optional[str] = None # relative to the stamped file's directory + expected_exit: int = 0 + output_digest: Optional[str] = None + input_hash: Optional[str] = None + + class Evidence(BaseModel): """A piece of evidence supporting a claim.""" @@ -186,6 +204,7 @@ class Evidence(BaseModel): detail: str timestamp: Optional[str] = Field(None, validation_alias=AliasChoices("at", "timestamp")) tool: Optional[str] = None + replay: Optional[Replay] = None def to_dict(self, compact: bool = False) -> dict: d = _strip_none(self.model_dump()) diff --git a/python/akf/stamp.py b/python/akf/stamp.py index d4c50d1..52d1a6b 100644 --- a/python/akf/stamp.py +++ b/python/akf/stamp.py @@ -169,6 +169,7 @@ def stamp_file( classification: str = "internal", ai_generated: bool = True, evidence: Optional[list] = None, + replay: Optional[Union[str, dict]] = None, **kwargs, ) -> AKF: """Stamp a file with AKF trust metadata. @@ -245,6 +246,26 @@ def stamp_file( pinned.append(claim.model_copy(update={"src_hash": h}) if h else claim) unit = unit.model_copy(update={"claims": pinned}) + # Replay recipe (#128): record how to falsify the claim, with the input + # closure fingerprint pinned at issuance so a later replay can tell + # "confirmed against the original inputs" from "confirmed against drift". + if replay is not None: + from .deps import input_fingerprint + from .models import Replay + + recipe = Replay(command=replay) if isinstance(replay, str) else Replay(**replay) + recipe.input_hash = input_fingerprint( + dep_hashes, [c.src_hash for c in unit.claims]) + ev = Evidence( + type=parse_evidence_string(recipe.command).type, + detail=f"replayable: {recipe.command}", + timestamp=datetime.now(timezone.utc).isoformat(), + replay=recipe, + ) + first = unit.claims[0] + updated = first.model_copy(update={"evidence": [*(first.evidence or []), ev]}) + unit = unit.model_copy(update={"claims": [updated, *unit.claims[1:]]}) + # Embed into the file using universal format layer from .universal import embed as _embed _embed(filepath, metadata=unit.to_dict(compact=True)) diff --git a/python/akf/verify.py b/python/akf/verify.py new file mode 100644 index 0000000..0b639f6 --- /dev/null +++ b/python/akf/verify.py @@ -0,0 +1,166 @@ +"""Replay verification — turn claimed evidence into checked evidence (#128). + +``akf check`` reads what a stamp *claims*; ``akf verify`` re-runs the probe +the stamp carries. Verdicts: + + CONFIRMED replay succeeded AND inputs match the issuance fingerprint + CONFIRMED_DRIFTED replay succeeded, but against inputs that diverged since + issuance — provably reproducible, possibly reproducibly + wrong (credit: Mike Czerwinski) + REFUTED replay ran and did not produce the claimed result + UNREPLAYABLE no replay recipe recorded (or execution not requested) + +Running a recipe executes a command recorded inside the file's stamp. By +default ``verify_file`` only inspects (fingerprint drift + recipe display); +execution requires ``run=True`` — never replay stamps from untrusted files +without reading the command first. +""" + +from __future__ import annotations + +import hashlib +import os +import subprocess +from dataclasses import dataclass, field +from typing import List, Optional + +from .models import AKF +from .deps import input_fingerprint + +_REPLAY_TIMEOUT_S = 300 + + +@dataclass +class VerifyResult: + file: str + verdict: str # CONFIRMED | CONFIRMED_DRIFTED | REFUTED | UNREPLAYABLE | REPLAY_AVAILABLE + exit_code: int + command: Optional[str] = None + inputs_drifted: Optional[bool] = None + detail: Optional[str] = None + executed: bool = False + + def to_dict(self) -> dict: + return { + "file": self.file, + "verdict": self.verdict, + "exit_code": self.exit_code, + "command": self.command, + "inputs_drifted": self.inputs_drifted, + "detail": self.detail, + "executed": self.executed, + } + + def summary_line(self) -> str: + parts = [self.verdict] + if self.inputs_drifted is not None: + parts.append("inputs=" + ("drifted" if self.inputs_drifted else "intact")) + if self.command: + parts.append(f'replay="{self.command}"') + if self.detail: + parts.append(self.detail) + return " ".join(parts) + + +def _current_fingerprint(filepath: str, unit: AKF) -> str: + from .deps import hash_source + + base_dir = os.path.dirname(os.path.abspath(filepath)) + recorded = (unit.meta or {}).get("deps") or {} + current_deps = {} + for rel in recorded: + dep_path = os.path.join(base_dir, rel) + try: + with open(dep_path, "rb") as f: + current_deps[rel] = "sha256:" + hashlib.sha256(f.read()).hexdigest()[:16] + except OSError: + current_deps[rel] = "missing" + current_srcs = [hash_source(c.source, base_dir) or "missing" + for c in unit.claims if getattr(c, "src_hash", None)] + return input_fingerprint(current_deps, current_srcs) + + +def _find_replay(unit: AKF): + for claim in unit.claims: + for ev in claim.evidence or []: + replay = getattr(ev, "replay", None) + if replay and getattr(replay, "command", None): + return replay + return None + + +def verify_file(filepath: str, run: bool = False) -> VerifyResult: + """Verify a file's stamped claims against their replay recipe. + + Without ``run``, reports drift + the recipe (safe inspection). With + ``run``, executes the recorded command in the file's directory and + returns CONFIRMED / CONFIRMED_DRIFTED / REFUTED. + """ + from . import universal + from .core import load + + meta = universal.extract(filepath) + if meta is not None: + unit = AKF(**meta) + elif filepath.endswith(".akf"): + try: + unit = load(filepath) + except Exception: + unit = None + else: + unit = None + + if unit is None: + return VerifyResult(file=filepath, verdict="UNREPLAYABLE", exit_code=3, + detail="no_metadata") + + replay = _find_replay(unit) + if replay is None: + return VerifyResult(file=filepath, verdict="UNREPLAYABLE", exit_code=3, + detail="no_replay_recipe") + + drifted: Optional[bool] = None + if replay.input_hash: + drifted = _current_fingerprint(filepath, unit) != replay.input_hash + + if not run: + return VerifyResult( + file=filepath, + verdict="REPLAY_AVAILABLE", + exit_code=1 if drifted else 0, + command=replay.command, + inputs_drifted=drifted, + detail="pass run=True (--run) to execute", + ) + + base_dir = os.path.dirname(os.path.abspath(filepath)) + cwd = os.path.join(base_dir, replay.cwd) if replay.cwd else base_dir + try: + proc = subprocess.run( + replay.command, shell=True, cwd=cwd, + capture_output=True, text=True, timeout=_REPLAY_TIMEOUT_S, + ) + except subprocess.TimeoutExpired: + return VerifyResult(file=filepath, verdict="REFUTED", exit_code=2, + command=replay.command, inputs_drifted=drifted, + detail=f"timeout>{_REPLAY_TIMEOUT_S}s", executed=True) + + exit_ok = proc.returncode == replay.expected_exit + output_ok = True + if replay.output_digest: + digest = "sha256:" + hashlib.sha256(proc.stdout.encode()).hexdigest()[:16] + output_ok = digest == replay.output_digest + + if exit_ok and output_ok: + if drifted: + return VerifyResult(file=filepath, verdict="CONFIRMED_DRIFTED", exit_code=1, + command=replay.command, inputs_drifted=True, + detail="replay succeeded against diverged inputs", executed=True) + return VerifyResult(file=filepath, verdict="CONFIRMED", exit_code=0, + command=replay.command, inputs_drifted=drifted, executed=True) + + detail = f"exit={proc.returncode} expected={replay.expected_exit}" if not exit_ok \ + else "output digest mismatch" + return VerifyResult(file=filepath, verdict="REFUTED", exit_code=2, + command=replay.command, inputs_drifted=drifted, + detail=detail, executed=True) diff --git a/python/tests/test_verify.py b/python/tests/test_verify.py new file mode 100644 index 0000000..ba91566 --- /dev/null +++ b/python/tests/test_verify.py @@ -0,0 +1,112 @@ +"""Tests for replay evidence + akf verify (#128).""" + +import json +import os +import time + +import pytest +from click.testing import CliRunner + +from akf.cli import main +from akf.stamp import stamp_file +from akf.verify import verify_file + + +@pytest.fixture +def runner(): + return CliRunner() + + +@pytest.fixture +def replayable(tmp_path): + """A stamped file whose replay probe checks its dep's content.""" + dep = tmp_path / "helper.py" + dep.write_text("VALUE = 1\n") + f = tmp_path / "main.py" + f.write_text("import helper\n") + stamp_file( + str(f), agent="claude-code", evidence=["tests pass"], + replay="python3 -c \"import helper; raise SystemExit(0 if helper.VALUE == 1 else 1)\"", + ) + return f + + +class TestReplayRecording: + def test_recipe_recorded_with_input_hash(self, replayable): + unit_ev = None + from akf import universal + from akf.models import AKF + unit = AKF(**universal.extract(str(replayable))) + for ev in unit.claims[0].evidence or []: + if getattr(ev, "replay", None): + unit_ev = ev + assert unit_ev is not None + assert unit_ev.replay.command.startswith("python3 -c") + assert unit_ev.replay.expected_exit == 0 + assert unit_ev.replay.input_hash.startswith("sha256:") + + def test_replay_dict_form(self, tmp_path): + f = tmp_path / "x.md" + f.write_text("# x\n") + unit = stamp_file(str(f), agent="a", + replay={"command": "true", "expected_exit": 0}) + assert any(getattr(e, "replay", None) for e in unit.claims[0].evidence) + + +class TestVerify: + def test_default_is_inspection_not_execution(self, replayable): + r = verify_file(str(replayable)) + assert r.verdict == "REPLAY_AVAILABLE" + assert r.executed is False + assert r.inputs_drifted is False + assert r.exit_code == 0 + + def test_run_confirms(self, replayable): + r = verify_file(str(replayable), run=True) + assert r.verdict == "CONFIRMED" + assert r.exit_code == 0 + assert r.executed + + def test_drifted_inputs_split_the_verdict(self, replayable): + # Change the dep so its content still satisfies the probe's assertion + # semantics differently: VALUE stays 1 but file content changes. + (replayable.parent / "helper.py").write_text("VALUE = 1 # edited\n") + r = verify_file(str(replayable), run=True) + assert r.verdict == "CONFIRMED_DRIFTED" + assert r.exit_code == 1 + assert r.inputs_drifted is True + + def test_refuted_when_probe_fails(self, replayable): + (replayable.parent / "helper.py").write_text("VALUE = 2\n") + r = verify_file(str(replayable), run=True) + assert r.verdict == "REFUTED" + assert r.exit_code == 2 + + def test_unreplayable_without_recipe(self, tmp_path): + f = tmp_path / "plain.md" + f.write_text("# p\n") + stamp_file(str(f), agent="a", evidence=["tests pass"]) + r = verify_file(str(f), run=True) + assert r.verdict == "UNREPLAYABLE" + assert r.exit_code == 3 + + +class TestVerifyCli: + def test_cli_inspection_shows_command(self, runner, replayable): + result = runner.invoke(main, ["replay", str(replayable)]) + assert result.exit_code == 0 + assert "REPLAY_AVAILABLE" in result.output + assert "replay=" in result.output + + def test_cli_run_json(self, runner, replayable): + result = runner.invoke(main, ["replay", str(replayable), "--run", "--json"]) + assert result.exit_code == 0 + assert json.loads(result.output)["verdict"] == "CONFIRMED" + + def test_stamp_cli_replay_flag(self, runner, tmp_path): + f = tmp_path / "r.md" + f.write_text("# r\n") + res = runner.invoke(main, ["stamp", str(f), "--agent", "a", "--replay", "true"]) + assert res.exit_code == 0 + out = runner.invoke(main, ["replay", str(f), "--run"]) + assert "CONFIRMED" in out.output