Agent Commerce Network

"""Procurement agent: reasons about security review procurement using the Agents SDK."""

from __future__ import annotations

import asyncio
import hashlib
import json
import logging
import os
from typing import Any

import httpx

log = logging.getLogger("commerce-network")


PROCUREMENT_PROMPT = """\
You are a procurement agent for a platform security team. Your job is to
procure security reviews from an external provider through a governed API.

Available tools:
- request_quote: Submit a quote request for a security review scope
- create_job: Accept a quote and create a governed job (will check budget)
- get_job: Check the status of a job
- dispute_job: Open a dispute if the review quality is insufficient

Workflow:
1. Request a quote for the specified scope and target
2. Evaluate the quote (price, scope match, approval requirements)
3. Create a job to accept the quote (set budget appropriately)
4. If approval is required, stop and return pending_approval for an independent operator
5. Check the job status to confirm fulfillment
6. If the review has issues, consider disputing

Respond with JSON:
{
  "quote_id": "...",
  "job_id": "...",
  "final_status": "fulfilled|denied_budget|disputed|pending_approval",
  "price_minor": <int>,
  "currency": "USD",
  "rationale": "why you made these decisions"
}"""

FAILED_OUTPUT = {
    "final_status": "outcome_unknown",
    "rationale": "The model run did not establish a completed job",
}


def _get_provider() -> str:
    selected = os.getenv("COMMERCE_PROVIDER")
    if selected:
        if selected not in {"direct", "openai", "anthropic"}:
            raise ValueError("COMMERCE_PROVIDER must be direct, openai or anthropic")
        return selected
    if os.getenv("OPENAI_API_KEY"):
        return "openai"
    if os.getenv("ANTHROPIC_API_KEY") or os.getenv("ANTHROPIC_AUTH_TOKEN"):
        return "anthropic"
    return "direct"


def run_procurement_agent(
    *,
    buyer_url: str,
    auth_token: str,
    capability_token: dict | None = None,
    scope: str,
    target: str,
    budget_minor: int,
    release_window: str | None = None,
    max_turns: int = 10,
) -> dict[str, Any]:
    """Run the procurement agent. Uses Agents SDK, Anthropic, or the direct workflow."""
    provider = _get_provider()
    http = httpx.Client(timeout=30.0)
    call_log: list[dict] = []

    # Serialize capability token for the Chio sidecar header
    cap_header = json.dumps(capability_token, separators=(",", ":")) if capability_token else None

    def _call_buyer(method: str, path: str, body: dict | None = None) -> dict:
        headers: dict[str, str] = {"Authorization": f"Bearer {auth_token}"}
        if cap_header:
            headers["X-Chio-Capability"] = cap_header
        if method == "GET":
            r = http.get(f"{buyer_url}{path}", headers=headers)
        else:
            r = http.post(f"{buyer_url}{path}", headers=headers, json=body)
        r.raise_for_status()
        result = r.json()
        receipt_id = r.headers.get("X-Chio-Receipt-Id")
        if not receipt_id:
            raise RuntimeError("The buyer response has no Chio HTTP receipt association")
        result["_chio_http"] = {
            "receipt_id": receipt_id,
            "method": method,
            "path": path,
            "body": body,
            "body_sha256": hashlib.sha256(r.request.content).hexdigest()
            if r.request.content
            else None,
            "capability_id": capability_token["id"] if capability_token else None,
            "status": r.status_code,
        }
        return result

    def request_quote(
        service_family: str = "security-review",
        target: str = target,
        requested_scope: str = scope,
        release_window: str | None = release_window,
    ) -> dict:
        return _call_buyer(
            "POST",
            "/procurement/quote-requests",
            {
                "service_family": service_family,
                "target": target,
                "requested_scope": requested_scope,
                "release_window": release_window,
            },
        )

    def create_job(
        quote_id: str,
        provider_id: str = "vanguard-security",
        service_family: str = "security-review",
        budget_minor: int = budget_minor,
    ) -> dict:
        return _call_buyer(
            "POST",
            "/procurement/jobs",
            {
                "quote_id": quote_id,
                "provider_id": provider_id,
                "service_family": service_family,
                "budget_minor": budget_minor,
            },
        )

    def approve_job(*args, **kwargs):
        raise PermissionError("Approval belongs to an independent operator")

    def get_job(job_id: str) -> dict:
        return _call_buyer("GET", f"/procurement/jobs/{job_id}")

    def dispute_job(
        job_id: str, reason_code: str = "quality", summary: str = "insufficient findings"
    ) -> dict:
        return _call_buyer(
            "POST",
            f"/procurement/jobs/{job_id}/disputes",
            {
                "reason_code": reason_code,
                "summary": summary,
            },
        )

    try:
        if provider == "direct":
            return _direct_flow(request_quote, create_job, approve_job, get_job, call_log, scope)

        if provider == "openai":
            return _agents_sdk_flow(
                request_quote,
                create_job,
                approve_job,
                get_job,
                dispute_job,
                call_log,
                scope,
                target,
                budget_minor,
                max_turns,
            )

        return _anthropic_flow(
            request_quote,
            create_job,
            approve_job,
            get_job,
            dispute_job,
            call_log,
            scope,
            target,
            budget_minor,
            max_turns,
        )
    finally:
        http.close()


def _direct_flow(request_quote, create_job, approve_job, get_job, call_log, scope):
    """Deterministic flow for CI -- no LLM needed."""
    log.info("[procurement] direct workflow")
    quote_resp = request_quote(requested_scope=scope)
    call_log.append({"tool": "request_quote", "output": quote_resp})

    quote = quote_resp["quote"]
    job_resp = create_job(quote_id=quote["quote_id"])
    call_log.append({"tool": "create_job", "output": job_resp})

    return {
        "quote_id": quote["quote_id"],
        "job_id": job_resp["job_id"],
        "final_status": job_resp["status"],
        "price_minor": quote["price_minor"],
        "currency": quote.get("currency", "USD"),
        "rationale": "direct workflow",
        "mode": "direct",
        "tool_calls": call_log,
    }


def _agents_sdk_flow(
    request_quote,
    create_job,
    approve_job,
    get_job,
    dispute_job,
    call_log,
    scope,
    target,
    budget_minor,
    max_turns,
):
    """Run via OpenAI Agents SDK."""
    try:
        from agents import (
            Agent,
            FunctionTool,
            OpenAIChatCompletionsModel,
            Runner,
            set_tracing_disabled,
        )
        from openai import AsyncOpenAI
    except ImportError:
        raise RuntimeError(
            "Install the locked project dependencies for OpenAI Agents SDK"
        ) from None

    def _wrap(fn, name, desc, schema):
        async def invoke(ctx, args_json):
            args = json.loads(args_json) if args_json else {}
            log.info("[procurement] %s(%s)", name, json.dumps(args)[:200])
            try:
                out = fn(**args)
            except Exception as exc:
                out = {"error": str(exc)}
            call_log.append({"tool": name, "input": args, "output": out})
            return json.dumps(out)

        return FunctionTool(
            name=name,
            description=desc,
            params_json_schema=schema,
            on_invoke_tool=invoke,
            strict_json_schema=False,
        )

    tools = [
        _wrap(
            request_quote,
            "request_quote",
            "Request a security review quote",
            {
                "type": "object",
                "properties": {
                    "requested_scope": {
                        "type": "string",
                        "enum": [
                            "hotfix-review",
                            "release-review",
                            "release-plus-cloud-review",
                            "full-estate-review",
                        ],
                    },
                    "target": {"type": "string"},
                    "service_family": {"type": "string", "default": "security-review"},
                },
                "required": ["requested_scope", "target"],
            },
        ),
        _wrap(
            create_job,
            "create_job",
            "Accept a quote and create a governed procurement job",
            {
                "type": "object",
                "properties": {
                    "quote_id": {"type": "string"},
                    "budget_minor": {
                        "type": "integer",
                        "description": "Budget in minor currency units (cents)",
                    },
                    "provider_id": {"type": "string", "default": "vanguard-security"},
                    "service_family": {"type": "string", "default": "security-review"},
                },
                "required": ["quote_id"],
            },
        ),
        _wrap(
            get_job,
            "get_job",
            "Check job status",
            {
                "type": "object",
                "properties": {"job_id": {"type": "string"}},
                "required": ["job_id"],
            },
        ),
        _wrap(
            dispute_job,
            "dispute_job",
            "Open a dispute against a fulfilled review",
            {
                "type": "object",
                "properties": {
                    "job_id": {"type": "string"},
                    "reason_code": {"type": "string"},
                    "summary": {"type": "string"},
                },
                "required": ["job_id", "reason_code", "summary"],
            },
        ),
    ]

    # Model transport is configured by the operator; tool authorization stays
    # in the buyer gateway and provider kernel across every model.
    set_tracing_disabled(True)

    async def execute():
        async with AsyncOpenAI(
            api_key=os.environ["OPENAI_API_KEY"], base_url=os.getenv("OPENAI_BASE_URL")
        ) as model_client:
            agent = Agent(
                name="procurement-agent",
                instructions=PROCUREMENT_PROMPT,
                tools=tools,
                model=OpenAIChatCompletionsModel(
                    model=os.getenv("OPENAI_MODEL", "gpt-4.1-mini"), openai_client=model_client
                ),
            )
            return await Runner.run(agent, user_msg, max_turns=max_turns)

    user_msg = json.dumps(
        {
            "task": "procure a security review",
            "scope": scope,
            "target": target,
            "budget_minor": budget_minor,
        }
    )

    try:
        result = asyncio.run(execute())
    except Exception as exc:
        log.error("[procurement] agents SDK error: %s", exc)
        return {
            **FAILED_OUTPUT,
            "mode": "openai_agents_sdk",
            "error": str(exc),
            "tool_calls": call_log,
        }

    raw = result.final_output or ""
    try:
        parsed = json.loads(raw) if isinstance(raw, str) else raw
    except (json.JSONDecodeError, TypeError):
        parsed = {**FAILED_OUTPUT, "raw_text": raw}
    if isinstance(parsed, dict):
        parsed["mode"] = "openai_agents_sdk"
        parsed["tool_calls"] = call_log
    return _verified_model_result(parsed, get_job, call_log)


def _anthropic_flow(
    request_quote,
    create_job,
    approve_job,
    get_job,
    dispute_job,
    call_log,
    scope,
    target,
    budget_minor,
    max_turns,
):
    """Anthropic SDK with manual tool loop."""
    try:
        import anthropic
    except ImportError as error:
        raise RuntimeError("Install the locked project dependencies for Anthropic") from error

    tool_fns = {
        "request_quote": request_quote,
        "create_job": create_job,
        "get_job": get_job,
        "dispute_job": dispute_job,
    }
    api_tools = [
        {
            "name": "request_quote",
            "description": "Request a security review quote",
            "input_schema": {
                "type": "object",
                "properties": {"requested_scope": {"type": "string"}, "target": {"type": "string"}},
                "required": ["requested_scope", "target"],
            },
        },
        {
            "name": "create_job",
            "description": "Accept a quote and create a job",
            "input_schema": {
                "type": "object",
                "properties": {"quote_id": {"type": "string"}, "budget_minor": {"type": "integer"}},
                "required": ["quote_id"],
            },
        },
        {
            "name": "get_job",
            "description": "Check job status",
            "input_schema": {
                "type": "object",
                "properties": {"job_id": {"type": "string"}},
                "required": ["job_id"],
            },
        },
        {
            "name": "dispute_job",
            "description": "Dispute a fulfilled review",
            "input_schema": {
                "type": "object",
                "properties": {
                    "job_id": {"type": "string"},
                    "reason_code": {"type": "string"},
                    "summary": {"type": "string"},
                },
                "required": ["job_id", "reason_code", "summary"],
            },
        },
    ]

    client = anthropic.Anthropic()
    messages = [
        {
            "role": "user",
            "content": json.dumps(
                {
                    "task": "procure a security review",
                    "scope": scope,
                    "target": target,
                    "budget_minor": budget_minor,
                }
            ),
        }
    ]

    for _ in range(max_turns):
        resp = client.messages.create(
            model=os.getenv("ANTHROPIC_MODEL", "claude-sonnet-4-20250514"),
            max_tokens=4096,
            system=PROCUREMENT_PROMPT,
            tools=api_tools,
            messages=messages,
        )
        if resp.stop_reason != "tool_use":
            raw = "\n".join(b.text for b in resp.content if b.type == "text")
            try:
                parsed = json.loads(raw)
            except json.JSONDecodeError:
                parsed = {**FAILED_OUTPUT, "raw_text": raw}
            parsed["mode"] = "anthropic"
            parsed["tool_calls"] = call_log
            return _verified_model_result(parsed, get_job, call_log)

        asst, results = [], []
        for block in resp.content:
            if block.type == "tool_use":
                fn = tool_fns.get(block.name)
                try:
                    out = fn(**block.input) if fn else {"error": f"unknown tool: {block.name}"}
                except Exception as exc:
                    out = {"error": str(exc)}
                call_log.append({"tool": block.name, "input": block.input, "output": out})
                asst.append(
                    {"type": "tool_use", "id": block.id, "name": block.name, "input": block.input}
                )
                results.append(
                    {"type": "tool_result", "tool_use_id": block.id, "content": json.dumps(out)}
                )
            elif block.type == "text":
                asst.append({"type": "text", "text": block.text})
        messages.append({"role": "assistant", "content": asst})
        messages.append({"role": "user", "content": results})

    return {**FAILED_OUTPUT, "mode": "model_limit_reached", "tool_calls": call_log}


def _verified_model_result(parsed, get_job, call_log):
    if not isinstance(parsed, dict):
        return {**FAILED_OUTPUT, "tool_calls": call_log, "error": "Model output was not an object"}
    created = {
        call["output"].get("job_id")
        for call in call_log
        if call["tool"] == "create_job" and isinstance(call.get("output"), dict)
    }
    if parsed.get("job_id") not in created or parsed.get("job_id") is None:
        return {
            **FAILED_OUTPUT,
            "tool_calls": call_log,
            "error": "Model did not identify a job created in this run",
        }
    actual = get_job(parsed["job_id"])
    call_log.append({"tool": "get_job", "output": actual})
    parsed.update(
        final_status=actual["status"],
        price_minor=actual["quote"]["price_minor"],
        currency=actual["quote"]["currency"],
    )
    return parsed

"""Durable procurement state, reservations and balanced settlement entries."""

from __future__ import annotations

import json
import sqlite3
from contextlib import contextmanager
from pathlib import Path
from threading import RLock


class ProcurementStore:
    def __init__(self, path: str, budget_minor: int):
        if budget_minor < 0:
            raise ValueError("The operator budget must be nonnegative")
        if path != ":memory:":
            Path(path).parent.mkdir(parents=True, exist_ok=True)
        self.lock = RLock()
        self.db = sqlite3.connect(path, isolation_level=None, check_same_thread=False, timeout=10)
        self.db.execute("PRAGMA journal_mode=WAL")
        self.db.execute("PRAGMA synchronous=FULL")
        self.db.executescript("""
          CREATE TABLE IF NOT EXISTS configuration(id INTEGER PRIMARY KEY CHECK(id=1), budget INTEGER NOT NULL CHECK(budget>=0));
          CREATE TABLE IF NOT EXISTS quotes(id TEXT PRIMARY KEY, body TEXT NOT NULL);
          CREATE TABLE IF NOT EXISTS jobs(id TEXT PRIMARY KEY, quote_id TEXT UNIQUE NOT NULL, body TEXT NOT NULL);
          CREATE TABLE IF NOT EXISTS positions(job_id TEXT PRIMARY KEY, amount INTEGER NOT NULL CHECK(amount>=0), status TEXT NOT NULL);
          CREATE TABLE IF NOT EXISTS allocations(id TEXT PRIMARY KEY, amount INTEGER NOT NULL CHECK(amount>0), reason TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP);
          CREATE TABLE IF NOT EXISTS ledger(job_id TEXT NOT NULL, account TEXT NOT NULL, amount INTEGER NOT NULL, PRIMARY KEY(job_id,account));
        """)
        with self.transaction() as db:
            db.execute("INSERT OR IGNORE INTO configuration VALUES(1,?)", (budget_minor,))
            if db.execute("SELECT budget FROM configuration").fetchone()[0] != budget_minor:
                raise ValueError(
                    "The retained ledger has a different operator budget; do not reset it implicitly"
                )

    @contextmanager
    def transaction(self):
        with self.lock:
            self.db.execute("BEGIN IMMEDIATE")
            try:
                yield self.db
                self.db.execute("COMMIT")
            except BaseException:
                self.db.execute("ROLLBACK")
                raise

    @staticmethod
    def load(db, table, identity):
        if table not in {"jobs", "quotes"}:
            raise ValueError("Unknown record kind")
        row = db.execute(f"SELECT body FROM {table} WHERE id=?", (identity,)).fetchone()
        if row is None:
            raise KeyError(identity)
        return json.loads(row[0])

    @staticmethod
    def save_job(db, job):
        db.execute("UPDATE jobs SET body=? WHERE id=?", (json.dumps(job), job["job_id"]))

    @staticmethod
    def capacity(db):
        return (
            db.execute("SELECT budget FROM configuration").fetchone()[0]
            + db.execute("SELECT COALESCE(SUM(amount),0) FROM allocations").fetchone()[0]
        )

    @staticmethod
    def available(db):
        budget = ProcurementStore.capacity(db)
        committed = db.execute(
            "SELECT COALESCE(SUM(amount),0) FROM positions WHERE status IN ('reserved','settled')"
        ).fetchone()[0]
        return budget - committed

{
  "source": [
    {
      "path": "review_input.py",
      "sha256": "e369e32bb57c5c41d8a08587363149b71ed90a7c8a0823dde006e8c429a8f9e1",
      "bytes": 325
    }
  ],
  "findings": [
    {
      "rule": "dynamic-code",
      "severity": "high",
      "file": "review_input.py",
      "line": 8,
      "recommendation": "Replace dynamic Python execution with a bounded parser or explicit operations."
    },
    {
      "rule": "shell-command",
      "severity": "high",
      "file": "review_input.py",
      "line": 12,
      "recommendation": "Pass an argument list with shell=False and validate the executable and inputs."
    },
    {
      "rule": "weak-digest",
      "severity": "medium",
      "file": "review_input.py",
      "line": 16,
      "recommendation": "Use a modern digest where collision resistance is required."
    }
  ]
}

"""Perform bounded Python source checks and retain the actual reviewed artifacts."""

from __future__ import annotations

import ast
import hashlib
import json
import os
import re
import sqlite3
import tempfile
import uuid
from collections import Counter
from datetime import UTC, datetime
from pathlib import Path

from catalog import offer


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


def sync_directory(path: Path) -> None:
    descriptor = os.open(path, os.O_RDONLY)
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def durable_text(path: Path, text: str) -> None:
    with path.open("w") as output:
        output.write(text)
        output.flush()
        os.fsync(output.fileno())


def artifacts_root() -> Path:
    workspace = Path(
        os.environ.get("PROVIDER_WORKSPACE", Path(__file__).resolve().parents[1] / "workspace")
    )
    return Path(os.environ.get("PROVIDER_ARTIFACTS", workspace.parent / ".state" / "provider"))


def open_dispute(arguments: dict) -> dict:
    job_id = arguments["job_id"]
    if not re.fullmatch(r"[A-Za-z0-9_-]{1,100}", job_id):
        raise ValueError("Invalid job identifier")
    root = artifacts_root()
    if not (root / job_id / "fulfillment.json").is_file():
        raise ValueError("The provider has no retained fulfillment for this job")
    with sqlite3.connect(root / "disputes.sqlite3") as db:
        db.execute("PRAGMA journal_mode=WAL")
        db.execute("PRAGMA synchronous=FULL")
        db.execute(
            "CREATE TABLE IF NOT EXISTS disputes(job_id TEXT PRIMARY KEY, body TEXT NOT NULL)"
        )
        db.execute("BEGIN IMMEDIATE")
        row = db.execute("SELECT body FROM disputes WHERE job_id=?", (job_id,)).fetchone()
        if row:
            record = json.loads(row[0])
            if any(record[key] != arguments[key] for key in ("reason_code", "summary")):
                raise ValueError("This job already has a dispute with different terms")
            return record
        record = {
            "dispute_id": "dispute_" + uuid.uuid4().hex,
            "job_id": job_id,
            "reason_code": arguments["reason_code"],
            "summary": arguments["summary"],
            "status": "opened",
            "opened_at": datetime.now(UTC).isoformat(),
            "settlement_status": "reversal_pending",
        }
        db.execute("INSERT INTO disputes VALUES(?,?)", (job_id, json.dumps(record)))
        return record


def execute_review(arguments: dict) -> dict:
    root = Path(
        os.environ.get("PROVIDER_WORKSPACE", Path(__file__).resolve().parents[1] / "workspace")
    ).resolve(strict=True)
    target = Path(arguments["target"])
    if target.is_absolute() or ".." in target.parts:
        raise ValueError("The target must be a relative directory inside PROVIDER_WORKSPACE")
    target = root / target
    if target.is_symlink() or not target.resolve(strict=True).is_relative_to(root):
        raise ValueError("Review target escapes the configured workspace")
    target = target.resolve(strict=True)
    if not target.is_dir():
        raise ValueError("The review target must be a directory")
    selected = offer(arguments["requested_scope"])
    source = []
    findings = []
    total = 0
    for path in sorted(target.rglob("*")):
        if path.is_symlink():
            raise ValueError("Review inputs must not contain symbolic links")
        if not path.is_file():
            continue
        suffix = path.suffix
        include = (
            suffix == ".py"
            or (suffix == ".json" and "json-exposure-settings" in selected["checks"])
            or (
                path.name.startswith("requirements")
                and suffix == ".txt"
                and "dependency-pins" in selected["checks"]
            )
        )
        if not include:
            continue
        if len(source) >= selected["max_files"]:
            raise ValueError("Review exceeds the quoted file limit")
        with path.open("rb") as stream:
            data = stream.read(selected["max_bytes"] - total + 1)
        total += len(data)
        if total > selected["max_bytes"]:
            raise ValueError("Review exceeds the quoted byte limit")
        name = str(path.relative_to(target))
        source.append({"path": name, "sha256": sha256(data), "bytes": len(data)})
        if suffix == ".json":

            def inspect(value, pointer="", file_name=name):
                if isinstance(value, dict):
                    for key, item in value.items():
                        location = pointer + "/" + key.replace("~", "~0").replace("/", "~1")
                        if (
                            key.lower() in {"privileged", "publiclyaccessible", "publicaccess"}
                            and item is True
                        ) or (
                            key.lower() in {"cidr", "cidrip", "cidr_ipv4"} and item == "0.0.0.0/0"
                        ):
                            findings.append(
                                {
                                    "rule": "json-exposure",
                                    "severity": "high",
                                    "file": file_name,
                                    "line": 1,
                                    "pointer": location,
                                    "recommendation": "Review this public access or privileged setting against the intended deployment.",
                                }
                            )
                        inspect(item, location)
                elif isinstance(value, list):
                    for index, item in enumerate(value):
                        inspect(item, pointer + "/" + str(index))

            inspect(json.loads(data))
            continue
        if suffix == ".txt":
            for number, line in enumerate(data.decode().splitlines(), 1):
                requirement = line.strip()
                if (
                    requirement
                    and not requirement.startswith(("#", "-"))
                    and "==" not in requirement
                ):
                    findings.append(
                        {
                            "rule": "dependency-pin",
                            "severity": "low",
                            "file": name,
                            "line": number,
                            "recommendation": "Resolve this dependency through an exact version and a retained lockfile.",
                        }
                    )
            continue
        tree = ast.parse(data, filename=name)
        for node in ast.walk(tree):
            if (
                "exception-handling" in selected["checks"]
                and isinstance(node, ast.ExceptHandler)
                and (
                    node.type is None
                    or (
                        isinstance(node.type, ast.Name)
                        and node.type.id in {"Exception", "BaseException"}
                    )
                )
                and all(isinstance(statement, ast.Pass) for statement in node.body)
            ):
                findings.append(
                    {
                        "rule": "swallowed-exception",
                        "severity": "medium",
                        "file": name,
                        "line": node.lineno,
                        "recommendation": "Handle the expected exception explicitly and retain failure evidence.",
                    }
                )
            if not isinstance(node, ast.Call):
                continue
            function = node.func
            rule = None
            if isinstance(function, ast.Name) and function.id in {"eval", "exec"}:
                rule = (
                    "dynamic-code",
                    "high",
                    "Replace dynamic Python execution with a bounded parser or explicit operations.",
                )
            if isinstance(function, ast.Attribute) and isinstance(function.value, ast.Name):
                if function.value.id == "subprocess" and any(
                    keyword.arg == "shell"
                    and isinstance(keyword.value, ast.Constant)
                    and keyword.value.value is True
                    for keyword in node.keywords
                ):
                    rule = (
                        "shell-command",
                        "high",
                        "Pass an argument list with shell=False and validate the executable and inputs.",
                    )
                if function.value.id == "hashlib" and function.attr in {"md5", "sha1"}:
                    rule = (
                        "weak-digest",
                        "medium",
                        "Use a modern digest where collision resistance is required.",
                    )
            if rule:
                findings.append(
                    {
                        "rule": rule[0],
                        "severity": rule[1],
                        "file": name,
                        "line": node.lineno,
                        "recommendation": rule[2],
                    }
                )
    if not source:
        raise ValueError("The target contains no files supported by the selected checks")
    job_id = arguments["job_id"]
    if not re.fullmatch(r"[A-Za-z0-9_-]{1,100}", job_id):
        raise ValueError("Invalid job identifier")
    binding = {
        key: arguments[key]
        for key in ("job_id", "quote_id", "service_family", "requested_scope", "target")
    }
    binding["source"] = source
    binding["checks"] = selected["checks"]
    input_hash = sha256(json.dumps(binding, sort_keys=True, separators=(",", ":")).encode())
    output_root = artifacts_root()
    output_root.mkdir(parents=True, exist_ok=True)
    destination = output_root / job_id
    if destination.exists():
        prior = json.loads((destination / "fulfillment.json").read_text())
        if prior["input_sha256"] != input_hash:
            raise ValueError("Job identifier was already used with different source or terms")
        for artifact in prior["artifacts"]:
            if sha256((destination / artifact["name"]).read_bytes()) != artifact["sha256"]:
                raise ValueError("Retained fulfillment artifact has changed")
        return prior
    summary = dict(Counter(item["severity"] for item in findings))
    texts = {
        "findings.json": json.dumps({"source": source, "findings": findings}, indent=2) + "\n",
        "executive-summary.md": f"# Review of {arguments['target']}\n\nChecked {len(source)} files; found {len(findings)} matches.\n\nChecks: {', '.join(selected['checks'])}. This report describes those checks only.\n",
        "remediation-checklist.md": "# Remediation\n\n"
        + (
            "\n".join(
                f"- [ ] {item['file']}:{item['line']}: {item['recommendation']}"
                for item in findings
            )
            if findings
            else "No matching patterns found."
        )
        + "\n",
    }
    result = {
        "fulfillment_id": f"fulfillment_{job_id}",
        **binding,
        "input_sha256": input_hash,
        "status": "completed_with_findings" if findings else "completed",
        "severity_summary": {
            severity: summary.get(severity, 0) for severity in ("critical", "high", "medium", "low")
        },
        "deliverables": list(texts),
        "artifacts": [
            {"name": name, "content": text, "sha256": sha256(text.encode())}
            for name, text in texts.items()
        ],
    }
    with tempfile.TemporaryDirectory(dir=output_root) as temporary:
        staged = Path(temporary) / job_id
        staged.mkdir()
        for name, text in texts.items():
            durable_text(staged / name, text)
        durable_text(staged / "fulfillment.json", json.dumps(result, indent=2) + "\n")
        sync_directory(staged)
        try:
            staged.rename(destination)
            sync_directory(output_root)
        except FileExistsError:
            return execute_review(arguments)
    return result

{
  "association": {
    "tool_name": "execute_review",
    "session_id": "959ac5d9e90076874b60db0c3544ed105bdea9ce7dd447f98ef9e07e16333a37",
    "capability_ids": [
      "cap-01a08e28-9d74-7032-a077-953e7c6be7a4"
    ],
    "result": {
      "content": [
        {
          "text": "completed security-review for job job_3b29d30d0b",
          "type": "text"
        }
      ],
      "structuredContent": {
        "artifacts": [
          {
            "content": "{\n  \"source\": [\n    {\n      \"path\": \"review_input.py\",\n      \"sha256\": \"e369e32bb57c5c41d8a08587363149b71ed90a7c8a0823dde006e8c429a8f9e1\",\n      \"bytes\": 325\n    }\n  ],\n  \"findings\": [\n    {\n      \"rule\": \"dynamic-code\",\n      \"severity\": \"high\",\n      \"file\": \"review_input.py\",\n      \"line\": 8,\n      \"recommendation\": \"Replace dynamic Python execution with a bounded parser or explicit operations.\"\n    },\n    {\n      \"rule\": \"shell-command\",\n      \"severity\": \"high\",\n      \"file\": \"review_input.py\",\n      \"line\": 12,\n      \"recommendation\": \"Pass an argument list with shell=False and validate the executable and inputs.\"\n    },\n    {\n      \"rule\": \"weak-digest\",\n      \"severity\": \"medium\",\n      \"file\": \"review_input.py\",\n      \"line\": 16,\n      \"recommendation\": \"Use a modern digest where collision resistance is required.\"\n    }\n  ]\n}\n",
            "name": "findings.json",
            "sha256": "9600923b5bdcbb780710ab1c04d41794f837a91a4e6635424434e224bcb9f736"
          },
          {
            "content": "# Review of payments-api\n\nChecked 1 files; found 3 matches.\n\nChecks: python-call-patterns. This report describes those checks only.\n",
            "name": "executive-summary.md",
            "sha256": "c08d1b680a55ddfe68269b7fd8c7a44b5d7146590f7690e11f4d337db311b741"
          },
          {
            "content": "# Remediation\n\n- [ ] review_input.py:8: Replace dynamic Python execution with a bounded parser or explicit operations.\n- [ ] review_input.py:12: Pass an argument list with shell=False and validate the executable and inputs.\n- [ ] review_input.py:16: Use a modern digest where collision resistance is required.\n",
            "name": "remediation-checklist.md",
            "sha256": "0b01d70df3696e5a5ee0600e43045b08ac25ef44b5951eec868613bc4975b97d"
          }
        ],
        "checks": [
          "python-call-patterns"
        ],
        "deliverables": [
          "findings.json",
          "executive-summary.md",
          "remediation-checklist.md"
        ],
        "fulfillment_id": "fulfillment_job_3b29d30d0b",
        "input_sha256": "cef80b279f1905bf198ba6cceb6a7c3043b28f88243fe0259da1b49603e86a60",
        "job_id": "job_3b29d30d0b",
        "quote_id": "quote_e19826ebf7",
        "requested_scope": "hotfix-review",
        "service_family": "security-review",
        "severity_summary": {
          "critical": 0,
          "high": 2,
          "low": 0,
          "medium": 1
        },
        "source": [
          {
            "bytes": 325,
            "path": "review_input.py",
            "sha256": "e369e32bb57c5c41d8a08587363149b71ed90a7c8a0823dde006e8c429a8f9e1"
          }
        ],
        "status": "completed_with_findings",
        "target": "payments-api"
      },
      "isError": false
    },
    "receipt_id": "4559db0e686eebbc555af339f86280d1c68ac4908a4101ecb6424384f2cc7405",
    "request_id": "mcp-edge-req-d859d26519e347578015d808ca845baf",
    "arguments": {
      "job_id": "job_3b29d30d0b",
      "quote_id": "quote_e19826ebf7",
      "service_family": "security-review",
      "requested_scope": "hotfix-review",
      "target": "payments-api",
      "release_window": null
    },
    "is_error": false,
    "edge_base_url": "http://127.0.0.1:40343",
    "receipt": {
      "action": {
        "parameter_hash": "8717564b1141e9051de59a2c7cbc6e56956ce0f88fa486f9e7158ae71f18bca7",
        "parameters": {
          "job_id": "job_3b29d30d0b",
          "quote_id": "quote_e19826ebf7",
          "release_window": null,
          "requested_scope": "hotfix-review",
          "service_family": "security-review",
          "target": "payments-api"
        }
      },
      "boundary_class": "prevent",
      "capability_id": "cap-01a08e28-9d74-7032-a077-953e7c6be7a4",
      "content_hash": "1ccb1715b30cb62519cb7c113e1dced14da341c3115148d735992fd7ae0b8556",
      "decision": {
        "verdict": "allow"
      },
      "id": "4559db0e686eebbc555af339f86280d1c68ac4908a4101ecb6424384f2cc7405",
      "kernel_key": "b47fa16e1fb8bceededad15e120f3e7a99c258b6c27a6c384f5f31e50abeaa51",
      "metadata": {
        "admission_operation": {
          "compensation_status": "not_compensated",
          "coordinator_lease_epoch": 8,
          "coordinator_lease_id": "01a08e28-81e7-7ac3-9d18-b318aa129555",
          "operation_id": "7ad156683ea55f85a2a90a632be025749435d5d238baa6bab0c84a302cd6d73e",
          "projected_dispatch_state": "terminal",
          "projected_operation_version": 8,
          "projected_state": "completed",
          "request_binding_hash": "ca569287ae6340f7b95705f4f38dfcdbae36bb70a99ef22a867c290fa3c6db90",
          "request_id": "mcp-edge-req-d859d26519e347578015d808ca845baf",
          "request_namespace_digest": "c5a9cc313e2b59a442c5dd5bd334cc45abfe831154c72750acb9cd65e04260c2",
          "retained_dispatch_commit": {
            "committed_version": 6,
            "coordinator_lease_epoch": 8,
            "coordinator_lease_id": "01a08e28-81e7-7ac3-9d18-b318aa129555",
            "provider_attempt": {
              "attempt_id": "attempt:7ad156683ea55f85a2a90a632be025749435d5d238baa6bab0c84a302cd6d73e",
              "operation_id": "7ad156683ea55f85a2a90a632be025749435d5d238baa6bab0c84a302cd6d73e",
              "transport_id": "kernel-tool-server:provider-security-review",
              "transport_key_epoch": 8
            },
            "store_fence": {
              "lease_id": "01a08e28-81e7-7ac3-9d18-b318aa129555",
              "owner_epoch": 8,
              "store_uuid": "01a08d81-f818-7250-8d0e-daf69621b1e7"
            }
          },
          "schema": "chio.admission-receipt.v1",
          "store_fence": {
            "lease_id": "01a08e28-81e7-7ac3-9d18-b318aa129555",
            "owner_epoch": 8,
            "store_uuid": "01a08d81-f818-7250-8d0e-daf69621b1e7"
          },
          "tool_outcome_id": "b9799e05786579ac942f74d6484f56ef33a2a049e58e59fafa18d2015dca5833",
          "tool_outcome_version": 2,
          "trusted_time_unix_ms": 1789091424779
        },
        "attribution": {
          "delegation_depth": 0,
          "grant_index": 1,
          "issuer_key": "0b4779b514470d64a2a3156bc8a15ae780c899a65de4fff8ecf1f8cc8474b393",
          "subject_key": "45ff2f3af9c2e69436ebf2579bcb950ffdbd41c08573b6db693b926c5eec4306"
        },
        "chio_receipt_signing_nonce": "rcpt-01a08e28-aa64-78a1-bf04-31409c8685f2",
        "receipt_context": {
          "request_id": "mcp-edge-req-d859d26519e347578015d808ca845baf"
        }
      },
      "policy_hash": "92a87c7cedea75a5e4007ae28fc6ac2e9916c7b2586cdc5ef96a462de6fb75ac",
      "receipt_kind": "mediated_decision",
      "redaction_mode": "none",
      "signature": "e522f1480f043af35c7f24d8ae920a12f66c425643ac05771aff2a67592a8fd263877a27823133788e54953ff81d6764dcb8723fa7f29b6961a3bfb8b35b3a05",
      "timestamp": 1789091424,
      "tool_name": "execute_review",
      "tool_origin": "caller_executed",
      "tool_server": "provider-security-review",
      "trust_level": "mediated"
    },
    "capability_id": "cap-01a08e28-9d74-7032-a077-953e7c6be7a4"
  },
  "receipt": {
    "id": "4559db0e686eebbc555af339f86280d1c68ac4908a4101ecb6424384f2cc7405",
    "timestamp": 1789091424,
    "capability_id": "cap-01a08e28-9d74-7032-a077-953e7c6be7a4",
    "tool_server": "provider-security-review",
    "tool_name": "execute_review",
    "action": {
      "parameters": {
        "job_id": "job_3b29d30d0b",
        "quote_id": "quote_e19826ebf7",
        "release_window": null,
        "requested_scope": "hotfix-review",
        "service_family": "security-review",
        "target": "payments-api"
      },
      "parameter_hash": "8717564b1141e9051de59a2c7cbc6e56956ce0f88fa486f9e7158ae71f18bca7"
    },
    "decision": {
      "verdict": "allow"
    },
    "receipt_kind": "mediated_decision",
    "boundary_class": "prevent",
    "tool_origin": "caller_executed",
    "redaction_mode": "none",
    "content_hash": "1ccb1715b30cb62519cb7c113e1dced14da341c3115148d735992fd7ae0b8556",
    "policy_hash": "92a87c7cedea75a5e4007ae28fc6ac2e9916c7b2586cdc5ef96a462de6fb75ac",
    "metadata": {
      "admission_operation": {
        "compensation_status": "not_compensated",
        "coordinator_lease_epoch": 8,
        "coordinator_lease_id": "01a08e28-81e7-7ac3-9d18-b318aa129555",
        "operation_id": "7ad156683ea55f85a2a90a632be025749435d5d238baa6bab0c84a302cd6d73e",
        "projected_dispatch_state": "terminal",
        "projected_operation_version": 8,
        "projected_state": "completed",
        "request_binding_hash": "ca569287ae6340f7b95705f4f38dfcdbae36bb70a99ef22a867c290fa3c6db90",
        "request_id": "mcp-edge-req-d859d26519e347578015d808ca845baf",
        "request_namespace_digest": "c5a9cc313e2b59a442c5dd5bd334cc45abfe831154c72750acb9cd65e04260c2",
        "retained_dispatch_commit": {
          "committed_version": 6,
          "coordinator_lease_epoch": 8,
          "coordinator_lease_id": "01a08e28-81e7-7ac3-9d18-b318aa129555",
          "provider_attempt": {
            "attempt_id": "attempt:7ad156683ea55f85a2a90a632be025749435d5d238baa6bab0c84a302cd6d73e",
            "operation_id": "7ad156683ea55f85a2a90a632be025749435d5d238baa6bab0c84a302cd6d73e",
            "transport_id": "kernel-tool-server:provider-security-review",
            "transport_key_epoch": 8
          },
          "store_fence": {
            "lease_id": "01a08e28-81e7-7ac3-9d18-b318aa129555",
            "owner_epoch": 8,
            "store_uuid": "01a08d81-f818-7250-8d0e-daf69621b1e7"
          }
        },
        "schema": "chio.admission-receipt.v1",
        "store_fence": {
          "lease_id": "01a08e28-81e7-7ac3-9d18-b318aa129555",
          "owner_epoch": 8,
          "store_uuid": "01a08d81-f818-7250-8d0e-daf69621b1e7"
        },
        "tool_outcome_id": "b9799e05786579ac942f74d6484f56ef33a2a049e58e59fafa18d2015dca5833",
        "tool_outcome_version": 2,
        "trusted_time_unix_ms": 1789091424779
      },
      "attribution": {
        "delegation_depth": 0,
        "grant_index": 1,
        "issuer_key": "0b4779b514470d64a2a3156bc8a15ae780c899a65de4fff8ecf1f8cc8474b393",
        "subject_key": "45ff2f3af9c2e69436ebf2579bcb950ffdbd41c08573b6db693b926c5eec4306"
      },
      "chio_receipt_signing_nonce": "rcpt-01a08e28-aa64-78a1-bf04-31409c8685f2",
      "receipt_context": {
        "request_id": "mcp-edge-req-d859d26519e347578015d808ca845baf"
      }
    },
    "trust_level": "mediated",
    "kernel_key": "b47fa16e1fb8bceededad15e120f3e7a99c258b6c27a6c384f5f31e50abeaa51",
    "signature": "e522f1480f043af35c7f24d8ae920a12f66c425643ac05771aff2a67592a8fd263877a27823133788e54953ff81d6764dcb8723fa7f29b6961a3bfb8b35b3a05"
  }
}

from __future__ import annotations

import hashlib
import json
import os
import secrets
import uuid
from contextlib import asynccontextmanager
from copy import deepcopy
from pathlib import Path
from threading import RLock, local
from typing import Any, Literal, Protocol

import httpx
from chio.invariants import canonicalize_json, sha256_hex_utf8, verify_receipt_with_trusted_signers
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, Field

try:
    from .store import ProcurementStore
except ImportError:
    from store import ProcurementStore


CONTRACTS_DIR = Path(__file__).resolve().parents[1] / "contracts"
PROTOCOL_VERSION = "2025-11-25"
DEFAULT_APPROVAL_THRESHOLD_MINOR = 100_000
DEFAULT_BUDGET_MINOR = 150_000
DEFAULT_BUYER_ID = "lattice-platform-security"
DEFAULT_PROVIDER_ID = "vanguard-security"


def contract_template(name: str) -> dict[str, Any]:
    return json.loads((CONTRACTS_DIR / name).read_text())


def random_id(prefix: str) -> str:
    return f"{prefix}_{uuid.uuid4().hex[:10]}"


class QuoteRequestPayload(BaseModel):
    service_family: str = Field(pattern="^security-review$")
    target: str = Field(min_length=1, max_length=500)
    requested_scope: Literal[
        "hotfix-review", "release-review", "release-plus-cloud-review", "full-estate-review"
    ]
    release_window: str | None = Field(default=None, max_length=200)


class CreateJobPayload(BaseModel):
    quote_id: str = Field(min_length=1, max_length=200)
    provider_id: str = Field(min_length=1, max_length=200)
    service_family: Literal["security-review"]
    budget_minor: int | None = Field(default=None, ge=0)


class ApprovalPayload(BaseModel):
    approver: str = Field(min_length=1, max_length=200)
    reason: str = Field(min_length=1, max_length=2000)


class DisputePayload(BaseModel):
    reason_code: str = Field(min_length=1, max_length=100)
    summary: str = Field(min_length=1, max_length=2000)


class ProviderGateway(Protocol):
    def request_quote(self, payload: dict[str, Any]) -> dict[str, Any]: ...

    def execute_review(self, payload: dict[str, Any]) -> dict[str, Any]: ...

    def open_dispute(self, payload: dict[str, Any]) -> dict[str, Any]: ...


class ProviderReviewClient:
    def __init__(
        self,
        *,
        base_url: str,
        auth_token: str,
        timeout: float = 10.0,
        client: httpx.Client | None = None,
        trusted_kernel_key: str | None = None,
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.trusted_kernel_key = trusted_kernel_key
        self.auth_token = auth_token
        self.client = client or httpx.Client(timeout=timeout)
        self._trace = local()
        self._lock = RLock()
        self._session_id: str | None = None
        self._sequence = 1

    @property
    def last_trace(self) -> dict[str, Any] | None:
        return getattr(self._trace, "value", None)

    def request_quote(self, payload: dict[str, Any]) -> dict[str, Any]:
        return self._call_tool("request_quote", payload)

    def execute_review(self, payload: dict[str, Any]) -> dict[str, Any]:
        return self._call_tool("execute_review", payload)

    def open_dispute(self, payload: dict[str, Any]) -> dict[str, Any]:
        return self._call_tool("open_dispute", payload)

    def _call_tool(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]:
        with self._lock:
            self._trace.value = None
            if self._session_id is None:
                self._session_id = self._initialize_session()
                self._post_mcp(
                    {"jsonrpc": "2.0", "method": "notifications/initialized"},
                    session_id=self._session_id,
                    expect_response=False,
                )
            self._sequence += 1
            payload = self._post_mcp(
                {
                    "jsonrpc": "2.0",
                    "id": self._sequence,
                    "method": "tools/call",
                    "params": {"name": name, "arguments": arguments},
                },
                session_id=self._session_id,
                expect_response=True,
            )
            if payload.get("id") != self._sequence or "error" in payload:
                raise RuntimeError("Provider response did not complete the requested MCP operation")
            result = payload.get("result", {})
            metadata = result.get("_meta", {}).get("chioReceipt", {})
            receipt_id, request_id = metadata.get("receiptId"), metadata.get("requestId")
            if not receipt_id or not request_id:
                raise RuntimeError(
                    "Provider response is missing its authoritative receipt association"
                )
            capability_ids = self._session_capability_ids(self._session_id)
            mapped_result = {
                key: result[key]
                for key in ("content", "structuredContent", "isError")
                if key in result
            }
            mapped_result.setdefault("content", [])
            mapped_result.setdefault("isError", False)
            self._trace.value = {
                "tool_name": name,
                "session_id": self._session_id,
                "capability_ids": sorted(capability_ids),
                "result": mapped_result,
                "receipt_id": receipt_id,
                "request_id": request_id,
                "arguments": arguments,
                "is_error": bool(result.get("isError")),
                "edge_base_url": self.base_url,
            }
            receipt = self._verified_receipt(
                receipt_id, request_id, capability_ids, name, arguments, mapped_result
            )
            self._trace.value["receipt"] = receipt
            self._trace.value["capability_id"] = receipt["capability_id"]
            if result.get("isError"):
                raise RuntimeError(
                    f"Provider refused or failed the operation; receipt {receipt_id}"
                )
            structured = result.get("structuredContent")
            if not isinstance(structured, dict):
                raise RuntimeError("Provider result is missing its structured work product")
            return deepcopy(structured)

    def _verified_receipt(self, identity, request_id, capability_ids, name, arguments, result):
        if not self.trusted_kernel_key or not capability_ids:
            raise RuntimeError("Configure the provider kernel key before accepting its work")
        response = self.client.get(
            self.base_url + "/admin/receipts/tools",
            headers={"Authorization": "Bearer " + self.auth_token},
            params={"receiptId": identity},
        )
        response.raise_for_status()
        receipts = response.json().get("receipts", [])
        matches = [receipt for receipt in receipts if receipt.get("id") == identity]
        if len(matches) != 1:
            raise RuntimeError(
                "The exact provider receipt is not available; do not settle this job"
            )
        receipt = matches[0]
        checks = verify_receipt_with_trusted_signers(receipt, [self.trusted_kernel_key])
        if not checks["ok"] or receipt["metadata"]["receipt_context"]["request_id"] != request_id:
            raise RuntimeError(
                "Provider receipt is invalid, untrusted or belongs to another request"
            )
        if (
            receipt["action"]["parameters"] != arguments
            or receipt["tool_name"] != name
            or receipt["capability_id"] not in capability_ids
        ):
            raise RuntimeError("Provider receipt does not match the requested operation")
        if receipt["decision"]["verdict"] == "allow":
            if sha256_hex_utf8(canonicalize_json(result)) != receipt["content_hash"]:
                raise RuntimeError("Provider result does not match its signed output hash")
        elif not result.get("isError"):
            raise RuntimeError("Provider reported success for a refused operation")
        return receipt

    def _initialize_session(self) -> str:
        response = self._post_raw(
            {
                "jsonrpc": "2.0",
                "id": 1,
                "method": "initialize",
                "params": {
                    "protocolVersion": PROTOCOL_VERSION,
                    "capabilities": {},
                    "clientInfo": {
                        "name": "agent-commerce-network-buyer",
                        "version": "0.1.0",
                    },
                },
            }
        )
        _ = self._decode_response(response)
        session_id = response.headers.get("MCP-Session-Id")
        if not session_id:
            raise RuntimeError("provider edge did not return MCP-Session-Id")
        return session_id

    def _post_mcp(
        self,
        payload: dict[str, Any],
        *,
        session_id: str | None = None,
        expect_response: bool,
    ) -> dict[str, Any]:
        response = self._post_raw(payload, session_id=session_id)
        if not expect_response or not response.text.strip():
            return {}
        return self._decode_response(response)

    def _post_raw(
        self,
        payload: dict[str, Any],
        *,
        session_id: str | None = None,
    ) -> httpx.Response:
        headers = {
            "Authorization": f"Bearer {self.auth_token}",
            "Accept": "application/json, text/event-stream",
            "Content-Type": "application/json",
            "MCP-Protocol-Version": PROTOCOL_VERSION,
        }
        if session_id:
            headers["MCP-Session-Id"] = session_id
        response = self.client.post(f"{self.base_url}/mcp", headers=headers, json=payload)
        if response.is_error:
            raise RuntimeError(
                f"Provider MCP transport failed (HTTP {response.status_code}): {response.text[:2000]}"
            )
        return response

    def _session_capability_ids(self, session_id: str) -> set[str]:
        response = self.client.get(
            f"{self.base_url}/admin/sessions/{session_id}/trust",
            headers={"Authorization": f"Bearer {self.auth_token}"},
        )
        response.raise_for_status()
        return {item["capabilityId"] for item in response.json().get("capabilities", [])}

    @staticmethod
    def _decode_response(response: httpx.Response) -> dict[str, Any]:
        content_type = response.headers.get("content-type", "")
        if "application/json" in content_type:
            return response.json()
        data_lines: list[str] = []
        for raw_line in response.text.splitlines():
            line = raw_line.strip()
            if not line:
                if data_lines:
                    break
                continue
            if line.startswith("data:"):
                data_lines.append(line.split(":", 1)[1].lstrip())
        if not data_lines:
            raise RuntimeError("no JSON-RPC payload received from provider edge")
        return json.loads("\n".join(data_lines))


class ProcurementService:
    def __init__(
        self,
        provider: ProviderGateway,
        *,
        buyer_id: str = DEFAULT_BUYER_ID,
        default_budget_minor: int = DEFAULT_BUDGET_MINOR,
        database: str = ":memory:",
    ) -> None:
        self.provider = provider
        self.buyer_id = buyer_id
        self.default_budget_minor = default_budget_minor
        self.store = ProcurementStore(database, default_budget_minor)

    def submit_quote_request(self, payload: QuoteRequestPayload) -> dict[str, Any]:
        request_id = random_id("quote_req")
        quote = self.provider.request_quote(
            {"request_id": request_id, "buyer_id": self.buyer_id, **payload.model_dump()}
        )
        if (
            quote.get("request_id") != request_id
            or quote.get("service_family") != payload.service_family
            or type(quote.get("price_minor")) is not int
            or quote["price_minor"] <= 0
            or quote.get("currency") != "USD"
            or not quote.get("quote_id")
            or quote.get("provider_id") != DEFAULT_PROVIDER_ID
            or quote.get("offer_id") != payload.requested_scope
            or quote.get("target") != payload.target
        ):
            raise ValueError(
                "Provider quote does not match the request or supported monetary terms"
            )
        # The buyer owns its approval threshold; the provider cannot waive it.
        quote["approval_required"] = quote["price_minor"] > DEFAULT_APPROVAL_THRESHOLD_MINOR
        record = {
            "request_id": request_id,
            "request": payload.model_dump(),
            "quote": quote,
            "provider_trace": self._provider_trace(),
        }
        with self.store.transaction() as db:
            db.execute("INSERT INTO quotes VALUES(?,?)", (quote["quote_id"], json.dumps(record)))
        return {
            "request_id": request_id,
            "status": "quoted",
            "quote": quote,
            "provider_trace": record["provider_trace"],
        }

    def create_job(self, payload: CreateJobPayload) -> dict[str, Any]:
        with self.store.transaction() as db:
            record = self.store.load(db, "quotes", payload.quote_id)
            quote = record["quote"]
            if (
                payload.provider_id != quote.get("provider_id", DEFAULT_PROVIDER_ID)
                or payload.service_family != quote["service_family"]
            ):
                raise ValueError("Provider and service must match the accepted quote")
            if db.execute("SELECT 1 FROM jobs WHERE quote_id=?", (payload.quote_id,)).fetchone():
                raise ValueError("This quote has already been consumed; inspect the existing job")
            budget = (
                self.store.capacity(db) if payload.budget_minor is None else payload.budget_minor
            )
            if budget > self.store.capacity(db):
                raise ValueError("A caller cannot increase the operator budget")
            job = {
                "job_id": random_id("job"),
                "buyer_id": self.buyer_id,
                "provider_id": payload.provider_id,
                "service_family": payload.service_family,
                "budget_minor": budget,
                "quote": quote,
                "requested_scope": record["request"]["requested_scope"],
                "target": record["request"]["target"],
                "release_window": record["request"]["release_window"],
                "status": "ready",
                "approval_required": quote["approval_required"],
                "approval": None,
                "fulfillment": None,
                "fulfillment_trace": None,
                "settlement": None,
                "disputes": [],
                "quote_provider_trace": record["provider_trace"],
            }
            if quote["price_minor"] > min(budget, self.store.available(db)):
                job.update(
                    status="denied_budget",
                    denial_reason="Quoted work exceeds the requested or remaining operator budget",
                )
            else:
                db.execute(
                    "INSERT INTO positions VALUES(?,?,?)",
                    (job["job_id"], quote["price_minor"], "reserved"),
                )
                if quote["approval_required"]:
                    job["status"] = "pending_approval"
            db.execute(
                "INSERT INTO jobs VALUES(?,?,?)", (job["job_id"], payload.quote_id, json.dumps(job))
            )
        if job["status"] == "ready":
            self._execute_job(job["job_id"])
        return self.get_job(job["job_id"])

    def get_job(self, job_id: str) -> dict[str, Any]:
        with self.store.transaction() as db:
            return self.store.load(db, "jobs", job_id)

    def approve_job(self, job_id: str, payload: ApprovalPayload) -> dict[str, Any]:
        with self.store.transaction() as db:
            job = self.store.load(db, "jobs", job_id)
            if job["status"] != "pending_approval":
                raise ValueError("Job is not waiting for approval")
            job["approval"] = {
                "approver": payload.approver,
                "reason": payload.reason,
                "status": "approved",
            }
            job["status"] = "ready"
            self.store.save_job(db, job)
        self._execute_job(job_id)
        return self.get_job(job_id)

    def dispute_job(self, job_id: str, payload: DisputePayload) -> dict[str, Any]:
        with self.store.transaction() as db:
            job = self.store.load(db, "jobs", job_id)
            if job["status"] != "fulfilled":
                raise ValueError("Only fulfilled jobs can open a dispute")
            job["status"] = "dispute_submitting"
            self.store.save_job(db, job)
        # A failed or ambiguous provider response leaves the durable submission
        # state for investigation. It never pretends the debit was reversed.
        dispute = self.provider.open_dispute({"job_id": job_id, **payload.model_dump()})
        with self.store.transaction() as db:
            job = self.store.load(db, "jobs", job_id)
            job["disputes"].append({"record": dispute, "provider_trace": self._provider_trace()})
            job["status"] = "disputed"
            job["settlement"].update(
                status="reversal_pending",
                buyer_position="contested",
                provider_position="review_requested",
            )
            self.store.save_job(db, job)
        return job

    def _execute_job(self, job_id: str) -> None:
        with self.store.transaction() as db:
            job = self.store.load(db, "jobs", job_id)
            if job["status"] != "ready":
                raise ValueError("Job is already dispatched or is not authorized")
            job["status"] = "executing"
            self.store.save_job(db, job)
        try:
            fulfillment = self.provider.execute_review(
                {
                    "job_id": job_id,
                    "quote_id": job["quote"]["quote_id"],
                    "service_family": job["service_family"],
                    "requested_scope": job["requested_scope"],
                    "target": job["target"],
                    "release_window": job["release_window"],
                }
            )
            if fulfillment.get("job_id") != job_id or fulfillment.get("status") not in {
                "completed",
                "completed_with_findings",
            }:
                raise ValueError("Provider did not return completed work for this job")
            expected_terms = {
                "quote_id": job["quote"]["quote_id"],
                "service_family": job["service_family"],
                "requested_scope": job["requested_scope"],
                "target": job["target"],
            }
            if any(fulfillment.get(key) != value for key, value in expected_terms.items()):
                raise ValueError("Delivered work does not match the accepted quote and target")
            artifacts = fulfillment.get("artifacts", [])
            if not artifacts or {item["name"] for item in artifacts} != set(
                fulfillment.get("deliverables", [])
            ):
                raise ValueError("Provider fulfillment is missing its actual deliverables")
            for artifact in artifacts:
                if hashlib.sha256(artifact["content"].encode()).hexdigest() != artifact["sha256"]:
                    raise ValueError("Provider artifact digest mismatch")
        except Exception as error:
            with self.store.transaction() as db:
                job = self.store.load(db, "jobs", job_id)
                job.update(
                    status="outcome_unknown",
                    error=str(error),
                    fulfillment_trace=self._provider_trace(),
                )
                self.store.save_job(db, job)
            return
        with self.store.transaction() as db:
            job = self.store.load(db, "jobs", job_id)
            amount = job["quote"]["price_minor"]
            entries = [
                {"account": self.buyer_id, "amount_minor": -amount},
                {"account": job["provider_id"], "amount_minor": amount},
            ]
            for entry in entries:
                db.execute(
                    "INSERT INTO ledger VALUES(?,?,?)",
                    (job_id, entry["account"], entry["amount_minor"]),
                )
            db.execute(
                "UPDATE positions SET status='settled' WHERE job_id=? AND status='reserved'",
                (job_id,),
            )
            job["settlement"] = {
                "settlement_id": f"settlement_{job_id}",
                "job_id": job_id,
                "quoted_amount_minor": amount,
                "approved_amount_minor": amount,
                "settled_amount_minor": amount,
                "currency": job["quote"]["currency"],
                "status": "reconciled",
                "settlement_kind": "internal_book_entry",
                "entries": entries,
                "buyer_position": "accepted",
                "provider_position": "accepted",
                "remaining_budget_minor": self.store.available(db),
            }
            job.update(
                fulfillment=fulfillment,
                fulfillment_trace=self._provider_trace(),
                status="fulfilled",
            )
            self.store.save_job(db, job)

    def _provider_trace(self) -> dict[str, Any] | None:
        trace = getattr(self.provider, "last_trace", None)
        return deepcopy(trace) if trace is not None else None


class UnconfiguredProvider:
    def request_quote(self, payload):
        raise RuntimeError(
            "Set BUYER_PROVIDER_BASE_URL and BUYER_PROVIDER_AUTH_TOKEN to the governed provider edge"
        )


def provider_from_env() -> ProviderGateway:
    provider_base_url = os.environ.get("BUYER_PROVIDER_BASE_URL")
    provider_auth_token = os.environ.get("BUYER_PROVIDER_AUTH_TOKEN")
    if provider_base_url:
        if not provider_auth_token or not os.environ.get("BUYER_PROVIDER_KERNEL_KEY"):
            raise ValueError(
                "Configure the provider credential and its independently selected kernel key"
            )
        return ProviderReviewClient(
            base_url=provider_base_url,
            auth_token=provider_auth_token,
            trusted_kernel_key=os.environ.get("BUYER_PROVIDER_KERNEL_KEY"),
        )
    return UnconfiguredProvider()


def build_service(provider: ProviderGateway | None = None) -> ProcurementService:
    buyer_id = os.environ.get("BUYER_ID", DEFAULT_BUYER_ID)
    default_budget_minor = int(
        os.environ.get("BUYER_DEFAULT_BUDGET_MINOR", str(DEFAULT_BUDGET_MINOR))
    )
    return ProcurementService(
        provider or provider_from_env(),
        buyer_id=buyer_id,
        default_budget_minor=default_budget_minor,
        database=":memory:"
        if provider is not None
        else os.environ.get("BUYER_STATE_DB", str(CONTRACTS_DIR.parent / ".state" / "buyer.db")),
    )


def create_app(provider: ProviderGateway | None = None) -> FastAPI:
    @asynccontextmanager
    async def lifespan(app):
        service = build_service(provider)
        app.state.procurement_service = service
        try:
            yield
        finally:
            service.store.db.close()
            if isinstance(service.provider, ProviderReviewClient):
                service.provider.client.close()

    app = FastAPI(
        lifespan=lifespan,
        title="Lattice Procurement API",
        version="0.1.0",
        description="Buyer-side procurement service for the agent-commerce-network example.",
    )

    @app.get("/healthz")
    def healthz() -> dict[str, str]:
        mode = (
            "wrapped-mcp-provider"
            if isinstance(app.state.procurement_service.provider, ProviderReviewClient)
            else "unconfigured"
        )
        if isinstance(app.state.procurement_service.provider, UnconfiguredProvider):
            raise HTTPException(
                status_code=503, detail="Configure the governed provider before accepting jobs"
            )
        return {"status": "ok", "provider_mode": mode}

    @app.post("/procurement/quote-requests", status_code=202)
    def request_quote(payload: QuoteRequestPayload, request: Request) -> dict[str, Any]:
        service: ProcurementService = request.app.state.procurement_service
        return service.submit_quote_request(payload)

    @app.post("/procurement/jobs", status_code=202)
    def create_job(payload: CreateJobPayload, request: Request) -> dict[str, Any]:
        service: ProcurementService = request.app.state.procurement_service
        try:
            return service.create_job(payload)
        except KeyError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc

    @app.get("/procurement/jobs/{job_id}")
    def get_job(job_id: str, request: Request) -> dict[str, Any]:
        service: ProcurementService = request.app.state.procurement_service
        try:
            return service.get_job(job_id)
        except KeyError as exc:
            raise HTTPException(status_code=404, detail=f"unknown job: {job_id}") from exc

    @app.post("/procurement/jobs/{job_id}/approve")
    def approve_job(job_id: str, payload: ApprovalPayload, request: Request) -> dict[str, Any]:
        expected = os.environ.get("BUYER_APPROVAL_TOKEN", "")
        actual = request.headers.get("X-Buyer-Approval", "")
        if not expected or not secrets.compare_digest(expected, actual):
            raise HTTPException(
                status_code=403, detail="An independent operator approval credential is required"
            )
        service: ProcurementService = request.app.state.procurement_service
        try:
            return service.approve_job(job_id, payload)
        except KeyError as exc:
            raise HTTPException(status_code=404, detail=f"unknown job: {job_id}") from exc
        except ValueError as exc:
            raise HTTPException(status_code=409, detail=str(exc)) from exc

    @app.post("/procurement/jobs/{job_id}/disputes", status_code=202)
    def dispute_job(job_id: str, payload: DisputePayload, request: Request) -> dict[str, Any]:
        service: ProcurementService = request.app.state.procurement_service
        try:
            return service.dispute_job(job_id, payload)
        except KeyError as exc:
            raise HTTPException(status_code=404, detail=f"unknown job: {job_id}") from exc

    @app.exception_handler(ValueError)
    async def invalid_operation(request: Request, error: ValueError):
        from fastapi.responses import JSONResponse

        return JSONResponse(status_code=409, content={"detail": str(error)})

    @app.exception_handler(RuntimeError)
    @app.exception_handler(httpx.HTTPError)
    async def provider_failure(request: Request, error: Exception):
        from fastapi.responses import JSONResponse

        return JSONResponse(
            status_code=502,
            content={
                "detail": str(error),
                "recovery": "Inspect the retained job and provider receipt before retrying an operation",
            },
        )

    return app


app = create_app()

"""Independently verify the live work, signatures and balanced book settlement."""

from __future__ import annotations

import hashlib
import json
from pathlib import Path

from chio.invariants import (
    canonicalize_json,
    sha256_hex_utf8,
    verify_http_receipt_with_trusted_signers,
    verify_receipt_with_trusted_signers,
)


def verify_bundle(bundle_path, *, trusted_signers):
    root = Path(bundle_path)
    errors = []

    def require(condition, message):
        if not condition:
            raise ValueError(message)

    try:
        run = json.loads((root / "agent-output.json").read_text())
        summary = json.loads((root / "summary.json").read_text())
        receipts = json.loads((root / "receipts.json").read_text())
        require(
            run["final_status"] == summary["final_status"] == "fulfilled",
            "The run did not fulfill its job",
        )
        require(receipts["http"] and receipts["mcp"], "Both HTTP and MCP records are required")
        outputs = {}
        ids = set()
        for item in receipts["mcp"]:
            receipt, association = item["receipt"], item["association"]
            require(
                verify_receipt_with_trusted_signers(receipt, trusted_signers)["ok"],
                "Invalid or untrusted provider receipt",
            )
            require(
                receipt["id"] == association["receipt_id"] and receipt["id"] not in ids,
                "Provider receipt association reused or changed",
            )
            ids.add(receipt["id"])
            require(
                receipt["metadata"]["receipt_context"]["request_id"] == association["request_id"],
                "Wrong provider request association",
            )
            require(
                receipt["tool_name"] == association["tool_name"]
                and receipt["action"]["parameters"] == association["arguments"],
                "Wrong provider tool or arguments",
            )
            require(
                receipt["decision"]["verdict"] == "allow", "Provider operation was not authorized"
            )
            require(
                sha256_hex_utf8(canonicalize_json(association["result"]))
                == receipt["content_hash"],
                "Provider output differs from the signed output hash",
            )
            outputs[receipt["tool_name"]] = association["result"]["structuredContent"]
        for item in receipts["http"]:
            receipt, association = item["receipt"], item["association"]
            require(
                verify_http_receipt_with_trusted_signers(receipt, trusted_signers)["ok"],
                "Invalid or untrusted HTTP receipt",
            )
            require(
                receipt["id"] == association["receipt_id"] and receipt["id"] not in ids,
                "HTTP receipt association reused or changed",
            )
            ids.add(receipt["id"])
            require(
                receipt["method"] == association["method"]
                and receipt["response_status"] == association["status"],
                "HTTP method or outcome changed",
            )
            binding = {
                "method": association["method"],
                "route_pattern": receipt["route_pattern"],
                "path": association["path"],
                "query": {},
                "body_hash": association["body_sha256"],
            }
            # HTTPX writes JSON compactly with ensure_ascii=False. Recreate
            # those bytes instead of trusting a supplied body hash alone.
            expected_body_hash = (
                hashlib.sha256(
                    json.dumps(
                        association["body"],
                        ensure_ascii=False,
                        separators=(",", ":"),
                        allow_nan=False,
                    ).encode()
                ).hexdigest()
                if association["body"] is not None
                else None
            )
            require(
                expected_body_hash == association["body_sha256"],
                "HTTP request bytes differ from the retained body",
            )
            require(
                sha256_hex_utf8(canonicalize_json(binding)) == receipt["content_hash"],
                "HTTP receipt belongs to different request content",
            )
            require(
                receipt["capability_id"] == association["capability_id"],
                "HTTP receipt names another grant",
            )
            require(
                receipt["metadata"]["chio_http_status_scope"] == "final",
                "HTTP record is not a final response",
            )
        quote = json.loads((root / "contracts/quote-response.json").read_text())
        fulfillment = json.loads((root / "contracts/fulfillment-package.json").read_text())
        settlement = json.loads((root / "contracts/settlement-reconciliation.json").read_text())
        # approval_required is a buyer decision, so compare the provider's
        # immutable quote identity and price independently of that local field.
        require(
            quote["quote_id"] == outputs["request_quote"]["quote_id"]
            and quote["price_minor"] == outputs["request_quote"]["price_minor"],
            "Quote differs from provider evidence",
        )
        require(
            fulfillment == outputs["execute_review"],
            "Fulfillment differs from signed provider output",
        )
        require(
            fulfillment["job_id"] == settlement["job_id"] == run["job_id"],
            "Work and settlement belong to different jobs",
        )
        require(fulfillment["artifacts"], "No work product was delivered")
        for artifact in fulfillment["artifacts"]:
            require(
                hashlib.sha256(artifact["content"].encode()).hexdigest() == artifact["sha256"],
                "Delivered artifact hash mismatch",
            )
        amount = quote["price_minor"]
        require(
            settlement["status"] == "reconciled"
            and settlement["settlement_kind"] == "internal_book_entry",
            "Missing completed book settlement",
        )
        require(settlement["settled_amount_minor"] == amount, "Settlement price differs from quote")
        entries = settlement["entries"]
        require(
            sorted(entry["amount_minor"] for entry in entries) == [-amount, amount],
            "Settlement entries are not balanced",
        )
        ledger = json.loads((root / "financial/book-ledger.json").read_text())
        stored = [entry for entry in ledger if entry["job_id"] == run["job_id"]]
        require(
            sorted((entry["account"], entry["amount_minor"]) for entry in stored)
            == sorted((entry["account"], entry["amount_minor"]) for entry in entries),
            "Settlement differs from the retained ledger",
        )
        return {
            "ok": True,
            "agent_status": run["final_status"],
            "agent_mode": run["mode"],
            "verified_receipts": len(ids),
            "artifacts": len(fulfillment["artifacts"]),
            "settled_amount_minor": amount,
            "settlement_kind": "internal_book_entry",
            "errors": [],
        }
    except (OSError, ValueError, KeyError, TypeError) as error:
        errors.append(str(error))
    return {"ok": False, "errors": errors}

#!/usr/bin/env python3
"""Run a real procurement application with retained state, work products and receipts."""

from __future__ import annotations

import argparse
import fcntl
import hashlib
import json
import os
import secrets
import shutil
import socket
import subprocess
import sys
import time
import urllib.error
import urllib.request
import uuid
from contextlib import contextmanager
from pathlib import Path

import httpx
from nacl.signing import SigningKey

ROOT = Path(__file__).resolve().parent


def request(url, body=None, *, token=None, capability=None, headers=None):
    fields = {"Content-Type": "application/json", **(headers or {})}
    if token:
        fields["Authorization"] = "Bearer " + token
    if capability:
        fields["X-Chio-Capability"] = json.dumps(capability, separators=(",", ":"))
    query = urllib.request.Request(
        url, headers=fields, data=None if body is None else json.dumps(body).encode()
    )
    try:
        response = urllib.request.urlopen(query, timeout=30)
    except urllib.error.HTTPError as error:
        response = error
    with response:
        return response.status, response.headers, json.loads(response.read())


def seed(path):
    if not path.exists():
        with os.fdopen(os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600), "w") as output:
            output.write(SigningKey.generate().encode().hex() + "\n")
    return SigningKey(bytes.fromhex(path.read_text().strip())).verify_key.encode().hex()


def free_address():
    with socket.socket() as listener:
        listener.bind(("127.0.0.1", 0))
        return "http://127.0.0.1:" + str(listener.getsockname()[1])


@contextmanager
def services(state, run):
    binary = os.getenv("CHIO_BIN") or shutil.which("chio")
    if not binary:
        raise RuntimeError("Install the Chio CLI revision documented in README.md, or set CHIO_BIN")
    binary = str(Path(binary).resolve())
    state.mkdir(parents=True, mode=0o700, exist_ok=True)
    state.chmod(0o700)
    run.mkdir(parents=True, mode=0o700, exist_ok=True)
    lock = (state / "operator.lock").open("a")
    try:
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError as error:
        lock.close()
        raise RuntimeError(
            "Another instance owns this application state; stop it before continuing"
        ) from error
    children, logs = [], []
    try:
        signers = {name: seed(state / (name + ".seed")) for name in ("trust", "buyer", "caller")}
        token, edge_token, approval_token = [secrets.token_hex(32) for _ in range(3)]
        control, provider, api, buyer = [free_address() for _ in range(4)]
        environment = {
            **os.environ,
            "CHIO_BIN": binary,
            "CHIO_CONTROL_URL": control,
            "CHIO_CONTROL_TOKEN": token,
            "CHIO_EDGE_TOKEN": edge_token,
            "CHIO_TRUSTED_ISSUER_KEY": signers["trust"],
            "PROVIDER_EDGE_LISTEN": provider.removeprefix("http://"),
            "PROVIDER_SESSION_DB": str(state / "provider-sessions.sqlite3"),
            "PROVIDER_ARTIFACTS": str(state / "provider-deliverables"),
            "BUYER_STATE_DB": str(state / "buyer.db"),
            "BUYER_APPROVAL_TOKEN": approval_token,
            "BUYER_PROVIDER_BASE_URL": provider,
            "BUYER_PROVIDER_AUTH_TOKEN": edge_token,
            "BUYER_AUTHORITY_SEED_FILE": str(state / "buyer.seed"),
            "BUYER_STATE_DIR": str(state / "buyer"),
            "BUYER_UPSTREAM_URL": api,
            "BUYER_SIDECAR_LISTEN": buyer.removeprefix("http://"),
            "BUYER_RECEIPT_STORE": str(state / "buyer-receipts.sqlite3"),
        }

        def start(name, command):
            log = (run / (name + ".log")).open("w")
            logs.append(log)
            child = subprocess.Popen(
                command, cwd=ROOT, env=environment, stdin=subprocess.DEVNULL, stdout=log, stderr=log
            )
            children.append(child)
            return child

        def ready(url=None, tcp=None):
            for _ in range(300):
                if any(child.poll() is not None for child in children):
                    raise RuntimeError(f"A service stopped; inspect {run}/*.log")
                try:
                    if url and request(url)[0] == 200:
                        return
                    if tcp:
                        with socket.create_connection(
                            ("127.0.0.1", int(tcp.rsplit(":", 1)[1])), timeout=0.2
                        ):
                            return
                except (OSError, ValueError):
                    pass
                time.sleep(0.1)
            raise RuntimeError(f"Service startup timed out; inspect {run}/*.log")

        trust_process = start(
            "trust",
            [
                binary,
                "trust",
                "serve",
                "--listen",
                control.removeprefix("http://"),
                "--service-token",
                token,
                "--receipt-db",
                str(state / "trust-receipts.sqlite3"),
                "--authority-seed-file",
                str(state / "trust.seed"),
                "--session-db",
                str(state / "trust-joint.sqlite3"),
            ],
        )
        ready(control + "/health")
        provider_process = start("provider", ["bash", "provider/run-edge.sh"])
        ready(tcp=provider)
        # This is the public key file of the kernel this operator owns. It is
        # selected before the buyer accepts any provider response or receipt.
        signers["provider"] = (state / "provider-sessions.sqlite3.kernel.pub").read_text().strip()
        if len(bytes.fromhex(signers["provider"])) != 32:
            raise ValueError("Invalid provider kernel public key")
        pin = state / "operator-signers.json"
        if pin.exists() and json.loads(pin.read_text()) != signers:
            raise ValueError(
                "A retained kernel identity changed; inspect the state before proceeding"
            )
        pin.write_text(json.dumps(signers, indent=2) + "\n")
        (run / "trusted-signers.json").write_text(json.dumps(signers, indent=2) + "\n")
        environment["BUYER_PROVIDER_KERNEL_KEY"] = signers["provider"]
        start(
            "buyer-api",
            [
                sys.executable,
                "-m",
                "uvicorn",
                "buyer.app:app",
                "--host",
                "127.0.0.1",
                "--port",
                api.rsplit(":", 1)[1],
            ],
        )
        ready(api + "/healthz")
        start("buyer-gateway", ["bash", "buyer/run-sidecar.sh"])
        ready(buyer + "/chio/health")
        yield {
            "control": control,
            "provider": provider,
            "api": api,
            "buyer": buyer,
            "service_token": token,
            "edge_token": edge_token,
            "approval_token": approval_token,
            "signers": signers,
            "trust_process": trust_process,
            "provider_process": provider_process,
        }
    finally:
        for child in reversed(children):
            if child.poll() is None:
                child.terminate()
        for child in reversed(children):
            try:
                child.wait(timeout=10)
            except subprocess.TimeoutExpired:
                child.kill()
                child.wait()
        for log in logs:
            log.close()
        lock.close()


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "action", nargs="?", default="buy", choices=["buy", "check", "approve", "fund", "status"]
    )
    parser.add_argument(
        "value", nargs="?", help="job ID for approve; additional allowance in cents for fund"
    )
    parser.add_argument(
        "--target", default="payments-api", help="relative directory inside workspace/"
    )
    parser.add_argument(
        "--scope",
        default="hotfix-review",
        choices=[
            "hotfix-review",
            "release-review",
            "release-plus-cloud-review",
            "full-estate-review",
        ],
    )
    parser.add_argument(
        "--budget",
        type=int,
        default=90_000,
        help="maximum purchase amount in USD cents; cannot raise the operator budget",
    )
    parser.add_argument("--provider", choices=["direct", "openai", "anthropic"], default="direct")
    parser.add_argument("--model", help="model identifier accepted by your configured provider")
    args = parser.parse_args()
    if args.budget < 0:
        parser.error("--budget must be nonnegative")
    if args.provider == "openai" and not os.getenv("OPENAI_API_KEY"):
        parser.error("Set OPENAI_API_KEY for the selected provider")
    if args.provider == "anthropic" and not (
        os.getenv("ANTHROPIC_API_KEY") or os.getenv("ANTHROPIC_AUTH_TOKEN")
    ):
        parser.error("Set ANTHROPIC_API_KEY or ANTHROPIC_AUTH_TOKEN for the selected provider")
    os.environ["COMMERCE_PROVIDER"] = args.provider
    if args.model:
        os.environ["OPENAI_MODEL" if args.provider == "openai" else "ANTHROPIC_MODEL"] = args.model
    os.umask(0o077)
    state = ROOT / ".state"
    if args.action == "check":
        os.environ["BUYER_DEFAULT_BUDGET_MINOR"] = "500000"
        state = state / "qualification" / str(uuid.uuid4())
    if args.action in {"fund", "status"}:
        from buyer.store import ProcurementStore

        state.mkdir(parents=True, mode=0o700, exist_ok=True)
        store = ProcurementStore(
            str(state / "buyer.db"), int(os.getenv("BUYER_DEFAULT_BUDGET_MINOR", "150000"))
        )
        try:
            with store.transaction() as db:
                if args.action == "fund":
                    if (
                        not args.value
                        or not args.value.isdigit()
                        or not 0 < int(args.value) <= 1_000_000_000
                    ):
                        parser.error(
                            "fund requires an additional allowance between 1 and 1000000000 cents"
                        )
                    db.execute(
                        "INSERT INTO allocations(id,amount,reason) VALUES(?,?,?)",
                        (str(uuid.uuid4()), int(args.value), "Explicit local operator allocation"),
                    )
                print(
                    json.dumps(
                        {
                            "allowance_minor": store.capacity(db),
                            "available_minor": store.available(db),
                            "jobs": [
                                {
                                    key: job[key]
                                    for key in ("job_id", "status", "target", "requested_scope")
                                }
                                for row in db.execute("SELECT body FROM jobs")
                                for job in [json.loads(row[0])]
                            ],
                        },
                        indent=2,
                    )
                )
        finally:
            store.db.close()
        return 0
    if args.action == "approve" and not args.value:
        parser.error("approve requires the pending job ID")
    run = state / "runs" / str(uuid.uuid4())
    with services(state, run) as endpoints:
        if args.action == "check":
            from qualify import qualify

            qualify(state, run, endpoints)
            return 0
        if args.action == "approve":
            from commerce_network.chio import TrustControl

            trust = TrustControl(endpoints["control"], endpoints["service_token"])
            grant = trust.issue_capability(
                subject_pk=endpoints["signers"]["caller"],
                ttl=300,
                scope={
                    "grants": [
                        {
                            "server_id": "chio_http_authority",
                            "tool_name": "authorize_http_request",
                            "operations": ["invoke"],
                            "constraints": [],
                        }
                    ],
                    "resource_grants": [],
                    "prompt_grants": [],
                },
            )
            path = "/procurement/jobs/" + args.value + "/approve"
            body = {
                "approver": "local-operator",
                "reason": "Explicit approval using the operator launcher",
            }
            with httpx.Client(timeout=30) as client:
                response = client.post(
                    endpoints["buyer"] + path,
                    json=body,
                    headers={
                        "X-Chio-Capability": json.dumps(grant, separators=(",", ":")),
                        "X-Buyer-Approval": endpoints["approval_token"],
                    },
                )
            job = response.json()
            if response.status_code != 200:
                raise ValueError(f"Approval refused: HTTP {response.status_code}: {job}")
            job["_chio_http"] = {
                "receipt_id": response.headers.get("X-Chio-Receipt-Id"),
                "method": "POST",
                "path": path,
                "body": body,
                "body_sha256": hashlib.sha256(response.request.content).hexdigest(),
                "capability_id": grant["id"],
                "status": response.status_code,
            }
            agent = {
                "mode": "operator-approval",
                "final_status": job["status"],
                "job_id": job["job_id"],
                "tool_calls": [{"tool": "approve_job", "output": job}],
            }
            (run / "agent-output.json").write_text(json.dumps(agent, indent=2) + "\n")
            (run / "summary.json").write_text(
                json.dumps({"final_status": job["status"]}, indent=2) + "\n"
            )
            contracts = run / "contracts"
            contracts.mkdir()
            for name, key in [
                ("quote-response", "quote"),
                ("fulfillment-package", "fulfillment"),
                ("settlement-reconciliation", "settlement"),
            ]:
                (contracts / (name + ".json")).write_text(json.dumps(job[key], indent=2) + "\n")
            from commerce_network.verify import verify_bundle
            from export_evidence import export

            export(run, state)
            verification = verify_bundle(run, trusted_signers=list(endpoints["signers"].values()))
            (run / "verification.json").write_text(json.dumps(verification, indent=2) + "\n")
            if not verification["ok"]:
                raise ValueError(
                    "Approval verification failed: " + "; ".join(verification["errors"])
                )
            print(
                f"Job {job['job_id']}: {job['status']}\nVerified approval, delivered work and settlement.\nEvidence: {run.relative_to(ROOT)}"
            )
            return 0
        from orchestrate import main as procure

        result = procure(
            [
                "--control-url",
                endpoints["control"],
                "--service-token",
                endpoints["service_token"],
                "--buyer-url",
                endpoints["buyer"],
                "--buyer-auth-token",
                endpoints["edge_token"],
                "--scope",
                args.scope,
                "--target",
                args.target,
                "--budget-minor",
                str(args.budget),
                "--artifact-dir",
                str(run),
            ]
        )
        from export_evidence import export

        export(run, state)
        summary = json.loads((run / "summary.json").read_text())
        if summary["final_status"] == "fulfilled":
            from commerce_network.verify import verify_bundle

            verification = verify_bundle(run, trusted_signers=list(endpoints["signers"].values()))
            (run / "verification.json").write_text(json.dumps(verification, indent=2) + "\n")
            if not verification["ok"]:
                raise ValueError("Run verification failed: " + "; ".join(verification["errors"]))
            print(
                f"Verified {verification['verified_receipts']} signed receipts, {verification['artifacts']} delivered artifacts, and balanced book entries."
            )
        print(
            f"Status: {summary['final_status']}\nEvidence: {run.relative_to(ROOT)}\nApplication data and budget are retained in .state/."
        )
        return result


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (ValueError, RuntimeError, OSError, subprocess.CalledProcessError) as error:
        print(str(error), file=sys.stderr)
        raise SystemExit(1) from None