diff --git a/examples/external-checkpoint-bridge/README.md b/examples/external-checkpoint-bridge/README.md new file mode 100644 index 000000000..c061c25dd --- /dev/null +++ b/examples/external-checkpoint-bridge/README.md @@ -0,0 +1,98 @@ +# 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 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. + +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. + +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: + +```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_ref": "..." +} +``` + +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 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 + +```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_ref": "...", + "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. +- 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. +- 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..5483b994f --- /dev/null +++ b/examples/external-checkpoint-bridge/demo.py @@ -0,0 +1,276 @@ +#!/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 json +import os +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): + action_ref: 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_ref: str + + +def stable_json(value: Any) -> str: + """Serialize JSON deterministically for checkpoint review.""" + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + allow_nan=False, + ) + + +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, + runtime: str, + tool_name: str, + proposed_action: str, + arguments: dict[str, Any], + policy_id: str, +) -> ActionEnvelope: + """Build an action envelope whose stable ref excludes review metadata.""" + ref_input = { + "actor": actor, + "runtime": runtime, + "tool_name": tool_name, + "proposed_action": proposed_action, + "arguments": arguments, + "policy_id": policy_id, + } + return { + "action_ref": action_ref_for(ref_input), + **ref_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['tool_name']}", + "action_ref": envelope["action_ref"], + } + + +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, + data=body, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(request, timeout=10) as response: + payload = response.read().decode("utf-8") + + 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, + "reason": reason, + "decision_id": decision_id, + "action_ref": envelope["action_ref"], + } + + +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" + if verdict == "deny": + return "block" + raise ValueError(f"Unsupported checkpoint verdict: {verdict!r}") + + +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_ref": verdict["action_ref"], + "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() diff --git a/tests/test_external_checkpoint_bridge.py b/tests/test_external_checkpoint_bridge.py new file mode 100644 index 000000000..0005e2965 --- /dev/null +++ b/tests/test_external_checkpoint_bridge.py @@ -0,0 +1,256 @@ +# 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 math +import sys +from pathlib import Path +from typing import Any + +import pytest + +_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", _EXAMPLE_DIR / "demo.py" +) +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) + + +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", + 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", + ) + + +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() + + +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", + [ + "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_ref_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_ref": "different-action-ref", + } + ).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_ref"): + demo.remote_checkpoint("https://checkpoint.example.com/review", envelope) + + assert observed == { + "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_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"), + [ + ("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]