Agent Harness 实战：状态、权限、工具与恢复

CREATE TABLE agent_runs (
  run_id TEXT PRIMARY KEY,
  state_version INTEGER NOT NULL,
  status TEXT NOT NULL,
  principal_id TEXT NOT NULL,
  tenant_id TEXT NOT NULL,
  policy_version TEXT NOT NULL,
  checkpoint_json TEXT NOT NULL
);

CREATE TABLE agent_effects (
  operation_key TEXT PRIMARY KEY,
  run_id TEXT NOT NULL,
  call_id TEXT NOT NULL,
  tool_name TEXT NOT NULL,
  tool_version TEXT NOT NULL,
  argument_digest TEXT NOT NULL,
  status TEXT NOT NULL,
  result_digest TEXT,
  external_reference TEXT
);

prepared -> dispatched -> committed
                       -> failed
                       -> unknown -> reconciled_committed | reconciled_absent

from __future__ import annotations

from dataclasses import dataclass, field
from hashlib import sha256
import json
from typing import Callable


@dataclass(frozen=True)
class Principal:
    subject: str
    tenant: str
    scopes: frozenset[str]


@dataclass(frozen=True)
class Proposal:
    call_id: str
    tool: str
    tool_version: str
    arguments: dict[str, object]


@dataclass(frozen=True)
class Approval:
    proposal_digest: str
    subject: str
    approver: str
    expires_at: int
    policy_version: str


@dataclass
class Run:
    run_id: str
    principal: Principal
    policy_version: str
    authorized_approvers: frozenset[str]
    max_steps: int
    steps: int = 0
    effects: dict[str, dict[str, object]] = field(default_factory=dict)
    events: list[dict[str, object]] = field(default_factory=list)


TOOLS = {
    "read_ticket": {"scope": "ticket:read", "risk": "read"},
    "close_ticket": {"scope": "ticket:write", "risk": "write"},
}


def digest(proposal: Proposal) -> str:
    payload = {
        "call_id": proposal.call_id,
        "tool": proposal.tool,
        "tool_version": proposal.tool_version,
        "arguments": proposal.arguments,
    }
    encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
    return sha256(encoded).hexdigest()


def authorize(run: Run, proposal: Proposal) -> str:
    metadata = TOOLS.get(proposal.tool)
    if metadata is None:
        return "deny"
    if metadata["scope"] not in run.principal.scopes:
        return "deny"
    if proposal.arguments.get("tenant") != run.principal.tenant:
        return "deny"
    return "approval_required" if metadata["risk"] == "write" else "allow"


def execute(
    run: Run,
    proposal: Proposal,
    tool: Callable[[dict[str, object], str], dict[str, object]],
    *,
    now: int,
    approval: Approval | None = None,
) -> dict[str, object]:
    run.steps += 1
    if run.steps > run.max_steps:
        raise RuntimeError("step budget exceeded")

    decision = authorize(run, proposal)
    run.events.append({"kind": "policy", "call_id": proposal.call_id,
                       "decision": decision})
    if decision == "deny":
        raise PermissionError("tool proposal denied")

    proposal_digest = digest(proposal)
    if decision == "approval_required":
        valid = (
            approval is not None
            and approval.proposal_digest == proposal_digest
            and approval.subject == run.principal.subject
            and approval.approver in run.authorized_approvers
            and approval.policy_version == run.policy_version
            and approval.expires_at >= now
        )
        if not valid:
            raise PermissionError("valid approval required")

    operation_key = (
        f"{run.run_id}:{proposal.call_id}:"
        f"{proposal.tool}:{proposal.tool_version}"
    )
    previous = run.effects.get(operation_key)
    if previous is not None:
        return previous

    run.effects[operation_key] = {"status": "dispatched"}
    try:
        result = tool(proposal.arguments, operation_key)
    except TimeoutError:
        run.effects[operation_key] = {"status": "unknown"}
        raise
    except Exception:
        run.effects[operation_key] = {"status": "failed"}
        raise

    record = {
        "status": "committed",
        "result": result,
        "result_digest": sha256(
            json.dumps(result, sort_keys=True).encode()
        ).hexdigest(),
    }
    run.effects[operation_key] = record
    run.events.append({"kind": "effect", "operation_key": operation_key,
                       "status": "committed"})
    return record


external_calls: dict[str, dict[str, object]] = {}


def close_ticket(
    arguments: dict[str, object],
    operation_key: str,
) -> dict[str, object]:
    if operation_key not in external_calls:
        external_calls[operation_key] = {
            "ticket_id": arguments["ticket_id"],
            "status": "closed",
        }
    return external_calls[operation_key]


principal = Principal(
    subject="user-42",
    tenant="acme",
    scopes=frozenset({"ticket:write"}),
)
run = Run(
    run_id="run-7",
    principal=principal,
    policy_version="policy-3",
    authorized_approvers=frozenset({"reviewer-5"}),
    max_steps=3,
)
proposal = Proposal(
    call_id="call-9",
    tool="close_ticket",
    tool_version="2",
    arguments={"tenant": "acme", "ticket_id": "T-100"},
)
approval = Approval(
    proposal_digest=digest(proposal),
    subject="user-42",
    approver="reviewer-5",
    expires_at=2_000_000_000,
    policy_version="policy-3",
)

first = execute(run, proposal, close_ticket, now=1_900_000_000,
                approval=approval)
replayed = execute(run, proposal, close_ticket, now=1_900_000_001,
                   approval=approval)

assert first == replayed
assert len(external_calls) == 1
assert first["status"] == "committed"
print(json.dumps({"result": first, "external_calls": len(external_calls)}))

{"result": {"status": "committed", "result": {"ticket_id": "T-100", "status": "closed"}, "result_digest": "<sha256>"}, "external_calls": 1}

{
  "server_id": "support-read",
  "tool_name": "get_ticket",
  "tool_version": "7",
  "schema_digest": "sha256:...",
  "risk_class": "sensitive_read",
  "required_scope": "ticket:read",
  "allowed_tenants": ["acme"],
  "max_result_bytes": 65536
}

{
  "approval_id": "approval-17",
  "run_id": "run-7",
  "call_id": "call-9",
  "actor": "user-42",
  "approver": "reviewer-5",
  "resource": "ticket:T-100",
  "tool_schema_digest": "sha256:...",
  "argument_digest": "sha256:...",
  "policy_version": "policy-3",
  "decision": "approve",
  "expires_at": "<timestamp>"
}

{
  "event": "tool_effect_resolved",
  "run_id": "run-7",
  "step_id": "step-4",
  "call_id": "call-9",
  "operation_key": "run-7:call-9:close_ticket:2",
  "tool": "close_ticket",
  "tool_version": "2",
  "policy_version": "policy-3",
  "policy_decision": "allow_after_approval",
  "effect_status": "committed",
  "result_digest": "sha256:...",
  "latency_ms": 84,
  "redactions": ["ticket_body", "access_token"]
}