From 120947695f837580daa3f2b4d82def821302179f Mon Sep 17 00:00:00 2001 From: patrickschmied Date: Fri, 12 Jun 2026 01:26:42 -0400 Subject: [PATCH] feat(transport): multi-host Transport seam + /verify endpoint Add a transport abstraction so clients can reach a CHP host over any backend (HTTP first; Zenoh/gRPC pluggable later): - chp_core/transport.py: Transport Protocol (async ainvoke_envelope, discover, replay_result, health, supports) + LocalTransport (in-process) and HttpTransport (stdlib HTTP via RemoteCapabilityHost, run off-loop). - http.py: RemoteCapabilityHost.invoke_envelope() and .verify(); new GET /verify/{correlation_id} route returning the SHA256 chain result. - Export Transport / LocalTransport / HttpTransport from the package. Additive; existing surface unchanged. 752 tests green. Co-Authored-By: Claude Opus 4.8 --- packages/python/chp_core/__init__.py | 4 + packages/python/chp_core/cli/_host.py | 2 +- packages/python/chp_core/http.py | 20 ++++ packages/python/chp_core/transport.py | 132 ++++++++++++++++++++++++++ 4 files changed, 157 insertions(+), 1 deletion(-) create mode 100644 packages/python/chp_core/transport.py diff --git a/packages/python/chp_core/__init__.py b/packages/python/chp_core/__init__.py index 3fbd09b..65ae587 100644 --- a/packages/python/chp_core/__init__.py +++ b/packages/python/chp_core/__init__.py @@ -20,6 +20,7 @@ ) from .host import CapabilityExecutionContext, LocalCapabilityHost from .http import CapabilityHostHTTPServer, RemoteCapabilityHost, create_http_server, serve_http +from .transport import HttpTransport, LocalTransport, Transport from .store import SQLiteEvidenceStore from .decorators import capability from .codex import ( @@ -197,6 +198,9 @@ "InvocationEnvelope", "InvocationResult", "LocalCapabilityHost", + "Transport", + "LocalTransport", + "HttpTransport", "PolicyDescriptor", "ReplayQuery", "ReplayResult", diff --git a/packages/python/chp_core/cli/_host.py b/packages/python/chp_core/cli/_host.py index 7217b2f..e598e75 100644 --- a/packages/python/chp_core/cli/_host.py +++ b/packages/python/chp_core/cli/_host.py @@ -85,7 +85,7 @@ def cmd_serve_http(args: argparse.Namespace) -> int: bind: str = args.bind port: int = args.port print(f"Serving CHP host {host.host_id!r} at http://{bind}:{port}") - print("Routes: GET /health, GET /host, GET /capabilities, POST /invoke, GET /replay/{id}") + print("Routes: GET /health, GET /host, GET /capabilities, POST /invoke, GET /replay/{id}, GET /verify/{id}") try: serve_http(host, bind=bind, port=port) except KeyboardInterrupt: diff --git a/packages/python/chp_core/http.py b/packages/python/chp_core/http.py index 43fb8b7..d0a4114 100644 --- a/packages/python/chp_core/http.py +++ b/packages/python/chp_core/http.py @@ -4,6 +4,7 @@ import asyncio import json +from dataclasses import asdict from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any @@ -57,6 +58,11 @@ def do_GET(self) -> None: correlation_id = unquote(path.removeprefix("/replay/")) self._write_json(self.server.chp_host.replay_result(correlation_id).to_dict()) return + if path.startswith("/verify/"): + correlation_id = unquote(path.removeprefix("/verify/")) + result = self.server.chp_host.store.verify_chain(correlation_id) + self._write_json(asdict(result)) + return self._write_error(HTTPStatus.NOT_FOUND, "not_found", f"Unknown route: {path}") def do_POST(self) -> None: @@ -282,6 +288,11 @@ def invoke( ) -> InvocationResult: return asyncio.run(self.ainvoke(capability_id, payload, **kwargs)) + def invoke_envelope(self, envelope: InvocationEnvelope) -> InvocationResult: + """Invoke from a pre-built envelope (synchronous; mirrors the server's /invoke).""" + data = self._post("/invoke", envelope.to_dict()) + return self._parse_result(data) + def discover(self, **filter_kwargs: Any) -> JSON: """Return the host descriptor, optionally filtering capabilities.""" descriptor = self._get("/host") @@ -308,3 +319,12 @@ def replay_result(self, query: "str | ReplayQuery | JSON") -> JSON: def health(self) -> JSON: """Return the /health response from the remote host.""" return self._get("/health") + + def verify(self, correlation_id: str) -> JSON: + """Return the SHA256 chain verification result for *correlation_id*. + + Shape mirrors ``ChainVerificationResult``: ``correlation_id``, + ``event_count``, ``verified_count``, ``unverified_count``, ``valid``, + ``first_broken_sequence``. + """ + return self._get(f"/verify/{correlation_id}") diff --git a/packages/python/chp_core/transport.py b/packages/python/chp_core/transport.py new file mode 100644 index 0000000..848d63a --- /dev/null +++ b/packages/python/chp_core/transport.py @@ -0,0 +1,132 @@ +"""Transport seam for CHP hosts. + +A ``Transport`` is the uniform async surface a client uses to talk to *one* host — +whether that host is in-process (`LocalTransport`) or reached over a wire +(`HttpTransport`, and later Zenoh/gRPC). The router in ``chp-host`` composes a +list of transports into a multi-host pool. + +The contract is deliberately thin — request/response plus discovery, replay, and +health — so new transports are cheap to add. Streaming / evidence pub-sub are +*optional* extensions advertised via ``supports(feature)`` and an optional +``subscribe_evidence`` hook, so adding them later (e.g. for a Zenoh mesh) never +breaks the core contract. + +This module depends only on the stdlib + chp-core internals: no transport here +pulls a third-party dependency. Heavier transports (Zenoh, gRPC) live in +downstream packages and implement this same ``Transport`` protocol. +""" + +from __future__ import annotations + +import asyncio +from typing import Protocol, runtime_checkable + +from .host import LocalCapabilityHost +from .http import RemoteCapabilityHost +from .types import InvocationEnvelope, InvocationResult, JSON, ReplayQuery + + +@runtime_checkable +class Transport(Protocol): + """Uniform async surface for talking to a single CHP host. + + Implementations: :class:`LocalTransport` (in-process), + :class:`HttpTransport` (stdlib HTTP). Downstream packages may add others + (Zenoh, gRPC) by satisfying this same protocol. + """ + + name: str + + async def ainvoke_envelope(self, envelope: InvocationEnvelope) -> InvocationResult: + """Invoke a capability from a pre-built envelope and return its result.""" + ... + + async def discover(self) -> JSON: + """Return the host descriptor (id, capabilities, evidence policy, ...).""" + ... + + async def replay_result(self, query: "str | ReplayQuery | JSON") -> JSON: + """Return a replay result (as a dict) for a correlation id or query.""" + ... + + async def health(self) -> JSON: + """Return a health snapshot: status, host_id, protocol, capability_count.""" + ... + + def supports(self, feature: str) -> bool: + """Report whether an optional feature (e.g. ``"streaming"``) is available.""" + ... + + +def _health_from_descriptor(descriptor: JSON) -> JSON: + """Build a /health-shaped snapshot from a host descriptor.""" + return { + "status": "ok", + "host_id": descriptor.get("id", "unknown"), + "protocol": "chp", + "version": "0.1", + "capability_count": len(descriptor.get("capabilities", [])), + } + + +class LocalTransport: + """Transport over an in-process :class:`LocalCapabilityHost`. + + Lets a local host participate in a multi-host router alongside remote hosts + on exactly the same seam. Invocation is awaited directly; the synchronous + discover/replay calls are fast in-memory operations. + """ + + def __init__(self, host: LocalCapabilityHost, *, name: str | None = None) -> None: + self._host = host + self.name = name or host.host_id + + async def ainvoke_envelope(self, envelope: InvocationEnvelope) -> InvocationResult: + return await self._host.ainvoke_envelope(envelope) + + async def discover(self) -> JSON: + return self._host.discover() + + async def replay_result(self, query: "str | ReplayQuery | JSON") -> JSON: + return self._host.replay_result(query).to_dict() + + async def health(self) -> JSON: + return _health_from_descriptor(self._host.discover()) + + def supports(self, feature: str) -> bool: + return False + + +class HttpTransport: + """Transport over CHP's stdlib HTTP surface. + + Wraps :class:`RemoteCapabilityHost` (blocking ``urllib``) and runs each call + in a worker thread so it never blocks the router's event loop. A + ``ConnectionError`` from the underlying client propagates unchanged so the + router can fail over to another host. + """ + + def __init__( + self, + base_url: str, + *, + name: str | None = None, + timeout: int = 30, + ) -> None: + self._remote = RemoteCapabilityHost(base_url, timeout=timeout) + self.name = name or base_url.rstrip("/") + + async def ainvoke_envelope(self, envelope: InvocationEnvelope) -> InvocationResult: + return await asyncio.to_thread(self._remote.invoke_envelope, envelope) + + async def discover(self) -> JSON: + return await asyncio.to_thread(self._remote.discover) + + async def replay_result(self, query: "str | ReplayQuery | JSON") -> JSON: + return await asyncio.to_thread(self._remote.replay_result, query) + + async def health(self) -> JSON: + return await asyncio.to_thread(self._remote.health) + + def supports(self, feature: str) -> bool: + return False