Verify one OpenAI Agents SDK tool boundary before release

py -3.10 -m venv .venv
.\.venv\Scripts\python -m pip install "openai-agents @ git+https://github.com/openai/openai-agents-python.git@f53c62451c7f34c15f0c87fa745e05404025420f"
.\.venv\Scripts\python -m pip show openai-agents openai
.\.venv\Scripts\python -m pip freeze > resolved-requirements.txt

def build_agent(config: dict) -> Agent:
    """Construct the real existing static agent and original FunctionTool.

    Importing this module and building the agent cause no model request, tool
    effect or remote write. Preserve its actual schema and implementation.
    Use the OpenAI Responses route, a string model ID, static instructions,
    ModelSettings(preserve_raw_usage=True) and default run-again behavior.
    """

def assert_staging_target(config: dict) -> str:
    """Independently check actual project, target and permission identity.

    Return config['stagingTarget'] only when they match; cause no effect.
    """

def observe_effect(config: dict, case: dict, phase: str) -> dict:
    """Read an authoritative effect/audit record without causing the effect.

    Return {'target': exact_target,
            'count': nonnegative_integer_total_for_selected_action,
            'lastArguments': actual_argument_object_or_None,
            'effect': actual_effect_object_or_None}.
    Raise on absent, uncertain or negative counts. A verified after count
    must equal the before count plus one, with matching arguments/effect.
    """

.\.venv\Scripts\python boundary_check.py inspect --dir .
.\.venv\Scripts\python boundary_check.py test --dir .

.\.venv\Scripts\python boundary_check.py stage --dir . --authorize-exact-run

.\.venv\Scripts\python boundary_check.py finalize --dir .

"""Reference recipe for one existing OpenAI Agents SDK FunctionTool.

Requires the SDK source pinned in RECIPE.md and reader-owned policy.json,
cases.json, config.json and reader_adapter.py. No fixture or account is bundled.
Never run `stage` merely to test this reference file.
"""

from __future__ import annotations

import argparse
import asyncio
import dataclasses
from datetime import datetime, timezone
import hashlib
import importlib.util
import importlib.metadata as metadata
import json
import os
from pathlib import Path
import secrets
import sys
from typing import Any

from agents import Agent, RunConfig, Runner, ToolGuardrailFunctionOutput, flush_traces, trace
from agents.exceptions import ToolInputGuardrailTripwireTriggered
from agents.testing import ScriptedModel, assistant_message, function_call
from agents.tool import FunctionTool
from agents.tool_guardrails import tool_input_guardrail


def digest(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def canonical(value: Any) -> bytes:
    return json.dumps(value, sort_keys=True, ensure_ascii=False, allow_nan=False,
                      separators=(",", ":")).encode("utf-8")


def no_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate JSON key")
        result[key] = value
    return result


def bad_constant(value: str) -> None:
    raise ValueError("non-finite JSON value")


def parse_json(raw: str) -> Any:
    return json.loads(raw, object_pairs_hook=no_duplicate_keys,
                      parse_constant=bad_constant)


def load_exact(path: Path, keys: set[str]) -> tuple[dict[str, Any], str]:
    raw = path.read_bytes()
    data = parse_json(raw.decode("utf-8"))
    if not isinstance(data, dict) or set(data) != keys:
        raise ValueError(f"invalid fields in {path.name}")
    return data, digest(raw)


def required_text(value: Any) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ValueError("missing reader-supplied text")
    return value


def require_args(value: Any) -> dict[str, Any]:
    if not isinstance(value, dict) or not value or any(not isinstance(k, str) for k in value):
        raise ValueError("tool arguments must be a nonempty JSON object")
    canonical(value)  # rejects non-finite numbers
    return value


def reader_files(directory: Path) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any], dict[str, str]]:
    policy, ph = load_exact(directory / "policy.json", {"toolName", "toolSchemaSha256", "allowedArguments", "consumptionPlan"})
    cases, ch = load_exact(directory / "cases.json", {"allowed", "denied"})
    config, gh = load_exact(directory / "config.json", {"agentName", "modelId", "projectId",
            "stagingTarget", "approvedBy", "privacyApprovedAtUtc", "traceDestination"})
    required_text(policy["toolName"])
    if not isinstance(policy["toolSchemaSha256"], str) or len(policy["toolSchemaSha256"]) != 64:
        raise ValueError("tool schema SHA-256 required")
    require_args(policy["allowedArguments"])
    plan = policy["consumptionPlan"]
    if not isinstance(plan, dict) or set(plan) != {"maxRequests", "maxInputTokens", "maxOutputTokens", "approvedSpendDescription", "pricingSource", "billingCaveat"}:
        raise ValueError("preapproved consumption plan required")
    if any(type(plan[k]) is not int or plan[k] < 0 for k in ("maxRequests", "maxInputTokens", "maxOutputTokens")):
        raise ValueError("consumption allowances must be nonnegative integers")
    for k in ("approvedSpendDescription", "pricingSource", "billingCaveat"):
        required_text(plan[k])
    for case_name, fields in (("allowed", {"prompt", "arguments", "expectedEffect"}),
                              ("denied", {"prompt", "arguments"})):
        case = cases[case_name]
        if not isinstance(case, dict) or set(case) != fields:
            raise ValueError(f"invalid {case_name} case fields")
        required_text(case["prompt"])
        require_args(case["arguments"])
    if canonical(cases["allowed"]["arguments"]) != canonical(policy["allowedArguments"]):
        raise ValueError("allowed case differs from approved policy")
    if canonical(cases["denied"]["arguments"]) == canonical(policy["allowedArguments"]):
        raise ValueError("denied case is approved by policy")
    if not isinstance(cases["allowed"]["expectedEffect"], dict):
        raise ValueError("expectedEffect must be a reader-approved object")
    for value in config.values():
        required_text(value)
    if config["traceDestination"] != "openai-dashboard":
        raise ValueError("this recipe supports only approved OpenAI dashboard traces")
    return policy, cases, config, {"policy": ph, "cases": ch, "config": gh}


def load_adapter(path: Path) -> Any:
    spec = importlib.util.spec_from_file_location("reader_adapter", path)
    if spec is None or spec.loader is None:
        raise ValueError("reader adapter could not be loaded")
    module = importlib.util.module_from_spec(spec)
    sys.path.insert(0, str(path.parent))
    try:
        spec.loader.exec_module(module)
    finally:
        sys.path.pop(0)
    for name in ("build_agent", "assert_staging_target", "observe_effect"):
        if not callable(getattr(module, name, None)):
            raise ValueError(f"reader adapter must define {name}")
    return module


def actual_agent_state(agent: Agent[Any], tool: FunctionTool) -> dict[str, Any]:
    # Unsupported/non-JSON settings fail closed. No repr(), which may hide drift.
    value = {"instructions": agent.instructions, "modelSettings": dataclasses.asdict(agent.model_settings),
             "toolDescription": tool.description, "toolStrict": tool.strict_json_schema,
             "toolSchema": tool.params_json_schema, "resetToolChoice": agent.reset_tool_choice,
             "runner": {"maxTurns": 2, "stageTracingDisabled": False,
                        "traceSensitiveData": False, "testTracingDisabled": True}}
    canonical(value)
    return value


def runtime_state() -> dict[str, Any]:
    sdk = metadata.distribution("openai-agents")
    direct = json.loads(sdk.read_text("direct_url.json") or "null")
    commit = (direct or {}).get("vcs_info", {}).get("commit_id")
    if commit != "f53c62451c7f34c15f0c87fa745e05404025420f" or sdk.version != "0.23.1":
        raise ValueError("SDK source commit/version differs from reviewed pin")
    packages = sorted((d.metadata["Name"].lower(), d.version) for d in metadata.distributions())
    return {"sdkCommit": commit, "sdkVersion": sdk.version,
            "openaiVersion": metadata.version("openai"),
            "dependenciesSha256": digest(canonical(packages))}


def application_lock(directory: Path, agent: Agent[Any], tool: FunctionTool) -> tuple[str, dict[str, Any]]:
    lock, _ = load_exact(directory / "application-lock.json",
        {"agentStateSha256", "runtimeSha256", "sourceFiles", "toolImplementationFile",
         "reviewedBy", "reviewedAtUtc", "attestsComplete"})
    if lock["attestsComplete"] is not True:
        raise ValueError("owner has not attested complete application inventory")
    required_text(lock["reviewedBy"]); required_text(lock["reviewedAtUtc"])
    files = lock["sourceFiles"]
    if not isinstance(files, dict) or not files or not isinstance(lock["toolImplementationFile"], str):
        raise ValueError("source inventory and original tool implementation file required")
    if lock["toolImplementationFile"] not in files or "reader_adapter.py" not in files:
        raise ValueError("tool implementation and adapter must be inventoried")
    for name, expected in files.items():
        if not isinstance(name, str) or not isinstance(expected, str) or len(expected) != 64:
            raise ValueError("invalid source inventory entry")
        path = (directory / name).resolve(strict=True)
        if not path.is_file() or digest(path.read_bytes()) != expected:
            raise ValueError(f"application source changed: {name}")
    if digest(canonical(actual_agent_state(agent, tool))) != lock["agentStateSha256"]:
        raise ValueError("actual built agent instructions/settings/tool changed")
    if digest(canonical(runtime_state())) != lock["runtimeSha256"]:
        raise ValueError("installed SDK/dependencies changed")
    return digest(canonical(lock)), lock


def provider_usage_evidence(result: Any) -> list[dict[str, Any]]:
    """Keep response identities, explicit field presence and a private payload hash."""
    evidence: list[dict[str, Any]] = []
    for response in result.raw_responses:
        raw = response.raw_usage
        raw_hash = digest(canonical(raw)) if isinstance(raw, dict) else None
        evidence.append({"responseId": response.response_id,
                         "requestId": response.request_id,
                         "rawUsageSha256": raw_hash,
                         "inputTokens": raw.get("input_tokens") if isinstance(raw, dict) else None,
                         "outputTokens": raw.get("output_tokens") if isinstance(raw, dict) else None})
    return evidence


def usage_acceptable(usage: Any, evidence: Any, plan: dict[str, Any], review: dict[str, Any]) -> bool:
    keys = ("requests", "inputTokens", "outputTokens")
    return (isinstance(usage, dict) and set(usage) == set(keys)
            and all(type(usage[k]) is int and usage[k] >= 0 for k in keys)
            and usage["requests"] >= 1
            and isinstance(evidence, list) and len(evidence) == usage["requests"]
            and all(isinstance(item, dict)
                    and set(item) == {"responseId", "requestId", "rawUsageSha256", "inputTokens", "outputTokens"}
                    and isinstance(item["responseId"], str) and bool(item["responseId"])
                    and (item["requestId"] is None or isinstance(item["requestId"], str))
                    and isinstance(item["rawUsageSha256"], str) and len(item["rawUsageSha256"]) == 64
                    and type(item["inputTokens"]) is int and item["inputTokens"] >= 0
                    and type(item["outputTokens"]) is int and item["outputTokens"] >= 0
                    for item in evidence)
            and len({item["responseId"] for item in evidence}) == len(evidence)
            and sum(item["inputTokens"] for item in evidence) == usage["inputTokens"]
            and sum(item["outputTokens"] for item in evidence) == usage["outputTokens"]
            and all(usage[k] <= plan[m] for k, m in zip(keys, ("maxRequests", "maxInputTokens", "maxOutputTokens")))
            and review["usageAccepted"] is True
            and review["usageSha256"] == digest(canonical(usage))
            and review["usageEvidenceSha256"] == digest(canonical(evidence))
            and isinstance(review["billingReview"], str)
            and bool(review["billingReview"].strip()))


def valid_effect_count(value: Any) -> bool:
    return type(value) is int and value >= 0


def inspect(directory: Path) -> None:
    config, _ = load_exact(directory / "config.json", {"agentName", "modelId", "projectId",
            "stagingTarget", "approvedBy", "privacyApprovedAtUtc", "traceDestination"})
    adapter = load_adapter(directory / "reader_adapter.py")
    agent = adapter.build_agent(config)
    if (not isinstance(agent, Agent) or not isinstance(agent.model, str)
            or len(agent.tools) != 1 or not isinstance(agent.tools[0], FunctionTool)):
        raise ValueError("expected one existing agent with one FunctionTool")
    tool = agent.tools[0]
    print(json.dumps({"agentName": agent.name, "modelId": agent.model,
                      "toolName": tool.name, "toolSchemaSha256": digest(canonical(tool.params_json_schema)),
                      "agentStateSha256": digest(canonical(actual_agent_state(agent, tool))),
                      "runtimeSha256": digest(canonical(runtime_state()))},
                     sort_keys=True))


def selected_tool(agent: Agent[Any], policy: dict[str, Any], config: dict[str, Any]) -> FunctionTool:
    if not isinstance(agent, Agent) or agent.name != config["agentName"]:
        raise ValueError("unexpected agent identity")
    if not isinstance(agent.model, str) or agent.model != config["modelId"]:
        raise ValueError("model identity cannot be verified against approved config")
    if agent.model_settings.preserve_raw_usage is not True:
        raise ValueError("agent must preserve raw provider usage before testing or staging")
    if agent.handoffs or agent.mcp_servers or len(agent.tools) != 1:
        raise ValueError("agent has another reachable tool, MCP server or handoff")
    if (agent.input_guardrails or agent.output_guardrails or agent.hooks is not None
            or agent.prompt is not None or agent.output_type is not None
            or agent.tool_use_behavior != "run_llm_again"
            or not isinstance(agent.instructions, str)):
        raise ValueError("dynamic agent path or prior guardrail requires a separate review")
    tool = agent.tools[0]
    if (not isinstance(tool, FunctionTool) or tool.name != policy["toolName"]
            or tool.qualified_name != tool.name):
        raise ValueError("selected FunctionTool identity differs")
    if tool.is_enabled is not True or tool.needs_approval is not False:
        raise ValueError("dynamic availability or approval needs a separate recipe")
    if tool.tool_input_guardrails or tool.tool_output_guardrails:
        raise ValueError("prior tool guardrails need a separate effect and ordering review")
    if digest(canonical(tool.params_json_schema)) != policy["toolSchemaSha256"]:
        raise ValueError("selected tool schema differs from approved schema")
    return tool


def guarded_agent(agent: Agent[Any], tool: FunctionTool, policy: dict[str, Any],
                  invoke: Any) -> tuple[Agent[Any], dict[str, int]]:
    state = {"allowed": 0}

    @tool_input_guardrail(name="exact_approved_tool_arguments")
    def exact_gate(data: Any) -> ToolGuardrailFunctionOutput:
        try:
            args = require_args(parse_json(data.context.tool_arguments))
            permitted = canonical(args) == canonical(policy["allowedArguments"]) and state["allowed"] == 0
        except (ValueError, TypeError, UnicodeError):
            permitted = False
        if not permitted:
            return ToolGuardrailFunctionOutput.raise_exception(output_info={"decision": "denied"})
        state["allowed"] += 1
        return ToolGuardrailFunctionOutput.allow(output_info={"decision": "allowed"})

    existing = list(tool.tool_input_guardrails or [])
    attached = dataclasses.replace(tool, on_invoke_tool=invoke,
                                   tool_input_guardrails=[*existing, exact_gate])
    clone = agent.clone(tools=[attached])
    # Parse/duplicate/non-finite gates are pure and deny before SDK invocation.
    for malformed in ('{"x":1,"x":2}', '{"x":NaN}', '[]'):
        try:
            parsed = require_args(parse_json(malformed))
            assert parsed != policy["allowedArguments"]
        except (ValueError, TypeError):
            pass
    return clone, state


async def deterministic(directory: Path) -> dict[str, Any]:
    policy, cases, config, hashes = reader_files(directory)
    adapter_path = directory / "reader_adapter.py"
    adapter = load_adapter(adapter_path)
    original = adapter.build_agent(config)
    tool = selected_tool(original, policy, config)
    app_fingerprint, _ = application_lock(directory, original, tool)
    calls: list[dict[str, Any]] = []

    async def fake_effect(context: Any, raw: str) -> str:
        args = require_args(parse_json(raw))
        calls.append(args)
        return "controlled test effect"

    allowed_model = ScriptedModel([
        [function_call(tool.name, cases["allowed"]["arguments"], call_id="allowed_call")],
        [assistant_message("controlled completion")],
    ])
    allowed_base = original.clone(model=allowed_model)
    allowed_agent, allowed_state = guarded_agent(allowed_base, tool, policy, fake_effect)
    allowed_result = await Runner.run(allowed_agent, cases["allowed"]["prompt"],
                    max_turns=2, run_config=RunConfig(tracing_disabled=True))
    allowed_model.assert_complete()
    if allowed_result.final_output != "controlled completion" or calls != [policy["allowedArguments"]] or allowed_state["allowed"] != 1:
        raise AssertionError("allowed tool invocation/effect mismatch")

    calls.clear()
    denied_model = ScriptedModel([
        [function_call(tool.name, cases["denied"]["arguments"], call_id="denied_call")],
    ])
    denied_base = original.clone(model=denied_model)
    denied_agent, denied_state = guarded_agent(denied_base, tool, policy, fake_effect)
    try:
        await Runner.run(denied_agent, cases["denied"]["prompt"], max_turns=2,
                         run_config=RunConfig(tracing_disabled=True))
        raise AssertionError("denied tool call did not trip")
    except ToolInputGuardrailTripwireTriggered:
        pass
    denied_model.assert_complete()
    if calls or denied_state["allowed"] != 0:
        raise AssertionError("denied path executed selected tool")

    return {"status": "deterministic-pass", "hashes": hashes,
            "applicationFingerprint": app_fingerprint,
            "adapterSha256": digest(adapter_path.read_bytes()),
            "harnessSha256": digest(Path(__file__).read_bytes()),
            "toolSchemaSha256": policy["toolSchemaSha256"],
            "allowedInvocations": 1, "deniedInvocations": 0,
            "limit": "SDK loop and guarded tool only; provider and external effect untested"}


def write_exclusive(path: Path, value: dict[str, Any]) -> None:
    payload = canonical(value) + b"\n"
    with path.open("xb") as output:
        output.write(payload)
        output.flush()
        os.fsync(output.fileno())


def finalize(directory: Path) -> None:
    stage_path = directory / "stage-receipt.json"
    stage_data, stage_hash = load_exact(stage_path,
        {"status", "completedAtUtc", "intentSha256", "traceId", "traceDestination", "traceFlush",
         "failureType", "invoked", "gateAllows", "before", "after", "usage", "usageEvidence",
         "effectVerified", "finalOutputSha256", "next"})
    intent_path = directory / "stage-intent.json"
    intent, intent_hash = load_exact(intent_path,
        {"startedAtUtc", "traceId", "hashes", "adapterSha256", "harnessSha256",
         "testReceiptSha256", "applicationFingerprint", "target", "decision"})
    _, _, config, current_hashes = reader_files(directory)
    current_adapter = load_adapter(directory / "reader_adapter.py")
    current_agent = current_adapter.build_agent(config)
    current_policy, _, _, _ = reader_files(directory)
    current_tool = selected_tool(current_agent, current_policy, config)
    current_app_fingerprint, _ = application_lock(directory, current_agent, current_tool)
    if (stage_data["intentSha256"] != intent_hash or stage_data["traceId"] != intent["traceId"]
            or intent["hashes"] != current_hashes
            or intent["applicationFingerprint"] != current_app_fingerprint
            or intent["adapterSha256"] != digest((directory / "reader_adapter.py").read_bytes())
            or intent["harnessSha256"] != digest(Path(__file__).read_bytes())
            or intent["testReceiptSha256"] != digest((directory / "deterministic-receipt.json").read_bytes())
            or intent["target"] != config["stagingTarget"]):
        raise ValueError("staging intent or bound code/config/cases changed")
    review, review_hash = load_exact(directory / "owner-review.json",
        {"stageReceiptSha256", "traceId", "traceSeen", "traceSensitiveDataAbsent",
         "finalOutputSha256", "finalOutputAccepted", "reviewer", "reviewedAtUtc",
         "usageSha256", "usageEvidenceSha256", "usageAccepted", "billingReview"})
    if review["stageReceiptSha256"] != stage_hash or review["traceId"] != stage_data["traceId"]:
        raise ValueError("owner review does not bind this staging attempt")
    if review["finalOutputSha256"] != stage_data["finalOutputSha256"]:
        raise ValueError("owner review does not bind this private output")
    if review["finalOutputSha256"] != digest((directory / "final-output-private.json").read_bytes()):
        raise ValueError("private output changed after staging")
    required_text(review["reviewer"])
    required_text(review["reviewedAtUtc"])
    try:
        reviewed_at = datetime.fromisoformat(review["reviewedAtUtc"].replace("Z", "+00:00"))
        completed_at = datetime.fromisoformat(stage_data["completedAtUtc"].replace("Z", "+00:00"))
        if reviewed_at.tzinfo is None or reviewed_at < completed_at:
            raise ValueError("review predates stage completion")
    except (TypeError, ValueError) as error:
        raise ValueError("review timestamp is not a valid later UTC time") from error
    passes = (stage_data["status"] == "HOLD" and stage_data["failureType"] is None
              and stage_data["effectVerified"] is True and stage_data["gateAllows"] == 1
              and usage_acceptable(stage_data["usage"], stage_data["usageEvidence"],
                                   current_policy["consumptionPlan"], review)
              and stage_data["traceFlush"] == "completed"
              and stage_data["traceDestination"] == "openai-dashboard"
              and review["traceSeen"] is True and review["traceSensitiveDataAbsent"] is True
              and review["finalOutputAccepted"] is True)
    write_exclusive(directory / "acceptance-record.json",
        {"decision": "ACCEPT" if passes else "HOLD",
         "stageReceiptSha256": stage_hash, "ownerReviewSha256": review_hash,
         "applicationFingerprint": current_app_fingerprint,
         "traceId": stage_data["traceId"], "reviewer": review["reviewer"],
         "scope": "one selected FunctionTool, one approved case and exact staging target only"})
    print("Bounded one-tool decision:", "ACCEPT" if passes else "HOLD")


async def stage(directory: Path, authorize: bool) -> None:
    if not authorize:
        raise ValueError("stage requires --authorize-exact-run from the reader's approved decision")
    test_receipt, _ = load_exact(directory / "deterministic-receipt.json",
        {"status", "hashes", "applicationFingerprint", "adapterSha256", "harnessSha256", "toolSchemaSha256",
         "allowedInvocations", "deniedInvocations", "limit"})
    fresh_test = await deterministic(directory)
    if test_receipt != fresh_test:
        raise ValueError("deterministic receipt changed; do not run staging")
    policy, cases, config, hashes = reader_files(directory)
    adapter_path = directory / "reader_adapter.py"
    adapter = load_adapter(adapter_path)
    agent = adapter.build_agent(config)
    tool = selected_tool(agent, policy, config)
    app_fingerprint, _ = application_lock(directory, agent, tool)
    if app_fingerprint != test_receipt["applicationFingerprint"]:
        raise ValueError("staging agent/application differs from tested revision")
    target = adapter.assert_staging_target(config)
    if target != config["stagingTarget"]:
        raise ValueError("staging target differs from approved config")
    before = adapter.observe_effect(config, cases["allowed"], "before")
    if not isinstance(before, dict) or before.get("target") != target or not valid_effect_count(before.get("count")):
        raise ValueError("staging effect baseline missing or unverified")

    # Exclusive intent is durable before any provider request. Never replay it automatically.
    trace_id = "trace_" + secrets.token_hex(16)
    intent = {"startedAtUtc": utc_now(), "traceId": trace_id, "hashes": hashes,
              "applicationFingerprint": app_fingerprint,
              "adapterSha256": digest(adapter_path.read_bytes()),
              "harnessSha256": digest(Path(__file__).read_bytes()),
              "testReceiptSha256": digest((directory / "deterministic-receipt.json").read_bytes()),
              "target": target, "decision": "one authorized staging attempt; uncertain effects require reconciliation"}
    write_exclusive(directory / "stage-intent.json", intent)

    invoked: list[dict[str, Any]] = []
    async def real_effect(context: Any, raw: str) -> Any:
        invoked.append(require_args(parse_json(raw)))
        return await tool.on_invoke_tool(context, raw)

    stage_agent, gate_state = guarded_agent(agent, tool, policy, real_effect)
    failure_type: str | None = None
    output_hash: str | None = None
    usage: dict[str, Any] = {"requests": "unknown", "inputTokens": "unknown", "outputTokens": "unknown"}
    usage_evidence: list[dict[str, Any]] = []
    try:
        # The owner has approved this destination and redacted spans before this call.
        with trace("one_tool_boundary_stage", trace_id=trace_id):
            result = await Runner.run(stage_agent, cases["allowed"]["prompt"], max_turns=2,
                      run_config=RunConfig(tracing_disabled=False, trace_include_sensitive_data=False))
        final_text = str(result.final_output)
        write_exclusive(directory / "final-output-private.json", {"output": final_text})
        output_hash = digest(canonical({"output": final_text}) + b"\n")
        measured = result.context_wrapper.usage
        for key, field in (("requests", "requests"), ("inputTokens", "input_tokens"),
                           ("outputTokens", "output_tokens")):
            value = getattr(measured, field, None)
            usage[key] = value if type(value) is int else "unknown"
        usage_evidence = provider_usage_evidence(result)
    except Exception as error:
        failure_type = type(error).__name__
    finally:
        try:
            flush_traces()
            trace_flush = "completed"
        except Exception:
            trace_flush = "failed"
        try:
            after = adapter.observe_effect(config, cases["allowed"], "after")
        except Exception:
            after = {"error": "effect readback failed"}
        receipt = {"status": "HOLD", "completedAtUtc": utc_now(),
                   "intentSha256": digest((directory / "stage-intent.json").read_bytes()),
                   "traceId": trace_id, "traceDestination": config["traceDestination"],
                   "traceFlush": trace_flush, "failureType": failure_type,
                   "invoked": invoked, "gateAllows": gate_state["allowed"],
                   "before": before, "after": after, "usage": usage,
                   "usageEvidence": usage_evidence,
                   "effectVerified": (failure_type is None and len(invoked) == 1
                       and canonical(invoked[0]) == canonical(policy["allowedArguments"])
                       and isinstance(after, dict) and after.get("target") == target
                       and valid_effect_count(after.get("count"))
                       and after["count"] == before["count"] + 1
                       and canonical(after.get("lastArguments")) == canonical(policy["allowedArguments"])
                       and canonical(after.get("effect")) == canonical(cases["allowed"]["expectedEffect"])),
                   "finalOutputSha256": output_hash,
                   "next": "Inspect exact staging effect, private output and dashboard trace; never replay uncertain action"}
        write_exclusive(directory / "stage-receipt.json", receipt)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("command", choices=["inspect", "test", "stage", "finalize"])
    parser.add_argument("--dir", type=Path, required=True)
    parser.add_argument("--authorize-exact-run", action="store_true")
    args = parser.parse_args()
    directory = args.dir.resolve(strict=True)
    if args.command == "inspect":
        inspect(directory)
    elif args.command == "test":
        result = asyncio.run(deterministic(directory))
        write_exclusive(directory / "deterministic-receipt.json", result)
        print("Deterministic allowed/denied tool boundary passed; receipt saved.")
    elif args.command == "stage":
        asyncio.run(stage(directory, args.authorize_exact_run))
        print("One staging attempt recorded as HOLD pending owner effect and trace review.")
    else:
        finalize(directory)


if __name__ == "__main__":
    main()