Add thin Attest receipt observation CLI

examples/attest_receipt_observe.py

@@ -0,0 +1,118 @@
+"""Print an unverified SyberWork observation from an Attest v0 receipt file.
+
+Attest writes these receipts under `<attest_home>/receipts/<case_id>.json`.
+This example only translates the JSON into Work.observe-compatible arguments;
+it does not append to a case or verify receipt authority.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+RECEIPT_SCHEMA = "attest.receipt/v0"
+OBSERVATION_KEY = "attest_receipt"
+OBSERVATION_SOURCE = "attest"
+OBSERVATION_ACTOR = "attest-bridge"
+REQUIRED_RECEIPT_FIELDS = {
+    "schema",
+    "case_id",
+    "created_at",
+    "doc",
+    "claim",
+    "authority",
+    "proposal",
+    "approval",
+    "evidence",
+    "handoff",
+    "disposition",
+    "verified",
+}
+OPTIONAL_RECEIPT_FIELDS = {"ticket", "app", "worker"}
+
+
+def receipt_to_observation(receipt: Any) -> dict[str, Any]:
+    """Return SyberWork `observe` keyword fields for a live Attest receipt."""
+    if not isinstance(receipt, dict):
+        raise ValueError("Attest receipt must be a JSON object")
+    if receipt.get("schema") != RECEIPT_SCHEMA:
+        raise ValueError(f"receipt schema must be {RECEIPT_SCHEMA}")
+
+    missing = REQUIRED_RECEIPT_FIELDS - receipt.keys()
+    if missing:
+        raise ValueError("receipt is missing required fields: " + ", ".join(sorted(missing)))
+    allowed_fields = REQUIRED_RECEIPT_FIELDS | OPTIONAL_RECEIPT_FIELDS
+    unexpected = receipt.keys() - allowed_fields
+    if unexpected:
+        raise ValueError("receipt has unexpected fields: " + ", ".join(sorted(unexpected)))
+    if receipt["verified"] is not False:
+        raise ValueError("receipt verified must be false")
+
+    value = receipt
+    if receipt["disposition"] == "implemented":
+        ticket = receipt.get("ticket")
+        app = receipt.get("app")
+        if not isinstance(ticket, dict) or not isinstance(ticket.get("title"), str) or not ticket["title"]:
+            raise ValueError("implemented receipt must include ticket.title")
+        if not isinstance(app, dict) or not isinstance(app.get("patch_sha256"), str) or not app["patch_sha256"]:
+            raise ValueError("implemented receipt must include app.patch_sha256")
+        value = dict(receipt)
+        value["implementation"] = {
+            "ticket_title": ticket["title"],
+            "app_patch_sha256": app["patch_sha256"],
+        }
+
+    evidence = receipt.get("evidence")
+    after = evidence.get("after") if isinstance(evidence, dict) else None
+    if isinstance(after, dict) and after.get("status") == "proven":
+        if value is receipt:
+            value = dict(receipt)
+        value["destination_verification"] = {
+            "status": "proven",
+            "method": "Kane re-check",
+        }
+
+    return {
+        "key": OBSERVATION_KEY,
+        "value": value,
+        "source": OBSERVATION_SOURCE,
+        "version": RECEIPT_SCHEMA,
+        "actor": OBSERVATION_ACTOR,
+        "verified": False,
+    }
+
+
+def load_receipt_observation(receipt_path: str | Path) -> dict[str, Any]:
+    """Load one receipt from disk and translate it to an unverified observation."""
+    path = Path(receipt_path)
+    try:
+        receipt = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ValueError(f"cannot read a UTF-8 JSON receipt from {path}") from exc
+    return receipt_to_observation(receipt)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description="Print a SyberWork observation for an Attest attest.receipt/v0 file"
+    )
+    parser.add_argument(
+        "receipt",
+        type=Path,
+        help="receipt path (Attest writes <attest_home>/receipts/<case_id>.json)",
+    )
+    args = parser.parse_args(argv)
+    try:
+        observation = load_receipt_observation(args.receipt)
+    except ValueError as exc:
+        print(f"attest_receipt_observe: {exc}", file=sys.stderr)
+        return 2
+    print(json.dumps(observation, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

examples/fixtures/attest_receipt_v0_sample.json

@@ -0,0 +1,48 @@
+{
+  "schema": "attest.receipt/v0",
+  "case_id": "a1812ca4f8bd-counter.promise-1",
+  "created_at": "2026-10-24T22:25:56+00:00",
+  "doc": {
+    "source": "demo/counter.md",
+    "sha256_before": "a1812ca4f8bd68df897b1eb86705648e0000b366000000000000000000000000",
+    "sha256_after": null
+  },
+  "claim": {
+    "id": "counter.promise",
+    "kind": "browser",
+    "lines": [10, 10],
+    "text": "increments by one",
+    "expects_before": {"label": "1"},
+    "expects_after": {"label": "1"}
+  },
+  "authority": {
+    "policy": "demo/attest.policy.yaml",
+    "decision": "doc_repair_denied",
+    "rule": "match: increments by one"
+  },
+  "proposal": {
+    "agent": null,
+    "patch_sha256": null,
+    "patch": null
+  },
+  "approval": {
+    "actor": "policy:demo/attest.policy.yaml",
+    "proposal_sha256": null
+  },
+  "evidence": {
+    "before": {
+      "status": "failed",
+      "note": "label showed 2",
+      "screenshot": null
+    },
+    "after": null
+  },
+  "handoff": {
+    "claim": "counter.promise",
+    "doc_text": "increments by one",
+    "observed": {"note": "label showed 2"},
+    "instruction": "the application must change, not the doc"
+  },
+  "disposition": "handoff_implementation",
+  "verified": false
+}

examples/fixtures/attest_receipt_v0_spec_sample.json

@@ -0,0 +1,62 @@
+{
+  "schema": "attest.receipt/v0",
+  "case_id": "a1812ca4f8bd-counter.promise-2",
+  "created_at": "2026-10-24T22:26:56+00:00",
+  "doc": {
+    "source": "demo/counter.md",
+    "sha256_before": "a1812ca4f8bd68df897b1eb86705648e0000b366000000000000000000000000",
+    "sha256_after": null
+  },
+  "claim": {
+    "id": "counter.promise",
+    "kind": "browser",
+    "lines": [10, 10],
+    "text": "increments by one",
+    "expects_before": {"label": "1"},
+    "expects_after": {"label": "2"}
+  },
+  "authority": {
+    "policy": "demo/attest.policy.yaml",
+    "decision": "implementation_required",
+    "rule": "the application must change, not the doc"
+  },
+  "proposal": {
+    "agent": "worker",
+    "patch_sha256": "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd",
+    "patch": "diff --git a/app.js b/app.js\n--- a/app.js\n+++ b/app.js\n@@ -1 +1 @@\n-const count = 2;\n+const count = 1;"
+  },
+  "approval": {
+    "actor": "policy:demo/attest.policy.yaml",
+    "proposal_sha256": "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"
+  },
+  "evidence": {
+    "before": {
+      "status": "failed",
+      "observed": {"label": "2"},
+      "note": "label showed 2",
+      "screenshot": null
+    },
+    "after": {
+      "status": "proven",
+      "observed": {"label": "1"},
+      "note": "Kane re-checked the candidate application",
+      "screenshot": null
+    }
+  },
+  "handoff": null,
+  "disposition": "implemented",
+  "verified": false,
+  "ticket": {
+    "path": "demo/tickets/a1812ca4f8bd-counter.promise-2.json",
+    "title": "Missing capability: counter increments by one"
+  },
+  "app": {
+    "original": "demo/app",
+    "candidate": "demo/candidate",
+    "patch_sha256": "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"
+  },
+  "worker": {
+    "type": "openai",
+    "model": "gpt-5-mini"
+  }
+}

tests/test_attest_receipt_observe_example.py

@@ -0,0 +1,90 @@
+"""Tests for the standalone Attest receipt observation example."""
+
+import json
+import subprocess
+import sys
+import unittest
+from pathlib import Path
+
+from examples.attest_receipt_observe import (
+    RECEIPT_SCHEMA,
+    load_receipt_observation,
+    receipt_to_observation,
+)
+
+ROOT = Path(__file__).resolve().parents[1]
+EXAMPLE = ROOT / "examples" / "attest_receipt_observe.py"
+FIXTURE = ROOT / "examples" / "fixtures" / "attest_receipt_v0_sample.json"
+SPEC_FIXTURE = ROOT / "examples" / "fixtures" / "attest_receipt_v0_spec_sample.json"
+
+
+class AttestReceiptObserveExample(unittest.TestCase):
+    def setUp(self):
+        self.receipt = json.loads(FIXTURE.read_text(encoding="utf-8"))
+
+    def test_happy_path_maps_receipt_to_syberwork_observation(self):
+        observation = load_receipt_observation(FIXTURE)
+        self.assertEqual(observation["key"], "attest_receipt")
+        self.assertEqual(observation["value"], self.receipt)
+        self.assertEqual(observation["source"], "attest")
+        self.assertEqual(observation["version"], RECEIPT_SCHEMA)
+        self.assertEqual(observation["actor"], "attest-bridge")
+        self.assertIs(observation["verified"], False)
+
+    def test_spec_mode_implemented_receipt_maps_ticket_patch_and_kane_recheck(self):
+        receipt = json.loads(SPEC_FIXTURE.read_text(encoding="utf-8"))
+        observation = load_receipt_observation(SPEC_FIXTURE)
+        value = observation["value"]
+
+        self.assertEqual(value["implementation"], {
+            "ticket_title": receipt["ticket"]["title"],
+            "app_patch_sha256": receipt["app"]["patch_sha256"],
+        })
+        self.assertEqual(value["destination_verification"], {
+            "status": "proven",
+            "method": "Kane re-check",
+        })
+        self.assertIs(value["verified"], False)
+        self.assertIs(observation["verified"], False)
+
+    def test_implemented_receipt_requires_ticket_title_and_patch_digest(self):
+        receipt = json.loads(SPEC_FIXTURE.read_text(encoding="utf-8"))
+        receipt["app"]["patch_sha256"] = None
+        with self.assertRaisesRegex(ValueError, "must include app.patch_sha256"):
+            receipt_to_observation(receipt)
+
+    def test_wrong_schema_is_rejected(self):
+        self.receipt["schema"] = "attest.receipt/v1"
+        with self.assertRaisesRegex(ValueError, "schema must be attest.receipt/v0"):
+            receipt_to_observation(self.receipt)
+
+    def test_unexpected_fields_are_rejected(self):
+        self.receipt["unexpected"] = True
+        with self.assertRaisesRegex(ValueError, "unexpected fields: unexpected"):
+            receipt_to_observation(self.receipt)
+
+    def test_import_never_marks_receipt_or_observation_verified(self):
+        observation = receipt_to_observation(self.receipt)
+        self.assertIs(self.receipt["verified"], False)
+        self.assertIs(observation["verified"], False)
+
+        claimed_verified = dict(self.receipt, verified=True)
+        with self.assertRaisesRegex(ValueError, "verified must be false"):
+            receipt_to_observation(claimed_verified)
+
+    def test_example_cli_prints_observation_and_exits_successfully(self):
+        result = subprocess.run(
+            [sys.executable, str(EXAMPLE), str(FIXTURE)],
+            cwd=ROOT,
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        self.assertEqual(result.returncode, 0, result.stderr)
+        observation = json.loads(result.stdout)
+        self.assertEqual(observation["value"], self.receipt)
+        self.assertIs(observation["verified"], False)
+
+
+if __name__ == "__main__":
+    unittest.main()