Hello A2A

{
  "name": "Chio document analyst",
  "description": "Count caller-provided text under a bounded signed grant",
  "supportedInterfaces": [
    {
      "url": "http://127.0.0.1:33067",
      "protocolBinding": "JSONRPC",
      "protocolVersion": "1.0"
    }
  ],
  "version": "0.1.0",
  "capabilities": {
    "streaming": false,
    "pushNotifications": false
  },
  "defaultInputModes": [
    "application/json"
  ],
  "defaultOutputModes": [
    "application/json",
    "text/plain"
  ],
  "skills": [
    {
      "id": "count_words",
      "name": "count_words",
      "description": "Count whitespace-separated words and UTF-8 bytes",
      "inputModes": [
        "application/json"
      ],
      "outputModes": [
        "application/json",
        "text/plain"
      ]
    }
  ]
}

{
  "method": "SendMessage",
  "params": {
    "message": {
      "messageId": "4d04eaf4-407f-439b-bee3-4ee84b36f85a",
      "role": "ROLE_USER",
      "parts": [
        {
          "data": {
            "text": "Chio gives agent operations explicit authority and evidence"
          }
        }
      ]
    },
    "metadata": {
      "chio": {
        "targetSkillId": "count_words"
      }
    }
  },
  "id": "5217355c-5550-4563-a280-2e7ed06d5d4d",
  "jsonrpc": "2.0"
}

{
  "trust_level": "mediated",
  "boundary_class": "prevent",
  "action": {
    "parameter_hash": "f0d237fe0bfdf5ee7bde0d96d6555723e2effa1281eb6d1d4ce514494b165e77",
    "parameters": {
      "text": "Chio gives agent operations explicit authority and evidence"
    }
  },
  "redaction_mode": "none",
  "tool_name": "count_words",
  "policy_hash": "56f393294872b8f8213479eff52ad3fb3e61581cc62a08410ddbbad21c4e745a",
  "kernel_key": "76cbb3c9ea1166015df0e8a21f61d3f3114486ee16da30989651c5760c2fc383",
  "tool_server": "documents",
  "content_hash": "8acf8908e7aeb801d9ee99e2965eb052c1d90e2e04f97d50be02ae149d079bd6",
  "metadata": {
    "route_selection": {
      "selectedProtocols": [
        "a2a",
        "native"
      ],
      "selectedRouteId": "native-route",
      "requestedTargetProtocol": "native",
      "selectedTargetProtocol": "native",
      "candidates": [
        {
          "selectedProtocols": [
            "a2a",
            "native"
          ],
          "routeId": "native-route",
          "available": true,
          "targetProtocol": "native"
        }
      ],
      "decision": "select",
      "routeSelectionId": "fe26d5b5a64ff93b186824e0cc61311ed384ac1dcfeeafb8d3d399b6fc33a0d0",
      "sourceProtocol": "a2a"
    },
    "admission_operation": {
      "retained_dispatch_commit": {
        "committed_version": 6,
        "provider_attempt": {
          "transport_key_epoch": 1,
          "attempt_id": "attempt:5cd58f78cab93490cd27f7590ec734ea33284f55e3d250ff1dd687ba21deaef5",
          "operation_id": "5cd58f78cab93490cd27f7590ec734ea33284f55e3d250ff1dd687ba21deaef5",
          "transport_id": "kernel-tool-server:documents"
        },
        "coordinator_lease_epoch": 1,
        "store_fence": {
          "lease_id": "01a08ea1-cb39-77d9-85fc-7a57d23c2ea3",
          "store_uuid": "01a08ea1-c9d3-77f1-84e1-57c471839cb6",
          "owner_epoch": 1
        },
        "coordinator_lease_id": "01a08ea1-cb39-77d9-85fc-7a57d23c2ea3"
      },
      "tool_outcome_id": "2df324d9b326f2ade4bb9c97a674cf2d7345b3031f2c1d16cf840622af0692f9",
      "trusted_time_unix_ms": 1789099363717,
      "projected_operation_version": 8,
      "projected_state": "completed",
      "schema": "chio.admission-receipt.v1",
      "request_namespace_digest": "08d7f6a8ba0a6a0ebe77a058222095ae5782509e270c73f75e7fb5fcc0c87ebc",
      "coordinator_lease_epoch": 1,
      "operation_id": "5cd58f78cab93490cd27f7590ec734ea33284f55e3d250ff1dd687ba21deaef5",
      "compensation_status": "not_compensated",
      "request_binding_hash": "b9483b689b03e16d9cd0b41ba788d3a880574b0884877dcbef158c35a8a7c6c0",
      "tool_outcome_version": 2,
      "projected_dispatch_state": "terminal",
      "request_id": "a2a-message:814d212f0f40657b29caadb395b494fd6bb07eb0e431ca43d05b03271c5fd825:4d04eaf4-407f-439b-bee3-4ee84b36f85a",
      "store_fence": {
        "lease_id": "01a08ea1-cb39-77d9-85fc-7a57d23c2ea3",
        "store_uuid": "01a08ea1-c9d3-77f1-84e1-57c471839cb6",
        "owner_epoch": 1
      },
      "coordinator_lease_id": "01a08ea1-cb39-77d9-85fc-7a57d23c2ea3"
    },
    "receipt_context": {
      "request_id": "a2a-message:814d212f0f40657b29caadb395b494fd6bb07eb0e431ca43d05b03271c5fd825:4d04eaf4-407f-439b-bee3-4ee84b36f85a"
    },
    "attribution": {
      "grant_index": 0,
      "delegation_depth": 0,
      "subject_key": "814d212f0f40657b29caadb395b494fd6bb07eb0e431ca43d05b03271c5fd825",
      "issuer_key": "76cbb3c9ea1166015df0e8a21f61d3f3114486ee16da30989651c5760c2fc383"
    },
    "stream": {
      "chunks_received": 2,
      "chunks_expected": 2,
      "total_bytes": 77,
      "chunk_hashes": [
        "c59ae49b669759807ff89ce98ab61530df38c047104676a7defc2e7d301324e3",
        "b2538f4ac125b83068662cf8f4dc7966e50d1bcbfce089b5824edea204d91624"
      ]
    },
    "chio_receipt_signing_nonce": "rcpt-01a08ea1-cdaa-748c-a529-317ec183de4d"
  },
  "id": "630d12d93d82f7773ba04bae94376d77818db661c69231e4ffe88753b70ccaa0",
  "tool_origin": "caller_executed",
  "capability_id": "cap-01a08ea1-cbce-7282-b0ab-190a5b50e499",
  "signature": "f9e2165a32ff8182770dfb750e28277c5668b250080ace54c6a5146c3ac59d8ae02d1e697709944bac2ac00691dfa3dd495f541c90a33ae6ce9fc181bfd34507",
  "decision": {
    "verdict": "allow"
  },
  "timestamp": 1789099363,
  "receipt_kind": "mediated_decision"
}

{
  "trust_level": "mediated",
  "boundary_class": "prevent",
  "action": {
    "parameter_hash": "f0d237fe0bfdf5ee7bde0d96d6555723e2effa1281eb6d1d4ce514494b165e77",
    "parameters": {
      "text": "Chio gives agent operations explicit authority and evidence"
    }
  },
  "redaction_mode": "none",
  "tool_name": "count_words",
  "policy_hash": "56f393294872b8f8213479eff52ad3fb3e61581cc62a08410ddbbad21c4e745a",
  "kernel_key": "76cbb3c9ea1166015df0e8a21f61d3f3114486ee16da30989651c5760c2fc383",
  "tool_server": "documents",
  "content_hash": "74234e98afe7498fb5daf1f36ac2d78acc339464f950703b8c019892f982b90b",
  "metadata": {
    "route_selection": {
      "selectedProtocols": [
        "a2a",
        "native"
      ],
      "selectedRouteId": "native-route",
      "requestedTargetProtocol": "native",
      "selectedTargetProtocol": "native",
      "candidates": [
        {
          "selectedProtocols": [
            "a2a",
            "native"
          ],
          "routeId": "native-route",
          "available": true,
          "targetProtocol": "native"
        }
      ],
      "decision": "select",
      "routeSelectionId": "9ed946ad973f8e988786f0cd80309546d67d773f44fe50c3c70c4341aa6819ee",
      "sourceProtocol": "a2a"
    },
    "receipt_context": {
      "request_id": "a2a-message:814d212f0f40657b29caadb395b494fd6bb07eb0e431ca43d05b03271c5fd825:0c31844e-c67f-4477-9420-b36e1fa6efa9"
    },
    "attribution": {
      "delegation_depth": 0,
      "subject_key": "814d212f0f40657b29caadb395b494fd6bb07eb0e431ca43d05b03271c5fd825",
      "issuer_key": "76cbb3c9ea1166015df0e8a21f61d3f3114486ee16da30989651c5760c2fc383"
    },
    "budget_authority": {
      "guarantee_level": "single_node_atomic",
      "metering_profile": "max_cost_preauthorize_then_reconcile_actual",
      "authority_profile": "authoritative_hold_event"
    },
    "chio_receipt_signing_nonce": "rcpt-01a08ea1-cf53-7211-8b17-ff6d409298d9"
  },
  "id": "ab00d1978888195456fb894b842506d65caebd08c31734a03b55d0778e5f104e",
  "tool_origin": "caller_executed",
  "capability_id": "cap-01a08ea1-cbce-7282-b0ab-190a5b50e499",
  "signature": "e76242b95a04cbb9628addab399033f95c71ef570b5bbc4a749fbb3e5a2be365c941d946f9b2e80034ad1d6fed84cf55ab832ab8433df835f49ad0d8eb650700",
  "decision": {
    "guard": "kernel",
    "verdict": "deny",
    "reason": "invocation budget exhausted for capability cap-01a08ea1-cbce-7282-b0ab-190a5b50e499"
  },
  "timestamp": 1789099364,
  "receipt_kind": "mediated_decision"
}

{
  "schema": "document-tools.authority.v1",
  "tool": "count_words",
  "max_invocations": 2,
  "lifetime_seconds": 1800
}

use chio_kernel::{
    KernelError, NestedFlowBridge, ToolCallChunk, ToolCallStream, ToolServerConnection,
    ToolServerStreamResult,
};
use chio_manifest::{ToolDefinition, ToolManifest};
use serde_json::{json, Value};

pub const SERVER: &str = "documents";
pub const TOOL: &str = "count_words";
pub const TEXT_LIMIT: usize = 65_536;

pub struct DocumentTools;

fn count(tool: &str, arguments: &Value) -> Result<Value, KernelError> {
    if tool != TOOL {
        return Err(KernelError::ToolNotRegistered(tool.to_owned()));
    }
    let text = arguments
        .get("text")
        .and_then(Value::as_str)
        .ok_or_else(|| KernelError::RequestIncomplete("text must be a string".into()))?;
    if text.len() > TEXT_LIMIT {
        return Err(KernelError::RequestIncomplete(
            "text exceeds 65536 UTF-8 bytes".into(),
        ));
    }
    Ok(json!({"words": text.split_whitespace().count(), "bytes": text.len()}))
}

#[async_trait::async_trait]
impl ToolServerConnection for DocumentTools {
    fn server_id(&self) -> &str {
        SERVER
    }
    fn tool_names(&self) -> Vec<String> {
        vec![TOOL.into()]
    }

    async fn invoke(
        &self,
        tool: &str,
        arguments: Value,
        _nested: Option<&mut dyn NestedFlowBridge>,
    ) -> Result<Value, KernelError> {
        count(tool, &arguments)
    }

    async fn invoke_stream(
        &self,
        tool: &str,
        arguments: Value,
        _nested: Option<&mut dyn NestedFlowBridge>,
    ) -> Result<Option<ToolServerStreamResult>, KernelError> {
        let result = count(tool, &arguments)?;
        // These chunks are collated by the edge into one terminal response.
        Ok(Some(ToolServerStreamResult::Complete(ToolCallStream {
            chunks: vec![
                ToolCallChunk { data: result },
                ToolCallChunk {
                    data: json!({"content": [
                        {"type": "text", "text": "Document counted"}
                    ]}),
                },
            ],
        })))
    }
}

pub fn manifest(public_key: String) -> ToolManifest {
    ToolManifest {
        schema: "chio.manifest.v1".into(),
        server_id: SERVER.into(),
        name: "Document tools".into(),
        version: "1.0.0".into(),
        description: Some("Count words in caller-provided text".into()),
        tools: vec![ToolDefinition {
            name: TOOL.into(),
            description: "Count whitespace-separated words and UTF-8 bytes".into(),
            input_schema: json!({
                "type": "object", "properties": {"text": {"type": "string"}},
                "required": ["text"], "x-chio-streaming": true,
                "x-chio-partial-output": true, "x-chio-cancellation": true
            }),
            output_schema: None,
            pricing: None,
            has_side_effects: false,
            latency_hint: None,
        }],
        server_tools: Vec::new(),
        required_permissions: None,
        public_key,
    }
}

#[path = "../host.rs"]
mod host;
#[path = "../service.rs"]
mod service;

use axum::{
    extract::{DefaultBodyLimit, State},
    http::{HeaderMap, StatusCode},
    routing::{get, post},
    Json, Router,
};
use chio_a2a_edge::{A2aEdgeConfig, A2aKernelExecutionContext, ChioA2aEdge};
use serde_json::{json, Value};
use std::{
    path::PathBuf,
    sync::{Arc, Mutex},
};

struct App {
    edge: Mutex<ChioA2aEdge>,
    kernel: chio_kernel::ChioKernel,
    execution: A2aKernelExecutionContext,
    token: String,
    card: Value,
    blocking: chio_example_runtime::tasks::BlockingPool,
}

async fn invoke(
    State(app): State<Arc<App>>,
    headers: HeaderMap,
    Json(message): Json<Value>,
) -> Result<Json<Value>, (StatusCode, &'static str)> {
    let supplied = headers
        .get("authorization")
        .and_then(|value| value.to_str().ok())
        .and_then(|value| value.strip_prefix("Bearer "))
        .unwrap_or("");
    if !chio_example_runtime::credential_matches(supplied, &app.token) {
        return Err((
            StatusCode::UNAUTHORIZED,
            "A valid application credential is required",
        ));
    }
    let pool = app.blocking.clone();
    let response = pool
        .run(move || {
            let mut edge = app
                .edge
                .lock()
                .map_err(|_| anyhow::anyhow!("Protocol state is unavailable"))?;
            Ok(edge
                .handle_v1_jsonrpc(message, &app.kernel, &app.execution)
                .into_value())
        })
        .await
        .map_err(|error| {
            eprintln!("Protocol execution refused: {error}");
            (
                StatusCode::SERVICE_UNAVAILABLE,
                "Protocol execution unavailable",
            )
        })?;
    response
        .map(Json)
        .ok_or((StatusCode::BAD_REQUEST, "A2A requests require an ID"))
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let directory = PathBuf::from(
        std::env::args()
            .nth(1)
            .ok_or_else(|| anyhow::anyhow!("Usage: a2a-host NEW_SESSION_DIRECTORY"))?,
    );
    let host = host::boot(&directory)?;
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
    let endpoint = format!("http://{}", listener.local_addr()?);
    let edge = ChioA2aEdge::new(
        A2aEdgeConfig {
            agent_name: "Chio document analyst".into(),
            agent_description: "Count caller-provided text under a bounded signed grant".into(),
            endpoint_url: endpoint.clone(),
            ..A2aEdgeConfig::default()
        },
        vec![host.manifest],
    )?;
    let token = chio_core::crypto::Keypair::generate().public_key().to_hex();
    std::fs::write(directory.join("client-token.txt"), &token)?;
    let execution = A2aKernelExecutionContext {
        agent_id: host.capability.subject.to_hex(),
        capability: host.capability,
        dpop_proof: None,
        execution_nonce: None,
        governed_intent: None,
        approval_token: None,
        approval_tokens: Vec::new(),
        threshold_approval_proposal: None,
        supplemental_authorization: None,
        model_metadata: None,
    };
    let app = Arc::new(App {
        card: edge.agent_card_v1(),
        edge: Mutex::new(edge),
        kernel: host.kernel,
        execution,
        token,
        blocking: chio_example_runtime::tasks::BlockingPool::new(2),
    });
    let router = Router::new()
        .route("/", post(invoke))
        .route(
            "/.well-known/agent-card.json",
            get(|State(app): State<Arc<App>>| async move { Json(app.card.clone()) }),
        )
        .layer(DefaultBodyLimit::max(65_536))
        .with_state(app.clone());
    std::fs::write(
        directory.join("server.json"),
        serde_json::to_vec_pretty(&json!({"endpoint":endpoint}))?,
    )?;
    eprintln!(
        "A2A 1.0 listening at {endpoint}; credential: {}",
        directory.join("client-token.txt").display()
    );
    let signal = chio_example_runtime::shutdown_signal()?;
    let served = axum::serve(listener, router)
        .with_graceful_shutdown(signal)
        .await;
    let drained = app.blocking.drain(std::time::Duration::from_secs(15)).await;
    served?;
    drained?;
    Ok(())
}

use chio_core::{
    capability::{
        scope::{ChioScope, Operation, ToolGrant},
        token::CapabilityToken,
    },
    crypto::Keypair,
};
use chio_kernel::{
    ChioKernel, KernelConfig, DEFAULT_CHECKPOINT_BATCH_SIZE, DEFAULT_MAX_STREAM_DURATION_SECS,
    DEFAULT_MAX_STREAM_TOTAL_BYTES,
};
use chio_manifest::ToolManifest;
use std::path::Path;

pub struct Host {
    pub kernel: ChioKernel,
    pub capability: CapabilityToken,
    pub manifest: ToolManifest,
}

pub fn boot(directory: &Path) -> anyhow::Result<Host> {
    let authority_bytes = std::fs::read("authority.json")?;
    let config: serde_json::Value = serde_json::from_slice(&authority_bytes)?;
    let fields = config
        .as_object()
        .ok_or_else(|| anyhow::anyhow!("Authority must be an object"))?;
    anyhow::ensure!(
        fields.keys().all(
            |key| ["schema", "tool", "max_invocations", "lifetime_seconds"].contains(&key.as_str())
        ),
        "Unknown authority field"
    );
    anyhow::ensure!(
        config["schema"] == "document-tools.authority.v1",
        "Unknown authority schema"
    );
    anyhow::ensure!(
        config["tool"] == crate::service::TOOL,
        "This host grants only count_words"
    );
    let call_limit = config["max_invocations"]
        .as_u64()
        .filter(|v| (1..=50).contains(v))
        .ok_or_else(|| anyhow::anyhow!("max_invocations must be an integer from 1 to 50"))?
        as u32;
    let lifetime = config["lifetime_seconds"]
        .as_u64()
        .filter(|v| (1..=3600).contains(v))
        .ok_or_else(|| anyhow::anyhow!("lifetime_seconds must be an integer from 1 to 3600"))?;
    // A new application run gets a fresh session. Refuse to reuse its directory.
    std::fs::create_dir(directory)?;
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(directory, std::fs::Permissions::from_mode(0o700))?;
    }
    std::fs::write(directory.join("authority.json"), &authority_bytes)?;
    let authority =
        chio_control_plane::DurableAdmissionRuntime::open(&directory.join("admission.db"))?;
    let keypair = authority.kernel_keypair();
    // The host chooses this key before any receipt exists.
    std::fs::write(
        directory.join("kernel-public-key.txt"),
        keypair.public_key().to_hex(),
    )?;
    let mut kernel = ChioKernel::new(KernelConfig {
        keypair,
        ca_public_keys: Vec::new(),
        max_delegation_depth: 3,
        policy_hash: chio_core::sha256_hex(&authority_bytes),
        allow_sampling: false,
        allow_sampling_tool_use: false,
        allow_elicitation: false,
        max_stream_duration_secs: DEFAULT_MAX_STREAM_DURATION_SECS,
        max_stream_total_bytes: DEFAULT_MAX_STREAM_TOTAL_BYTES,
        require_web3_evidence: false,
        allow_ephemeral_receipt_log: true,
        allow_ephemeral_revocation_store: true,
        checkpoint_batch_size: DEFAULT_CHECKPOINT_BATCH_SIZE,
        retention_config: None,
        memory_budget: chio_kernel::MemoryBudgetConfig::defaults(),
        deadlines: chio_kernel::HotPathDeadlineConfig::default(),
    });
    kernel.set_receipt_store(Box::new(chio_store_sqlite::SqliteReceiptStore::open(
        directory.join("receipts.db"),
    )?))?;
    authority.attach(&mut kernel)?;
    let service_key = Keypair::generate();
    let manifest = crate::service::manifest(service_key.public_key().to_hex());
    let signed = chio_manifest::sign_manifest(&manifest, &service_key)?;
    chio_manifest::verify_manifest(&signed, &service_key.public_key())?;
    std::fs::write(
        directory.join("manifest.json"),
        serde_json::to_vec_pretty(&signed)?,
    )?;
    kernel.register_tool_server(Box::new(crate::service::DocumentTools));
    let agent = Keypair::generate();
    let capability = kernel.issue_capability(
        &agent.public_key(),
        ChioScope {
            grants: vec![ToolGrant {
                server_id: crate::service::SERVER.into(),
                tool_name: crate::service::TOOL.into(),
                operations: vec![Operation::Invoke],
                constraints: Vec::new(),
                max_invocations: Some(call_limit),
                max_cost_per_invocation: None,
                max_total_cost: None,
                dpop_required: None,
            }],
            ..ChioScope::default()
        },
        lifetime,
    )?;
    std::fs::write(
        directory.join("capability.json"),
        serde_json::to_vec_pretty(&capability)?,
    )?;
    Ok(Host {
        kernel,
        capability,
        manifest,
    })
}

# /// script
# requires-python = ">=3.11"
# dependencies = ["a2a-sdk==1.1.2", "chio-sdk"]
# [tool.uv.sources]
# chio-sdk = { path = "sdk/chio-py" }
# ///
"""Call a real Chio HTTP agent with the official A2A Python client."""
import argparse
import asyncio
import json
import os
from pathlib import Path
import subprocess
import time
import uuid

import httpx
from a2a.client import A2ACardResolver
from a2a.client.transports.jsonrpc import JsonRpcTransport
from a2a.types import GetTaskRequest, Message, Part, Role, SendMessageRequest
from google.protobuf.json_format import MessageToDict, ParseDict
from google.protobuf.struct_pb2 import Value as ProtobufValue
from chio.invariants import canonicalize_json, sha256_hex_utf8, verify_receipt_with_trusted_signers

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


def projected_parts(value):
    if isinstance(value, str):
        return [{"text": value}]
    if isinstance(value, dict) and isinstance(value.get("content"), list):
        return [{"text": part["text"]} for part in value["content"] if isinstance(part.get("text"), str)]
    if isinstance(value, (dict, list)):
        return [{"data": value}]
    return [{"text": json.dumps(value, separators=(",", ":"))}]


def verify_task(task, text, key):
    receipt = task["metadata"]["chio"]["receipt"]
    if not verify_receipt_with_trusted_signers(receipt, [key])["ok"]:
        raise ValueError("Invalid receipt or untrusted kernel")
    if task["metadata"]["chio"]["receiptId"] != receipt["id"]:
        raise ValueError("Task references another receipt")
    if receipt["action"]["parameters"] != {"text": text}:
        raise ValueError("Receipt binds different tool input")
    if sha256_hex_utf8(canonicalize_json({"text": text})) != receipt["action"]["parameter_hash"]:
        raise ValueError("Input hash differs")
    allowed = receipt["decision"]["verdict"] == "allow"
    if allowed != (task["status"]["state"] == "TASK_STATE_COMPLETED"):
        raise ValueError("Task status differs from the signed decision")
    if allowed:
        output = task["artifacts"][0]["parts"][0]["data"]
        if output != {"words": len(text.split()), "bytes": len(text.encode())}:
            raise ValueError("The agent did not return the actual document count")
        retained = task["metadata"]["chio"]["retainedOutput"]
        if retained["kind"] == "stream":
            hashes = [sha256_hex_utf8(canonicalize_json(chunk)) for chunk in retained["chunks"]]
            if hashes != receipt["metadata"]["stream"]["chunk_hashes"]:
                raise ValueError("A retained stream chunk differs from its signed hash")
            content_hash = sha256_hex_utf8("".join(hashes))
            expected_parts = [part for chunk in retained["chunks"] for part in projected_parts(chunk)]
            if retained["chunks"][0] != output:
                raise ValueError("A2A projected another document count")
        else:
            content_hash = sha256_hex_utf8(canonicalize_json(retained["value"]))
            expected_parts = projected_parts(retained["value"])
            if retained["value"] != output:
                raise ValueError("A2A projected another tool output")
        if task["artifacts"][0]["parts"] != expected_parts:
            raise ValueError("A projected artifact changed after the kernel signed its output")
        if content_hash != receipt["content_hash"]:
            raise ValueError("Tool output differs from its signed hash")
    return receipt


async def exercise(endpoint, token, key, capability, text, directory):
    async with httpx.AsyncClient(timeout=30) as public:
        refused = await public.post(endpoint, json={"jsonrpc": "2.0", "id": 1, "method": "SendMessage", "params": {}})
        if refused.status_code != 401:
            raise ValueError("Missing application credential was not refused")
    wire = []
    async def capture(response):
        await response.aread()
        if response.request.method == "POST":
            wire.append({"request": json.loads(response.request.content), "response": response.json()})
            (directory / "wire.json").write_text(json.dumps(wire, indent=2) + "\n")
    async with httpx.AsyncClient(headers={"Authorization": "Bearer " + token}, timeout=30,
                                 event_hooks={"response": [capture]}) as http:
        card = await A2ACardResolver(http, endpoint).get_agent_card()
        client = JsonRpcTransport(http, card, endpoint)
        records = []
        for number in range(3):
            message_id = str(uuid.uuid4())
            request = SendMessageRequest(message=Message(message_id=message_id, role=Role.ROLE_USER,
                parts=[Part(data=ParseDict({"text": text}, ProtobufValue()))]), metadata={"chio": {"targetSkillId": "count_words"}})
            response = await client.send_message(request)
            task = MessageToDict(response)["task"]
            receipt = verify_task(task, text, key)
            if receipt["decision"]["verdict"] != ("allow" if number < 2 else "deny"):
                raise ValueError("The third call must hit the two-invocation grant")
            expected_request = "a2a-message:" + capability["subject"] + ":" + message_id
            if receipt["capability_id"] != capability["id"]:
                raise ValueError("Receipt belongs to another capability")
            if receipt["metadata"]["receipt_context"]["request_id"] != expected_request:
                raise ValueError("Receipt belongs to another protocol message")
            lookup = MessageToDict(await client.get_task(GetTaskRequest(id=task["id"])))
            if lookup != task:
                raise ValueError("GetTask returned another task or changed its evidence")
            if number == 0:
                retry = MessageToDict(await client.send_message(request))["task"]
                if retry != task:
                    raise ValueError("An identical retry did not retain the original result")
            records.append({"message_id": message_id, "task": task, "verified": True})
            print(task["status"]["state"] + ": " + receipt["id"])
        capture_record = {"client": {"package": "a2a-sdk", "version": "1.1.2", "transport": "JsonRpcTransport"},
            "trusted_kernel_key": key, "agent_card": MessageToDict(card), "runs": records,
            "wire": wire, "verification": {"ok": True, "missing_credential_refused": True,
                "identical_retry_retained": True, "task_lookup_bound": True, "quota_refused": True}}
        (directory / "a2a-capture.json").write_text(json.dumps(capture_record, indent=2) + "\n")


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("text", nargs="?", default="Chio gives agent operations explicit authority and evidence")
    args = parser.parse_args()
    if not 0 < len(args.text.encode()) <= 60_000:
        parser.error("text must contain 1 to 60000 UTF-8 bytes")
    subprocess.run(["cargo", "build", "--locked", "--bin", "a2a-host"], cwd=ROOT, check=True)
    target = Path(os.environ.get("CARGO_TARGET_DIR", ROOT / "target"))
    if not target.is_absolute():
        target = ROOT / target
    os.umask(0o077)
    (ROOT / ".state").mkdir(exist_ok=True)
    directory = ROOT / ".state" / ("a2a-" + str(uuid.uuid4()))
    with (ROOT / ".state" / (directory.name + ".log")).open("w") as log:
        process = subprocess.Popen([str(target / "debug/a2a-host"), str(directory)], cwd=ROOT,
                                   stdout=log, stderr=log, stdin=subprocess.DEVNULL)
        try:
            deadline = time.monotonic() + 30
            while not (directory / "server.json").exists():
                if process.poll() is not None or time.monotonic() > deadline:
                    raise RuntimeError("A2A host did not become ready; inspect its .state log")
                time.sleep(.1)
            endpoint = json.loads((directory / "server.json").read_text())["endpoint"]
            token = (directory / "client-token.txt").read_text().strip()
            key = (directory / "kernel-public-key.txt").read_text().strip()
            capability = json.loads((directory / "capability.json").read_text())
            asyncio.run(exercise(endpoint, token, key, capability, args.text, directory))
            print("Evidence:", directory.relative_to(ROOT) / "a2a-capture.json")
        finally:
            if process.poll() is None:
                process.terminate()
            try:
                process.wait(timeout=10)
            except subprocess.TimeoutExpired:
                process.kill()
                process.wait()


if __name__ == "__main__":
    main()