feat(replay): falsifiable-by-replay evidence + akf replay

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,

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",

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,
     )
 

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))

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())

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))

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)