From 957d6b7906de9c68cb5dd4e296ff293ecb422a03 Mon Sep 17 00:00:00 2001 From: Eric Law <39393654+acn-ericlaw@users.noreply.github.com> Date: Sat, 22 Aug 2026 19:42:21 -0700 Subject: [PATCH 1/8] feat: primitive in-process event bus - the single dispatch pipeline Ratified design (Eric, 2026-08-23): every invocation reaches a function the same way - a per-route FIFO mailbox consumed by 'instances' worker tasks (the parameter becomes faithful; the semaphore is gone). The HTTP host and the local side of PostOffice are thin ingress adapters over the bus. - bus.py (internal): deliver (RPC, ttl-bounded -> 408; dead-work skip for queued calls whose caller already timed out) and publish (drop-n-forget, 202-shape ack). Worker owns the invocation pipeline: per-delivery trace context, AppException -> portable error envelope, exec_time, sync handlers via the executor, error logging for un-replied publishes. - PostOffice without an endpoint = local ingress with the engines' semantics: private routes callable in-app (the wire keeps its 403), headers verbatim, reply envelope identical to the remote path; unregistered route -> 404 envelope. With an endpoint: unchanged wire client. - No spill tier, no queue cap (ratified): back-pressure belongs to the tier that owns recovery - the engines' flows and graphs; a leaf host fails fast by deadline. In-memory only; send() has no ttl valve (engine parity). - Demo gains the local-composition pair (public hello.chain -> private demo.suffix.helper); README documents local calls, faithful private and the workflow boundary; scope statement amended. Pins: tests/test_bus.py - local RPC public+private, unregistered 404, FIFO order, instances=2 concurrency peak, local 408 + dead-work skip proven, trace chained caller -> entry -> private helper, local send ack. The whole existing host suite now exercises the bus path implicitly. Gates: ruff clean, basedpyright 0 errors, tests 40/40. Co-Authored-By: Claude Code --- CHANGELOG.md | 10 ++ README.md | 30 ++++- examples/demo_app.py | 16 +++ src/mercury_composable/bus.py | 175 +++++++++++++++++++++++++++++ src/mercury_composable/client.py | 56 +++++++-- src/mercury_composable/registry.py | 4 + src/mercury_composable/server.py | 85 ++------------ tests/test_bus.py | 156 +++++++++++++++++++++++++ 8 files changed, 444 insertions(+), 88 deletions(-) create mode 100644 src/mercury_composable/bus.py create mode 100644 tests/test_bus.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d881fdb..c96230b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,16 @@ ## 0.1.0 (unreleased) +- Primitive in-process event bus - the single dispatch pipeline: one FIFO mailbox per + route consumed by `instances` worker tasks (the parameter is faithful); RPC deliveries + are ttl-bounded with a dead-work skip; drop-n-forget returns the 202-shape ack. The + HTTP host and the local side of PostOffice are thin ingress adapters over it. No spill + tier and no queue cap by design - back-pressure belongs to the engines' flows/graphs; + a leaf host fails fast by deadline. +- PostOffice without an endpoint delivers locally (engine semantics): private routes are + callable in-app while the wire keeps its 403; headers pass verbatim; the reply envelope + shape is identical to the remote path. + Repository repurposed for the Mercury Composable **polyglot initiative** (August 2026). - Lightweight Event-over-HTTP function host (`POST /api/event`) mirroring the engines' diff --git a/README.md b/README.md index 1a89343..0071ebf 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,8 @@ This package is a deliberately **lightweight wrapper of the Event-over-HTTP prot registered functions, - a **thin client** (`PostOffice`) to call functions on peer applications the same way, - the **standard event envelope wire format** codec (language-neutral MsgPack), and +- a **primitive in-process event bus** — the single dispatch pipeline: one FIFO mailbox + per route consumed by `instances` worker tasks, and - the **minimalist utilities** shared with the engines for consistency: configuration management, logging in the engines' presentation format, and distributed-trace context. @@ -80,6 +82,24 @@ synchronous handlers run in a thread-pool executor so the event loop never block - Functions must be stateless; anything you must keep belongs to the caller's flow model or state machine. +### Local function calls + +`PostOffice` **without an endpoint** delivers through this application's own event bus — +the engines' semantics for an in-app `po` call: + +- `private=True` means exactly what it means in the engines: callable **in-app only**. + Local calls reach private and public routes alike; the HTTP host keeps answering 403 + for private targets from the wire. +- `instances` is faithful: each route has one FIFO mailbox consumed by that many worker + tasks. RPC waits are bounded by `timeout_ms` (the standard 408 envelope on breach), and + a queued call whose caller already timed out is skipped, never wastefully executed. +- There is **no spill tier and no queue cap** by design: back-pressure belongs to the tier + that owns recovery — the engines' flows and graphs. A leaf host fails fast by deadline + instead of hoarding work. + +Local eventing is for simple leaf-side composition. Workflow processing belongs in Event +Script and Knowledge Graph on the engines — that boundary is the architecture. + ## Configuration, logging, telemetry The same conventions as the engines, so a polyglot installation stays uniform: @@ -122,10 +142,12 @@ millisecond precision; binary payloads use MsgPack `bin`. ## Scope -This package intentionally contains **no event bus, no flows, no graphs and no -orchestration** — those live in the engines. It provides functions plus the minimalist -foundation utilities, keeping Python fast to prototype with while the composable core -guarantees the architecture. +This package intentionally contains **no orchestration: no flows, no graphs, no +persistence, no pub/sub broadcast** — those live in the engines. What it does carry is +deliberately minimal: functions, a primitive in-process event bus (route mailboxes + +workers, RPC and drop-n-forget — nothing more), and the minimalist foundation utilities, +keeping Python fast to prototype with while the composable core guarantees the +architecture. ## Development diff --git a/examples/demo_app.py b/examples/demo_app.py index d0f3c45..42afb7a 100644 --- a/examples/demo_app.py +++ b/examples/demo_app.py @@ -43,5 +43,21 @@ async def declarative_echo(headers: dict[str, str], body: Body): return {"body": body, "headers": headers, "language": "python"} +@preload(route="demo.suffix.helper", instances=10, private=True) +async def suffix_helper(_headers: dict[str, str], body: Body): + """Private helper - callable in-app only (the HTTP host answers 403 for it).""" + assert isinstance(body, dict) + return {"text": f"{body.get('text', '')}!", "language": "python"} + + +@preload(route="hello.chain", instances=10) +async def chain(_headers: dict[str, str], body: Body): + """Local composition: a public function calls a private sibling through the bus.""" + from mercury_composable import PostOffice + + reply = await PostOffice().request("demo.suffix.helper", body=body, timeout_ms=5000) + return reply.body + + if __name__ == "__main__": platform.run() diff --git a/src/mercury_composable/bus.py b/src/mercury_composable/bus.py new file mode 100644 index 0000000..a27ddd8 --- /dev/null +++ b/src/mercury_composable/bus.py @@ -0,0 +1,175 @@ +""" +The primitive in-process event bus - the single dispatch pipeline. + +Every invocation reaches a function the same way: through a per-route FIFO +mailbox consumed by ``instances`` worker tasks (the engines' semantics - the +parameter is faithful). The HTTP host and the local side of PostOffice are +thin ingress adapters over this bus; neither has its own invocation path. + +Deliberately primitive, riding asyncio's native machinery: + +- Two operations only: :meth:`EventBus.deliver` (RPC - enqueue with a reply + future, bounded by the caller's ttl) and :meth:`EventBus.publish` + (drop-n-forget - enqueue and return the 202-shape acknowledgement). +- **No spill tier and no queue cap**: back-pressure belongs to the tier that + owns recovery - the engines' flows and graphs. A leaf host fails fast by + deadline (the 408 envelope) instead of hoarding work, and a queued RPC + delivery whose caller already timed out is skipped (dead-work check). +- In-memory only: in-flight events die with the process, exactly like the + engines' own in-memory bus; at-least-once comes from flow-level retries. +- No orchestration, no flows, no persistence, no pub/sub broadcast. + +The bus is internal: application code uses ``@preload`` and ``PostOffice``, +never this module - the same way engine developers never touch the engine bus. +""" + +from __future__ import annotations + +import asyncio +import contextvars +import time +import traceback +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any + +from .envelope import EventEnvelope, iso_utc +from .exceptions import AppException +from .log import get_logger +from .trace import TraceInfo, _reset_trace, _set_trace + +if TYPE_CHECKING: + from .registry import ServiceDef + +log = get_logger("mercury.bus") + + +class DeliveryTimeout(Exception): + """An RPC delivery missed its deadline; adapters shape the 408 for their protocol.""" + + def __init__(self, ttl_ms: int): + super().__init__(f"Timeout for {ttl_ms} ms") + self.ttl_ms = ttl_ms + + +def async_ack() -> EventEnvelope: + """The 202 drop-n-forget acknowledgement (EventApiService shape).""" + return EventEnvelope().set_status(202).set_body( + {"type": "async", "delivered": True, "time": iso_utc()}) + + +@dataclass +class _Delivery: + service: ServiceDef + headers: dict[str, str] + body: Any + trace_id: str | None + trace_path: str | None + cid: str | None + reply: asyncio.Future[EventEnvelope] | None # None = drop-n-forget + + +class EventBus: + """Per-registry bus: one FIFO mailbox and N workers per registered route.""" + + def __init__(self) -> None: + self._mailboxes: dict[str, asyncio.Queue[_Delivery]] = {} + self._workers: dict[str, list[asyncio.Task[None]]] = {} + + def _mailbox(self, service: ServiceDef) -> asyncio.Queue[_Delivery]: + mailbox = self._mailboxes.get(service.route) + if mailbox is None: + # lazy: mailbox and workers bind to the running event loop on first use + mailbox = asyncio.Queue() + self._mailboxes[service.route] = mailbox + self._workers[service.route] = [ + asyncio.get_running_loop().create_task( + self._run_worker(mailbox), name=f"mercury-bus-{service.route}-{n}") + for n in range(service.instances) + ] + return mailbox + + async def deliver(self, service: ServiceDef, headers: dict[str, str], body: Any, + ttl_ms: int, *, trace_id: str | None = None, + trace_path: str | None = None, cid: str | None = None) -> EventEnvelope: + """RPC: enqueue and await the reply envelope within the ttl.""" + reply: asyncio.Future[EventEnvelope] = asyncio.get_running_loop().create_future() + self._mailbox(service).put_nowait(_Delivery( + service=service, headers=headers, body=body, + trace_id=trace_id, trace_path=trace_path, cid=cid, reply=reply)) + try: + return await asyncio.wait_for(reply, timeout=max(100, ttl_ms) / 1000) + except asyncio.TimeoutError: + raise DeliveryTimeout(ttl_ms) from None + + def publish(self, service: ServiceDef, headers: dict[str, str], body: Any, *, + trace_id: str | None = None, trace_path: str | None = None, + cid: str | None = None) -> EventEnvelope: + """Drop-n-forget: enqueue and return the 202-shape acknowledgement.""" + self._mailbox(service).put_nowait(_Delivery( + service=service, headers=headers, body=body, + trace_id=trace_id, trace_path=trace_path, cid=cid, reply=None)) + return async_ack() + + async def close(self) -> None: + """Cancel all workers (tests and orderly shutdown).""" + for workers in self._workers.values(): + for worker in workers: + worker.cancel() + for workers in self._workers.values(): + for worker in workers: + try: + await worker + except asyncio.CancelledError: + pass + self._workers.clear() + self._mailboxes.clear() + + async def _run_worker(self, mailbox: asyncio.Queue[_Delivery]) -> None: + while True: + delivery = await mailbox.get() + # dead-work check: the caller of a queued RPC already gave up (408 sent) - + # skip instead of computing a reply nobody reads + if delivery.reply is not None and delivery.reply.done(): + continue + reply = await self._execute(delivery) + if delivery.reply is not None: + if not delivery.reply.done(): + delivery.reply.set_result(reply) + elif reply.has_error(): + log.warning("Async event %s ended with status %d - %s", + delivery.service.route, reply.get_status(), reply.body) + + async def _execute(self, delivery: _Delivery) -> EventEnvelope: + """Run the handler under its trace context and shape the outcome as a reply.""" + service = delivery.service + info = TraceInfo(trace_id=delivery.trace_id, trace_path=delivery.trace_path, + cid=delivery.cid) + token = _set_trace(info) + start = time.perf_counter() + # noinspection PyBroadException + try: + if service.is_async: + result = await service.handler(delivery.headers, delivery.body) + else: + # copy_context() carries the trace contextvar into the executor thread + context = contextvars.copy_context() + result = await asyncio.get_running_loop().run_in_executor( + None, lambda: context.run(service.handler, delivery.headers, + delivery.body)) + reply = result if isinstance(result, EventEnvelope) else EventEnvelope(body=result) + except AppException as e: + reply = EventEnvelope().set_status(e.status).set_body(e.message) + except asyncio.CancelledError: + raise + except Exception as e: # noqa: BLE001 + # any handler failure becomes the portable error contract + # (status 500 + message + stack), mirroring the engines + reply = EventEnvelope().set_status(500).set_body(str(e)) + reply.stack = traceback.format_exc(limit=20) + finally: + _reset_trace(token) + reply.sender = reply.sender or service.route + reply.exec_time = round((time.perf_counter() - start) * 1000, 3) + if info.annotations: + reply.annotations.update(info.annotations) + return reply diff --git a/src/mercury_composable/client.py b/src/mercury_composable/client.py index 93af7d3..986a70b 100644 --- a/src/mercury_composable/client.py +++ b/src/mercury_composable/client.py @@ -1,14 +1,24 @@ """ -Thin Event-over-HTTP client — the PostOffice analog. +The PostOffice — local and remote function calls with one envelope contract. -Sends an event envelope to a peer's ``/api/event`` endpoint (a Java or Rust -engine application, or another polyglot function host) with the same HTTP -contract as the engines' relay: ``content-type: application/octet-stream``, +**Remote** (an ``endpoint`` is given, on the constructor or per call): sends +the event envelope to a peer's ``/api/event`` — a Java or Rust engine +application, or another polyglot function host — with the same HTTP contract +as the engines' relay: ``content-type: application/octet-stream``, ``accept: */*``, ``x-no-stream: true``, ``x-ttl`` (ms), ``x-async: true`` for -drop-n-forget, optional security headers, and trace headers -(``X-Trace-Id`` plus a W3C ``traceparent`` when the trace id is W3C-shaped). - -The decoded reply envelope is authoritative: an error from the target rides +drop-n-forget, optional security headers, and trace headers (``X-Trace-Id`` +plus a W3C ``traceparent`` when the trace id is W3C-shaped). The remote +target must be public (its host answers 403 for private routes). + +**Local** (no endpoint): the call goes through this application's primitive +event bus to a locally registered function — private OR public, the engines' +semantics (``private`` means in-app only). Headers are delivered verbatim +(ingress hygiene applies to the wire, not to in-app calls), the ttl bounds +the wait with the standard 408 envelope, and the reply shape is identical to +the remote path. Local eventing is for simple leaf-side composition; workflow +processing belongs in Event Script and Knowledge Graph on the engines. + +The decoded reply envelope is authoritative in both modes: an error rides back as a normal envelope with status >= 400 — inspect ``reply.get_status()``. """ @@ -19,8 +29,10 @@ import aiohttp +from .bus import DeliveryTimeout from .envelope import EventEnvelope from .exceptions import AppException +from .registry import FunctionRegistry, default_registry from .trace import get_trace _W3C_TRACE_ID = re.compile(r"^[0-9a-f]{32}$") @@ -46,9 +58,11 @@ class PostOffice: """Event-over-HTTP client for calling functions on peer applications.""" def __init__(self, endpoint: str | None = None, - security_headers: dict[str, str] | None = None): + security_headers: dict[str, str] | None = None, + registry: FunctionRegistry | None = None): self.endpoint = endpoint self.security_headers = dict(security_headers or {}) + self._registry = registry or default_registry self._session: aiohttp.ClientSession | None = None def _get_session(self) -> aiohttp.ClientSession: @@ -91,13 +105,33 @@ def _http_headers(self, timeout_ms: int, is_async: bool, headers["traceparent"] = f"00-{event.trace_id}-{event.span_id}-01" return headers + async def _call_local(self, route: str, body: Any, headers: dict[str, str] | None, + timeout_ms: int, is_async: bool, from_route: str | None, + cid: str | None) -> EventEnvelope: + """In-app delivery through the primitive event bus (private OR public).""" + service = self._registry.get(route) + if service is None: + return EventEnvelope().set_status(404).set_body(f"Route {route} not found") + event = _build_event(route, body, headers, from_route, cid) + bus = self._registry.bus + if is_async: + return bus.publish(service, event.headers, event.body, trace_id=event.trace_id, + trace_path=event.trace_path, cid=event.cid) + try: + return await bus.deliver(service, event.headers, event.body, timeout_ms, + trace_id=event.trace_id, trace_path=event.trace_path, + cid=event.cid) + except DeliveryTimeout: + return EventEnvelope().set_status(408).set_body(f"Timeout for {timeout_ms} ms") + async def _call(self, route: str, body: Any, headers: dict[str, str] | None, timeout_ms: int, endpoint: str | None, is_async: bool, from_route: str | None, cid: str | None) -> EventEnvelope: url = endpoint or self.endpoint if not url: - raise ValueError("Missing event endpoint - " - "e.g. PostOffice(endpoint='http://peer:8085/api/event')") + # no endpoint = local: the engines' semantics for an in-app po call + return await self._call_local(route, body, headers, timeout_ms, + is_async, from_route, cid) event = _build_event(route, body, headers, from_route, cid) session = self._get_session() # +100 ms cushion so the HTTP client does not time out before the target diff --git a/src/mercury_composable/registry.py b/src/mercury_composable/registry.py index 00e1576..ce176b9 100644 --- a/src/mercury_composable/registry.py +++ b/src/mercury_composable/registry.py @@ -18,6 +18,7 @@ from dataclasses import dataclass from typing import Any +from .bus import EventBus from .envelope import Body # the function contract: (headers, body) in, reply body (or EventEnvelope) out - @@ -48,6 +49,9 @@ class ServiceDef: class FunctionRegistry: def __init__(self) -> None: self._services: dict[str, ServiceDef] = {} + # the registry's own dispatch pipeline (see bus.py) - shared by the + # HTTP host and the local side of PostOffice + self.bus = EventBus() def register(self, route: str, handler: Handler, *, instances: int = 10, private: bool = False) -> ServiceDef: diff --git a/src/mercury_composable/server.py b/src/mercury_composable/server.py index cecd290..4677cd5 100644 --- a/src/mercury_composable/server.py +++ b/src/mercury_composable/server.py @@ -20,19 +20,13 @@ from __future__ import annotations -import asyncio -import contextvars -import time -import traceback - from aiohttp import web +from .bus import DeliveryTimeout from .config import app_config -from .envelope import EventEnvelope, iso_utc -from .exceptions import AppException +from .envelope import EventEnvelope from .log import get_logger -from .registry import FunctionRegistry, ServiceDef, default_registry -from .trace import TraceInfo, _reset_trace, _set_trace +from .registry import FunctionRegistry, default_registry OCTET_STREAM = "application/octet-stream" X_TTL = "x-ttl" @@ -59,49 +53,10 @@ def _handler_headers(event: EventEnvelope) -> dict[str, str]: class EventApiServer: + """Thin ingress: protocol guards + header hygiene, then the registry's bus.""" + def __init__(self, registry: FunctionRegistry | None = None): self.registry = registry or default_registry - self._semaphores: dict[str, asyncio.Semaphore] = {} - - def _semaphore(self, service: ServiceDef) -> asyncio.Semaphore: - semaphore = self._semaphores.get(service.route) - if semaphore is None: - semaphore = asyncio.Semaphore(service.instances) - self._semaphores[service.route] = semaphore - return semaphore - - async def _invoke(self, service: ServiceDef, event: EventEnvelope, - headers: dict[str, str]) -> EventEnvelope: - """Run the handler under its trace context and shape the outcome as a reply.""" - info = TraceInfo(trace_id=event.trace_id, trace_path=event.trace_path, cid=event.cid) - token = _set_trace(info) - start = time.perf_counter() - try: - async with self._semaphore(service): - if service.is_async: - result = await service.handler(headers, event.body) - else: - # copy_context() carries the trace contextvar into the executor thread - context = contextvars.copy_context() - result = await asyncio.get_running_loop().run_in_executor( - None, lambda: context.run(service.handler, headers, event.body)) - reply = result if isinstance(result, EventEnvelope) else EventEnvelope(body=result) - except AppException as e: - reply = EventEnvelope().set_status(e.status).set_body(e.message) - except asyncio.CancelledError: - raise - except Exception as e: # noqa: BLE001 - the host converts ANY handler failure - # into the portable error contract (envelope status 500 + message + stack), - # mirroring the engines; letting it propagate would drop the reply - reply = EventEnvelope().set_status(500).set_body(str(e)) - reply.stack = traceback.format_exc(limit=20) - finally: - _reset_trace(token) - reply.sender = reply.sender or service.route - reply.exec_time = round((time.perf_counter() - start) * 1000, 3) - if info.annotations: - reply.annotations.update(info.annotations) - return reply async def handle_event(self, request: web.Request) -> web.Response: raw = await request.read() @@ -124,38 +79,22 @@ async def handle_event(self, request: web.Request) -> web.Response: if service.private: return _transport_error(403, f"{event.to} is private") headers = _handler_headers(event) + bus = self.registry.bus if is_async: - task = asyncio.get_running_loop().create_task( - self._invoke(service, event, headers)) - task.add_done_callback(self._log_async_outcome(event.to)) - ack = EventEnvelope().set_status(202).set_body( - {"type": "async", "delivered": True, "time": iso_utc()}) + ack = bus.publish(service, headers, event.body, trace_id=event.trace_id, + trace_path=event.trace_path, cid=event.cid) return web.Response(status=202, body=ack.to_bytes(), content_type=OCTET_STREAM) try: - reply = await asyncio.wait_for( - self._invoke(service, event, headers), timeout=ttl / 1000) - except asyncio.TimeoutError: + reply = await bus.deliver(service, headers, event.body, ttl, + trace_id=event.trace_id, trace_path=event.trace_path, + cid=event.cid) + except DeliveryTimeout: log.warning("Event %s timeout for %d ms (trace_id=%s)", event.to, ttl, event.trace_id) return _transport_error(408, f"Timeout for {ttl} ms") log.info("Handled %s status=%d exec_time=%sms trace_id=%s", event.to, reply.get_status(), reply.exec_time, event.trace_id) return web.Response(status=200, body=reply.to_bytes(), content_type=OCTET_STREAM) - @staticmethod - def _log_async_outcome(route: str): - def callback(task: asyncio.Task[EventEnvelope]) -> None: - # noinspection PyBroadException - try: - reply = task.result() - if reply.has_error(): - log.warning("Async event %s ended with status %d - %s", - route, reply.get_status(), reply.body) - except Exception: - # deliberate log-only sink: a drop-n-forget event has no requester - # to answer, so any failure is logged with its traceback, never raised - log.exception("Async event %s failed", route) - return callback - # aiohttp handlers must be coroutines - async is the framework contract # even though this one has nothing to await @staticmethod diff --git a/tests/test_bus.py b/tests/test_bus.py new file mode 100644 index 0000000..c192c4b --- /dev/null +++ b/tests/test_bus.py @@ -0,0 +1,156 @@ +"""Primitive event bus pins: local RPC (public+private), FIFO, workers, deadlines.""" + +import asyncio +from collections.abc import AsyncIterator + +import pytest_asyncio + +from mercury_composable import Body, EventEnvelope, FunctionRegistry, PostOffice, trace_context +from mercury_composable.trace import get_trace + + +@pytest_asyncio.fixture +async def registry() -> AsyncIterator[FunctionRegistry]: + fresh = FunctionRegistry() + yield fresh + await fresh.bus.close() + + +async def test_local_rpc_to_public_route(registry: FunctionRegistry): + async def echo(headers: dict[str, str], body: Body): + return {"headers": headers, "body": body} + + registry.register("bus.echo", echo) + po = PostOffice(registry=registry) + reply = await po.request("bus.echo", body={"a": 1}, headers={"h1": "v1"}, timeout_ms=5000) + assert reply.get_status() == 200 + assert reply.body["body"] == {"a": 1} + # local delivery passes headers verbatim (hygiene is a wire-ingress concern) + assert reply.body["headers"] == {"h1": "v1"} + assert reply.sender == "bus.echo" + assert reply.exec_time is not None + + +async def test_local_rpc_reaches_private_route(registry: FunctionRegistry): + # the engines' semantics: private = callable in-app only; the HTTP host + # still answers 403 for private targets (pinned in test_server) + async def secret(_headers: dict[str, str], _body: Body): + return {"secret": "ok"} + + registry.register("bus.secret", secret, private=True) + po = PostOffice(registry=registry) + reply = await po.request("bus.secret", body={}, timeout_ms=5000) + assert reply.get_status() == 200 + assert reply.body == {"secret": "ok"} + + +async def test_unregistered_local_route_404(registry: FunctionRegistry): + po = PostOffice(registry=registry) + reply = await po.request("bus.no.where", body={}, timeout_ms=5000) + assert reply.get_status() == 404 + assert reply.body == "Route bus.no.where not found" + + +async def test_fifo_ordering_with_one_worker(registry: FunctionRegistry): + processed: list[int] = [] + done = asyncio.Event() + + async def collector(_headers: dict[str, str], body: Body): + assert isinstance(body, dict) + processed.append(int(str(body["n"]))) + if len(processed) == 3: + done.set() + + registry.register("bus.fifo", collector, instances=1) + po = PostOffice(registry=registry) + for n in (1, 2, 3): + ack = await po.send("bus.fifo", body={"n": n}) + assert ack.get_status() == 202 + assert ack.body["delivered"] is True + await asyncio.wait_for(done.wait(), timeout=5) + assert processed == [1, 2, 3] + + +async def test_instances_bounds_concurrency(registry: FunctionRegistry): + active = 0 + peak = 0 + + async def slow(_headers: dict[str, str], _body: Body): + nonlocal active, peak + active += 1 + peak = max(peak, active) + await asyncio.sleep(0.15) + active -= 1 + return {"ok": True} + + registry.register("bus.slow", slow, instances=2) + po = PostOffice(registry=registry) + replies = await asyncio.gather(*( + po.request("bus.slow", body={}, timeout_ms=5000) for _ in range(4))) + assert all(r.get_status() == 200 for r in replies) + assert peak == 2 # instances = the number of concurrent workers, faithfully + + +async def test_local_timeout_408_and_dead_work_skip(registry: FunctionRegistry): + executed: list[str] = [] + release = asyncio.Event() + + async def gate(_headers: dict[str, str], body: Body): + assert isinstance(body, dict) + executed.append(str(body["id"])) + await release.wait() + return {"ok": True} + + registry.register("bus.gate", gate, instances=1) + po = PostOffice(registry=registry) + first = asyncio.create_task(po.request("bus.gate", body={"id": "first"}, timeout_ms=5000)) + await asyncio.sleep(0.05) # the single worker is now blocked inside 'first' + # the second RPC waits in the mailbox and times out before a worker frees up + reply = await po.request("bus.gate", body={"id": "second"}, timeout_ms=200) + assert reply.get_status() == 408 + assert reply.body == "Timeout for 200 ms" + release.set() + assert (await first).get_status() == 200 + await asyncio.sleep(0.05) # give the worker a chance to reach the dead delivery + # dead-work check: the timed-out delivery was skipped, never executed + assert executed == ["first"] + + +async def test_trace_chain_through_local_private_sibling(registry: FunctionRegistry): + async def helper(_headers: dict[str, str], _body: Body): + info = get_trace() + assert info is not None + return {"helper_trace": info.trace_id, "helper_cid": info.cid} + + async def entry(_headers: dict[str, str], _body: Body): + info = get_trace() + assert info is not None + po = PostOffice(registry=registry) + inner = await po.request("bus.helper", body={}, timeout_ms=5000) + assert isinstance(inner.body, dict) + return {"entry_trace": info.trace_id, **inner.body} + + registry.register("bus.helper", helper, private=True) + registry.register("bus.entry", entry) + po = PostOffice(registry=registry) + with trace_context("trace-bus-1", "TEST /bus", cid="cid-bus-1"): + reply = await po.request("bus.entry", body={}, timeout_ms=5000) + assert reply.get_status() == 200 + # one trace id flows: caller context -> entry handler -> private helper + assert reply.body == {"entry_trace": "trace-bus-1", "helper_trace": "trace-bus-1", + "helper_cid": "cid-bus-1"} + + +async def test_local_send_returns_ack_envelope(registry: FunctionRegistry): + seen = asyncio.Event() + + async def sink(_headers: dict[str, str], _body: Body): + seen.set() + + registry.register("bus.sink", sink) + po = PostOffice(registry=registry) + ack = await po.send("bus.sink", body={"fire": "forget"}) + assert isinstance(ack, EventEnvelope) + assert ack.get_status() == 202 + assert ack.body["type"] == "async" + await asyncio.wait_for(seen.wait(), timeout=5) From 300f74fdef57ecf1ddfab498d627a957bdcef0a3 Mon Sep 17 00:00:00 2001 From: Eric Law <39393654+acn-ericlaw@users.noreply.github.com> Date: Sat, 22 Aug 2026 19:52:31 -0700 Subject: [PATCH 2/8] chore: Sonar round on the bus - comment wording, close() idiom, static execute Eric's Sonar scan of the new code: the trailing 'None = drop-n-forget' comment parsed as commented-out code (the S125 wording lesson); close() now uses gather(return_exceptions=True) - the workers' own CancelledError outcomes are collected while a cancellation of close() itself still propagates; _execute is a staticmethod (uses nothing from self). Gates: ruff clean, basedpyright 0 errors, tests 40/40. Co-Authored-By: Claude Code --- src/mercury_composable/bus.py | 20 +++++++++----------- 1 file changed, 9 insertions(+), 11 deletions(-) diff --git a/src/mercury_composable/bus.py b/src/mercury_composable/bus.py index a27ddd8..ef0d436 100644 --- a/src/mercury_composable/bus.py +++ b/src/mercury_composable/bus.py @@ -65,7 +65,7 @@ class _Delivery: trace_id: str | None trace_path: str | None cid: str | None - reply: asyncio.Future[EventEnvelope] | None # None = drop-n-forget + reply: asyncio.Future[EventEnvelope] | None # drop-n-forget deliveries carry no reply future class EventBus: @@ -112,15 +112,12 @@ def publish(self, service: ServiceDef, headers: dict[str, str], body: Any, *, async def close(self) -> None: """Cancel all workers (tests and orderly shutdown).""" - for workers in self._workers.values(): - for worker in workers: - worker.cancel() - for workers in self._workers.values(): - for worker in workers: - try: - await worker - except asyncio.CancelledError: - pass + cancelled = [worker for workers in self._workers.values() for worker in workers] + for worker in cancelled: + worker.cancel() + # return_exceptions collects the workers' own CancelledError outcomes; + # a cancellation of close() itself still propagates from the gather + await asyncio.gather(*cancelled, return_exceptions=True) self._workers.clear() self._mailboxes.clear() @@ -139,7 +136,8 @@ async def _run_worker(self, mailbox: asyncio.Queue[_Delivery]) -> None: log.warning("Async event %s ended with status %d - %s", delivery.service.route, reply.get_status(), reply.body) - async def _execute(self, delivery: _Delivery) -> EventEnvelope: + @staticmethod + async def _execute(delivery: _Delivery) -> EventEnvelope: """Run the handler under its trace context and shape the outcome as a reply.""" service = delivery.service info = TraceInfo(trace_id=delivery.trace_id, trace_path=delivery.trace_path, From 2ff28a4f8382af0dce62cbfc96f6f5d6326300c4 Mon Sep 17 00:00:00 2001 From: Eric Law <39393654+acn-ericlaw@users.noreply.github.com> Date: Sat, 22 Aug 2026 19:53:40 -0700 Subject: [PATCH 3/8] chore: unshadow the inner PostOffice in the trace-chain test Eric's IDE review: the entry handler's local 'po' shadowed the test's outer 'po' - renamed inner_po; node twin aligned in its own commit. Co-Authored-By: Claude Code --- tests/test_bus.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_bus.py b/tests/test_bus.py index c192c4b..79a2600 100644 --- a/tests/test_bus.py +++ b/tests/test_bus.py @@ -125,8 +125,8 @@ async def helper(_headers: dict[str, str], _body: Body): async def entry(_headers: dict[str, str], _body: Body): info = get_trace() assert info is not None - po = PostOffice(registry=registry) - inner = await po.request("bus.helper", body={}, timeout_ms=5000) + inner_po = PostOffice(registry=registry) + inner = await inner_po.request("bus.helper", body={}, timeout_ms=5000) assert isinstance(inner.body, dict) return {"entry_trace": info.trace_id, **inner.body} From 338cbfa36fb6ec29f5fb82a25027d0b830b09693 Mon Sep 17 00:00:00 2001 From: Eric Law <39393654+acn-ericlaw@users.noreply.github.com> Date: Sat, 22 Aug 2026 19:55:41 -0700 Subject: [PATCH 4/8] session 2026-08-23 [Claude Code] --- memory/continuity.md | 8 ++++-- memory/sessions/2026-08-23-024221.md | 43 ++++++++++++++++++++++++++++ 2 files changed, 48 insertions(+), 3 deletions(-) create mode 100644 memory/sessions/2026-08-23-024221.md diff --git a/memory/continuity.md b/memory/continuity.md index 2600130..fb3fcda 100644 --- a/memory/continuity.md +++ b/memory/continuity.md @@ -17,7 +17,7 @@ polyglot initiative: a lightweight Event-over-HTTP function host + thin client, repurposed August 2026 (legacy language pack in git history only) - **last_enabled:** 2026-08-22 -- **last_session:** 2026-08-23 | agent: Claude Code (2026-08-23-005709) +- **last_session:** 2026-08-23 | agent: Claude Code (2026-08-23-024221) - **last_review:** (none yet) - **last_invariant_check:** (none yet) - **repo:** ~/sandbox/mercury-python (origin: github.com/Accenture/mercury-python) @@ -91,8 +91,10 @@ ## Open Threads -- [ ] (feature — design RATIFIED by Eric 2026-08-23 in the quality-round conversation; - implementation next on `feature/primitive-event-bus`, LOCK-STEP with mercury-nodejs) +- [ ] (feature — design RATIFIED by Eric 2026-08-23; **IMPLEMENTED same day on + `feature/primitive-event-bus`, commit `957d6b7`, tests 40/40 incl. the 8 bus pins + + live wire proof (chain → private via bus; wire → private = 403); node twin `da8ce4c` + 39/39; PENDING Eric's PR gate**) **Primitive in-process event bus — the single dispatch pipeline.** Ratified shape: per-route FIFO mailbox (asyncio.Queue; node = queue + worker loops) with **instances = N worker tasks** (replaces the semaphore — the parameter becomes faithful); diff --git a/memory/sessions/2026-08-23-024221.md b/memory/sessions/2026-08-23-024221.md new file mode 100644 index 0000000..3118c7c --- /dev/null +++ b/memory/sessions/2026-08-23-024221.md @@ -0,0 +1,43 @@ +# Session (2026-08-23T02:42:21.000Z) + +**Agent:** Claude Code + +## Summary + +**The primitive in-process event bus implemented (the ratified design in +[[thread-primitive-event-bus]]), branch `feature/primitive-event-bus`, commit `957d6b7` — +lock-step with mercury-nodejs (`da8ce4c` there).** + +- `bus.py` (internal): per-route FIFO mailbox (asyncio.Queue) + `instances` worker tasks — + the parameter is faithful; the semaphore is gone. `deliver` (RPC, ttl-bounded → 408, + dead-work skip) + `publish` (drop-n-forget, 202-shape ack). The worker owns the whole + invocation pipeline (trace context, AppException → portable error, exec_time, executor + for sync handlers, error logging for un-replied publishes). +- The HTTP host became thin ingress (guards + hygiene → bus); PostOffice without an + endpoint = local ingress with engine semantics — **private is faithful** (in-app + callable; the wire keeps its 403). PostOffice gained a `registry` parameter (tests use + fresh registries; apps default). +- Demo: public `hello.chain` → private `demo.suffix.helper` through the bus; README + documents local calls, the no-spill/no-cap posture and the workflow boundary; scope + statement amended per the ratified fence. +- **Pins (tests/test_bus.py, 8)**: local RPC public+private, unregistered 404, FIFO order, + instances=2 concurrency peak==2, local 408 + dead-work skip proven (second delivery + never executed), trace chained caller→entry→private helper, local send ack. The whole + existing host suite now exercises the bus path implicitly. +- **Live wire proof**: demo app on 18086 — `hello.chain` over `/api/event` returned + `{'text': 'polyglot!', 'language': 'python'}` (wire → public → bus → private) and the + direct call to the private route answered 403. Same proof green against the node twin. + +Follow-up commits on the branch: `300f74f` (Eric's Sonar round on the new code — comment +wording that parsed as commented-out code, `close()` via gather(return_exceptions=True) +so close's own cancellation still propagates, static `_execute`) and `2ff28a4` (unshadowed +inner PostOffice in the trace-chain test). Node twin aligned (`31c8b0b` + `3d866f1`). + +Gates: ruff clean, basedpyright 0 errors, **tests 40/40**. Branch = 3 commits, pending +Eric's PR gate. + +## Memory References + +- referenced: thread-primitive-event-bus (implemented), conv-python-quality-gates + (gates held throughout) +- updated: thread-primitive-event-bus (implementation state) From 56c002ce563bb5d7e81089a516f6815507b9d7da Mon Sep 17 00:00:00 2001 From: Eric Law <39393654+acn-ericlaw@users.noreply.github.com> Date: Sat, 22 Aug 2026 20:15:57 -0700 Subject: [PATCH 5/8] feat: actuator endpoints - the engines' operational surface for Kubernetes GET /info, /info/routes, /env, /health, /livenessprobe on the Event API host, mirroring the Java ActuatorServices shapes as ported to Rust. Health-check functions are normal registered functions speaking the engines' type=info / type=health contract, listed in mandatory/optional.health.dependencies and called through the event bus; /livenessprobe follows the most recent /health outcome (400 when down, engine parity). Version moved to version.py (single Python-side source). Co-Authored-By: Claude Code --- CHANGELOG.md | 7 + README.md | 31 ++++ examples/demo_app.py | 12 ++ src/mercury_composable/__init__.py | 3 +- src/mercury_composable/actuator.py | 227 +++++++++++++++++++++++++++++ src/mercury_composable/server.py | 17 ++- src/mercury_composable/version.py | 3 + tests/test_actuator.py | 197 +++++++++++++++++++++++++ tests/test_server.py | 16 +- 9 files changed, 497 insertions(+), 16 deletions(-) create mode 100644 src/mercury_composable/actuator.py create mode 100644 src/mercury_composable/version.py create mode 100644 tests/test_actuator.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c96230b..a3b7728 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,13 @@ ## 0.1.0 (unreleased) +- Actuator endpoints `/info`, `/info/routes`, `/env`, `/health` and `/livenessprobe` - + the engines' operational surface, for Kubernetes probes and one-dashboard monitoring + of polyglot installations. Health-check functions are normal registered functions + speaking the engines' `type=info` / `type=health` interface contract, listed in + `mandatory.health.dependencies` / `optional.health.dependencies` and called through + the event bus. `/health` answers `UP` (200) / `DOWN` (400); `/livenessprobe` follows + the most recent health outcome. - Primitive in-process event bus - the single dispatch pipeline: one FIFO mailbox per route consumed by `instances` worker tasks (the parameter is faithful); RPC deliveries are ttl-bounded with a dead-work skip; drop-n-forget returns the 202-shape ack. The diff --git a/README.md b/README.md index 0071ebf..0dddbc3 100644 --- a/README.md +++ b/README.md @@ -127,6 +127,37 @@ Log lines follow the Java reference engine's pattern for one-aggregation consist 2026-08-22 10:15:30.123 INFO my_app:42 - Loaded PUBLIC hello.python, instances=10 ``` +## Actuator endpoints + +The host serves the engines' operational endpoints on the same port as `/api/event`, so +Kubernetes probes and dashboards treat a Python app exactly like a Java or Rust engine app: + +| Endpoint | Purpose | +|----------|---------| +| `GET /info` | app identity, runtime, origin id, start time, uptime | +| `GET /info/routes` | registered routes split by visibility, with instance counts | +| `GET /env` | selected environment variables and configuration parameters | +| `GET /health` | dependency health checks — `UP` (HTTP 200) or `DOWN` (HTTP 400) | +| `GET /livenessprobe` | `OK` while the last health outcome was good, else HTTP 400 | + +Configuration keys carry the engines' names: `info.app.version`, `info.app.description`, +`show.env.variables` and `show.application.properties` (opt-in lists — secrets are never +dumped wholesale), and `mandatory.health.dependencies` / `optional.health.dependencies` +(routes of health-check functions; optional ones never change the overall status). A +health-check function is a normal registered function — usually private — speaking the +engines' interface contract, called through the event bus: + +```python +@preload("demo.health", private=True) +async def health(headers: dict[str, str], _body: Body) -> Body: + if headers.get("type") == "info": + return {"service": "demo.service", "href": "http://127.0.0.1"} + return "demo.service is running fine" # a non-200 reply marks it down +``` + +Kubernetes wiring: point `livenessProbe` at `/livenessprobe` and `readinessProbe` at +`/health`. + ## Wire compatibility The codec implements the diff --git a/examples/demo_app.py b/examples/demo_app.py index 42afb7a..999d51a 100644 --- a/examples/demo_app.py +++ b/examples/demo_app.py @@ -59,5 +59,17 @@ async def chain(_headers: dict[str, str], body: Body): return reply.body +@preload(route="demo.health", instances=5, private=True) +async def health_check(headers: dict[str, str], _body: Body): + """Health check speaking the engines' interface contract (type=info / type=health). + + Activate it for the /health actuator endpoint: + mercury-serve examples/demo_app.py -Dmandatory.health.dependencies=demo.health + """ + if headers.get("type") == "info": + return {"service": "demo.service", "href": "http://127.0.0.1"} + return "demo.service is running fine" + + if __name__ == "__main__": platform.run() diff --git a/src/mercury_composable/__init__.py b/src/mercury_composable/__init__.py index 050ef12..d192fbb 100644 --- a/src/mercury_composable/__init__.py +++ b/src/mercury_composable/__init__.py @@ -17,8 +17,7 @@ from .registry import FunctionRegistry, Handler, default_registry, preload from .server import EventApiServer, Platform, platform from .trace import TraceInfo, annotate_trace, get_trace, trace_context - -__version__ = "0.1.0" +from .version import __version__ __all__ = [ "AppConfig", diff --git a/src/mercury_composable/actuator.py b/src/mercury_composable/actuator.py new file mode 100644 index 0000000..e65365e --- /dev/null +++ b/src/mercury_composable/actuator.py @@ -0,0 +1,227 @@ +""" +Actuator endpoints for operations and Kubernetes deployment. + +The same operational surface as the engines (the Java ``ActuatorServices`` +and its Rust port), so a polyglot installation monitors every app one way: + +- ``GET /info`` - application identity (name, version, description), runtime, + origin id, start/current time and uptime. +- ``GET /info/routes`` - the local routing table split by visibility + (``routing.public`` / ``routing.private``, route -> instance count). +- ``GET /env`` - selected environment variables (``show.env.variables``) and + selected configuration parameters (``show.application.properties``) - + opt-in lists, so secrets are never dumped wholesale (engine parity). +- ``GET /health`` - runs the health-check functions listed in + ``mandatory.health.dependencies`` / ``optional.health.dependencies``. + All mandatory up -> ``UP`` (HTTP 200); any mandatory down -> ``DOWN`` + (HTTP 400, engine parity). The outcome feeds the liveness state. +- ``GET /livenessprobe`` - ``OK`` (text) while the last health outcome is + good, else HTTP 400 ``Unhealthy. Please check '/health' endpoint.`` + +A health-check function is a normal registered function (usually private) +speaking the engines' interface contract - called through the same event bus +that serves PostOffice, first with header ``type=info`` (an advisory identity +map merged into its dependency entry), then with ``type=health`` (a status +text or map; a non-200 reply marks the dependency down):: + + @preload("demo.health", private=True) + async def health(headers: dict[str, str], _body: Body) -> Body: + if headers.get("type") == "info": + return {"service": "demo.service", "href": "http://127.0.0.1"} + return "demo.service is running fine" + +Engine deltas (deliberate, wrapper-scale): no ``/info/lib`` (a wrapper app +has no runtime dependency manifest - deferred on the Rust port too), no XML +responses, and no 5-second info cache (dependencies are in-process +functions, so the ``type=info`` lookup costs nothing). +""" + +from __future__ import annotations + +import contextlib +import os +import platform as runtime_platform +import re +import time +import uuid +from datetime import datetime, timezone +from typing import Any + +from aiohttp import web + +from .bus import DeliveryTimeout +from .config import app_config +from .envelope import iso_utc +from .log import get_logger +from .registry import FunctionRegistry +from .version import __version__ + +log = get_logger("mercury.actuator") + +INFO_TIMEOUT_MS = 3000 # engine value for the advisory type=info lookup +HEALTH_TIMEOUT_MS = 10000 # engine value for the type=health probe +UNHEALTHY = "Unhealthy. Please check '/health' endpoint." + +_SPLIT = re.compile(r"[,\s]+") + +_origin: str | None = None + + +def app_origin() -> str: + """Unique instance id, minted once per process (the Java reference + engine's format: UTC yyyyMMdd date prefix + 32-hex uuid).""" + global _origin + if _origin is None: + _origin = time.strftime("%Y%m%d", time.gmtime()) + uuid.uuid4().hex + return _origin + + +def elapsed_time(milliseconds: float) -> str: + """Human-readable duration matching the engines' rendering + (including their strict boundary behavior, kept verbatim for parity).""" + one_second = 1000 + one_minute = 60 * one_second + one_hour = 60 * one_minute + one_day = 24 * one_hour + remaining = int(milliseconds) + parts: list[str] = [] + if remaining > one_day: + days = remaining // one_day + parts.append(f"{days} day" if days == 1 else f"{days} days") + remaining -= days * one_day + if remaining > one_hour: + hours = remaining // one_hour + parts.append(f"{hours} hour" if hours == 1 else f"{hours} hours") + remaining -= hours * one_hour + if remaining > one_minute: + minutes = remaining // one_minute + parts.append(f"{minutes} minute" if minutes == 1 else f"{minutes} minutes") + remaining -= minutes * one_minute + if remaining >= one_second: + seconds = remaining // one_second + parts.append(f"{seconds} second" if seconds == 1 else f"{seconds} seconds") + return " ".join(parts) if parts else f"{remaining} ms" + + +def _as_list(value: Any) -> list[str]: + """A comma/space-separated string (engine syntax) or a YAML list.""" + if isinstance(value, list): + items = [str(item).strip() for item in value] + else: + items = _SPLIT.split(str(value or "")) + return [item for item in items if item] + + +class Actuator: + """HTTP handlers for the actuator endpoints (wired by EventApiServer).""" + + def __init__(self, registry: FunctionRegistry): + config = app_config() + self.registry = registry + self.start = datetime.now(timezone.utc) + self.healthy = True # liveness follows the most recent /health outcome + self.app_name = config.get_property("application.name", "application") or "application" + self.app_version = config.get_property("info.app.version", __version__) or __version__ + self.description = (config.get_property("info.app.description", self.app_name) + or self.app_name) + self.required = _as_list(config.get("mandatory.health.dependencies")) + self.optional = _as_list(config.get("optional.health.dependencies")) + if self.required: + log.info("Mandatory service dependencies - %s", self.required) + if self.optional: + log.info("Optional services dependencies - %s", self.optional) + + def _app_block(self) -> dict[str, Any]: + return {"name": self.app_name, "version": self.app_version, + "description": self.description} + + # aiohttp handlers must be coroutines - async is the framework contract + # even when a handler has nothing to await + async def handle_info(self, _request: web.Request) -> web.Response: + now = datetime.now(timezone.utc) + return web.json_response({ + "app": self._app_block(), + "runtime": { + "language": "python", + "python": runtime_platform.python_version(), + "mercury_composable": __version__, + }, + "origin": app_origin(), + "time": {"start": iso_utc(self.start), "current": iso_utc(now)}, + "up_time": elapsed_time((now - self.start).total_seconds() * 1000), + }) + + async def handle_routes(self, _request: web.Request) -> web.Response: + public: dict[str, int] = {} + private: dict[str, int] = {} + for route, service in sorted(self.registry.routes().items()): + target = private if service.private else public + target[route] = service.instances + return web.json_response({ + "app": self._app_block(), + "routing": {"public": public, "private": private}, + }) + + async def handle_env(self, _request: web.Request) -> web.Response: + config = app_config() + environment = {name: os.environ.get(name, "") + for name in _as_list(config.get("show.env.variables"))} + properties = {name: config.get_property(name) or "" + for name in _as_list(config.get("show.application.properties"))} + return web.json_response({ + "app": self._app_block(), + "env": {"environment": environment, "properties": properties}, + }) + + async def handle_health(self, _request: web.Request) -> web.Response: + dependency: list[dict[str, Any]] = [] + # optional services never affect the overall status (engine semantics) + await self._check_services(self.optional, required=False, dependency=dependency) + up = await self._check_services(self.required, required=True, dependency=dependency) + self.healthy = up + result: dict[str, Any] = {} + if not dependency: + result["message"] = ("Did you forget to define mandatory.health.dependencies " + "or optional.health.dependencies") + result["dependency"] = dependency + result["status"] = "UP" if up else "DOWN" + result["origin"] = app_origin() + result["name"] = self.app_name + return web.json_response(result, status=200 if up else 400) + + async def handle_livenessprobe(self, _request: web.Request) -> web.Response: + if self.healthy: + return web.Response(text="OK") + return web.Response(status=400, text=UNHEALTHY) + + async def _check_services(self, services: list[str], *, required: bool, + dependency: list[dict[str, Any]]) -> bool: + all_up = True + for route in services: + entry: dict[str, Any] = {"route": route, "required": required} + dependency.append(entry) + service = self.registry.get(route) + if service is None: + all_up = False + entry["status_code"] = 404 + entry["message"] = f"Please check - Route {route} not found" + continue + bus = self.registry.bus + # info is advisory - merge whatever the service reports about + # itself; the health probe below decides the status + with contextlib.suppress(DeliveryTimeout): + info = await bus.deliver(service, {"type": "info"}, None, INFO_TIMEOUT_MS) + if isinstance(info.body, dict): + entry.update(info.body) + try: + reply = await bus.deliver(service, {"type": "health"}, None, HEALTH_TIMEOUT_MS) + entry["status_code"] = reply.get_status() + if isinstance(reply.body, (str, dict)): + entry["message"] = reply.body + if reply.has_error(): + all_up = False + except DeliveryTimeout as e: + all_up = False + entry["status_code"] = 408 + entry["message"] = f"Please check - {e}" + return all_up diff --git a/src/mercury_composable/server.py b/src/mercury_composable/server.py index 4677cd5..808ec05 100644 --- a/src/mercury_composable/server.py +++ b/src/mercury_composable/server.py @@ -16,12 +16,16 @@ guard) and transported ``my_*`` keys are removed from the handler's header view; the ``my_cid`` tag is injected as the read-only ``my_correlation_id`` header, per the wire-format contract. +- The host also serves the engines' actuator endpoints (``/info``, + ``/info/routes``, ``/env``, ``/health``, ``/livenessprobe``) for + operations and Kubernetes probes - see :mod:`mercury_composable.actuator`. """ from __future__ import annotations from aiohttp import web +from .actuator import Actuator from .bus import DeliveryTimeout from .config import app_config from .envelope import EventEnvelope @@ -57,6 +61,7 @@ class EventApiServer: def __init__(self, registry: FunctionRegistry | None = None): self.registry = registry or default_registry + self.actuator = Actuator(self.registry) async def handle_event(self, request: web.Request) -> web.Response: raw = await request.read() @@ -95,16 +100,14 @@ async def handle_event(self, request: web.Request) -> web.Response: event.to, reply.get_status(), reply.exec_time, event.trace_id) return web.Response(status=200, body=reply.to_bytes(), content_type=OCTET_STREAM) - # aiohttp handlers must be coroutines - async is the framework contract - # even though this one has nothing to await - @staticmethod - async def handle_health(_request: web.Request) -> web.Response: - return web.Response(text="OK") - def create_app(self) -> web.Application: app = web.Application(client_max_size=16 * 1024 * 1024) app.router.add_post("/api/event", self.handle_event) - app.router.add_get("/health", self.handle_health) + app.router.add_get("/info", self.actuator.handle_info) + app.router.add_get("/info/routes", self.actuator.handle_routes) + app.router.add_get("/env", self.actuator.handle_env) + app.router.add_get("/health", self.actuator.handle_health) + app.router.add_get("/livenessprobe", self.actuator.handle_livenessprobe) return app diff --git a/src/mercury_composable/version.py b/src/mercury_composable/version.py new file mode 100644 index 0000000..58fc848 --- /dev/null +++ b/src/mercury_composable/version.py @@ -0,0 +1,3 @@ +"""Package version - the single Python-side source (pyproject.toml mirrors it).""" + +__version__ = "0.1.0" diff --git a/tests/test_actuator.py b/tests/test_actuator.py new file mode 100644 index 0000000..c9356c8 --- /dev/null +++ b/tests/test_actuator.py @@ -0,0 +1,197 @@ +"""Actuator endpoint pins: engine-parity operational surface for Kubernetes.""" + +import re +from collections.abc import AsyncIterator, Iterator +from contextlib import asynccontextmanager +from typing import Any + +import aiohttp +import pytest +from aiohttp import web + +from mercury_composable import AppException, Body, FunctionRegistry, __version__, app_config +from mercury_composable.actuator import app_origin, elapsed_time +from mercury_composable.server import EventApiServer + +ORIGIN_SHAPE = r"\d{8}[0-9a-f]{32}" # UTC yyyyMMdd + 32-hex uuid (Java reference format) + +CONFIG_KEYS = [ + "mandatory.health.dependencies", "optional.health.dependencies", + "show.env.variables", "show.application.properties", + "application.name", "info.app.description", "info.app.version", +] + + +@pytest.fixture(autouse=True) +def clean_config() -> Iterator[None]: + yield + config = app_config() + for key in CONFIG_KEYS: + config.set(key, "") # empty override = unset (the Actuator treats "" as absent) + + +@asynccontextmanager +async def actuator_server(registry: FunctionRegistry) -> AsyncIterator[str]: + # the Actuator reads its configuration at construction (engine semantics), + # so each test sets config BEFORE entering this context + server = EventApiServer(registry) + runner = web.AppRunner(server.create_app()) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + try: + yield f"http://127.0.0.1:{runner.addresses[0][1]}" + finally: + await runner.cleanup() + await registry.bus.close() + + +async def get_json(url: str) -> tuple[int, Any]: + async with aiohttp.ClientSession() as session, session.get(url) as response: + return response.status, await response.json() + + +async def get_text(url: str) -> tuple[int, str]: + async with aiohttp.ClientSession() as session, session.get(url) as response: + return response.status, await response.text() + + +def engine_contract_registry() -> FunctionRegistry: + """A health-check function speaking the engines' type=info/type=health contract.""" + registry = FunctionRegistry() + + async def health(headers: dict[str, str], _body: Body): + if headers.get("type") == "info": + return {"service": "demo.service", "href": "http://127.0.0.1"} + return "demo.service is running fine" + + registry.register("demo.health", health, private=True) + return registry + + +async def test_info_reports_identity_runtime_origin(): + config = app_config() + config.set("application.name", "unit-app") + config.set("info.app.description", "actuator test app") + async with actuator_server(FunctionRegistry()) as url: + status, info = await get_json(f"{url}/info") + assert status == 200 + assert info["app"] == {"name": "unit-app", "version": __version__, + "description": "actuator test app"} + assert info["runtime"]["language"] == "python" + assert info["runtime"]["mercury_composable"] == __version__ + assert re.fullmatch(ORIGIN_SHAPE, info["origin"]) + assert info["time"]["start"] <= info["time"]["current"] + assert "up_time" in info + + +async def test_info_routes_splits_by_visibility(): + registry = FunctionRegistry() + + async def noop(_headers: dict[str, str], _body: Body): + return None + + registry.register("unit.public.route", noop, instances=8) + registry.register("unit.private.route", noop, instances=2, private=True) + async with actuator_server(registry) as url: + status, result = await get_json(f"{url}/info/routes") + assert status == 200 + assert result["routing"] == {"public": {"unit.public.route": 8}, + "private": {"unit.private.route": 2}} + assert result["app"]["name"] == "application" + + +async def test_env_shows_only_opted_in_values(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("MERCURY_UNIT_ENV", "unit-value") + config = app_config() + config.set("show.env.variables", "MERCURY_UNIT_ENV, MERCURY_UNIT_ABSENT") + config.set("show.application.properties", "application.name") + config.set("application.name", "unit-app") + async with actuator_server(FunctionRegistry()) as url: + status, result = await get_json(f"{url}/env") + assert status == 200 + # a missing environment variable renders as an empty string (engine parity) + assert result["env"]["environment"] == {"MERCURY_UNIT_ENV": "unit-value", + "MERCURY_UNIT_ABSENT": ""} + assert result["env"]["properties"] == {"application.name": "unit-app"} + + +async def test_health_up_with_engine_contract_dependency(): + app_config().set("mandatory.health.dependencies", "demo.health") + async with actuator_server(engine_contract_registry()) as url: + status, health = await get_json(f"{url}/health") + live = await get_text(f"{url}/livenessprobe") + assert status == 200 + assert health["status"] == "UP" + assert health["name"] == "application" + assert re.fullmatch(ORIGIN_SHAPE, health["origin"]) + # the info map merges into the dependency entry; health decides the status + assert health["dependency"] == [{ + "route": "demo.health", "required": True, + "service": "demo.service", "href": "http://127.0.0.1", + "status_code": 200, "message": "demo.service is running fine", + }] + assert live == (200, "OK") + + +async def test_health_down_missing_dependency_drives_liveness(): + app_config().set("mandatory.health.dependencies", "no.such.route") + async with actuator_server(FunctionRegistry()) as url: + assert await get_text(f"{url}/livenessprobe") == (200, "OK") # healthy until proven + status, health = await get_json(f"{url}/health") + live_status, live_text = await get_text(f"{url}/livenessprobe") + assert status == 400 + assert health["status"] == "DOWN" + assert health["dependency"] == [{ + "route": "no.such.route", "required": True, + "status_code": 404, "message": "Please check - Route no.such.route not found", + }] + assert live_status == 400 + assert live_text == "Unhealthy. Please check '/health' endpoint." + + +async def test_optional_failure_never_downs_health(): + registry = engine_contract_registry() + + async def broken(_headers: dict[str, str], _body: Body): + raise AppException(500, "backend down") + + registry.register("broken.health", broken, private=True) + config = app_config() + config.set("mandatory.health.dependencies", "demo.health") + config.set("optional.health.dependencies", "broken.health") + async with actuator_server(registry) as url: + status, health = await get_json(f"{url}/health") + assert status == 200 + assert health["status"] == "UP" + broken_dep = next(d for d in health["dependency"] if d["route"] == "broken.health") + assert broken_dep["required"] is False + assert broken_dep["status_code"] == 500 + assert broken_dep["message"] == "backend down" + + +async def test_health_without_dependencies_teaches(): + async with actuator_server(FunctionRegistry()) as url: + status, health = await get_json(f"{url}/health") + assert status == 200 + assert health["status"] == "UP" + assert health["dependency"] == [] + assert health["message"].startswith("Did you forget to define") + + +def test_elapsed_time_matches_engine_rendering(): + assert elapsed_time(0) == "0 ms" + assert elapsed_time(500) == "500 ms" + assert elapsed_time(1000) == "1 second" + assert elapsed_time(61_000) == "1 minute 1 second" + # the engines' strict boundary behavior, kept verbatim + assert elapsed_time(60_000) == "60 seconds" + assert elapsed_time(3_600_000) == "60 minutes" + assert elapsed_time(86_400_000) == "24 hours" + assert elapsed_time(120_000) == "2 minutes" + assert elapsed_time(90_061_000) == "1 day 1 hour 1 minute 1 second" + + +def test_origin_is_stable_and_engine_shaped(): + assert app_origin() == app_origin() # minted once per process + assert re.fullmatch(ORIGIN_SHAPE, app_origin()) diff --git a/tests/test_server.py b/tests/test_server.py index 23bd389..4d170f7 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -187,10 +187,12 @@ async def test_async_drop_n_forget_202_ack(server_url: str): assert "time" in reply.body -async def test_health_endpoint(server_url: str): - async with ( - aiohttp.ClientSession() as session, - session.get(f"{server_url}/health") as response, - ): - assert response.status == 200 - assert await response.text() == "OK" +async def test_actuator_endpoints_are_wired(server_url: str): + # shapes are pinned in test_actuator.py - this pins the host wiring only + async with aiohttp.ClientSession() as session: + async with session.get(f"{server_url}/livenessprobe") as response: + assert response.status == 200 + assert await response.text() == "OK" + async with session.get(f"{server_url}/info") as response: + assert response.status == 200 + assert (await response.json())["runtime"]["language"] == "python" From a6741985746c8b19cbb1ef3185c70a5a2887596e Mon Sep 17 00:00:00 2001 From: Eric Law <39393654+acn-ericlaw@users.noreply.github.com> Date: Sat, 22 Aug 2026 20:23:11 -0700 Subject: [PATCH 6/8] chore: IDE/Sonar round on the actuator - single async dispatcher, narrowing, spelling One async handle() dispatcher (awaits the /health branch; the constant endpoints render synchronously) replaces four async-without-await handlers - and now mirrors the node twin's dispatcher shape. app_origin uses the local-narrowing pattern (checkers do not narrow module globals). 'health check' spelling; origin test no longer compares an expression to itself. Co-Authored-By: Claude Code --- CHANGELOG.md | 2 +- README.md | 4 +-- src/mercury_composable/actuator.py | 41 +++++++++++++++++++++--------- src/mercury_composable/server.py | 8 +++--- tests/test_actuator.py | 7 ++--- 5 files changed, 39 insertions(+), 23 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a3b7728..8fae590 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,7 @@ - Actuator endpoints `/info`, `/info/routes`, `/env`, `/health` and `/livenessprobe` - the engines' operational surface, for Kubernetes probes and one-dashboard monitoring - of polyglot installations. Health-check functions are normal registered functions + of polyglot installations. Health check functions are normal registered functions speaking the engines' `type=info` / `type=health` interface contract, listed in `mandatory.health.dependencies` / `optional.health.dependencies` and called through the event bus. `/health` answers `UP` (200) / `DOWN` (400); `/livenessprobe` follows diff --git a/README.md b/README.md index 0dddbc3..cdc089c 100644 --- a/README.md +++ b/README.md @@ -143,8 +143,8 @@ Kubernetes probes and dashboards treat a Python app exactly like a Java or Rust Configuration keys carry the engines' names: `info.app.version`, `info.app.description`, `show.env.variables` and `show.application.properties` (opt-in lists — secrets are never dumped wholesale), and `mandatory.health.dependencies` / `optional.health.dependencies` -(routes of health-check functions; optional ones never change the overall status). A -health-check function is a normal registered function — usually private — speaking the +(routes of health check functions; optional ones never change the overall status). A +health check function is a normal registered function — usually private — speaking the engines' interface contract, called through the event bus: ```python diff --git a/src/mercury_composable/actuator.py b/src/mercury_composable/actuator.py index e65365e..8ae799b 100644 --- a/src/mercury_composable/actuator.py +++ b/src/mercury_composable/actuator.py @@ -11,14 +11,14 @@ - ``GET /env`` - selected environment variables (``show.env.variables``) and selected configuration parameters (``show.application.properties``) - opt-in lists, so secrets are never dumped wholesale (engine parity). -- ``GET /health`` - runs the health-check functions listed in +- ``GET /health`` - runs the health check functions listed in ``mandatory.health.dependencies`` / ``optional.health.dependencies``. All mandatory up -> ``UP`` (HTTP 200); any mandatory down -> ``DOWN`` (HTTP 400, engine parity). The outcome feeds the liveness state. - ``GET /livenessprobe`` - ``OK`` (text) while the last health outcome is good, else HTTP 400 ``Unhealthy. Please check '/health' endpoint.`` -A health-check function is a normal registered function (usually private) +A health check function is a normal registered function (usually private) speaking the engines' interface contract - called through the same event bus that serves PostOffice, first with header ``type=info`` (an advisory identity map merged into its dependency entry), then with ``type=health`` (a status @@ -71,9 +71,11 @@ def app_origin() -> str: """Unique instance id, minted once per process (the Java reference engine's format: UTC yyyyMMdd date prefix + 32-hex uuid).""" global _origin - if _origin is None: - _origin = time.strftime("%Y%m%d", time.gmtime()) + uuid.uuid4().hex - return _origin + origin = _origin # local narrowing - checkers do not narrow module globals + if origin is None: + origin = time.strftime("%Y%m%d", time.gmtime()) + uuid.uuid4().hex + _origin = origin + return origin def elapsed_time(milliseconds: float) -> str: @@ -135,9 +137,24 @@ def _app_block(self) -> dict[str, Any]: return {"name": self.app_name, "version": self.app_version, "description": self.description} - # aiohttp handlers must be coroutines - async is the framework contract - # even when a handler has nothing to await - async def handle_info(self, _request: web.Request) -> web.Response: + async def handle(self, request: web.Request) -> web.Response: + """The single GET dispatcher (aiohttp handlers must be coroutines): + /health awaits its dependency probes; the rest render synchronously.""" + match request.path: + case "/info": + return self._info() + case "/info/routes": + return self._routes() + case "/env": + return self._env() + case "/health": + return await self._health() + case "/livenessprobe": + return self._liveness_probe() + case _: # the router only maps the five paths above here + return web.Response(status=404, text="Not found") + + def _info(self) -> web.Response: now = datetime.now(timezone.utc) return web.json_response({ "app": self._app_block(), @@ -151,7 +168,7 @@ async def handle_info(self, _request: web.Request) -> web.Response: "up_time": elapsed_time((now - self.start).total_seconds() * 1000), }) - async def handle_routes(self, _request: web.Request) -> web.Response: + def _routes(self) -> web.Response: public: dict[str, int] = {} private: dict[str, int] = {} for route, service in sorted(self.registry.routes().items()): @@ -162,7 +179,7 @@ async def handle_routes(self, _request: web.Request) -> web.Response: "routing": {"public": public, "private": private}, }) - async def handle_env(self, _request: web.Request) -> web.Response: + def _env(self) -> web.Response: config = app_config() environment = {name: os.environ.get(name, "") for name in _as_list(config.get("show.env.variables"))} @@ -173,7 +190,7 @@ async def handle_env(self, _request: web.Request) -> web.Response: "env": {"environment": environment, "properties": properties}, }) - async def handle_health(self, _request: web.Request) -> web.Response: + async def _health(self) -> web.Response: dependency: list[dict[str, Any]] = [] # optional services never affect the overall status (engine semantics) await self._check_services(self.optional, required=False, dependency=dependency) @@ -189,7 +206,7 @@ async def handle_health(self, _request: web.Request) -> web.Response: result["name"] = self.app_name return web.json_response(result, status=200 if up else 400) - async def handle_livenessprobe(self, _request: web.Request) -> web.Response: + def _liveness_probe(self) -> web.Response: if self.healthy: return web.Response(text="OK") return web.Response(status=400, text=UNHEALTHY) diff --git a/src/mercury_composable/server.py b/src/mercury_composable/server.py index 808ec05..ad88103 100644 --- a/src/mercury_composable/server.py +++ b/src/mercury_composable/server.py @@ -103,11 +103,9 @@ async def handle_event(self, request: web.Request) -> web.Response: def create_app(self) -> web.Application: app = web.Application(client_max_size=16 * 1024 * 1024) app.router.add_post("/api/event", self.handle_event) - app.router.add_get("/info", self.actuator.handle_info) - app.router.add_get("/info/routes", self.actuator.handle_routes) - app.router.add_get("/env", self.actuator.handle_env) - app.router.add_get("/health", self.actuator.handle_health) - app.router.add_get("/livenessprobe", self.actuator.handle_livenessprobe) + # the engines' actuator endpoints (see actuator.py) + for path in ("/info", "/info/routes", "/env", "/health", "/livenessprobe"): + app.router.add_get(path, self.actuator.handle) return app diff --git a/tests/test_actuator.py b/tests/test_actuator.py index c9356c8..e854a59 100644 --- a/tests/test_actuator.py +++ b/tests/test_actuator.py @@ -57,7 +57,7 @@ async def get_text(url: str) -> tuple[int, str]: def engine_contract_registry() -> FunctionRegistry: - """A health-check function speaking the engines' type=info/type=health contract.""" + """A health check function speaking the engines' type=info/type=health contract.""" registry = FunctionRegistry() async def health(headers: dict[str, str], _body: Body): @@ -193,5 +193,6 @@ def test_elapsed_time_matches_engine_rendering(): def test_origin_is_stable_and_engine_shaped(): - assert app_origin() == app_origin() # minted once per process - assert re.fullmatch(ORIGIN_SHAPE, app_origin()) + minted = app_origin() + assert minted == app_origin() # minted once per process + assert re.fullmatch(ORIGIN_SHAPE, minted) From 7d5ca6c0adf4dfb9570ff6f8afe6f0d6ae3b5bf1 Mon Sep 17 00:00:00 2001 From: Eric Law <39393654+acn-ericlaw@users.noreply.github.com> Date: Sat, 22 Aug 2026 20:30:07 -0700 Subject: [PATCH 7/8] feat: log.format compact (JSONL) + sample resources/application.yml for the demo log.format now carries the engines' three presentations: text (default), json (pretty-printed, matching the engine's pretty serializer) and compact (single-line JSONL for log aggregators). The demo app gains examples/resources/application.yml - the engines' resources convention, auto-loaded next to the app file - wiring the demo.health dependency and demonstrating the well-known keys. README quick start teaches the -D override syntax instead of --port. Co-Authored-By: Claude Code --- CHANGELOG.md | 4 ++++ README.md | 10 ++++++---- examples/demo_app.py | 10 +++++++--- examples/resources/application.yml | 22 ++++++++++++++++++++++ src/mercury_composable/config.py | 3 ++- src/mercury_composable/log.py | 20 +++++++++++++++----- tests/test_log.py | 12 ++++++++++++ 7 files changed, 68 insertions(+), 13 deletions(-) create mode 100644 examples/resources/application.yml diff --git a/CHANGELOG.md b/CHANGELOG.md index 8fae590..4b70bf9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ `mandatory.health.dependencies` / `optional.health.dependencies` and called through the event bus. `/health` answers `UP` (200) / `DOWN` (400); `/livenessprobe` follows the most recent health outcome. +- `log.format` carries the engines' three presentations: `text` (default), `json` + (pretty-printed) and `compact` (single-line JSONL for log aggregators). A sample + `examples/resources/application.yml` demonstrates the resources convention and the + well-known keys. - Primitive in-process event bus - the single dispatch pipeline: one FIFO mailbox per route consumed by `instances` worker tasks (the parameter is faithful); RPC deliveries are ttl-bounded with a dead-work skip; drop-n-forget returns the 202-shape ack. The diff --git a/README.md b/README.md index cdc089c..55ecfca 100644 --- a/README.md +++ b/README.md @@ -44,7 +44,7 @@ Run it: ```bash pip install -e '.[dev]' -mercury-serve app.py --port 8086 +mercury-serve app.py -Drest.server.port=8086 ``` Call it from a Mercury engine application with two configuration entries and no code — @@ -108,12 +108,14 @@ The same conventions as the engines, so a polyglot installation stays uniform: |-----|---------|---------| | `application.name` | application identity in logs | `application` | | `rest.server.port` | Event API port | `8085` | -| `log.format` | `text` or `json` | `text` | +| `log.format` | `text`, `json` (pretty-printed) or `compact` (single-line JSONL) | `text` | | `log.level` | log level (`LOG_LEVEL` env var wins) | `INFO` | Configuration lives in the `resources` folder, mirroring the engines: -`resources/application.yml` (or `.yaml` / `.properties`), or an explicit `--config` path. -Values support `${ENV_VAR:default}` substitution. Runtime parameter overrides use the +`resources/application.yml` (or `.yaml` / `.properties`) in the working directory or next +to the application file, or an explicit `--config` path — see +[`examples/resources/application.yml`](examples/resources/application.yml) for a worked +sample. Values support `${ENV_VAR:default}` substitution. Runtime parameter overrides use the same `-D` syntax as the Java engine and the Rust port — checked first on every read (`AppConfig.set(key, value)` does the same programmatically, the `f:setConfig` analog): diff --git a/examples/demo_app.py b/examples/demo_app.py index 999d51a..e6bdacb 100644 --- a/examples/demo_app.py +++ b/examples/demo_app.py @@ -1,7 +1,11 @@ """ Demo polyglot functions. -Run: mercury-serve examples/demo_app.py --port 8086 +Run: mercury-serve examples/demo_app.py + +Configuration comes from examples/resources/application.yml (the engines' +"resources" convention - port 8086, the demo.health dependency, log format); +override any key with -Dkey=value, e.g. -Drest.server.port=8090. Then map a route from a Mercury engine application (event-over-http.yaml): @@ -63,8 +67,8 @@ async def chain(_headers: dict[str, str], body: Body): async def health_check(headers: dict[str, str], _body: Body): """Health check speaking the engines' interface contract (type=info / type=health). - Activate it for the /health actuator endpoint: - mercury-serve examples/demo_app.py -Dmandatory.health.dependencies=demo.health + Activated for the /health actuator endpoint by mandatory.health.dependencies + in examples/resources/application.yml (or a -D override). """ if headers.get("type") == "info": return {"service": "demo.service", "href": "http://127.0.0.1"} diff --git a/examples/resources/application.yml b/examples/resources/application.yml new file mode 100644 index 0000000..f295e90 --- /dev/null +++ b/examples/resources/application.yml @@ -0,0 +1,22 @@ +# Sample configuration - the engines' "resources" convention. +# mercury-serve loads resources/application.yml from the working directory, or from +# the resources folder next to the application file (this one). Any key can be +# overridden at run time with the engines' -D syntax, e.g.: +# mercury-serve examples/demo_app.py -Drest.server.port=8090 -Dlog.format=compact + +application.name: 'demo-app' +info.app.description: 'Mercury Composable polyglot demo' + +rest.server.port: 8086 + +# text (default) | json (pretty-printed) | compact (single-line JSONL) +log.format: text +log.level: INFO + +# Actuator /health dependencies - routes of health check functions speaking the +# engines' type=info / type=health contract (demo.health in demo_app.py). +mandatory.health.dependencies: 'demo.health' + +# Opt-in lists for the /env endpoint - nothing is ever dumped wholesale. +show.env.variables: 'LOG_LEVEL' +show.application.properties: 'application.name, rest.server.port' diff --git a/src/mercury_composable/config.py b/src/mercury_composable/config.py index 926d0a4..120bfe8 100644 --- a/src/mercury_composable/config.py +++ b/src/mercury_composable/config.py @@ -22,7 +22,8 @@ - ``application.name`` — application identity used in logs. - ``rest.server.port`` — the Event API port (default 8085). -- ``log.format`` — ``text`` (default) or ``json``. +- ``log.format`` — ``text`` (default), ``json`` (pretty-printed) or + ``compact`` (single-line JSONL), the engines' three presentations. - ``log.level`` — default INFO; the ``LOG_LEVEL`` environment variable wins, mirroring the engines' log4j2 setting. """ diff --git a/src/mercury_composable/log.py b/src/mercury_composable/log.py index 55ac4a4..0614f0e 100644 --- a/src/mercury_composable/log.py +++ b/src/mercury_composable/log.py @@ -13,9 +13,10 @@ - The level comes from the ``LOG_LEVEL`` environment variable when set (mirroring the engines), else the ``log.level`` configuration key, else INFO. -- ``log.format=json`` switches to one JSON object per line with the same - information (time, level, logger, message, and trace_id when a trace - context is active). +- ``log.format`` carries the engines' three presentations: ``text`` + (default), ``json`` (pretty-printed JSON with time, level, logger, + message, and trace_id when a trace context is active) and ``compact`` + (the same object on a single line - JSONL - for log aggregators). """ from __future__ import annotations @@ -43,6 +44,12 @@ def format(self, record: logging.LogRecord) -> str: class EngineJsonFormatter(logging.Formatter): + """Engine JSON presentations: json = pretty-printed, compact = one line (JSONL).""" + + def __init__(self, *, compact: bool = False): + super().__init__() + self._indent = None if compact else 2 + def format(self, record: logging.LogRecord) -> str: ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(record.created)) entry = { @@ -58,7 +65,7 @@ def format(self, record: logging.LogRecord) -> str: entry["trace_id"] = info.trace_id if record.exc_info: entry["exception"] = self.formatException(record.exc_info) - return json.dumps(entry, ensure_ascii=False) + return json.dumps(entry, ensure_ascii=False, indent=self._indent) def _setup() -> None: @@ -69,7 +76,10 @@ def _setup() -> None: level_name = os.environ.get("LOG_LEVEL") or str(config.get("log.level", "INFO")) log_format = str(config.get("log.format", "text")).lower() handler = logging.StreamHandler(sys.stdout) - handler.setFormatter(EngineJsonFormatter() if log_format == "json" else EngineTextFormatter()) + if log_format in ("json", "compact"): + handler.setFormatter(EngineJsonFormatter(compact=log_format == "compact")) + else: + handler.setFormatter(EngineTextFormatter()) root = logging.getLogger() root.handlers.clear() root.addHandler(handler) diff --git a/tests/test_log.py b/tests/test_log.py index abb3e55..fc39894 100644 --- a/tests/test_log.py +++ b/tests/test_log.py @@ -33,3 +33,15 @@ def test_json_formatter_carries_exception(): assert entry["logger"] == "unit.test:42" assert entry["message"] == "Async event demo.route failed" assert "RuntimeError: boom-x" in entry["exception"] + + +def test_json_is_pretty_and_compact_is_jsonl(): + import json + + record = _record_with_exception() + pretty = EngineJsonFormatter().format(record) + compact = EngineJsonFormatter(compact=True).format(record) + # engine presentations: json = pretty-printed, compact = single line (JSONL) + assert "\n" in pretty + assert "\n" not in compact + assert json.loads(pretty) == json.loads(compact) # same information, two renderings From 1931f01c436c2788378399802839879319894c77 Mon Sep 17 00:00:00 2001 From: Eric Law <39393654+acn-ericlaw@users.noreply.github.com> Date: Sat, 22 Aug 2026 21:12:08 -0700 Subject: [PATCH 8/8] session 2026-08-23 [Claude Code] --- memory/continuity.md | 19 ++++++- memory/sessions/2026-08-23-031558.md | 80 ++++++++++++++++++++++++++++ 2 files changed, 98 insertions(+), 1 deletion(-) create mode 100644 memory/sessions/2026-08-23-031558.md diff --git a/memory/continuity.md b/memory/continuity.md index fb3fcda..dc24651 100644 --- a/memory/continuity.md +++ b/memory/continuity.md @@ -17,7 +17,7 @@ polyglot initiative: a lightweight Event-over-HTTP function host + thin client, repurposed August 2026 (legacy language pack in git history only) - **last_enabled:** 2026-08-22 -- **last_session:** 2026-08-23 | agent: Claude Code (2026-08-23-024221) +- **last_session:** 2026-08-23 | agent: Claude Code (2026-08-23-031558) - **last_review:** (none yet) - **last_invariant_check:** (none yet) - **repo:** ~/sandbox/mercury-python (origin: github.com/Accenture/mercury-python) @@ -114,6 +114,23 @@ hosted→local-private. README boundary statement: leaf-side composition here; workflow processing = Event Script / Knowledge Graph. + +- [ ] (feature — Eric's directive 2026-08-23, **IMPLEMENTED same day on + `feature/primitive-event-bus`, commits `56c002c` + `a674198` (IDE/Sonar round); + node twin `342a854` + `7a8b12c`; PENDING Eric's PR gate together with the bus**) + **Actuator endpoints — the engines' operational surface for Kubernetes PODs.** + GET `/info`, `/info/routes`, `/env`, `/health`, `/livenessprobe` on the Event API port; + shapes mirror the Rust engine's actuator (the approved minimalist port of Java + `ActuatorServices`). Health check functions are normal registered functions speaking + the engines' `type=info` / `type=health` interface contract (Eric's ruling), listed in + `mandatory.health.dependencies` / `optional.health.dependencies` and called through the + event bus; `/health` = UP 200 / DOWN 400 (Java parity); `/livenessprobe` follows the + most recent health outcome. Engine formats verbatim (origin = UTC yyyyMMdd + 32-hex + uuid per the Java reference; elapsed-time boundary quirks pinned). Documented deltas: + no `/info/lib`, no XML, no info cache. One async `handle()` dispatcher (S7503-clean, + mirrors the node twin). 10 pins + live demo drives on both wrappers. + Relates [[thread-primitive-event-bus]]; serves [[bp-publish-interop-gate]]. + > Mark completed items `- [x]` and leave them in place — the review sweeps them to > the archive once older than `archive_window` sessions. Don't archive them by hand. diff --git a/memory/sessions/2026-08-23-031558.md b/memory/sessions/2026-08-23-031558.md new file mode 100644 index 0000000..8dc8a91 --- /dev/null +++ b/memory/sessions/2026-08-23-031558.md @@ -0,0 +1,80 @@ +# Session (2026-08-23T03:15:58.000Z) + +**Agent:** Claude Code + +Actuator endpoints on `feature/primitive-event-bus` — Eric's directive right after the +bus landed: the wrapper apps deploy as Kubernetes PODs, so they need the engines' +operational surface. Commits `56c002c` (feature) + `a674198` (IDE/Sonar round). +Node twin: `342a854` + `7a8b12c` (see mercury-nodejs memory, same-day log). + +## What shipped + +- `actuator.py`: GET `/info`, `/info/routes`, `/env`, `/health`, `/livenessprobe` on the + Event API port. Response shapes mirror the Rust engine's actuator — itself the + maintainer-approved minimalist port of Java `ActuatorServices` — so DevSecOps sees one + surface across the estate: `/info` = `{app{name,version,description}, + runtime{language,python,mercury_composable}, origin, time{start,current}, up_time}`; + `/info/routes` = `routing.public/private` (route → instances, sorted); `/env` = opt-in + `show.env.variables` / `show.application.properties` (absent env var → `""`); + `/health` = dependency entries `{route, required, …info map, status_code, message}`, + UP 200 / DOWN 400 (Java parity), teaching message when no dependencies configured; + `/livenessprobe` follows the most recent health outcome (`OK` / 400 + `Unhealthy. Please check '/health' endpoint.`). +- **Health check functions = normal registered functions speaking the engines' interface + contract (Eric's mid-session ruling, matching my derivation):** called through the + event bus with header `type=info` (advisory identity map, merged into the entry) then + `type=health` (non-200 = down); listed in `mandatory.health.dependencies` / + `optional.health.dependencies` (optional never affects overall status); unregistered + route → 404 `Please check - Route X not found`; DeliveryTimeout → 408 + `Please check - Timeout for 10000 ms`. Engine timeouts kept (info 3 s, health 10 s). +- Engine formats verbatim: origin = UTC `yyyyMMdd` + 32-hex uuid (the JAVA reference + format — the Rust port simplified to a bare uuid, so the wrappers follow the + reference); `elapsed_time` with the engines' strict `>` boundary quirks (60 s → + "60 seconds", 3600 s → "60 minutes") — pinned with the same vectors the Rust port pins. +- Deliberate wrapper-scale deltas, documented in the module docstring: no `/info/lib` + (no runtime dependency manifest — deferred on the Rust port too), no XML responses, + no 5-second info cache (dependencies are in-process functions). +- `version.py` becomes the single Python-side version source (`__init__` re-exports) — + breaks the actuator → `__init__` import cycle. +- The pre-release `/health` "OK" scaffolding endpoint is replaced by the real one. +- IDE/Sonar round (Eric's screenshots): four S7503 async-without-await handlers → ONE + async `handle()` dispatcher (awaits the `/health` branch; constant endpoints render + synchronously) — the python Actuator now mirrors the node twin's dispatcher shape; + `app_origin` adopts the local-narrowing pattern (checkers don't narrow module + globals); "health check" spelling (Grazie); S5863 self-compare fixed in the origin + test; the `handle_livenessprobe` name (typo dictionary) dissolved by the dispatcher. + +## Verification + +- 49/49 pytest (10 actuator pins: five shapes, liveness-follows-health round trip, + optional-failure-never-downs, teach-message, elapsed/origin format vectors), ruff + clean, basedpyright 0 — re-run after the Sonar round. +- Live drive: `mercury-serve examples/demo_app.py --port 8399 + -Dmandatory.health.dependencies=demo.health` → all five endpoints engine-shaped; + `/health` UP with the full `demo.health` dependency entry; node twin symmetric on 8398. +- `examples/demo_app.py` gained `demo.health` (private, contract-speaking); README + "Actuator endpoints" section (+ K8s wiring: livenessProbe → `/livenessprobe`, + readinessProbe → `/health`); CHANGELOG entry. + +## Follow-up round (Eric's three asks, commit `7d5ca6c`; node twin `8ed4266`) + +- **`log.format` gains `compact`** — the engines' third presentation. Verified against + the Java engine's own appenders (`JsonAppender` = Gson pretty serializer, + `CompactAppender` = single-line serializer, same data map): so wrapper `json` is now + PRETTY-PRINTED (indent 2) and `compact` is the single-line JSONL the old `json` + emitted. Pinned: json has newlines, compact has none, both parse to the same entry. +- **Sample `examples/resources/application.yml`** (both wrappers) — the engines' + resources convention, auto-loaded by the CLI's next-to-app fallback: demo-app + identity, port (8086 py / 8087 node), `mandatory.health.dependencies: demo.health`, + opt-in /env lists, log keys. The demo now runs with plain + `mercury-serve examples/demo_app.py` — no flags. +- **README `-D` syntax** — quick starts teach `-Drest.server.port=8086` (engine syntax) + instead of `--port`; the `--port`/`--config` CLI flags remain as conveniences. +- Live drives re-run through the yml on both wrappers: /health UP under the demo-app + identity, /env shows the opted-in keys, `-Dlog.format=compact` emits JSONL. Gates: + 50/50 pytest + ruff + basedpyright 0; node 48/48. + +## Memory References + +- created: thread-actuator-endpoints +- referenced: thread-primitive-event-bus, bp-publish-interop-gate, vision-mercury-python