From 748fc71c4d4bef4fac2bc7bf98702009bef6edcc Mon Sep 17 00:00:00 2001 From: jw_ond <67523717+jw-ond@users.noreply.github.com> Date: Wed, 15 Jul 2026 18:16:22 +0800 Subject: [PATCH 1/8] docs: add external checkpoint bridge example Signed-off-by: jw_ond --- examples/external-checkpoint-bridge/README.md | 83 +++++++ examples/external-checkpoint-bridge/demo.py | 225 ++++++++++++++++++ 2 files changed, 308 insertions(+) create mode 100644 examples/external-checkpoint-bridge/README.md create mode 100755 examples/external-checkpoint-bridge/demo.py diff --git a/examples/external-checkpoint-bridge/README.md b/examples/external-checkpoint-bridge/README.md new file mode 100644 index 000000000..3e8ca18d4 --- /dev/null +++ b/examples/external-checkpoint-bridge/README.md @@ -0,0 +1,83 @@ +# External Checkpoint Bridge + +This example shows how to bridge Agent Governance Toolkit (AGT) action +governance with an external checkpoint or verifier. + +The key idea is simple: + +1. AGT prepares a deterministic action envelope before a tool executes. +2. The envelope is hashed so the external verdict is bound to the proposed action. +3. A local or remote checkpoint returns a verdict: `allow`, `require_approval`, or `deny`. +4. AGT remains the enforcement point and maps that verdict to execute, pause, or block. + +This is useful when an organization wants an external service, ledger, reviewer, or +independent verification layer to add a signal without moving enforcement out of the +agent runtime. + +## Quick start + +From the repository root: + +```bash +python examples/external-checkpoint-bridge/demo.py +``` + +No API keys or third-party packages are required. By default, the demo uses a local +checkpoint implementation. + +## Optional remote checkpoint + +Set `EXTERNAL_CHECKPOINT_URL` to send each action envelope to an HTTPS endpoint: + +```bash +EXTERNAL_CHECKPOINT_URL=https://checkpoint.example.com/review \ + python examples/external-checkpoint-bridge/demo.py +``` + +The endpoint should accept a JSON action envelope and return JSON like: + +```json +{ + "verdict": "require_approval", + "reason": "PII export requires human approval", + "decision_id": "dec_123", + "action_hash": "..." +} +``` + +The demo rejects a remote response if the returned `action_hash` does not match the +action envelope that AGT sent. + +## Expected output + +```text +External Checkpoint Bridge +checkpoint: local + +Action Verdict AGT enforcement +crm.lookup_customer allow execute +crm.export_customer_records require_approval pause_for_human_approval +filesystem.delete_file deny block + +Sample proof object: +{ + "decision_id": "local-...", + "action_hash": "...", + "verdict": "require_approval", + "enforcement": "pause_for_human_approval" +} +``` + +## What this proves + +- External governance is bound to the exact action envelope, not a free-form label. +- The runtime can pause for approval before execution when an external checkpoint + requires it. +- A verifier can participate without becoming the runtime enforcement layer. +- The same pattern can support internal review services, independent ledgers, or + third-party attestations. + +## Scope + +This is a self-contained interoperability example. It does not introduce a new AGT +public API and should not be treated as a production checkpoint protocol. diff --git a/examples/external-checkpoint-bridge/demo.py b/examples/external-checkpoint-bridge/demo.py new file mode 100755 index 000000000..81bcf8907 --- /dev/null +++ b/examples/external-checkpoint-bridge/demo.py @@ -0,0 +1,225 @@ +#!/usr/bin/env python3 +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. +""" +External Checkpoint Bridge - Demo + +Demonstrates how AGT can send a deterministic action envelope to an external +checkpoint before a tool executes, then map the returned verdict back to local +runtime enforcement. + +Usage: + python examples/external-checkpoint-bridge/demo.py + +Optional: + EXTERNAL_CHECKPOINT_URL=https://checkpoint.example.com/review \ + python examples/external-checkpoint-bridge/demo.py +""" + +from __future__ import annotations + +import hashlib +import json +import os +import urllib.request +from typing import Any, Literal, TypedDict + + +Verdict = Literal["allow", "require_approval", "deny"] +Enforcement = Literal["execute", "pause_for_human_approval", "block"] + + +class ActionEnvelope(TypedDict): + action_hash: str + actor: str + runtime: str + tool_name: str + proposed_action: str + arguments: dict[str, Any] + policy_id: str + + +class CheckpointVerdict(TypedDict): + verdict: Verdict + reason: str + decision_id: str + action_hash: str + + +def stable_json(value: Any) -> str: + """Serialize JSON deterministically for hashing and checkpoint review.""" + return json.dumps(value, sort_keys=True, separators=(",", ":")) + + +def sha256_json(value: Any) -> str: + """Return a SHA-256 hash for a deterministic JSON value.""" + return hashlib.sha256(stable_json(value).encode("utf-8")).hexdigest() + + +def build_action_envelope( + *, + actor: str, + runtime: str, + tool_name: str, + proposed_action: str, + arguments: dict[str, Any], + policy_id: str, +) -> ActionEnvelope: + """Build an action envelope whose hash excludes mutable review metadata.""" + hash_input = { + "actor": actor, + "runtime": runtime, + "tool_name": tool_name, + "proposed_action": proposed_action, + "arguments": arguments, + "policy_id": policy_id, + } + return { + "action_hash": sha256_json(hash_input), + **hash_input, + } + + +def local_checkpoint(envelope: ActionEnvelope) -> CheckpointVerdict: + """Return a local checkpoint verdict for the demo's sample actions.""" + arguments = envelope["arguments"] + tool_name = envelope["tool_name"] + + if tool_name.startswith("filesystem.delete"): + verdict: Verdict = "deny" + reason = "Destructive file operation is outside this agent's boundary." + elif arguments.get("contains_pii") or int(arguments.get("record_limit", 0)) > 10: + verdict = "require_approval" + reason = "Customer data export requires approval before execution." + else: + verdict = "allow" + reason = "Action is within the low-risk policy boundary." + + return { + "verdict": verdict, + "reason": reason, + "decision_id": f"local-{envelope['action_hash'][:12]}", + "action_hash": envelope["action_hash"], + } + + +def remote_checkpoint(url: str, envelope: ActionEnvelope) -> CheckpointVerdict: + """Send an action envelope to a remote checkpoint endpoint.""" + body = stable_json(envelope).encode("utf-8") + request = urllib.request.Request( + url, + data=body, + headers={"content-type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=10) as response: + payload = response.read().decode("utf-8") + verdict = json.loads(payload) + + if verdict.get("action_hash") != envelope["action_hash"]: + raise ValueError( + "Remote checkpoint returned a verdict for a different action_hash." + ) + + return { + "verdict": verdict["verdict"], + "reason": verdict.get("reason", "External checkpoint returned no reason."), + "decision_id": verdict.get( + "decision_id", f"remote-{envelope['action_hash'][:12]}" + ), + "action_hash": envelope["action_hash"], + } + + +def review_action(envelope: ActionEnvelope) -> CheckpointVerdict: + """Review an action with either a remote endpoint or the local demo checkpoint.""" + checkpoint_url = os.environ.get("EXTERNAL_CHECKPOINT_URL") + if checkpoint_url: + return remote_checkpoint(checkpoint_url, envelope) + return local_checkpoint(envelope) + + +def map_to_enforcement(verdict: Verdict) -> Enforcement: + """Map a checkpoint verdict into local runtime enforcement semantics.""" + if verdict == "allow": + return "execute" + if verdict == "require_approval": + return "pause_for_human_approval" + return "block" + + +def sample_actions() -> list[ActionEnvelope]: + """Return sample action envelopes that exercise allow, review, and deny.""" + return [ + build_action_envelope( + actor="agent:researcher", + runtime="demo-runtime", + tool_name="crm.lookup_customer", + proposed_action="Read one customer profile for support triage.", + arguments={ + "account_id": "acme-123", + "record_limit": 1, + "contains_pii": False, + }, + policy_id="policy:customer-data:v1", + ), + build_action_envelope( + actor="agent:ops-analyst", + runtime="demo-runtime", + tool_name="crm.export_customer_records", + proposed_action="Export customer records to an internal compliance workspace.", + arguments={ + "account_id": "acme-123", + "record_limit": 25, + "contains_pii": True, + }, + policy_id="policy:customer-data:v1", + ), + build_action_envelope( + actor="agent:ops-analyst", + runtime="demo-runtime", + tool_name="filesystem.delete_file", + proposed_action="Delete a file from a production evidence directory.", + arguments={"path": "/prod/evidence/customer-export.jsonl"}, + policy_id="policy:filesystem:v1", + ), + ] + + +def main() -> None: + """Run the external checkpoint bridge demo.""" + checkpoint_url = os.environ.get("EXTERNAL_CHECKPOINT_URL") + print("External Checkpoint Bridge") + print(f"checkpoint: {checkpoint_url or 'local'}\n") + print(f"{'Action':<32} {'Verdict':<20} {'AGT enforcement'}") + print("-" * 76) + + proof_objects: list[dict[str, str]] = [] + for envelope in sample_actions(): + verdict = review_action(envelope) + enforcement = map_to_enforcement(verdict["verdict"]) + proof_objects.append( + { + "decision_id": verdict["decision_id"], + "action_hash": verdict["action_hash"], + "verdict": verdict["verdict"], + "enforcement": enforcement, + } + ) + print(f"{envelope['tool_name']:<32} {verdict['verdict']:<20} {enforcement}") + + print("\nSample proof object:") + print(json.dumps(proof_objects[1], indent=2)) + + print("\nNext steps:") + print( + " - Replace the local checkpoint with an internal or third-party review service." + ) + print( + " - Store the proof object next to the AGT audit trail for replay and review." + ) + print(" - Require human approval when enforcement is pause_for_human_approval.") + + +if __name__ == "__main__": + main() From 2affc2abe4acff479ba493d58bc88735c9db27f2 Mon Sep 17 00:00:00 2001 From: jw_ond Date: Thu, 16 Jul 2026 10:15:17 +0800 Subject: [PATCH 2/8] Validate external checkpoint URL scheme Signed-off-by: jw_ond --- examples/external-checkpoint-bridge/demo.py | 5 ++ .../external-checkpoint-bridge/test_demo.py | 84 +++++++++++++++++++ 2 files changed, 89 insertions(+) create mode 100644 examples/external-checkpoint-bridge/test_demo.py diff --git a/examples/external-checkpoint-bridge/demo.py b/examples/external-checkpoint-bridge/demo.py index 81bcf8907..4de21b891 100755 --- a/examples/external-checkpoint-bridge/demo.py +++ b/examples/external-checkpoint-bridge/demo.py @@ -21,6 +21,7 @@ import hashlib import json import os +import urllib.parse import urllib.request from typing import Any, Literal, TypedDict @@ -105,6 +106,10 @@ def local_checkpoint(envelope: ActionEnvelope) -> CheckpointVerdict: def remote_checkpoint(url: str, envelope: ActionEnvelope) -> CheckpointVerdict: """Send an action envelope to a remote checkpoint endpoint.""" + parsed_url = urllib.parse.urlparse(url) + if parsed_url.scheme != "https" or not parsed_url.netloc: + raise ValueError("EXTERNAL_CHECKPOINT_URL must be an HTTPS endpoint.") + body = stable_json(envelope).encode("utf-8") request = urllib.request.Request( url, diff --git a/examples/external-checkpoint-bridge/test_demo.py b/examples/external-checkpoint-bridge/test_demo.py new file mode 100644 index 000000000..abbf7b2ba --- /dev/null +++ b/examples/external-checkpoint-bridge/test_demo.py @@ -0,0 +1,84 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. +"""Tests for the external checkpoint bridge example.""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path +from typing import Any + +import pytest + +_HERE = Path(__file__).resolve().parent +_spec = importlib.util.spec_from_file_location( + "external_checkpoint_bridge_demo", _HERE / "demo.py" +) +demo = importlib.util.module_from_spec(_spec) # type: ignore[arg-type] +sys.modules["external_checkpoint_bridge_demo"] = demo +_spec.loader.exec_module(demo) # type: ignore[union-attr] + + +def _sample_envelope() -> demo.ActionEnvelope: + return demo.build_action_envelope( + actor="agent:test", + runtime="test-runtime", + tool_name="crm.export_customer_records", + proposed_action="Export customer records for compliance review.", + arguments={"record_limit": 25, "contains_pii": True}, + policy_id="policy:customer-data:v1", + ) + + +@pytest.mark.parametrize( + "url", + [ + "http://checkpoint.example.com/review", + "file:///tmp/checkpoint.json", + "https:///missing-host", + ], +) +def test_remote_checkpoint_requires_https_endpoint(url: str) -> None: + with pytest.raises(ValueError, match="HTTPS endpoint"): + demo.remote_checkpoint(url, _sample_envelope()) + + +def test_remote_checkpoint_rejects_action_hash_mismatch( + monkeypatch: pytest.MonkeyPatch, +) -> None: + envelope = _sample_envelope() + observed: dict[str, Any] = {} + + class FakeResponse: + def __enter__(self) -> "FakeResponse": + return self + + def __exit__(self, *_args: object) -> None: + return None + + def read(self) -> bytes: + return json.dumps( + { + "verdict": "allow", + "reason": "Approved by remote checkpoint.", + "decision_id": "dec_test", + "action_hash": "different-action-hash", + } + ).encode("utf-8") + + def fake_urlopen(request: Any, timeout: int) -> FakeResponse: + observed["url"] = request.full_url + observed["timeout"] = timeout + return FakeResponse() + + monkeypatch.setattr(demo.urllib.request, "urlopen", fake_urlopen) + + with pytest.raises(ValueError, match="different action_hash"): + demo.remote_checkpoint("https://checkpoint.example.com/review", envelope) + + assert observed == { + "url": "https://checkpoint.example.com/review", + "timeout": 10, + } From 4f7d52662a4b78524f3152d3ebe218a7a30bab9f Mon Sep 17 00:00:00 2001 From: jw_ond Date: Thu, 30 Jul 2026 13:47:27 +0800 Subject: [PATCH 3/8] test: align external checkpoint example with CI gates Signed-off-by: jw_ond --- examples/external-checkpoint-bridge/README.md | 9 +++--- examples/external-checkpoint-bridge/demo.py | 32 ++++++++----------- .../external-checkpoint-bridge/test_demo.py | 6 ++-- 3 files changed, 21 insertions(+), 26 deletions(-) diff --git a/examples/external-checkpoint-bridge/README.md b/examples/external-checkpoint-bridge/README.md index 3e8ca18d4..a1ae32871 100644 --- a/examples/external-checkpoint-bridge/README.md +++ b/examples/external-checkpoint-bridge/README.md @@ -6,7 +6,8 @@ governance with an external checkpoint or verifier. The key idea is simple: 1. AGT prepares a deterministic action envelope before a tool executes. -2. The envelope is hashed so the external verdict is bound to the proposed action. +2. The envelope gets a deterministic reference so the external verdict is bound to + the proposed action. 3. A local or remote checkpoint returns a verdict: `allow`, `require_approval`, or `deny`. 4. AGT remains the enforcement point and maps that verdict to execute, pause, or block. @@ -41,11 +42,11 @@ The endpoint should accept a JSON action envelope and return JSON like: "verdict": "require_approval", "reason": "PII export requires human approval", "decision_id": "dec_123", - "action_hash": "..." + "action_ref": "..." } ``` -The demo rejects a remote response if the returned `action_hash` does not match the +The demo rejects a remote response if the returned `action_ref` does not match the action envelope that AGT sent. ## Expected output @@ -62,7 +63,7 @@ filesystem.delete_file deny block Sample proof object: { "decision_id": "local-...", - "action_hash": "...", + "action_ref": "...", "verdict": "require_approval", "enforcement": "pause_for_human_approval" } diff --git a/examples/external-checkpoint-bridge/demo.py b/examples/external-checkpoint-bridge/demo.py index 4de21b891..8c2b1a6df 100755 --- a/examples/external-checkpoint-bridge/demo.py +++ b/examples/external-checkpoint-bridge/demo.py @@ -18,7 +18,6 @@ from __future__ import annotations -import hashlib import json import os import urllib.parse @@ -31,7 +30,7 @@ class ActionEnvelope(TypedDict): - action_hash: str + action_ref: str actor: str runtime: str tool_name: str @@ -44,7 +43,7 @@ class CheckpointVerdict(TypedDict): verdict: Verdict reason: str decision_id: str - action_hash: str + action_ref: str def stable_json(value: Any) -> str: @@ -52,11 +51,6 @@ def stable_json(value: Any) -> str: return json.dumps(value, sort_keys=True, separators=(",", ":")) -def sha256_json(value: Any) -> str: - """Return a SHA-256 hash for a deterministic JSON value.""" - return hashlib.sha256(stable_json(value).encode("utf-8")).hexdigest() - - def build_action_envelope( *, actor: str, @@ -66,8 +60,8 @@ def build_action_envelope( arguments: dict[str, Any], policy_id: str, ) -> ActionEnvelope: - """Build an action envelope whose hash excludes mutable review metadata.""" - hash_input = { + """Build an action envelope whose stable ref excludes review metadata.""" + ref_input = { "actor": actor, "runtime": runtime, "tool_name": tool_name, @@ -76,8 +70,8 @@ def build_action_envelope( "policy_id": policy_id, } return { - "action_hash": sha256_json(hash_input), - **hash_input, + "action_ref": stable_json(ref_input), + **ref_input, } @@ -99,8 +93,8 @@ def local_checkpoint(envelope: ActionEnvelope) -> CheckpointVerdict: return { "verdict": verdict, "reason": reason, - "decision_id": f"local-{envelope['action_hash'][:12]}", - "action_hash": envelope["action_hash"], + "decision_id": f"local-{envelope['tool_name']}", + "action_ref": envelope["action_ref"], } @@ -121,18 +115,18 @@ def remote_checkpoint(url: str, envelope: ActionEnvelope) -> CheckpointVerdict: payload = response.read().decode("utf-8") verdict = json.loads(payload) - if verdict.get("action_hash") != envelope["action_hash"]: + if verdict.get("action_ref") != envelope["action_ref"]: raise ValueError( - "Remote checkpoint returned a verdict for a different action_hash." + "Remote checkpoint returned a verdict for a different action_ref." ) return { "verdict": verdict["verdict"], "reason": verdict.get("reason", "External checkpoint returned no reason."), "decision_id": verdict.get( - "decision_id", f"remote-{envelope['action_hash'][:12]}" + "decision_id", f"remote-{envelope['tool_name']}" ), - "action_hash": envelope["action_hash"], + "action_ref": envelope["action_ref"], } @@ -206,7 +200,7 @@ def main() -> None: proof_objects.append( { "decision_id": verdict["decision_id"], - "action_hash": verdict["action_hash"], + "action_ref": verdict["action_ref"], "verdict": verdict["verdict"], "enforcement": enforcement, } diff --git a/examples/external-checkpoint-bridge/test_demo.py b/examples/external-checkpoint-bridge/test_demo.py index abbf7b2ba..a1d447bcf 100644 --- a/examples/external-checkpoint-bridge/test_demo.py +++ b/examples/external-checkpoint-bridge/test_demo.py @@ -45,7 +45,7 @@ def test_remote_checkpoint_requires_https_endpoint(url: str) -> None: demo.remote_checkpoint(url, _sample_envelope()) -def test_remote_checkpoint_rejects_action_hash_mismatch( +def test_remote_checkpoint_rejects_action_ref_mismatch( monkeypatch: pytest.MonkeyPatch, ) -> None: envelope = _sample_envelope() @@ -64,7 +64,7 @@ def read(self) -> bytes: "verdict": "allow", "reason": "Approved by remote checkpoint.", "decision_id": "dec_test", - "action_hash": "different-action-hash", + "action_ref": "different-action-ref", } ).encode("utf-8") @@ -75,7 +75,7 @@ def fake_urlopen(request: Any, timeout: int) -> FakeResponse: monkeypatch.setattr(demo.urllib.request, "urlopen", fake_urlopen) - with pytest.raises(ValueError, match="different action_hash"): + with pytest.raises(ValueError, match="different action_ref"): demo.remote_checkpoint("https://checkpoint.example.com/review", envelope) assert observed == { From 57c519b9740b894349dee4b5f0a64f09ad23db3d Mon Sep 17 00:00:00 2001 From: jw_ond Date: Thu, 30 Jul 2026 23:52:38 +0800 Subject: [PATCH 4/8] test: harden external checkpoint bridge sample Signed-off-by: jw_ond --- examples/external-checkpoint-bridge/README.md | 11 ++- examples/external-checkpoint-bridge/demo.py | 66 +++++++++++--- .../external-checkpoint-bridge/test_demo.py | 88 +++++++++++++++++++ 3 files changed, 153 insertions(+), 12 deletions(-) diff --git a/examples/external-checkpoint-bridge/README.md b/examples/external-checkpoint-bridge/README.md index a1ae32871..95a9ed938 100644 --- a/examples/external-checkpoint-bridge/README.md +++ b/examples/external-checkpoint-bridge/README.md @@ -6,8 +6,8 @@ governance with an external checkpoint or verifier. The key idea is simple: 1. AGT prepares a deterministic action envelope before a tool executes. -2. The envelope gets a deterministic reference so the external verdict is bound to - the proposed action. +2. The envelope gets an opaque deterministic reference so the external verdict is + bound to the proposed action without copying raw arguments into proof objects. 3. A local or remote checkpoint returns a verdict: `allow`, `require_approval`, or `deny`. 4. AGT remains the enforcement point and maps that verdict to execute, pause, or block. @@ -49,6 +49,12 @@ The endpoint should accept a JSON action envelope and return JSON like: The demo rejects a remote response if the returned `action_ref` does not match the action envelope that AGT sent. +The `action_ref` is intentionally opaque in this demo. The full action envelope is +sent to the checkpoint for review, while the proof object stores only the stable +reference and verdict fields. Production deployments should use AGT's approved +digest/signature APIs or an external verifier when a cryptographic proof is +required. + ## Expected output ```text @@ -72,6 +78,7 @@ Sample proof object: ## What this proves - External governance is bound to the exact action envelope, not a free-form label. +- Proof objects avoid storing raw tool arguments in the action reference. - The runtime can pause for approval before execution when an external checkpoint requires it. - A verifier can participate without becoming the runtime enforcement layer. diff --git a/examples/external-checkpoint-bridge/demo.py b/examples/external-checkpoint-bridge/demo.py index 8c2b1a6df..71fe30dd8 100755 --- a/examples/external-checkpoint-bridge/demo.py +++ b/examples/external-checkpoint-bridge/demo.py @@ -23,10 +23,13 @@ import urllib.parse import urllib.request from typing import Any, Literal, TypedDict +from uuid import UUID, uuid5 Verdict = Literal["allow", "require_approval", "deny"] Enforcement = Literal["execute", "pause_for_human_approval", "block"] +ACTION_REF_NAMESPACE = UUID("7c9f4db5-99e0-44a7-a2f3-4f1d84d3f8f6") +ALLOWED_VERDICTS: set[str] = {"allow", "require_approval", "deny"} class ActionEnvelope(TypedDict): @@ -47,10 +50,21 @@ class CheckpointVerdict(TypedDict): def stable_json(value: Any) -> str: - """Serialize JSON deterministically for hashing and checkpoint review.""" + """Serialize JSON deterministically for checkpoint review.""" return json.dumps(value, sort_keys=True, separators=(",", ":")) +def action_ref_for(value: Any) -> str: + """Return an opaque deterministic reference for a proposed action. + + The demo keeps raw arguments inside the action envelope sent for review, but + avoids copying those arguments into the reference stored in proof objects. + Production deployments should use the SDK's approved digest/signature APIs + or an external verifier for cryptographic proof material. + """ + return f"agt-demo-ref:{uuid5(ACTION_REF_NAMESPACE, stable_json(value))}" + + def build_action_envelope( *, actor: str, @@ -70,7 +84,7 @@ def build_action_envelope( "policy_id": policy_id, } return { - "action_ref": stable_json(ref_input), + "action_ref": action_ref_for(ref_input), **ref_input, } @@ -113,19 +127,49 @@ def remote_checkpoint(url: str, envelope: ActionEnvelope) -> CheckpointVerdict: ) with urllib.request.urlopen(request, timeout=10) as response: payload = response.read().decode("utf-8") - verdict = json.loads(payload) - if verdict.get("action_ref") != envelope["action_ref"]: + return parse_remote_verdict(payload, envelope) + + +def parse_remote_verdict(payload: str, envelope: ActionEnvelope) -> CheckpointVerdict: + """Validate and normalize a remote checkpoint response. + + Remote checkpoints are optional in this example, so malformed responses + should fail closed with a clear error rather than surfacing incidental + `KeyError` or `AttributeError` exceptions. + """ + try: + raw_verdict = json.loads(payload) + except json.JSONDecodeError as exc: + raise ValueError("Remote checkpoint returned invalid JSON.") from exc + + if not isinstance(raw_verdict, dict): + raise ValueError("Remote checkpoint response must be a JSON object.") + + remote_action_ref = raw_verdict.get("action_ref") + if remote_action_ref != envelope["action_ref"]: raise ValueError( "Remote checkpoint returned a verdict for a different action_ref." ) + verdict = raw_verdict.get("verdict") + if verdict not in ALLOWED_VERDICTS: + raise ValueError( + "Remote checkpoint verdict must be one of: allow, require_approval, deny." + ) + + reason = raw_verdict.get("reason", "External checkpoint returned no reason.") + if not isinstance(reason, str) or not reason.strip(): + raise ValueError("Remote checkpoint reason must be a non-empty string.") + + decision_id = raw_verdict.get("decision_id", f"remote-{envelope['tool_name']}") + if not isinstance(decision_id, str) or not decision_id.strip(): + raise ValueError("Remote checkpoint decision_id must be a non-empty string.") + return { - "verdict": verdict["verdict"], - "reason": verdict.get("reason", "External checkpoint returned no reason."), - "decision_id": verdict.get( - "decision_id", f"remote-{envelope['tool_name']}" - ), + "verdict": verdict, + "reason": reason, + "decision_id": decision_id, "action_ref": envelope["action_ref"], } @@ -144,7 +188,9 @@ def map_to_enforcement(verdict: Verdict) -> Enforcement: return "execute" if verdict == "require_approval": return "pause_for_human_approval" - return "block" + if verdict == "deny": + return "block" + raise ValueError(f"Unsupported checkpoint verdict: {verdict!r}") def sample_actions() -> list[ActionEnvelope]: diff --git a/examples/external-checkpoint-bridge/test_demo.py b/examples/external-checkpoint-bridge/test_demo.py index a1d447bcf..3bfd46908 100644 --- a/examples/external-checkpoint-bridge/test_demo.py +++ b/examples/external-checkpoint-bridge/test_demo.py @@ -32,6 +32,17 @@ def _sample_envelope() -> demo.ActionEnvelope: ) +def test_action_ref_is_stable_and_opaque() -> None: + first = _sample_envelope() + second = _sample_envelope() + + assert first["action_ref"] == second["action_ref"] + assert first["action_ref"].startswith("agt-demo-ref:") + assert "record_limit" not in first["action_ref"] + assert "contains_pii" not in first["action_ref"] + assert "customer" not in first["action_ref"].lower() + + @pytest.mark.parametrize( "url", [ @@ -82,3 +93,80 @@ def fake_urlopen(request: Any, timeout: int) -> FakeResponse: "url": "https://checkpoint.example.com/review", "timeout": 10, } + + +@pytest.mark.parametrize( + ("payload", "match"), + [ + ("not-json", "invalid JSON"), + ("[]", "JSON object"), + ( + json.dumps( + { + "verdict": "escalate", + "reason": "Unsupported verdict.", + "decision_id": "dec_test", + "action_ref": "use-envelope-ref", + } + ), + "verdict must be one of", + ), + ( + json.dumps( + { + "verdict": "allow", + "reason": "", + "decision_id": "dec_test", + "action_ref": "use-envelope-ref", + } + ), + "reason must be", + ), + ( + json.dumps( + { + "verdict": "allow", + "reason": "Approved.", + "decision_id": "", + "action_ref": "use-envelope-ref", + } + ), + "decision_id must be", + ), + ], +) +def test_parse_remote_verdict_rejects_malformed_payloads( + payload: str, match: str +) -> None: + envelope = _sample_envelope() + payload = payload.replace("use-envelope-ref", envelope["action_ref"]) + + with pytest.raises(ValueError, match=match): + demo.parse_remote_verdict(payload, envelope) + + +def test_parse_remote_verdict_accepts_valid_payload() -> None: + envelope = _sample_envelope() + verdict = demo.parse_remote_verdict( + json.dumps( + { + "verdict": "require_approval", + "reason": "PII export requires human approval.", + "decision_id": "dec_test", + "action_ref": envelope["action_ref"], + } + ), + envelope, + ) + + assert verdict == { + "verdict": "require_approval", + "reason": "PII export requires human approval.", + "decision_id": "dec_test", + "action_ref": envelope["action_ref"], + } + + +def test_map_to_enforcement_rejects_unknown_verdict() -> None: + with pytest.raises(ValueError, match="Unsupported checkpoint verdict"): + demo.map_to_enforcement("escalate") # type: ignore[arg-type] From 6a016345eb564e826b3bab0136037fbde3cec907 Mon Sep 17 00:00:00 2001 From: jw_ond Date: Fri, 31 Jul 2026 00:15:00 +0800 Subject: [PATCH 5/8] docs: align external checkpoint proof object Signed-off-by: jw_ond --- examples/external-checkpoint-bridge/README.md | 7 ++++--- examples/external-checkpoint-bridge/demo.py | 8 +++++++- examples/external-checkpoint-bridge/test_demo.py | 14 ++++++++++++++ 3 files changed, 25 insertions(+), 4 deletions(-) diff --git a/examples/external-checkpoint-bridge/README.md b/examples/external-checkpoint-bridge/README.md index 95a9ed938..b041cdcfe 100644 --- a/examples/external-checkpoint-bridge/README.md +++ b/examples/external-checkpoint-bridge/README.md @@ -51,9 +51,10 @@ action envelope that AGT sent. The `action_ref` is intentionally opaque in this demo. The full action envelope is sent to the checkpoint for review, while the proof object stores only the stable -reference and verdict fields. Production deployments should use AGT's approved -digest/signature APIs or an external verifier when a cryptographic proof is -required. +reference plus the checkpoint decision fields shown in the sample output: +`decision_id`, `verdict`, and mapped `enforcement`. Production deployments should +use AGT's approved digest/signature APIs or an external verifier when a +cryptographic proof is required. ## Expected output diff --git a/examples/external-checkpoint-bridge/demo.py b/examples/external-checkpoint-bridge/demo.py index 71fe30dd8..ab58e95d7 100755 --- a/examples/external-checkpoint-bridge/demo.py +++ b/examples/external-checkpoint-bridge/demo.py @@ -51,7 +51,13 @@ class CheckpointVerdict(TypedDict): def stable_json(value: Any) -> str: """Serialize JSON deterministically for checkpoint review.""" - return json.dumps(value, sort_keys=True, separators=(",", ":")) + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + allow_nan=False, + ) def action_ref_for(value: Any) -> str: diff --git a/examples/external-checkpoint-bridge/test_demo.py b/examples/external-checkpoint-bridge/test_demo.py index 3bfd46908..0d47d8fc3 100644 --- a/examples/external-checkpoint-bridge/test_demo.py +++ b/examples/external-checkpoint-bridge/test_demo.py @@ -6,6 +6,7 @@ import importlib.util import json +import math import sys from pathlib import Path from typing import Any @@ -21,6 +22,19 @@ _spec.loader.exec_module(demo) # type: ignore[union-attr] +def test_stable_json_is_order_insensitive_and_unicode_preserving() -> None: + left = {"b": 2, "a": "東京"} + right = {"a": "東京", "b": 2} + + assert demo.stable_json(left) == demo.stable_json(right) + assert demo.stable_json(left) == '{"a":"東京","b":2}' + + +def test_stable_json_rejects_non_standard_numbers() -> None: + with pytest.raises(ValueError, match="Out of range float values"): + demo.stable_json({"value": math.nan}) + + def _sample_envelope() -> demo.ActionEnvelope: return demo.build_action_envelope( actor="agent:test", From 5f7222cdba24314c4e675f1d2cc5399d2f54cbf7 Mon Sep 17 00:00:00 2001 From: jw_ond Date: Fri, 31 Jul 2026 00:28:16 +0800 Subject: [PATCH 6/8] test: move external checkpoint coverage into root tests Signed-off-by: jw_ond --- examples/external-checkpoint-bridge/README.md | 6 ++++++ .../test_external_checkpoint_bridge.py | 17 +++++++++++++++-- 2 files changed, 21 insertions(+), 2 deletions(-) rename examples/external-checkpoint-bridge/test_demo.py => tests/test_external_checkpoint_bridge.py (91%) diff --git a/examples/external-checkpoint-bridge/README.md b/examples/external-checkpoint-bridge/README.md index b041cdcfe..c061c25dd 100644 --- a/examples/external-checkpoint-bridge/README.md +++ b/examples/external-checkpoint-bridge/README.md @@ -26,6 +26,12 @@ python examples/external-checkpoint-bridge/demo.py No API keys or third-party packages are required. By default, the demo uses a local checkpoint implementation. +To run the regression tests for this example: + +```bash +python -m pytest tests/test_external_checkpoint_bridge.py -q +``` + ## Optional remote checkpoint Set `EXTERNAL_CHECKPOINT_URL` to send each action envelope to an HTTPS endpoint: diff --git a/examples/external-checkpoint-bridge/test_demo.py b/tests/test_external_checkpoint_bridge.py similarity index 91% rename from examples/external-checkpoint-bridge/test_demo.py rename to tests/test_external_checkpoint_bridge.py index 0d47d8fc3..3078bdab0 100644 --- a/examples/external-checkpoint-bridge/test_demo.py +++ b/tests/test_external_checkpoint_bridge.py @@ -13,9 +13,10 @@ import pytest -_HERE = Path(__file__).resolve().parent +_REPO_ROOT = Path(__file__).resolve().parents[1] +_EXAMPLE_DIR = _REPO_ROOT / "examples" / "external-checkpoint-bridge" _spec = importlib.util.spec_from_file_location( - "external_checkpoint_bridge_demo", _HERE / "demo.py" + "external_checkpoint_bridge_demo", _EXAMPLE_DIR / "demo.py" ) demo = importlib.util.module_from_spec(_spec) # type: ignore[arg-type] sys.modules["external_checkpoint_bridge_demo"] = demo @@ -181,6 +182,18 @@ def test_parse_remote_verdict_accepts_valid_payload() -> None: } +@pytest.mark.parametrize( + ("verdict", "enforcement"), + [ + ("allow", "execute"), + ("require_approval", "pause_for_human_approval"), + ("deny", "block"), + ], +) +def test_map_to_enforcement(verdict: demo.Verdict, enforcement: demo.Enforcement) -> None: + assert demo.map_to_enforcement(verdict) == enforcement + + def test_map_to_enforcement_rejects_unknown_verdict() -> None: with pytest.raises(ValueError, match="Unsupported checkpoint verdict"): demo.map_to_enforcement("escalate") # type: ignore[arg-type] From 2d4936158caad6d81d7af96e6de85177ea8958f0 Mon Sep 17 00:00:00 2001 From: jw_ond Date: Fri, 31 Jul 2026 00:34:15 +0800 Subject: [PATCH 7/8] test: cover checkpoint review branches Signed-off-by: jw_ond --- tests/test_external_checkpoint_bridge.py | 55 ++++++++++++++++++++++++ 1 file changed, 55 insertions(+) diff --git a/tests/test_external_checkpoint_bridge.py b/tests/test_external_checkpoint_bridge.py index 3078bdab0..499691b96 100644 --- a/tests/test_external_checkpoint_bridge.py +++ b/tests/test_external_checkpoint_bridge.py @@ -58,6 +58,12 @@ def test_action_ref_is_stable_and_opaque() -> None: assert "customer" not in first["action_ref"].lower() +def test_local_checkpoint_returns_all_demo_verdicts() -> None: + verdicts = [demo.local_checkpoint(envelope)["verdict"] for envelope in demo.sample_actions()] + + assert verdicts == ["allow", "require_approval", "deny"] + + @pytest.mark.parametrize( "url", [ @@ -182,6 +188,55 @@ def test_parse_remote_verdict_accepts_valid_payload() -> None: } +def test_review_action_uses_local_checkpoint_by_default( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("EXTERNAL_CHECKPOINT_URL", raising=False) + envelope = _sample_envelope() + + verdict = demo.review_action(envelope) + + assert verdict["verdict"] == "require_approval" + assert verdict["action_ref"] == envelope["action_ref"] + + +def test_review_action_uses_remote_checkpoint_url( + monkeypatch: pytest.MonkeyPatch, +) -> None: + envelope = _sample_envelope() + monkeypatch.setenv("EXTERNAL_CHECKPOINT_URL", "https://checkpoint.example.com/review") + + class FakeResponse: + def __enter__(self) -> "FakeResponse": + return self + + def __exit__(self, *_args: object) -> None: + return None + + def read(self) -> bytes: + return json.dumps( + { + "verdict": "allow", + "reason": "Approved by remote checkpoint.", + "decision_id": "dec_remote", + "action_ref": envelope["action_ref"], + } + ).encode("utf-8") + + monkeypatch.setattr( + demo.urllib.request, "urlopen", lambda _request, timeout: FakeResponse() + ) + + verdict = demo.review_action(envelope) + + assert verdict == { + "verdict": "allow", + "reason": "Approved by remote checkpoint.", + "decision_id": "dec_remote", + "action_ref": envelope["action_ref"], + } + + @pytest.mark.parametrize( ("verdict", "enforcement"), [ From 71e709cbb0862051d6834d14cfc57086be1c6899 Mon Sep 17 00:00:00 2001 From: jw_ond Date: Fri, 31 Jul 2026 00:48:49 +0800 Subject: [PATCH 8/8] test: tidy external checkpoint test loader Signed-off-by: jw_ond --- examples/external-checkpoint-bridge/demo.py | 2 +- tests/test_external_checkpoint_bridge.py | 6 ++++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/examples/external-checkpoint-bridge/demo.py b/examples/external-checkpoint-bridge/demo.py index ab58e95d7..5483b994f 100755 --- a/examples/external-checkpoint-bridge/demo.py +++ b/examples/external-checkpoint-bridge/demo.py @@ -128,7 +128,7 @@ def remote_checkpoint(url: str, envelope: ActionEnvelope) -> CheckpointVerdict: request = urllib.request.Request( url, data=body, - headers={"content-type": "application/json"}, + headers={"Content-Type": "application/json"}, method="POST", ) with urllib.request.urlopen(request, timeout=10) as response: diff --git a/tests/test_external_checkpoint_bridge.py b/tests/test_external_checkpoint_bridge.py index 499691b96..0005e2965 100644 --- a/tests/test_external_checkpoint_bridge.py +++ b/tests/test_external_checkpoint_bridge.py @@ -18,9 +18,11 @@ _spec = importlib.util.spec_from_file_location( "external_checkpoint_bridge_demo", _EXAMPLE_DIR / "demo.py" ) -demo = importlib.util.module_from_spec(_spec) # type: ignore[arg-type] +assert _spec is not None +assert _spec.loader is not None +demo = importlib.util.module_from_spec(_spec) sys.modules["external_checkpoint_bridge_demo"] = demo -_spec.loader.exec_module(demo) # type: ignore[union-attr] +_spec.loader.exec_module(demo) def test_stable_json_is_order_insensitive_and_unicode_preserving() -> None: