diff --git a/CHANGELOG.md b/CHANGELOG.md index 469d91a..65bb8cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +1,62 @@ # Changelog -## 0.1.0 (unreleased) +## 4.12.0 (2026-08-30) +The progressive-rendering milestone release. The version aligns with the Mercury +Composable engine lock-step line (Java and Rust engines, Python and Node.js +language packs all at v4.12.0): token/event streaming end to end with full +OpenTelemetry lineage, business-correlation continuity and application log +context across all four runtimes - useful on its own, and the foundation for +the AI SDLC (agent, MCP and tool adapters as wrapper-side functions with +complete observability). + +- **Event streaming** - the platform-wide multi-shot reply contract, both halves. + Producer: `@preload(..., interceptor=True)` handlers receive the raw envelope + and stream through `EventStreamWriter` (the engines' exact API - `first`, + `write`, `write_named`, `close` with trailing metadata, `fail` with the standard + error key-values); the `/api/event` host answers a caller that accepts + `text/event-stream` with the platform's hybrid SSE dialect (envelope frames for + the head, the terminals and non-text segments; raw frames for text tokens), + refuses a non-accepting caller of a streaming function with the pinned 406, and + keeps single-shot replies over the capable path byte-identical. Consumer: + `PostOffice.stream()` (an async iterator yielding the same decoded envelopes an + engine reply route receives, with the dialect conformance guards) and + `PostOffice.stream_to()` (the relay form: forward your caller's reply address + and segments flow through verbatim - engine-parity composition). Under it all, + the primitive event bus gained the engines' reply_to mechanism: envelope-routed + delivery to a LOCAL function or per-request reply sink - simple routing, no + orchestration. Same keep-alive config key as the engines + (`event.stream.keep.alive`). Engine-identical wire and messages + (Java PR #299-#301 / Rust PR #216-#218 lineage). +- **Business correlation-id continuity** (the engines' PostOffice parity): the + client stamps the current context's business correlation-id onto outbound + events as the engine-managed `my_cid` tag, local bus deliveries inject the + read-only `my_correlation_id` header view exactly like the HTTP host, and + `get_trace()` / `trace_context()` carry `my_correlation_id` - so the business + correlation-id continues across engine⇄wrapper and wrapper⇄wrapper hops. +- **Span lineage** (the engines' telemetry model): every traced execution mints + a 16-hex span with the caller's span (from the inbound envelope) as its + parent, outbound events carry the current span so the next hop parents onto + it (`PostOffice.touch` parity, W3C `traceparent` included), streaming + segments carry the producer's span, and non-RPC executions emit the engines' + distributed-trace dataset record on the `distributed.tracing` log stream - + the same `{"trace": {...}, "annotations": {...}}` shape the Java engine + logs, so stdout log-ingest agents stitch spans across all four runtimes. + RPC round-trips are suppressed exactly like the engines (the new `rpc` + envelope tag rides `request()` calls). `trace_context()` accepts `span_id` + to parent onto an external OpenTelemetry span. Outbound events and stream + segments also fill their sender with the executing function's route, and + the `/api/event` host fills `event.api.service` for an anonymous caller - + the engines' sender-attribution rules. +- **Application log context** (the engines' app-log-context feature, on by + default via the packaged `default-log-context.yaml` - the engines' resource + twin): with `log.format` json/compact, every log line inside a traced + request carries a `context` block - cid (the business correlation-id), + traceId, tracePath, spanId, parentSpanId, service, timestamp - so app logs + and the distributed-trace records correlate end to end. Customize with + `resources/app-log-context.yaml` (reserved `$tokens` or constants with + `${ENV:default}`), opt out with `app.log.context=false`, and add + per-request key-values with `update_context()` (reserved keys guarded). - Documentation site (mkdocs-material, the engine repo's theme): the three-layer theme reference, rationale/design foundations, function-writing patterns, flow and knowledge-graph join chapters, a one-page AI agent guide with llms.txt, and diff --git a/docs/guides/function-patterns.md b/docs/guides/function-patterns.md index ce970fd..9ebf0de 100644 --- a/docs/guides/function-patterns.md +++ b/docs/guides/function-patterns.md @@ -95,23 +95,50 @@ message and a stack trace, mirroring the engines. Handler-level errors always ri HTTP 200; only transport-level failures (unknown route, private target, timeout, undecodable envelope) surface as HTTP status codes. -## Trace context +## Trace context and span lineage -Every delivery runs under its caller's trace: +Every delivery runs under its caller's trace, and every traced execution mints +its own **span** with the caller's span as its parent - the engines' exact +OpenTelemetry lineage model, so a chain like *user → engine flow → wrapper +function (agent, MCP tool) → engine* stays one connected trace tree: ```python from mercury_composable import annotate_trace, get_trace -info = get_trace() # trace_id, trace_path, cid - or None -annotate_trace("model", "v3") # rides back on the reply envelope +info = get_trace() # trace_id, trace_path, cid, my_correlation_id, + # span_id, parent_span_id - or None +annotate_trace("model", "v3") # rides back on the reply envelope AND the trace record ``` -Outside a hosted function (batch jobs, tests), establish context explicitly: +Outbound calls carry the current span (the receiver's parent), the business +correlation-id (`my_cid` tag), and a W3C `traceparent` header when the trace id +is W3C-shaped. Non-RPC executions emit the engines' distributed-trace dataset +on the `distributed.tracing` log stream - the same +`{"trace": {...}, "annotations": {...}}` record the Java engine logs - so a +stdout log-ingest agent (Dynatrace-style) or any log aggregation stitches the +span tree across all four runtimes. RPC round-trips fold into the caller's +view, exactly like the engines. + +**Application log context**: with `log.format` json/compact, every log line a +function writes inside a traced request carries a `context` block (the +engines' app-log-context feature, on by default) - the standard trace context +(`cid` = the business correlation-id, `traceId`, `tracePath`, `spanId`, +`parentSpanId`, `service`, `timestamp`) - so application logs and the +distributed-trace records correlate in one aggregation. Customize with your +own `resources/app-log-context.yaml` (`context:` section mapping output keys +to reserved `$tokens` or constants, `${ENV:default}` supported), opt out with +`app.log.context=false`, and add per-request key-values from a handler with +`update_context("tenant", "acme")` (a logging-only sink; reserved keys are +guarded; `None` removes). + +Outside a hosted function (batch jobs, tests), establish context explicitly - +including an external OpenTelemetry span to parent onto: ```python from mercury_composable import trace_context -with trace_context("trace-1", "BATCH /nightly", cid="order-42"): +with trace_context("4bf92f3577b34da6a3ce929d0e0e4736", "BATCH /nightly", + cid="order-42", span_id="00f067aa0ba902b7"): reply = await po.request("my.function", body={...}) ``` diff --git a/docs/guides/http-surface-reference.md b/docs/guides/http-surface-reference.md index b08cff8..836a5b5 100644 --- a/docs/guides/http-surface-reference.md +++ b/docs/guides/http-surface-reference.md @@ -22,7 +22,8 @@ Mirrors the engines' `event.api.service`: | Reply | always envelope bytes, `content-type: application/octet-stream` | | Handler outcome | rides **HTTP 200** with the status inside the envelope (including AppException and unexpected errors) | | Transport failures | set the HTTP status too: 400 undecodable / missing route field, 403 private target, 404 unknown route (`Route X not found`), 408 timeout (`Timeout for N ms`) | -| Header hygiene | inbound `x-event-api` and `my_*` removed; the `my_cid` tag becomes the read-only `my_correlation_id` header | +| Header hygiene | inbound `x-event-api` and `my_*` removed; the `my_cid` tag becomes the read-only `my_correlation_id` header (local bus deliveries inject the same view). Outbound, the client stamps the current context's business correlation-id back onto the event as the `my_cid` tag — the engines' PostOffice parity, so the business correlation-id continues across every hop | +| `accept: text/event-stream` | streaming-capable call to an interceptor target: a streamed reply rides the same call as SSE in the envelope-mode dialect; a single-shot reply stays byte-identical; a streaming reply to a NON-accepting caller → 406 `Streaming function requires a caller that accepts text/event-stream`. See [Event Streaming](streaming.md) | ## Actuator endpoints diff --git a/docs/guides/streaming.md b/docs/guides/streaming.md new file mode 100644 index 0000000..d91d28b --- /dev/null +++ b/docs/guides/streaming.md @@ -0,0 +1,128 @@ +# Event Streaming + +A function that produces its result progressively — an LLM relay emitting tokens, a +long-running job reporting progress — should not make its caller wait for the whole +answer. This chapter is the wrapper's half of the platform-wide streaming contract: +the same paradigm on all four runtimes (Java, Rust, Python, Node.js): + +> **The caller provides a reply address; the callee streams events to it until a +> terminal signal.** + +Each segment is one event to the caller's `reply_to`, marked with the reserved +envelope header `x-event-stream: data | eof | exception`. A calling engine renders +the segments out its HTTP edge, hands them to a flow, or relays them onward — your +Python function neither knows nor cares. + +## Write a streaming function + +A streaming producer is an **interceptor**: it receives the raw `EventEnvelope` +(so the caller's reply address travels the engines' way) and replies through +`EventStreamWriter` instead of a return value: + +```python +from mercury_composable import EventEnvelope, EventStreamWriter, preload + +@preload(route="hello.tokens", instances=10, interceptor=True) +async def stream_tokens(headers: dict[str, str], event: EventEnvelope): + out = EventStreamWriter.from_request(event) + out.first(200, "text/event-stream") # head control rides the first event + out.write("The answer is") # data segment + out.write_named("tokens", {"n": 2}) # named (typed) SSE event + out.close({"usage": {"tokens": 2}}) # end of transmission + trailing metadata + # or out.fail(e) # in-band failure +``` + +The writer is the engines' exact API. `first(status, content_type, ttl_seconds=None)` +declares the response head and, optionally, the idle allowance between segments; +`fail(e)` carries the standard error key-values +`'{"type": "error", "status": n, "message": text}'`; writes after `close()`/`fail()` +are dropped. Plain-`def` handlers can stream too — the writer bridges from the +executor thread back to the host loop. + +An interceptor's return value is never auto-replied. To answer single-shot from an +interceptor (a relay that sometimes buffers, for example), send a plain envelope to +`event.reply_to` yourself. An uncaught exception becomes the standard error envelope +to the caller — single-shot before the stream starts, in-band after. + +## How it crosses the wire + +When a calling engine (or `curl`) invokes your streaming function through +`POST /api/event` with `Accept: text/event-stream`, the host answers the same call +with a Server-Sent Events response in the platform's hybrid dialect: + +- **envelope frames** — the reserved SSE event name `envelope`, one base64-encoded + serialized envelope per frame — carry everything with envelope semantics: the + first event (head control), the `eof`/`exception` terminals, and any segment that + cannot round-trip as plain text (a dict or bytes body, text containing a carriage + return, an event name colliding with the reserved word); +- **raw SSE frames** carry plain text segments, so token relays stay near-zero + overhead. + +Everything degrades explicitly: a caller that did not opt in receives +`406 Streaming function requires a caller that accepts text/event-stream` instead of +a truncated reply; a non-streaming (single-shot) answer over the capable path is +byte-identical to a normal RPC reply; idle expiry fails the stream in-band with the +standard 408 error body. The `x-ttl` request header (ms) is the idle allowance +between segments — your `first(..., ttl_seconds=...)` can extend it for the whole +stream. While the producer is quiet, the host emits `: ping` keep-alive comments +(`event.stream.keep.alive`, the engines' config key — default 30s, `0` disables). + +## Consume a stream + +`PostOffice.stream()` is the consumer surface — an async iterator yielding the same +decoded envelopes an engine reply route receives: `data` segments, then the terminal. +It works against a remote peer's `/api/event` (an engine or another function host) +and against local functions alike, and opting in is always safe — a non-streaming +target simply yields its one classic reply: + +```python +from mercury_composable import PostOffice + +async with PostOffice() as po: + async for segment in po.stream("hello.tokens", None, + endpoint="http://127.0.0.1:8100/api/event", + timeout_ms=30000): + marker = segment.headers.get("x-event-stream") + if marker == "data": + print(segment.body) + elif marker == "exception": + raise RuntimeError(segment.body["message"]) + # eof: segment.body carries the trailing metadata, if any +``` + +`timeout_ms` is the idle allowance between segments. The consumer guards the dialect +for you: a malformed frame, a stream that ends without a terminal, or idle expiry +each yield the standard in-band exception envelope, then the iterator ends. + +## Compose a relay + +The pattern the whole streaming program is built on: forward **your own caller's** +reply address into a call against a remote streaming function, and the segments flow +`engine → your function → remote peer → back to the original caller` with no +buffering anywhere: + +```python +@preload(route="llm.relay", instances=10, interceptor=True) +async def relay(headers: dict[str, str], event: EventEnvelope): + async with PostOffice() as po: + await po.stream_to("remote.tokens", None, + reply_to=event.reply_to or "", + endpoint="http://peer:8085/api/event", + cid=event.cid, timeout_ms=30000) +``` + +`stream_to()` forwards every decoded envelope verbatim to the named LOCAL route +(here, the reply sink the host opened for your caller) and returns the terminal. +Combined with a calling engine's `stream: true` endpoint, this streams a remote +peer's tokens progressively out that engine's HTTP edge — with zero imperative +streaming code in between. + +## See also + +- The engines' HTTP Response Streaming guides (the same contract at the HTTP edge): + [Java](https://accenture.github.io/mercury-composable/guides/http-streaming/) · + [Rust](https://accenture.github.io/mercury/guides/http-streaming/) +- [Interop Test Report — Progressive Rendering](../test-reports/progressive-rendering-interop.md) — + the live four-runtime validation of this contract +- [HTTP Surface Reference](http-surface-reference.md) — the `/api/event` contract +- [Function Writing Patterns](function-patterns.md) diff --git a/docs/test-reports/progressive-rendering-interop.md b/docs/test-reports/progressive-rendering-interop.md new file mode 100644 index 0000000..4914920 --- /dev/null +++ b/docs/test-reports/progressive-rendering-interop.md @@ -0,0 +1,371 @@ +--- +title: Interop Test Report — Progressive Rendering, all four runtimes +summary: Permanent record of the live cross-runtime validation of the progressive + streaming contract - the multi-shot reply protocol and the Event-over-HTTP + envelope-mode SSE dialect - across the Java and Rust engines and the Python and + Node.js function hosts. +layer: reference +audience: [developer, architect] +keywords: [interop, streaming, sse, event over http, envelope mode, test report] +--- + +# Interop Test Report — Progressive Rendering, all four runtimes + +*Live cross-runtime validation of the progressive streaming contract across the four +Mercury Composable runtimes — the Java engine +([mercury-composable](https://github.com/Accenture/mercury-composable)), the Rust engine +([mercury](https://github.com/Accenture/mercury)), and the Python +([mercury-python](https://github.com/Accenture/mercury-python)) and Node.js +([mercury-nodejs](https://github.com/Accenture/mercury-nodejs)) function hosts — conducted +2026-08-30 (UTC) at the close of the streaming program's wrapper round. This report is a +permanent record in the tradition of the +[Event over HTTP interop report](https://accenture.github.io/mercury-composable/test-reports/event-over-http-interop/): what was tested, the +evidence, and the defects the round surfaced with their fixes.* + +## The contract under test + +One paradigm on all four runtimes: **the caller provides a reply address; the callee +streams events to it until a terminal signal.** Each segment is one event to the +caller's `reply_to`, marked with the reserved envelope header +`x-event-stream: data | eof | exception`. Across the Event-over-HTTP hop, the peer +answers the one POST with a Server-Sent Events response in the **hybrid envelope-mode +dialect**: envelope frames (the reserved SSE event name `envelope`, one base64-encoded +serialized envelope per frame) wherever envelope semantics matter — the head, the +terminals, and any segment that cannot round-trip as plain text — and raw SSE frames +for text tokens. The consuming client decodes the dialect and forwards each event to +the original reply address with the original correlation id, so a local stream and a +remote stream are indistinguishable to the consumer. + +Delivered by: Java engine PRs +[#299](https://github.com/Accenture/mercury-composable/pull/299) (edge streaming), +[#300](https://github.com/Accenture/mercury-composable/pull/300) (SSE consumption), +[#301](https://github.com/Accenture/mercury-composable/pull/301) (envelope mode) — +ADR-0018/0019; Rust engine PRs +[#216](https://github.com/Accenture/mercury/pull/216), +[#217](https://github.com/Accenture/mercury/pull/217), +[#218](https://github.com/Accenture/mercury/pull/218) — ADR-0015/0016; and the +matching wrapper round in mercury-python and mercury-nodejs (interceptor functions, +`EventStreamWriter`, streaming `/api/event` host, `stream()`/`stream_to()` client). + +## The live matrix + +Ten producer/consumer combinations were driven live with the SHIPPED demo applications +— no test shims. Producers pace their segments (300 ms in these drives), so +progressive delivery is directly observable in the arrival timestamps; buffered +delivery would show all segments arriving together. + +| # | Consumer | Producer | Result | +|---|----------|----------|--------| +| 1 | Node.js client | Python `hello.tokens` | segments at ~11/312/612/916 ms, eof metadata `{count, language: python}` | +| 2 | Python client | Node.js `hello.tokens` | segments at ~5/304/606/908 ms, eof metadata `{count, language: node.js}` | +| 3 | Python client | **Java engine** `hello.sse` | segments at ~103/407/708/1012 ms, terminal `eof` | +| 4 | Node.js client | **Java engine** `hello.sse` | segments at ~13/320/620/924 ms, terminal `eof` | +| 5 | Python client | **Rust engine** `hello.sse` | segments at ~3/306/607 ms, terminal `eof` | +| 6 | Node.js client | **Rust engine** `hello.sse` | segments at ~10/312/615 ms, terminal `eof` | +| 7 | **Java engine** edge (`/api/hello/remote`) | Python `hello.tokens` | SSE out the engine edge at ~195/493/793/1094 ms, terminal `event: done` with the Python eof metadata | +| 8 | **Java engine** edge | Node.js `hello.tokens` | ~217/516/817 ms, terminal metadata `{count, language: node.js}` | +| 9 | **Rust engine** edge | Python `hello.tokens` | ~20/323/624 ms, same shape | +| 10 | **Rust engine** edge | Node.js `hello.tokens` | ~19/320/622 ms, same shape | + +Common observations across all ten: + +- **Progressive on the wire** — arrival gaps match the producer's pacing; nothing + buffers end-to-end (rows 7–10 traverse the full chain: HTTP edge → engine relay + function → Event-over-HTTP → wrapper host → wrapper function and back). +- **Correlation** — the caller's correlation id is restored on every delivered + envelope (D7), and the engines' demo authentication (`authorization: demo`) rides + the per-target security headers unchanged. +- **Exact types** — eof trailing metadata arrives as a real map, not text, on every + path (the envelope-frame escape hatch). +- **Explicit degradation, observed live** — a missing token produced the engine's + 401 and a private target its 403, both delivered as clean envelopes through the + buffered fallback; a caller without the `accept: text/event-stream` opt-in receives + the pinned `406 Streaming function requires a caller that accepts + text/event-stream`. + +## Engine ⇄ engine and per-runtime coverage + +Because the protocol signatures are identical across the four runtimes, each +repository's unit suite exercises the full protocol against its own application +instance (client consuming its own `/api/event` host in one process): Java +`EventOverHttpStreamTest` (14 cases), Rust `event_over_http_stream.rs` (14), Python +`test_event_stream.py` (17), Node.js `event-stream.test.ts` (17). Each suite also +carries **misbehaving-peer fixtures** — a raw first frame, a transport end without a +decoded terminal, trailing frames after the terminal — because a self-loop alone +cannot catch a deviation implemented identically in both halves of one runtime; the +live matrix above is the cross-implementation conformance check, and it passed with +zero shims. + +## Trace continuity (OpenTelemetry span / parent-span verification) + +The drives were repeated with telemetry capture to verify distributed-trace +continuity across the streaming hop. Findings, with the observed evidence: + +- **One trace id end to end, both directions.** In the engine→wrapper drive, the + Java edge minted trace `f22af2e005f2445198563e2dc4f1ba54`; every engine span + carried it (the relay function, the HTTP client leg, each reply-lane segment + delivery), the Python function received it (trace id and path ride inside the + wire envelope), and the demo now echoes it in the eof trailing metadata - so + the terminal frame rendered out the engine's own edge carries the same trace + id the edge minted: continuity is self-documenting in the demo output. In the + wrapper→engine drive, a Python caller supplied W3C trace id + `4bf92f3577b34da6a3ce929d0e0e4736` with trace path `PY /stream-drive`; the + engine's `event.api.auth`, `event.api.service` and `hello.sse` spans all + carried that id, and the function's span carried the caller's trace path. +- **Span parenting chains on the engines.** The streaming target's span parents + onto the Event API service span - observed live: + `event.api.service span_id=bfcddb32dd597ddb` → + `hello.sse parent_span_id=bfcddb32dd597ddb`. Outbound, the engines' relay leg + sends the W3C `traceparent` header carrying the sending function's span id + (the trace-aware PostOffice stamps the span onto the outbound event), so a + receiving ENGINE parents its spans onto the caller's - the same cross-engine + parenting validated in the Event over HTTP interop report. +- **Wrapper executions are real spans** (closed at this round - the wrappers + originally propagated the trace id but minted no spans, which broke the + lineage into disconnected segments at every wrapper hop). The function hosts + now implement the engines' exact span model: every traced execution mints a + 16-hex span with the caller's span (from the inbound envelope) as its + parent; outbound calls and stream segments carry the current span onward; + and non-RPC executions emit the engines' distributed-trace dataset record on + the `distributed.tracing` log stream - the same + `{"trace": {...}, "annotations": {...}}` shape the Java engine logs - so one + log aggregation (or a stdout log-ingest agent forwarding to an observability + dashboard) stitches the full span tree across all four runtimes. RPC + round-trips emit no dataset, exactly like the engines: their metrics fold + into the caller's view. +- **The connected tree, live-proven in all three directions.** Engine→wrapper: + under one Java-edge trace, the relay function's span `8cbc0b4d4be75362` + became the Python `hello.tokens` span's parent (`span_id=7b90847f4b8b617a`, + `parent_span_id=8cbc0b4d4be75362` in the wrapper's own trace record), and + the Java reply-lane delivery spans then parented onto the Python span - + edge → engine function → wrapper function → engine deliveries, one unbroken + chain. Wrapper→engine: a Python caller carrying external span + `00f067aa0ba902b7` (the shape of a user-edge OpenTelemetry span) produced + `event.api.auth` and `event.api.service` spans parented on it, with + `hello.sse` parented on the service span. Wrapper⇄wrapper: a Node.js + execution's record showed the Python caller's span as its parent. This is + the lineage the AI SDLC requires - user → agent → MCP → tools in one tree. +- **One deliberate exception**: plain-text token segments ride raw SSE frames, + which carry no envelope metadata (the zero-overhead token path), so their + engine-side delivery spans join the trace unparented; a stream's head and + terminal segments ride envelope frames and parent correctly. The engines' + HTTP client leg (`async.http.request`) remains a sibling span - classic + Event-over-HTTP parity. +- **Correlation id**: the caller's cid rode every delivered envelope in every + drive (`cid-trace-drive` on all four segments of the supplied-trace drive). + +The check itself surfaced defect #5: the engine demo relay originally built its +forward event with the raw EventEmitter, which stamps no trace - the hop +continued cid but DROPPED the trace id. Fixed by using a trace-aware +PostOffice (its `touch` fill-stamps from/trace/span), and the wrapper demos' +`hello.tokens` now echo their received trace id in the eof metadata so the +continuity is visible in every future drive. + +## Business correlation-id continuity + +The same verification was run for the business correlation-id (`my_cid`) - the +engine-managed envelope tag captured at an engine's HTTP edge from the +configured header (default `X-Correlation-Id`) and injected into every +receiving function's input headers as the read-only `my_correlation_id` view. +The wrapper demos' `hello.tokens` echo the view in the eof metadata alongside +the trace id, so every future drive self-documents both continuity dimensions. +Live results: + +- **Java engine edge → Python**: `X-Correlation-Id: biz-e2w-001` on the edge + request came back in the terminal metadata rendered out the same edge - + edge header → `my_cid` tag → relay function's injected header view → + trace-aware PostOffice re-stamp → packed envelope over the hop → wrapper + host injection → function echo. +- **Rust engine edge → Python**: `biz-rust-005` echoed identically. The Rust + demo relay needed no change: the Rust engine has one PostOffice and + `apply_current_trace` always stamps trace and business correlation-id from + the ambient context. +- **Python → Node.js and Node.js → Python** (wrapper ⇄ wrapper): a caller-side + context (`trace_context(..., my_correlation_id=...)` / `runWithTrace({..., + myCorrelationId})`) produced `biz-w2w-002` / `biz-w2w-003` echoes from the + opposite wrapper - the tag crossed the hop and the receiving host injected + the view. +- **Wrapper → engine**: the wrappers stamp the identical tag bytes (the same + codec proven above), and the engines' pinned suites cover the receiving + half: an `/api/event` caller carrying the `my_cid` tag reaches the target + function as `po.getMyCorrelationId()` (Java + `EventHttpTest.eventOverHttpPropagatesTraceAndCorrelationId`; Rust twin). + +The check surfaced one parity gap in the (unreleased) wrapper round, fixed and +test-pinned in both wrappers: the wrapper clients inherited trace id/path and +the internal correlation id into outbound events but did not re-stamp the +business correlation-id as the `my_cid` tag, and local bus deliveries skipped +the header-view injection that the HTTP host performs. Both halves now mirror +the engines (`PostOffice.touch` / WorkerHandler parity): outbound events carry +the context's business correlation-id as the tag, local deliveries inject the +read-only view, and the trace context (`get_trace()` / `getTrace()`) exposes +it to handlers and relays. + +By design (confirmed at this verification round): an engine's `/api/event` +ingress serves LOCAL routes only - it answers 404 for a route it does not +host, even when its own `yaml.event.over.http` map points that route at a +peer. The map is caller-side routing for the app's own outbound calls; +forwarding inbound calls onward would make every application an +Event-over-HTTP relay and open routing loops (the `x-event-api` wire marker +exists precisely to prevent such re-forwarding). Callers address the owning +peer directly; deliberate hop-through composition is an explicit relay +function - the demo `hello.remote.relay` is exactly that pattern. + +## Defects surfaced by the round (fixed and re-verified) + +Honest engineering record — each was caught by the twin-building discipline before +any release: + +1. **Rust engine**: the envelope-mode single-shot reply initially reached the caller + unwrapped (rendered as plain JSON instead of the classic packed-envelope wire); + fixed by wrapping at the one `SingleShot` outcome site — caught by the twin suite's + first run. +2. **Node.js host**: a `Promise.race` against a queue waiter abandoned the losing + waiter, which would steal and drop the next envelope; fixed by reusing one pending + promise across keep-alive cycles (`raceMs` documents the rule) — caught in design + review before the first test run. +3. **Error-body contract**: the in-band exception bodies converged on the standard + error key-values `type` / `status` / `message` across all four runtimes (three + Java sites and their twins were missing the `type` key or used a plain-text body). +4. **Node.js**: object error bodies would have stringified as `[object Object]`; + fixed with JSON rendering (`errorText`). +5. **Java demo relay**: the forward event was built with the raw (untraced) + EventEmitter, silently dropping the distributed trace across the hop; fixed + with a trace-aware PostOffice - found by the trace-continuity verification + above. + +## Reproduce + +Every row of the matrix uses shipped demo applications: + +- **Engine as producer**: run the Java `lambda-example` (port 8085) or Rust + `hello-world` (8085; any port with `-Drest.server.port=…`); their public + `hello.sse` streams paced test messages. Consume from a wrapper with + `PostOffice.stream("hello.sse", …, endpoint="http://host:port/api/event")` and the + demo `authorization: demo` header. +- **Engine as consumer**: run a wrapper demo app (python `mercury-serve + examples/demo_app.py`, port 8086; node demo, 8087), then the engine demo with its + routing map (Java: `-Dyaml.event.over.http=classpath:/event-over-http.yaml`; Rust: + ships enabled) and `-Dpeer.demo.port` as needed, and watch + `curl -N -H 'accept: text/event-stream' + 'http://127.0.0.1:8085/api/hello/remote?delay=300&count=3'` render the wrapper's + tokens progressively out the engine's edge. +- **Wrapper ⇄ wrapper**: point either wrapper's `stream()` at the other demo's + `/api/event`. + +Baselines at the time of the drives: Java engine main after PR #301, Rust engine main +after PR #218, mercury-python and mercury-nodejs at the streaming feature round - +shipped together as the v4.12.0 milestone release across all four repositories. + +## Appendix - telemetry and app context example (live capture) + +One live request, captured end to end, showing how the telemetry stream and +the application log context connect the engine to the wrapper. Setup: the Java +`lambda-example` (port 8085) with its event-over-http map pointing +`hello.tokens` at the Python demo app (port 8086, `-Dlog.format=compact`). +The caller supplies a business correlation-id; the engine's HTTP edge mints +the trace: + +```bash +curl -N -H 'accept: text/event-stream' -H 'X-Correlation-Id: biz-e2e-777' \ + 'http://127.0.0.1:8085/api/hello/remote?delay=100&count=1' +``` + +**1. Java engine - the relay function's telemetry record** (the root span of +trace `710dcb0b706e4b5b949be138d0992b0b`, minted at the edge): + +```json +{ + "level": "INFO", + "time": "2026-08-29 21:24:17.734", + "source": "org.platformlambda.core.services.Telemetry.handleEvent(Telemetry.java:81)", + "thread": 762, + "message": { + "trace": { + "path": "GET /api/hello/remote?delay=100&count=1", + "span_id": "afa2897918e3871b", + "service": "hello.remote.relay", + "success": true, + "origin": "20260830a0dcb550832b4683b4853c609778cd82", + "start": "2026-08-30T04:24:17.733Z", + "exec_time": 0.728, + "from": "http.request", + "id": "710dcb0b706e4b5b949be138d0992b0b", + "status": 200 + } + } +} +``` + +**2. The wire** - the relay's trace-aware PostOffice stamps the outbound event +with the trace id and path, its own span id (`afa2897918e3871b`) and the +`my_cid` tag (`biz-e2e-777`); the whole envelope crosses `POST /api/event`, +with `X-Trace-Id` and the W3C `traceparent` on the HTTP headers. + +**3. Python wrapper - the function's own application log line.** The handler +runs `log.info("Streaming %d messages", count)`; the app-log-context feature +adds the `context` block. Note the join keys: the engine's trace id, the +business `cid` from the edge header, and this execution's `spanId` with the +relay's span as `parentSpanId`: + +```json +{"time": "2026-08-29 21:24:17.847", "level": "INFO", "logger": "mercury_user_app:93", "message": "Streaming 1 messages", "context": {"cid": "biz-e2e-777", "traceId": "710dcb0b706e4b5b949be138d0992b0b", "tracePath": "GET /api/hello/remote?delay=100&count=1", "spanId": "55ebd6464ffdf9c2", "parentSpanId": "afa2897918e3871b", "service": "hello.tokens", "timestamp": "2026-08-30T04:24:17.847Z"}} +``` + +**4. Python wrapper - the telemetry record for the same execution** (same +`span_id` as the app log line's `spanId` - the join key between the two +streams; `from` names the calling engine function): + +```json +{"time": "2026-08-29 21:24:17.949", "level": "INFO", "logger": "distributed.tracing:158", "message": {"trace": {"origin": "202608303cbae673831a4ab08cbb6f108fec0171", "id": "710dcb0b706e4b5b949be138d0992b0b", "path": "GET /api/hello/remote?delay=100&count=1", "service": "hello.tokens", "start": "2026-08-30T04:24:17.847Z", "success": true, "from": "hello.remote.relay", "exec_time": 101.673, "status": 200, "span_id": "55ebd6464ffdf9c2", "parent_span_id": "afa2897918e3871b"}}} +``` + +**5. Java engine - a reply-lane delivery record.** The wrapper's stream +segments carry its span, so the engine-side delivery span parents onto the +Python function (`parent_span_id = 55ebd6464ffdf9c2`): + +```json +{ + "level": "INFO", + "time": "2026-08-29 21:24:17.853", + "source": "org.platformlambda.core.services.Telemetry.handleEvent(Telemetry.java:81)", + "thread": 779, + "message": { + "trace": { + "path": "GET /api/hello/remote?delay=100&count=1", + "parent_span_id": "55ebd6464ffdf9c2", + "span_id": "a50c19b7415406df", + "service": "async.http.response.stream.499", + "success": true, + "origin": "20260830a0dcb550832b4683b4853c609778cd82", + "start": "2026-08-30T04:24:17.851Z", + "exec_time": 1.202, + "from": "hello.tokens", + "id": "710dcb0b706e4b5b949be138d0992b0b", + "status": 200 + } + } +} +``` + +**6. The caller's view** - the terminal SSE frame rendered back out the Java +edge echoes both continuity dimensions: + +```text +event: done +data: {"trace_id":"710dcb0b706e4b5b949be138d0992b0b","count":1,"my_correlation_id":"biz-e2e-777","language":"python"} +``` + +**Connectivity, in one paragraph.** Every record above carries the one trace +id the engine minted at its HTTP edge, and the span ids chain across the +runtime boundary in both directions: the edge request → the relay function's +span (`afa289...`) → the Python execution's span (`55ebd6...`, parented on the +relay) → the engine's reply-lane delivery spans (parented on the Python span). +The business correlation-id from the `X-Correlation-Id` edge header rides the +`my_cid` envelope tag through every hop and surfaces as the `cid` in the +wrapper's app-log context and in the demo's terminal metadata. The wrapper's +application log line and its telemetry record share the same `span_id`, which +is what lets a log aggregation (or a stdout log-ingest agent) attach a +function's own log output to the exact span of the distributed trace - the +complete telemetry and app-context story from user to engine to wrapper and +back, assembled entirely from the four runtimes' stdout logs. diff --git a/examples/demo_app.py b/examples/demo_app.py index b8463d5..4b77028 100644 --- a/examples/demo_app.py +++ b/examples/demo_app.py @@ -14,12 +14,17 @@ target: 'http://127.0.0.1:8086/api/event' """ +import asyncio + from mercury_composable import ( AppException, Body, + EventEnvelope, + EventStreamWriter, PostOffice, annotate_trace, get_logger, + get_trace, platform, preload, ) @@ -71,6 +76,36 @@ def sync_chain(_headers: dict[str, str], body: Body): return reply.body +@preload(route="hello.tokens", instances=10, interceptor=True) +async def stream_tokens(headers: dict[str, str], event: EventEnvelope): + """Streaming demo: paced test messages over the multi-shot reply contract. + + A calling engine consumes this progressively through Event-over-HTTP + (accept: text/event-stream on the outbound event) and can render it out + its own HTTP edge - engine-to-wrapper token streaming. Optional headers: + "delay" ms between messages (default 500, clamped 50-5000) and "count" + messages (default 5, clamped 1-100). + """ + delay = min(5000, max(50, int(headers.get("delay", "500") or 500))) / 1000 + count = min(100, max(1, int(headers.get("count", "5") or 5))) + # with log.format=json/compact, this line carries the application log + # "context" block (trace ids, business cid) - see the streaming guide + log.info("Streaming %d messages", count) + out = EventStreamWriter.from_request(event) + out.first(200, "text/event-stream") + out.write("The following messages are rendered slowly to demonstrate streaming:") + for n in range(1, count + 1): + await asyncio.sleep(delay) + out.write(f"test message {n} (python)") + # the trailing metadata echoes the distributed trace id and the business + # correlation-id, so a calling engine's edge shows both continuity + # dimensions end to end + info = get_trace() + out.close({"count": count, "language": "python", + "trace_id": info.trace_id if info else None, + "my_correlation_id": headers.get("my_correlation_id")}) + + @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). diff --git a/mkdocs.yml b/mkdocs.yml index 24b46cd..a4ef195 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -109,6 +109,7 @@ nav: - Write functions: - Function Writing Patterns: guides/function-patterns.md + - Event Streaming: guides/streaming.md - Configuration, Logging & Actuators: guides/config-logging-actuators.md - Testing Your Functions: guides/testing.md @@ -122,5 +123,6 @@ nav: - Reference: - Configuration Reference: guides/configuration-reference.md - HTTP Surface Reference: guides/http-surface-reference.md + - Interop Test Report — Progressive Rendering: test-reports/progressive-rendering-interop.md - Release Notes: https://github.com/Accenture/mercury-python/blob/main/CHANGELOG.md - Contributing: https://github.com/Accenture/mercury-python/blob/main/CONTRIBUTING.md diff --git a/pyproject.toml b/pyproject.toml index 17ff60f..92f252b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "mercury-composable" -version = "0.1.0" +version = "4.12.0" description = "Lightweight Event-over-HTTP function host and client for Mercury Composable" readme = "README.md" license = { file = "LICENSE.txt" } diff --git a/src/mercury_composable/__init__.py b/src/mercury_composable/__init__.py index d192fbb..e8140c0 100644 --- a/src/mercury_composable/__init__.py +++ b/src/mercury_composable/__init__.py @@ -12,11 +12,12 @@ from .client import PostOffice from .config import AppConfig, app_config, load_config from .envelope import Body, EventEnvelope, iso_utc +from .event_stream import EventStreamWriter from .exceptions import AppException, CompactFormatError from .log import get_logger from .registry import FunctionRegistry, Handler, default_registry, preload from .server import EventApiServer, Platform, platform -from .trace import TraceInfo, annotate_trace, get_trace, trace_context +from .trace import TraceInfo, annotate_trace, get_trace, trace_context, update_context from .version import __version__ __all__ = [ @@ -26,6 +27,7 @@ "CompactFormatError", "EventApiServer", "EventEnvelope", + "EventStreamWriter", "FunctionRegistry", "Handler", "Platform", @@ -42,4 +44,5 @@ "platform", "preload", "trace_context", + "update_context", ] diff --git a/src/mercury_composable/bus.py b/src/mercury_composable/bus.py index 904e52c..a758d82 100644 --- a/src/mercury_composable/bus.py +++ b/src/mercury_composable/bus.py @@ -27,20 +27,31 @@ import asyncio import contextvars +import secrets import time import traceback +import uuid from dataclasses import dataclass -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, cast from .envelope import EventEnvelope, iso_utc from .exceptions import AppException from .log import get_logger -from .trace import TraceInfo, _reset_trace, _set_trace +from .trace import ( + MY_CID_TAG, + MY_CORRELATION_ID, + RPC_TAG, + TraceInfo, + _reset_trace, + _set_trace, +) if TYPE_CHECKING: - from .registry import ServiceDef + from .registry import Handler, InterceptorHandler, ServiceDef log = get_logger("mercury.bus") +# the engines' distributed-trace log stream (Java Telemetry parity) +telemetry_log = get_logger("distributed.tracing") # The loop hosting the current sync handler - stamped by _execute before the # handler is dispatched to the executor thread (copy_context carries it), so @@ -77,6 +88,74 @@ class _Delivery: trace_path: str | None cid: str | None reply: asyncio.Future[EventEnvelope] | None # drop-n-forget deliveries carry no reply future + # the raw envelope, for interceptor handlers (they receive it verbatim - + # reply_to and correlation id travel the engines' way) + envelope: EventEnvelope | None = None + + +def _business_cid(delivery: _Delivery) -> str | None: + """The caller's business correlation-id at delivery: the engine-managed + my_cid envelope tag, else a my_correlation_id view already injected by an + HTTP host (the engines' WorkerHandler resolution order).""" + tag = delivery.envelope.tags.get(MY_CID_TAG) if delivery.envelope else None + return tag or delivery.headers.get(MY_CORRELATION_ID) + + +def _headers_view(delivery: _Delivery, my_cid: str | None) -> dict[str, str]: + """The handler's header view, with the read-only business correlation-id + injected at delivery (engine parity).""" + if my_cid and MY_CORRELATION_ID not in delivery.headers: + return {**delivery.headers, MY_CORRELATION_ID: my_cid} + return delivery.headers + + +def _trace_info(delivery: _Delivery, my_cid: str | None) -> TraceInfo: + """The execution's trace context. Under a trace, every execution mints its + own 16-hex span and records the caller's span (from the inbound envelope) + as its parent - the engines' WorkerHandler model.""" + span_id = secrets.token_hex(8) if delivery.trace_id else None + parent = delivery.envelope.span_id if delivery.envelope else None + return TraceInfo(route=delivery.service.route, + trace_id=delivery.trace_id, trace_path=delivery.trace_path, + cid=delivery.cid, my_correlation_id=my_cid, + span_id=span_id, parent_span_id=parent) + + +def _is_rpc(delivery: _Delivery) -> bool: + """True for an RPC round-trip: a local reply future, or the engines' rpc + envelope tag transported over the wire. RPC legs emit no trace dataset + (engine parity) - their metrics fold into the caller's view.""" + if delivery.reply is not None: + return True + return bool(delivery.envelope and delivery.envelope.tags.get(RPC_TAG)) + + +def _emit_trace(delivery: _Delivery, info: TraceInfo, start: str, + exec_time: float, status: int, success: bool, + exception: str | None) -> None: + """Emit the engines' distributed-trace dataset for a traced, non-RPC + execution - the same record shape the Java reference engine logs + (message = {"trace": {...}, "annotations": {...}}), so polyglot log + aggregation stitches spans across all runtimes.""" + if not info.trace_id or _is_rpc(delivery): + return + from .actuator import app_origin # late: actuator imports the registry chain + trace: dict[str, Any] = { + "origin": app_origin(), "id": info.trace_id, "path": info.trace_path, + "service": delivery.service.route, "start": start, "success": success, + "from": (delivery.envelope.sender if delivery.envelope else None) or "unknown", + "exec_time": exec_time, "status": status, + } + if not success and exception: + trace["exception"] = exception + if info.span_id: + trace["span_id"] = info.span_id + if info.parent_span_id: + trace["parent_span_id"] = info.parent_span_id + dataset: dict[str, Any] = {"trace": trace} + if info.annotations: + dataset["annotations"] = dict(info.annotations) + telemetry_log.info(dataset) class EventBus: @@ -85,6 +164,35 @@ class EventBus: def __init__(self) -> None: self._mailboxes: dict[str, asyncio.Queue[_Delivery]] = {} self._workers: dict[str, list[asyncio.Task[None]]] = {} + # per-request reply sinks (the engines' inbox idea): generated local + # route names backed by queues - the reply_to addressing of interceptor + # dispatch and of streaming responses. Local-only by design. + self._sinks: dict[str, asyncio.Queue[EventEnvelope]] = {} + # backref for reply routing (registry constructs and owns this bus) + self._registry: Any = None + + def bind_registry(self, registry: Any) -> None: + self._registry = registry + + def open_sink(self) -> tuple[str, asyncio.Queue[EventEnvelope]]: + """Open a per-request reply sink under a generated local route name.""" + route = f"inbox.{uuid.uuid4().hex}" + queue: asyncio.Queue[EventEnvelope] = asyncio.Queue() + self._sinks[route] = queue + return route, queue + + def close_sink(self, route: str) -> None: + self._sinks.pop(route, None) + + def offer_sink(self, route: str, event: EventEnvelope) -> bool: + """Deliver an envelope to a reply sink; False when the sink is gone + (a completed, timed-out or disconnected request) - late segments are + no-op drops, the engines' semantics.""" + queue = self._sinks.get(route) + if queue is None: + return False + queue.put_nowait(event) + return True def _mailbox(self, service: ServiceDef) -> asyncio.Queue[_Delivery]: mailbox = self._mailboxes.get(service.route) @@ -101,12 +209,18 @@ def _mailbox(self, service: ServiceDef) -> asyncio.Queue[_Delivery]: 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.""" + trace_path: str | None = None, cid: str | None = None, + envelope: EventEnvelope | None = None) -> EventEnvelope: + """RPC: enqueue and await the reply envelope within the ttl. + + ``envelope`` is delivery context only (engine-managed tags such as the + business correlation-id); the handler still receives headers + body. + """ 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)) + trace_id=trace_id, trace_path=trace_path, cid=cid, reply=reply, + envelope=envelope)) try: return await asyncio.wait_for(reply, timeout=max(100, ttl_ms) / 1000) except asyncio.TimeoutError: @@ -114,13 +228,24 @@ async def deliver(self, service: ServiceDef, headers: dict[str, str], body: Any, 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: + cid: str | None = None, + envelope: EventEnvelope | 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)) + trace_id=trace_id, trace_path=trace_path, cid=cid, reply=None, + envelope=envelope)) return async_ack() + def publish_envelope(self, service: ServiceDef, event: EventEnvelope) -> None: + """Route one envelope to a local function (the reply_to mechanism): + drop-n-forget delivery carrying the raw envelope, so an interceptor + handler receives reply_to and the correlation id the engines' way.""" + self._mailbox(service).put_nowait(_Delivery( + service=service, headers=dict(event.headers), body=event.body, + trace_id=event.trace_id, trace_path=event.trace_path, cid=event.cid, + reply=None, envelope=event)) + async def close(self) -> None: """Cancel all workers (tests and orderly shutdown).""" cancelled = [worker for workers in self._workers.values() for worker in workers] @@ -133,6 +258,10 @@ async def close(self) -> None: self._mailboxes.clear() async def _run_worker(self, mailbox: asyncio.Queue[_Delivery]) -> None: + # workers are long-lived tasks created lazily on first use, so they + # inherit the creating task's contextvars - clear the trace so nothing + # from an arbitrary first caller leaks into later executions' logs + _set_trace(None) while True: delivery = await mailbox.get() # dead-work check: the caller of a queued RPC already gave up (408 sent) - @@ -147,18 +276,22 @@ 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) - @staticmethod - async def _execute(delivery: _Delivery) -> EventEnvelope: + async def _execute(self, delivery: _Delivery) -> EventEnvelope: """Run the handler under its trace context and shape the outcome as a reply.""" + if delivery.service.interceptor: + return await self._execute_interceptor(delivery) service = delivery.service - info = TraceInfo(trace_id=delivery.trace_id, trace_path=delivery.trace_path, - cid=delivery.cid) + my_cid = _business_cid(delivery) + headers = _headers_view(delivery, my_cid) + info = _trace_info(delivery, my_cid) token = _set_trace(info) + start_iso = iso_utc() start = time.perf_counter() # noinspection PyBroadException try: + handler = cast("Handler", service.handler) if service.is_async: - result = await service.handler(delivery.headers, delivery.body) + result = await handler(headers, delivery.body) else: # stamp the host loop (for the PostOffice sync bridge), then let # copy_context() carry trace + loop into the executor thread @@ -169,8 +302,7 @@ async def _execute(delivery: _Delivery) -> EventEnvelope: finally: _HOST_LOOP.reset(loop_token) result = await loop.run_in_executor( - None, lambda: context.run(service.handler, delivery.headers, - delivery.body)) + None, lambda: context.run(handler, 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) @@ -187,4 +319,69 @@ async def _execute(delivery: _Delivery) -> EventEnvelope: reply.exec_time = round((time.perf_counter() - start) * 1000, 3) if info.annotations: reply.annotations.update(info.annotations) + _emit_trace(delivery, info, start_iso, reply.exec_time, + reply.get_status(), not reply.has_error(), + str(reply.body) if reply.has_error() else None) return reply + + async def _execute_interceptor(self, delivery: _Delivery) -> EventEnvelope: + """Run an interceptor handler: it receives the raw envelope, replies + manually through reply_to (the engines' @EventInterceptor contract), + and its return value is discarded. An uncaught exception becomes an + error envelope to the delivery's reply_to - so a caller waiting on a + reply sink sees it - and a streaming host renders it in-band.""" + service = delivery.service + event = delivery.envelope or EventEnvelope( + to=service.route, body=delivery.body, headers=dict(delivery.headers)) + my_cid = _business_cid(delivery) + headers = _headers_view(delivery, my_cid) + info = _trace_info(delivery, my_cid) + token = _set_trace(info) + start_iso = iso_utc() + start = time.perf_counter() + error: Exception | None = None + handler = cast("InterceptorHandler", service.handler) + try: + if service.is_async: + await handler(headers, event) + else: + loop = asyncio.get_running_loop() + loop_token = _HOST_LOOP.set(loop) + try: + context = contextvars.copy_context() + finally: + _HOST_LOOP.reset(loop_token) + await loop.run_in_executor( + None, lambda: context.run(handler, headers, event)) + except asyncio.CancelledError: + raise + except Exception as e: # noqa: BLE001 + error = e + self._reply_interceptor_error(service.route, event, e) + finally: + _reset_trace(token) + status = error.status if isinstance(error, AppException) \ + else (500 if error else 200) + _emit_trace(delivery, info, start_iso, + round((time.perf_counter() - start) * 1000, 3), + status, error is None, str(error) if error else None) + # an interceptor's own outcome is never auto-replied + return EventEnvelope() + + def _reply_interceptor_error(self, route: str, event: EventEnvelope, + e: Exception) -> None: + if isinstance(e, AppException): + error = EventEnvelope().set_status(e.status).set_body(e.message) + else: + error = EventEnvelope().set_status(500).set_body(str(e)) + error.stack = traceback.format_exc(limit=20) + error.sender = route + if event.cid: + error.set_correlation_id(event.cid) + reply_to = event.reply_to + delivered = bool( + reply_to and self._registry is not None + and self._registry.send_event(error.set_to(reply_to))) + if not delivered: + log.warning("Interceptor %s ended with status %d - %s", + route, error.get_status(), error.body) diff --git a/src/mercury_composable/client.py b/src/mercury_composable/client.py index 54e751e..cc15ce5 100644 --- a/src/mercury_composable/client.py +++ b/src/mercury_composable/client.py @@ -33,18 +33,33 @@ from __future__ import annotations import asyncio +import base64 import concurrent.futures +import json import re -from collections.abc import Callable, Coroutine +from collections.abc import AsyncIterator, Callable, Coroutine from typing import Any import aiohttp from .bus import DeliveryTimeout, get_host_loop from .envelope import EventEnvelope +from .event_stream import ( + DATA, + ENVELOPE, + EOF, + EXCEPTION, + STREAM_CALLER_REQUIRED, + TEXT_EVENT_STREAM, + X_EVENT_NAME, + X_EVENT_STREAM, + SseParser, + exception_envelope, + stream_signal, +) from .exceptions import AppException from .registry import FunctionRegistry, default_registry -from .trace import _reset_trace, _set_trace, get_trace +from .trace import MY_CID_TAG, RPC_TAG, _reset_trace, _set_trace, get_trace _W3C_TRACE_ID = re.compile(r"^[0-9a-f]{32}$") _W3C_SPAN_ID = re.compile(r"^[0-9a-f]{16}$") @@ -57,11 +72,23 @@ def _build_event(route: str, body: Any, headers: dict[str, str] | None, if from_route: event.set_from(from_route) info = get_trace() + # fill the sender with the executing function's route (touch parity) + if info and info.route and not event.sender: + event.set_from(info.route) if info and info.trace_id: event.set_trace(info.trace_id, info.trace_path or route) effective_cid = cid or (info.cid if info else None) if effective_cid: event.set_correlation_id(effective_cid) + # propagate the business correlation-id to the next touch point as the + # engine-managed my_cid tag (the engines' PostOffice.touch parity) - the + # receiving host injects it as the read-only my_correlation_id header + if info and info.my_correlation_id and MY_CID_TAG not in event.tags: + event.tags[MY_CID_TAG] = info.my_correlation_id + # carry this execution's span so the receiver stores it as its + # parent_span_id (touch parity) - also lights up the traceparent header + if info and info.span_id: + event.set_span_id(info.span_id) return event @@ -124,14 +151,40 @@ async def _call_local(self, route: str, body: Any, headers: dict[str, str] | Non 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) + if not is_async: + # the engines' RPC round-trip marker: an RPC leg emits no trace + # dataset - its metrics fold into the caller's view + event.tags.setdefault(RPC_TAG, str(timeout_ms)) bus = self._registry.bus + if service.interceptor: + if is_async: + bus.publish_envelope(service, event) + from .bus import async_ack + return async_ack() + # RPC to an interceptor: a per-request reply sink is the reply + # address; the first envelope classifies exactly like the engines - + # unmarked = the reply; marked = a streaming target refusing a + # single-shot caller (the pinned 406) + sink_route, queue = bus.open_sink() + try: + bus.publish_envelope(service, event.set_reply_to(sink_route)) + try: + first = await asyncio.wait_for(queue.get(), max(100, timeout_ms) / 1000) + except asyncio.TimeoutError: + return EventEnvelope().set_status(408).set_body( + f"Timeout for {timeout_ms} ms") + if stream_signal(first) is not None: + return EventEnvelope().set_status(406).set_body(STREAM_CALLER_REQUIRED) + return first + finally: + bus.close_sink(sink_route) if is_async: return bus.publish(service, event.headers, event.body, trace_id=event.trace_id, - trace_path=event.trace_path, cid=event.cid) + trace_path=event.trace_path, cid=event.cid, envelope=event) 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) + cid=event.cid, envelope=event) except DeliveryTimeout: return EventEnvelope().set_status(408).set_body(f"Timeout for {timeout_ms} ms") @@ -144,6 +197,9 @@ async def _call(self, route: str, body: Any, headers: dict[str, str] | None, return await self._call_local(route, body, headers, timeout_ms, is_async, from_route, cid) event = _build_event(route, body, headers, from_route, cid) + if not is_async: + # the engines' RPC round-trip marker (see _call_local) + event.tags.setdefault(RPC_TAG, str(timeout_ms)) session = self._get_session() # +100 ms cushion so the HTTP client does not time out before the target client_timeout = aiohttp.ClientTimeout(total=(max(100, timeout_ms) + 100) / 1000) @@ -177,6 +233,130 @@ async def send(self, route: str, body: Any = None, *, return await self._call(route, body, headers, timeout_ms, endpoint, True, from_route, cid) + async def stream(self, route: str, body: Any = None, *, + headers: dict[str, str] | None = None, + timeout_ms: int = 30000, + endpoint: str | None = None, + from_route: str | None = None, + cid: str | None = None) -> AsyncIterator[EventEnvelope]: + """Consume a streaming function progressively - the same decoded + envelopes an engine reply route receives: ``data`` segments, then the + ``eof`` or ``exception`` terminal. A non-streaming target yields its + one classic reply (opting in is always safe). ``timeout_ms`` is the + idle allowance between segments; expiry, a truncated stream and a + malformed dialect yield the in-band exception envelope, then end. + + Remote (an endpoint is given, or set on the constructor): the peer's + ``/api/event`` answers the one POST with the envelope-mode SSE dialect. + Local (no endpoint): the same first-envelope classification through a + per-request reply sink on the primitive bus. + """ + url = endpoint or self.endpoint + event = _build_event(route, body, headers, from_route, cid) + if not url: + async for reply in self._stream_local(route, event, timeout_ms): + yield reply + return + async for reply in self._stream_remote(url, event, timeout_ms): + yield reply + + async def stream_to(self, route: str, body: Any = None, *, + reply_to: str, + headers: dict[str, str] | None = None, + timeout_ms: int = 30000, + endpoint: str | None = None, + from_route: str | None = None, + cid: str | None = None) -> EventEnvelope: + """The relay form of :meth:`stream` for composition: every decoded + envelope forwards verbatim to the LOCAL ``reply_to`` route (typically + the caller's own reply address, handed through by an interceptor), so + segments flow remote peer -> this application -> the original caller + with no buffering. Awaits and returns the last envelope (normally the + terminal).""" + last = EventEnvelope().set_status(500).set_body("Stream produced no events") + async for segment in self.stream(route, body, headers=headers, + timeout_ms=timeout_ms, endpoint=endpoint, + from_route=from_route, cid=cid): + last = segment + forward = EventEnvelope.from_map(segment.to_map()).set_to(reply_to) + if not self._registry.send_event(forward): + # the local consumer is gone - late segments are no-op drops + break + return last + + async def _stream_local(self, route: str, event: EventEnvelope, + timeout_ms: int) -> AsyncIterator[EventEnvelope]: + service = self._registry.get(route) + if service is None: + yield EventEnvelope().set_status(404).set_body(f"Route {route} not found") + return + if not service.interceptor: + # a plain function cannot stream - its single reply is the stream + yield await self._call_local(route, event.body, event.headers, + timeout_ms, False, event.sender, event.cid) + return + bus = self._registry.bus + sink_route, queue = bus.open_sink() + try: + bus.publish_envelope(service, event.set_reply_to(sink_route)) + idle = max(100, timeout_ms) / 1000 + streaming = False + while True: + try: + reply = await asyncio.wait_for(queue.get(), idle) + except asyncio.TimeoutError: + seconds = max(100, timeout_ms) // 1000 + yield exception_envelope(408, f"Timeout for {seconds} seconds") + return + out, done = _classify_sink_reply(reply, streaming) + streaming = True + yield out + if done: + return + finally: + bus.close_sink(sink_route) + + async def _stream_remote(self, url: str, event: EventEnvelope, + timeout_ms: int) -> AsyncIterator[EventEnvelope]: + effective_cid = event.cid + http_headers = self._http_headers(timeout_ms, False, event) + http_headers["accept"] = TEXT_EVENT_STREAM + idle_seconds = max(1.0, timeout_ms / 1000) + # no total limit - a healthy stream may outlive any fixed total; the + # per-read socket timeout is the idle allowance between segments + client_timeout = aiohttp.ClientTimeout(total=None, sock_connect=10, + sock_read=idle_seconds) + session = self._get_session() + async with session.post(url, data=event.to_bytes(), headers=http_headers, + timeout=client_timeout) as response: + content_type = response.headers.get("content-type", "") + if not content_type.startswith(TEXT_EVENT_STREAM): + # the peer answered single-shot (a non-streaming target, or an + # edge error) - the classic reply, decoded tolerantly + yield _decode_single_shot(await response.read(), response.status) + return + parser = SseParser() + head_seen = False + try: + async for chunk in response.content.iter_any(): + for name, text in parser.feed(chunk): + reply, terminal = _decode_frame(name, text, head_seen, + effective_cid) + if reply is None: + continue + head_seen = True + yield reply + if terminal: + return # frames after the terminal are discarded + # the dialect ends with a decoded terminal - a bare transport + # end is a truncation + yield _relay_guard(500, "Event stream ended without eof", effective_cid) + except asyncio.TimeoutError: + yield _relay_guard(408, f"Timeout for {int(idle_seconds)} seconds", + effective_cid) + except aiohttp.ClientError as e: + yield _relay_guard(500, str(e) or type(e).__name__, effective_cid) + @staticmethod def _run_sync(factory: Callable[[], Coroutine[Any, Any, EventEnvelope]], timeout_ms: int) -> EventEnvelope: @@ -240,3 +420,82 @@ def send_sync(self, route: str, body: Any = None, *, lambda: self.send(route, body, headers=headers, timeout_ms=timeout_ms, endpoint=endpoint, from_route=from_route, cid=cid), timeout_ms) + + +def _classify_sink_reply(reply: EventEnvelope, + streaming: bool) -> tuple[EventEnvelope, bool]: + """Classify one reply-sink envelope exactly like the engines: unmarked + before any segment = the classic single-shot answer; unmarked mid-stream = + the bus's error contract for an uncaught interceptor exception (fails + in-band); marked = a stream segment, terminal on eof/exception.""" + marker = stream_signal(reply) + if marker is None: + if streaming: + message = str(reply.body) if reply.body is not None else "Stream failed" + return exception_envelope(reply.get_status(), message), True + return reply, True + return reply, marker in (EOF, EXCEPTION) + + +def _relay_guard(status: int, message: str, cid: str | None) -> EventEnvelope: + """An in-band exception envelope synthesized by the consuming relay.""" + event = exception_envelope(status, message) + if cid: + event.set_correlation_id(cid) + return event + + +def _decode_frame(name: str | None, text: str, head_seen: bool, + cid: str | None) -> tuple[EventEnvelope | None, bool]: + """Decode one SSE frame of the envelope-mode dialect: an ``envelope`` frame + is one base64-encoded serialized envelope (the head, the terminals and + non-text segments); any other frame is a raw text segment. Returns + (envelope-or-None, terminal). Dialect guards fail in-band: the first frame + must be an envelope frame, and a malformed frame ends the stream.""" + if name == ENVELOPE: + try: + # binascii.Error is a ValueError subclass - one catch covers both + decoded = EventEnvelope.from_bytes(base64.b64decode(text, validate=True)) + except ValueError: + return _relay_guard(500, "Invalid event stream - malformed envelope frame", + cid), True + decoded.to = None + decoded.reply_to = None + if cid: + decoded.set_correlation_id(cid) + marker = stream_signal(decoded) + return decoded, marker in (EOF, EXCEPTION) + if not head_seen: + # the dialect guarantees an envelope frame first (conformance guard) + return _relay_guard(500, "Invalid event stream - missing envelope head", + cid), True + segment = EventEnvelope(body=text).set_header(X_EVENT_STREAM, DATA) + if name: + segment.set_header(X_EVENT_NAME, name) + if cid: + segment.set_correlation_id(cid) + return segment, False + + +def _decode_single_shot(payload: bytes, http_status: int) -> EventEnvelope: + """Decode a single-shot Event-over-HTTP reply: a serialized envelope + normally, with the classic tolerant handling of an edge-level REST error + body ('{"type": "error", "status": n, "message": text}' JSON) and of a + payload that is not a serialized envelope at all.""" + if not payload: + return EventEnvelope().set_status(http_status) + try: + reply = EventEnvelope.from_bytes(payload) + except ValueError as e: + if http_status >= 400: + try: + data = json.loads(payload.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + data = None + if isinstance(data, dict) and data.get("type") == "error" \ + and isinstance(data.get("message"), str): + return EventEnvelope().set_status(http_status).set_body(data["message"]) + return EventEnvelope().set_status(400).set_body( + f"Invalid event-over-http response - {e}") + reply.reply_to = None + return reply diff --git a/src/mercury_composable/config.py b/src/mercury_composable/config.py index 120bfe8..f254e97 100644 --- a/src/mercury_composable/config.py +++ b/src/mercury_composable/config.py @@ -130,6 +130,12 @@ def get_property(self, key: str, default: str | None = None) -> str | None: value = self.get(key, default) return None if value is None else str(value) + def resolve_text(self, value: str) -> Any: + """Resolve ``${ENV:default}`` substitution in a text value - the same + rules as configuration values (used by companion config files such as + app-log-context.yaml).""" + return self._substitute(value) + def exists(self, key: str) -> bool: return key in self._overrides or key in self._store diff --git a/src/mercury_composable/default-log-context.yaml b/src/mercury_composable/default-log-context.yaml new file mode 100644 index 0000000..bfbb79d --- /dev/null +++ b/src/mercury_composable/default-log-context.yaml @@ -0,0 +1,19 @@ +# +# Built-in default application log context (the engines' default-log-context.yaml twin). +# +# The log-context feature is ON by default using this template. It applies to the +# structured JSON presentations (log.format=json or compact). +# +# To customize the context block, provide your own app-log-context.yaml in the +# application's resources folder (next to application.yml) - it replaces this +# default entirely. +# To turn the feature off, set app.log.context=false in the configuration. +# +context: + cid: $cid + traceId: $traceId + tracePath: $tracePath + spanId: $spanId + parentSpanId: $parentSpanId + service: $service + timestamp: $utc diff --git a/src/mercury_composable/envelope.py b/src/mercury_composable/envelope.py index 90da93a..ee523d0 100644 --- a/src/mercury_composable/envelope.py +++ b/src/mercury_composable/envelope.py @@ -117,6 +117,10 @@ def set_trace(self, trace_id: str, trace_path: str) -> EventEnvelope: self.trace_path = trace_path return self + def set_span_id(self, span_id: str) -> EventEnvelope: + self.span_id = span_id + return self + def set_reply_to(self, route: str | None) -> EventEnvelope: self.reply_to = route return self diff --git a/src/mercury_composable/event_stream.py b/src/mercury_composable/event_stream.py new file mode 100644 index 0000000..ab2bd3e --- /dev/null +++ b/src/mercury_composable/event_stream.py @@ -0,0 +1,340 @@ +""" +Event streaming: the multi-shot reply contract and the envelope-mode SSE dialect. + +The platform's native streaming pattern (all four runtimes): *the caller provides +a reply address; the callee streams events to it until a terminal signal*. Each +segment is one event to the caller's ``reply_to``, marked with the reserved +envelope header ``x-event-stream: data | eof | exception``. On the Event-over-HTTP +wire, the peer answers the one POST with a Server-Sent Events response in a hybrid +dialect: **envelope frames** (the reserved SSE event name ``envelope``, one +base64-encoded serialized envelope per frame) wherever envelope semantics matter - +the head, the terminals and non-text segments - and **raw SSE frames** for plain +text segments, so token relays stay near-zero overhead. + +:class:`EventStreamWriter` is the producer helper - the engines' exact API:: + + out = EventStreamWriter.from_request(event) # an interceptor's raw envelope + out.first(200, "text/event-stream") + out.write("hello") # data segment + out.write_named("tokens", {"n": 2}) # named (typed) SSE event + out.close({"usage": usage}) # end of transmission + metadata + # or out.fail(e) # in-band failure + +Writes after close/fail are dropped (debug log), mirroring the engines. An +in-band failure body carries the standard error key-values +``'{"type": "error", "status": n, "message": text}'``. +""" + +from __future__ import annotations + +import asyncio +import base64 +from typing import TYPE_CHECKING, Any + +from .config import app_config +from .envelope import EventEnvelope +from .exceptions import AppException +from .log import get_logger +from .trace import get_trace + +if TYPE_CHECKING: + from .registry import FunctionRegistry + +# reserved envelope header (internal protocol, never on the HTTP wire) +X_EVENT_STREAM = "x-event-stream" +# optional companion on a data event: maps to the SSE "event:" field +X_EVENT_NAME = "x-event-name" +# marker vocabulary - deliberately the engines' ObjectStream vocabulary +DATA = "data" +EOF = "eof" +EXCEPTION = "exception" +# reserved SSE event name of the envelope-mode wire dialect: a frame with this +# name carries one base64-encoded serialized EventEnvelope +ENVELOPE = "envelope" + +X_TTL = "x-ttl" +TEXT_EVENT_STREAM = "text/event-stream" +STREAM_CALLER_REQUIRED = "Streaming function requires a caller that accepts text/event-stream" + +# reserved envelope headers a raw SSE frame may carry without loss +_RESERVED_HEADERS = {X_EVENT_STREAM, X_EVENT_NAME, X_TTL} + +log = get_logger("mercury.stream") + + +def stream_signal(event: EventEnvelope) -> str | None: + """The x-event-stream marker (lowercased), or None for an unmarked envelope.""" + for key, value in event.headers.items(): + if key.lower() == X_EVENT_STREAM: + return value.lower() + return None + + +def stream_event_name(event: EventEnvelope) -> str | None: + """The x-event-name companion header (the SSE ``event:`` field), if any.""" + for key, value in event.headers.items(): + if key.lower() == X_EVENT_NAME: + return value + return None + + +def error_body(status: int, message: str) -> dict[str, Any]: + """The standard error key-values: '{"type": "error", "status": n, "message": text}'""" + return {"type": "error", "status": status, "message": message} + + +def exception_envelope(status: int, message: str) -> EventEnvelope: + """An in-band exception envelope with the standard error body.""" + return (EventEnvelope() + .set_header(X_EVENT_STREAM, EXCEPTION) + .set_status(status) + .set_body(error_body(status, message))) + + +def sse_frame(event_name: str | None, text: str) -> bytes: + """One SSE frame: optional ``event:`` line, one ``data:`` line per text line.""" + lines = [] + if event_name: + lines.append(f"event: {event_name}\n") + for line in text.split("\n"): + lines.append(f"data: {line}\n") + lines.append("\n") + return "".join(lines).encode("utf-8") + + +def envelope_frame(event: EventEnvelope) -> bytes: + """One envelope-mode wire frame: the envelope serialized verbatim - with the + host-internal addressing cleared, because the consuming relay rewrites + addressing to the original caller - as base64 under the reserved name.""" + clone = EventEnvelope.from_map(event.to_map()) + clone.to = None + clone.reply_to = None + encoded = base64.b64encode(clone.to_bytes()).decode("ascii") + return sse_frame(ENVELOPE, encoded) + + +def raw_streamable(event: EventEnvelope) -> bool: + """A data segment may ride a raw SSE frame only when the frame carries it + losslessly: a 200 status, no custom envelope headers, a user event name + clear of the reserved word, and a text (or empty) body without a carriage + return - SSE normalizes line endings. Everything else takes the + envelope-frame escape hatch.""" + if event.get_status() != 200: + return False + for key, value in event.headers.items(): + lowered = key.lower() + if lowered not in _RESERVED_HEADERS: + return False + if lowered == X_EVENT_NAME and value == ENVELOPE: + return False + body = event.body + return body is None or (isinstance(body, str) and "\r" not in body) + + +def data_frame(event: EventEnvelope, first_frame: bool) -> bytes: + """One envelope-mode data frame: the first event always rides an envelope + frame (it carries the head control); a losslessly raw-able text segment + rides a raw frame; a bare no-op segment carries nothing.""" + if first_frame or not raw_streamable(event): + return envelope_frame(event) + if event.body is None: + return b"" + return sse_frame(stream_event_name(event), event.body) + + +def keep_alive_ms() -> int: + """SSE keep-alive comment interval in ms (``event.stream.keep.alive``, + default 30s; 0 disables - the engines' config key).""" + raw = str(app_config().get_property("event.stream.keep.alive", "30s") or "30s") + raw = raw.strip().lower() + if raw in ("0", "0s", "0ms", "0m"): + return 0 + try: + if raw.endswith("ms"): + return int(raw[:-2]) + if raw.endswith("s"): + return int(raw[:-1]) * 1000 + if raw.endswith("m"): + return int(raw[:-1]) * 60_000 + return int(raw) * 1000 + except ValueError: + return 30_000 + + +class SseParser: + """Incremental SSE frame parser: byte-level line split (a newline is a + single byte, so this is UTF-8 safe), one-leading-space value strip, + comment/id/retry suppression, multi-line data joined per the SSE + specification. Mirrors the engines' parsers.""" + + def __init__(self) -> None: + self._pending = bytearray() + self._data_lines: list[str] = [] + self._event_name: str | None = None + + def feed(self, chunk: bytes) -> list[tuple[str | None, str]]: + """Feed one body chunk; return the completed (event_name, data) events.""" + self._pending.extend(chunk) + events: list[tuple[str | None, str]] = [] + buffer = bytes(self._pending) + start = 0 + for i, byte in enumerate(buffer): + if byte != 0x0A: # '\n' + continue + end = i - 1 if i > start and buffer[i - 1] == 0x0D else i + self._on_line(buffer[start:end].decode("utf-8", errors="replace"), events) + start = i + 1 + self._pending = bytearray(buffer[start:]) + return events + + def _on_line(self, line: str, events: list[tuple[str | None, str]]) -> None: + """One SSE line: a blank line dispatches the pending event; a comment + line (leading colon) is consumed, never forwarded; id, retry and + unknown fields are ignored (SSE specification).""" + if not line: + if self._data_lines: + events.append((self._event_name, "\n".join(self._data_lines))) + self._data_lines = [] + self._event_name = None + return + if line.startswith(":"): + return + colon = line.find(":") + field = line if colon == -1 else line[:colon] + value = "" if colon == -1 else line[colon + 1:] + value = value.removeprefix(" ") + if field == "data": + self._data_lines.append(value) + elif field == "event": + self._event_name = value + + +class EventStreamWriter: + """Producer helper for a multi-shot reply - the engines' exact API. + + Only an interceptor function can stream: it receives the raw envelope, so + the caller-provided reply address travels the engines' way + (``EventStreamWriter.from_request(event)`` reads ``reply_to`` and the + correlation id). Segments route to the LOCAL reply address through the + primitive event bus - simple routing to a local function or reply sink, + never across the wire (cross-wire replies ride the Event-over-HTTP SSE + response, exactly as on the engines). + """ + + def __init__(self, reply_to: str | None, correlation_id: str | None = None, *, + registry: FunctionRegistry | None = None): + if not reply_to: + raise AppException(400, "Streaming producer requires a reply_to address") + from .registry import default_registry + self._registry = registry or default_registry + self._reply_to = reply_to + self._cid = correlation_id + self._first_status = 200 + self._first_content_type: str | None = None + self._first_ttl_seconds = 0 + self._head_sent = False + self._closed = False + + @classmethod + def from_request(cls, event: EventEnvelope, *, + registry: FunctionRegistry | None = None) -> EventStreamWriter: + """Create a writer from the incoming request envelope (the usual form + for an interceptor function).""" + return cls(event.reply_to, event.cid, registry=registry) + + def first(self, status: int, content_type: str, + ttl_seconds: int | None = None) -> EventStreamWriter: + """Optional head control carried by the first outgoing event: response + status, content type, and an optional idle-allowance override in + seconds between segments.""" + self._first_status = int(status) + self._first_content_type = content_type + if ttl_seconds is not None: + self._first_ttl_seconds = int(ttl_seconds) + return self + + def write(self, segment: Any) -> None: + """Send one ``data`` segment (text, bytes, dict, list - any payload).""" + self._send(DATA, segment, None) + + def write_named(self, event_name: str, segment: Any) -> None: + """Send one named segment - the name maps to the SSE ``event:`` field.""" + self._send(DATA, segment, event_name) + + def close(self, trailing_metadata: Any = None) -> None: + """Declare end of transmission, with optional trailing metadata.""" + if self._closed: + return + self._closed = True + self._send(EOF, trailing_metadata, None, unchecked=True) + + def fail(self, error: Exception) -> None: + """Declare an in-band failure and end the stream.""" + if self._closed: + return + self._closed = True + status = error.status if isinstance(error, AppException) else 500 + status = status if status >= 400 else 500 + message = str(error) or type(error).__name__ + event = self._envelope(EXCEPTION, error_body(status, message), None) + event.set_status(status) + self._emit(event) + + @property + def closed(self) -> bool: + """True when the stream has been closed or failed.""" + return self._closed + + def _send(self, marker: str, body: Any, event_name: str | None, *, + unchecked: bool = False) -> None: + if self._closed and not unchecked: + log.debug("Segment to %s dropped - stream already closed", self._reply_to) + return + self._emit(self._envelope(marker, body, event_name)) + + def _envelope(self, marker: str, body: Any, event_name: str | None) -> EventEnvelope: + event = EventEnvelope(to=self._reply_to, body=body) + event.set_header(X_EVENT_STREAM, marker) + if self._cid: + event.set_correlation_id(self._cid) + if event_name: + event.set_header(X_EVENT_NAME, event_name) + # segments inherit the producer's identity, trace and span, so a + # consuming engine's per-segment delivery spans parent onto this + # function (the engines' po.send/touch parity) + info = get_trace() + if info and info.route: + event.set_from(info.route) + if info and info.trace_id: + event.set_trace(info.trace_id, info.trace_path or self._reply_to) + if info.span_id: + event.set_span_id(info.span_id) + if not self._head_sent: + self._head_sent = True + event.set_status(self._first_status) + if self._first_content_type: + event.set_header("content-type", self._first_content_type) + if self._first_ttl_seconds > 0: + event.set_header(X_TTL, str(self._first_ttl_seconds)) + return event + + def _emit(self, event: EventEnvelope) -> None: + """Deliver to the local reply address - safe from async handlers and + from plain-def handlers on executor threads (the sync bridge's host + loop carries the delivery back to the event loop).""" + try: + asyncio.get_running_loop() + except RuntimeError: + from .bus import get_host_loop + loop = get_host_loop() + if loop is None: + raise RuntimeError( + "No Mercury host event loop in context - the stream writer works " + "inside a hosted function") from None + loop.call_soon_threadsafe(self._deliver, event) + else: + self._deliver(event) + + def _deliver(self, event: EventEnvelope) -> None: + if not self._registry.send_event(event): + log.warning("Event dropped - route %s not found", self._reply_to) diff --git a/src/mercury_composable/log.py b/src/mercury_composable/log.py index 0614f0e..af7d4b7 100644 --- a/src/mercury_composable/log.py +++ b/src/mercury_composable/log.py @@ -14,9 +14,10 @@ (mirroring the engines), else the ``log.level`` configuration key, else INFO. - ``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). + (default), ``json`` (pretty-printed) and ``compact`` (the same object on a + single line - JSONL - for log aggregators). Inside a traced request, the + JSON presentations add the application log ``context`` block (the engines' + app-log-context feature - see :mod:`mercury_composable.log_context`). """ from __future__ import annotations @@ -32,12 +33,24 @@ _configured = False +def _message_of(record: logging.LogRecord) -> str | dict: + """A structured (dict) message stays structural in the JSON presentations + and renders as compact JSON in text mode - used by the distributed-trace + dataset records, which stdout log-ingest agents parse.""" + if isinstance(record.msg, dict) and not record.args: + return record.msg + return record.getMessage() + + class EngineTextFormatter(logging.Formatter): def format(self, record: logging.LogRecord) -> str: ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(record.created)) ms = int(record.msecs) level = f"{record.levelname:<5}" - line = f"{ts}.{ms:03d} {level} {record.name}:{record.lineno} - {record.getMessage()}" + message = _message_of(record) + if isinstance(message, dict): + message = json.dumps(message, ensure_ascii=False) + line = f"{ts}.{ms:03d} {level} {record.name}:{record.lineno} - {message}" if record.exc_info: line += "\n" + self.formatException(record.exc_info) return line @@ -56,13 +69,20 @@ def format(self, record: logging.LogRecord) -> str: "time": f"{ts}.{int(record.msecs):03d}", "level": record.levelname, "logger": f"{record.name}:{record.lineno}", - "message": record.getMessage(), + "message": _message_of(record), } - from .trace import get_trace # late import to avoid a cycle + # late imports to avoid a cycle (this module bootstraps logging) + from .log_context import log_context_config + from .trace import get_trace + # the application log context (the engines' app-log-context feature): + # a "context" block on every structured line inside a traced request, + # correlating app logs with the distributed-trace telemetry stream info = get_trace() if info and info.trace_id: - entry["trace_id"] = info.trace_id + context_config = log_context_config() + if context_config.enabled: + entry["context"] = context_config.render(info) if record.exc_info: entry["exception"] = self.formatException(record.exc_info) return json.dumps(entry, ensure_ascii=False, indent=self._indent) diff --git a/src/mercury_composable/log_context.py b/src/mercury_composable/log_context.py new file mode 100644 index 0000000..786b938 --- /dev/null +++ b/src/mercury_composable/log_context.py @@ -0,0 +1,172 @@ +""" +Application log context - the engines' app-log-context feature. + +When enabled (``app.log.context``, default true), the structured log +presentations (``log.format`` json/compact) add a ``context`` block to every +log line written inside a traced request, so application logs and the +distributed-trace telemetry stream correlate end to end in one aggregation. + +The context template mirrors the engines' contract exactly: + +- The built-in default template carries the standard trace context + (cid, traceId, tracePath, spanId, parentSpanId, service, timestamp). +- An application may replace it entirely with its own ``app-log-context.yaml`` + in the resources folder (next to application.yml), mapping each output key + to a reserved ``$token`` - resolved live per log line - or a constant + (a literal, or ``${ENV:default}`` resolved once at load). +- ``app.log.context=false`` opts out. +- The ``cid`` token is the BUSINESS correlation-id only (the engine-managed + my_cid tag); an internal routing id under the ``cid`` label would mislead + log aggregation. +- Developer-supplied key-values (:func:`mercury_composable.update_context`) + merge into the block; keys resolving to None are omitted, never "null". +""" + +from __future__ import annotations + +import importlib.resources +import os +import threading +from typing import Any + +import yaml + +from .config import app_config +from .envelope import iso_utc +from .log import get_logger +from .trace import RESERVED_CONTEXT_TOKENS, TraceInfo + +FEATURE_FLAG = "app.log.context" +CONFIG_FILE = "app-log-context.yaml" +# the built-in default template ships as a packaged resource, exactly like the +# engines' classpath:/default-log-context.yaml +DEFAULT_FILE = "default-log-context.yaml" + +log = get_logger("mercury.log") + + +def _context_section(data: Any) -> dict[str, Any] | None: + """The template's ``context:`` section, or None when absent/malformed.""" + section = data.get("context") if isinstance(data, dict) else None + if isinstance(section, dict): + return {str(k): v for k, v in section.items()} + return None + + +def default_template() -> dict[str, Any] | None: + """The built-in default template from the packaged default-log-context.yaml.""" + try: + resource = importlib.resources.files("mercury_composable").joinpath(DEFAULT_FILE) + data = yaml.safe_load(resource.read_text(encoding="utf-8")) or {} + except OSError: + return None + return _context_section(data) + + +def _token_value(info: TraceInfo, token: str) -> Any: + """Resolve a reserved token to its live value (None when absent).""" + if token == "utc": + return iso_utc() + return { + "cid": info.my_correlation_id, + "traceId": info.trace_id, + "tracePath": info.trace_path, + "spanId": info.span_id, + "parentSpanId": info.parent_span_id, + "service": info.route, + }.get(token) + + +class LogContextConfig: + """Parsed context template: output key -> reserved token or constant.""" + + def __init__(self, template: dict[str, Any] | None): + self.tokens: dict[str, str] = {} + self.constants: dict[str, Any] = {} + for output_key, raw in (template or {}).items(): + self._parse_entry(output_key, raw if isinstance(raw, str) else str(raw)) + self.enabled = bool(self.tokens or self.constants) + + def _parse_entry(self, output_key: str, value: str) -> None: + """One template entry: a reserved $token, or a constant (env-resolved + value or literal; an unset ${VAR} with no default resolves to None and + is dropped) - the engines' parseEntry.""" + if value.startswith("$") and not value.startswith("${"): + token = value[1:] + if token not in RESERVED_CONTEXT_TOKENS: + raise ValueError( + f"Invalid log context token '{value}' for key " + f"'{output_key}' - allowed tokens: " + f"{sorted(RESERVED_CONTEXT_TOKENS)}") + self.tokens[output_key] = token + return + resolved = app_config().resolve_text(value) + if resolved is not None: + self.constants[output_key] = resolved + + def render(self, info: TraceInfo) -> dict[str, Any]: + """The context block for one log line: template tokens resolved live, + constants, and the developer's custom key-values. Keys resolving to + None are omitted.""" + out: dict[str, Any] = {} + for output_key, token in self.tokens.items(): + value = _token_value(info, token) + if value is not None: + out[output_key] = value + out.update(self.constants) + for key, value in info.custom_context.items(): + if value is not None: + out[key] = value + return out + + +_lock = threading.Lock() +_instance: LogContextConfig | None = None + + +def _load() -> tuple[LogContextConfig, str | None]: + """Resolve the active template; returns (config, warning-or-None). Never + logs itself - the caller emits the warning AFTER installing the config, so + the log line (which renders through this feature) cannot re-enter.""" + config = app_config() + if (config.get_property(FEATURE_FLAG, "true") or "true").lower() == "false": + return LogContextConfig(None), None + # an application override replaces the default entirely - same resources + # convention as application.yml + source = config.source + folder = os.path.dirname(source) if source != "none" else "resources" + candidate = os.path.join(folder or "resources", CONFIG_FILE) + if os.path.isfile(candidate): + with open(candidate, "r", encoding="utf-8") as f: + section = _context_section(yaml.safe_load(f.read()) or {}) + if section is None: + # the engines log a warning and disable; mirror the outcome + return LogContextConfig(None), \ + f"Log context config has no 'context' section - feature disabled ({candidate})" + return LogContextConfig(section), None + template = default_template() + if template is None: + return LogContextConfig(None), \ + f"Built-in {DEFAULT_FILE} missing - log context feature disabled" + return LogContextConfig(template), None + + +def log_context_config() -> LogContextConfig: + """The shared context template (loaded on first structured log line).""" + global _instance + warning = None + with _lock: + instance = _instance + if instance is None: + instance, warning = _load() + _instance = instance + if warning: + log.warning(warning) + return instance + + +def reset_for_test() -> None: + """Test seam: reset so the next structured log line reloads the template.""" + global _instance + with _lock: + _instance = None diff --git a/src/mercury_composable/registry.py b/src/mercury_composable/registry.py index b7e7262..e8c16c5 100644 --- a/src/mercury_composable/registry.py +++ b/src/mercury_composable/registry.py @@ -26,12 +26,17 @@ from typing import Any from .bus import EventBus -from .envelope import Body +from .envelope import Body, EventEnvelope # the function contract: (headers, body) in, reply body (or EventEnvelope) out - # mirrors the node package's exported Handler type Handler = Callable[[dict[str, str], Body], Any] +# the engines' @EventInterceptor contract: (headers, raw envelope) in, replies +# sent manually via reply_to, return value discarded - the streaming producer +# and relay-function signature +InterceptorHandler = Callable[[dict[str, str], EventEnvelope], Any] + _ROUTE_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._-]*$") @@ -47,10 +52,15 @@ def validate_route(route: str) -> str: @dataclass class ServiceDef: route: str - handler: Handler + handler: Handler | InterceptorHandler instances: int = 10 private: bool = False is_async: bool = False + # the engines' @EventInterceptor flavor: the handler receives the raw + # EventEnvelope as its second argument (reply_to and correlation id travel + # the engines' way), replies manually via reply_to, and its return value + # is discarded. Streaming producers and relay functions are interceptors. + interceptor: bool = False class FunctionRegistry: @@ -59,9 +69,11 @@ def __init__(self) -> None: # the registry's own dispatch pipeline (see bus.py) - shared by the # HTTP host and the local side of PostOffice self.bus = EventBus() + self.bus.bind_registry(self) - def register(self, route: str, handler: Handler, *, - instances: int = 10, private: bool = False) -> ServiceDef: + def register(self, route: str, handler: Handler | InterceptorHandler, *, + instances: int = 10, private: bool = False, + interceptor: bool = False) -> ServiceDef: route = validate_route(route) service = ServiceDef( route=route, @@ -69,6 +81,7 @@ def register(self, route: str, handler: Handler, *, instances=max(1, int(instances)), private=bool(private), is_async=inspect.iscoroutinefunction(handler), + interceptor=bool(interceptor), ) self._services[route] = service return service @@ -82,12 +95,30 @@ def exists(self, route: str) -> bool: def routes(self) -> dict[str, ServiceDef]: return dict(self._services) + def send_event(self, event: EventEnvelope) -> bool: + """The reply_to mechanism: deliver one envelope to a LOCAL reply sink + or registered function, drop-n-forget (simple routing, never across + the wire - cross-wire replies ride the Event-over-HTTP SSE response). + Returns False when the target no longer exists, so a late segment is + a no-op drop, the engines' semantics.""" + route = event.to + if not route: + return False + if self.bus.offer_sink(route, event): + return True + service = self._services.get(route) + if service is None: + return False + self.bus.publish_envelope(service, event) + return True + # the default registry used by @preload and platform.run() default_registry = FunctionRegistry() -def preload(route: str, instances: int = 10, private: bool = False): +def preload(route: str, instances: int = 10, private: bool = False, + interceptor: bool = False): """Register a function handler under a route name (engine PreLoad analog). Usage:: @@ -95,8 +126,20 @@ def preload(route: str, instances: int = 10, private: bool = False): @preload(route="hello.python", instances=10) def handle_event(headers: dict[str, str], body): return {"text": body["text"].upper()} + + An ``interceptor=True`` handler receives the raw :class:`EventEnvelope` + as its second argument and replies manually (the engines' + ``@EventInterceptor``) - the streaming producer pattern:: + + @preload(route="hello.tokens", instances=10, interceptor=True) + async def stream_tokens(headers: dict[str, str], event: EventEnvelope): + out = EventStreamWriter.from_request(event) + out.first(200, "text/event-stream") + out.write("hello") + out.close() """ - def wrapper(fn: Handler) -> Handler: - default_registry.register(route, fn, instances=instances, private=private) + def wrapper(fn: Handler | InterceptorHandler) -> Handler | InterceptorHandler: + default_registry.register(route, fn, instances=instances, private=private, + interceptor=interceptor) return fn return wrapper diff --git a/src/mercury_composable/server.py b/src/mercury_composable/server.py index 8304c78..ac93e43 100644 --- a/src/mercury_composable/server.py +++ b/src/mercury_composable/server.py @@ -23,21 +23,35 @@ from __future__ import annotations +import asyncio +import contextlib + from aiohttp import web from .actuator import Actuator from .bus import DeliveryTimeout from .config import app_config from .envelope import EventEnvelope +from .event_stream import ( + DATA, + STREAM_CALLER_REQUIRED, + TEXT_EVENT_STREAM, + data_frame, + envelope_frame, + exception_envelope, + keep_alive_ms, + stream_signal, +) from .log import get_logger -from .registry import FunctionRegistry, default_registry +from .registry import FunctionRegistry, ServiceDef, default_registry +from .trace import MY_CID_TAG, MY_CORRELATION_ID OCTET_STREAM = "application/octet-stream" X_TTL = "x-ttl" X_ASYNC = "x-async" X_EVENT_API = "x-event-api" -MY_CID_TAG = "my_cid" -MY_CORRELATION_ID = "my_correlation_id" +# the engines' reserved route name for the Event-over-HTTP ingress +EVENT_API_SERVICE = "event.api.service" log = get_logger("mercury.server") @@ -63,7 +77,7 @@ 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: + async def handle_event(self, request: web.Request) -> web.StreamResponse: raw = await request.read() try: ttl = int(request.headers.get(X_TTL, "0") or 0) @@ -83,16 +97,27 @@ async def handle_event(self, request: web.Request) -> web.Response: return _transport_error(404, f"Route {event.to} not found") if service.private: return _transport_error(403, f"{event.to} is private") + if not event.sender: + # the engines' EventApiService parity: its PostOffice fills the + # sender with its own route when the wire envelope carries none + event.set_from(EVENT_API_SERVICE) headers = _handler_headers(event) bus = self.registry.bus if is_async: ack = bus.publish(service, headers, event.body, trace_id=event.trace_id, - trace_path=event.trace_path, cid=event.cid) + trace_path=event.trace_path, cid=event.cid, envelope=event) return web.Response(status=202, body=ack.to_bytes(), content_type=OCTET_STREAM) + if service.interceptor: + # interceptor dispatch (the reply_to mechanism): the handler + # receives the raw envelope with a per-request reply sink as its + # reply address and answers manually - single-shot or streaming + capable = TEXT_EVENT_STREAM in (request.headers.get("accept") or "") + return await self._dispatch_interceptor(request, service, event, headers, + ttl, capable) try: reply = await bus.deliver(service, headers, event.body, ttl, trace_id=event.trace_id, trace_path=event.trace_path, - cid=event.cid) + cid=event.cid, envelope=event) 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") @@ -100,6 +125,139 @@ 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) + async def _dispatch_interceptor(self, request: web.Request, service: ServiceDef, + event: EventEnvelope, headers: dict[str, str], + ttl: int, capable: bool) -> web.StreamResponse: + """Dispatch to an interceptor and classify its first reply exactly like + the engines: unmarked = the classic single-shot response, byte + identical; marked = the envelope-mode SSE dialect for a caller that + accepts text/event-stream, or the pinned 406 refusal for one that + does not.""" + bus = self.registry.bus + sink_route, queue = bus.open_sink() + try: + handler_event = EventEnvelope(to=event.to, body=event.body, headers=headers) + handler_event.set_reply_to(sink_route) + if event.tags: + # engine-managed tags (e.g. the business correlation-id) ride + # the delivered envelope verbatim, the engines' way + handler_event.tags = dict(event.tags) + if event.cid: + handler_event.set_correlation_id(event.cid) + if event.trace_id: + handler_event.set_trace(event.trace_id, event.trace_path or service.route) + if event.span_id: + # the caller's span - the handler's span parents onto it + handler_event.set_span_id(event.span_id) + if event.sender: + handler_event.set_from(event.sender) + bus.publish_envelope(service, handler_event) + try: + first = await asyncio.wait_for(queue.get(), ttl / 1000) + except asyncio.TimeoutError: + 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") + marker = stream_signal(first) + if marker is None: + # the classic single-shot reply (a manual answer, or the bus's + # error contract for an uncaught interceptor exception) + log.info("Handled %s status=%d exec_time=%sms trace_id=%s", + event.to, first.get_status(), first.exec_time, event.trace_id) + return web.Response(status=200, body=first.to_bytes(), + content_type=OCTET_STREAM) + if not capable: + # a streaming reply cannot ride a single-shot response + return _transport_error(406, STREAM_CALLER_REQUIRED) + return await self._stream_response(request, queue, first, marker, ttl) + finally: + bus.close_sink(sink_route) + + async def _stream_response(self, request: web.Request, + queue: asyncio.Queue[EventEnvelope], + first: EventEnvelope, first_marker: str, + ttl: int) -> web.StreamResponse: + """Render the envelope-mode SSE dialect: envelope frames for the head, + the terminals and non-text segments; raw frames for plain text. The + x-ttl allowance (overridable by the producer's head control, in + seconds) is the per-segment idle; expiry fails the stream in-band with + the standard 408 error body. Keep-alive comments ride while the + producer is quiet (event.stream.keep.alive, the engines' key).""" + idle_ms = ttl + for key, value in first.headers.items(): + if key.lower() == X_TTL: + with contextlib.suppress(ValueError): + seconds = int(str(value).strip()) + if seconds > 0: + idle_ms = seconds * 1000 + response = web.StreamResponse(status=first.get_status()) + response.headers["content-type"] = TEXT_EVENT_STREAM + response.headers.setdefault("cache-control", "no-cache") + await response.prepare(request) + try: + await response.write(envelope_frame(first)) + if first_marker == DATA: + await self._stream_segments(response, queue, idle_ms) + except ConnectionError: + # a disconnected client ends the stream; late segments are no-op drops + log.debug("Client disconnected from event stream") + with contextlib.suppress(ConnectionError): + await response.write_eof() + return response + + async def _stream_segments(self, response: web.StreamResponse, + queue: asyncio.Queue[EventEnvelope], idle_ms: int) -> None: + ping_ms = keep_alive_ms() + while True: + event = await self._next_segment(response, queue, idle_ms, ping_ms) + if event is None: + # idle expiry - fail in-band (the engines' housekeeper parity) + seconds = idle_ms // 1000 + frame = envelope_frame(exception_envelope(408, f"Timeout for {seconds} seconds")) + await response.write(frame) + return + marker = stream_signal(event) + if marker == DATA: + frame = data_frame(event, first_frame=False) + if frame: + await response.write(frame) + elif marker is not None: + # eof or exception: the terminal envelope frame ends the + # response cleanly - no cosmetic frames on this wire + await response.write(envelope_frame(event)) + return + elif event.has_error(): + # the bus's error contract for an uncaught interceptor + # exception mid-stream - fail in-band with the exact status + message = str(event.body) if event.body is not None else "Stream failed" + frame = envelope_frame(exception_envelope(event.get_status(), message)) + await response.write(frame) + return + else: + log.warning("Dropping event - invalid %s signal", "x-event-stream") + + @staticmethod + async def _next_segment(response: web.StreamResponse, + queue: asyncio.Queue[EventEnvelope], + idle_ms: int, ping_ms: int) -> EventEnvelope | None: + """Wait for the next segment within the idle allowance, emitting SSE + keep-alive comments while the producer is quiet (best-effort; pings + never extend the idle allowance).""" + loop = asyncio.get_running_loop() + deadline = loop.time() + idle_ms / 1000 + while True: + remaining = deadline - loop.time() + if remaining <= 0: + return None + wait = min(remaining, ping_ms / 1000) if ping_ms > 0 else remaining + try: + return await asyncio.wait_for(queue.get(), wait) + except asyncio.TimeoutError: + if loop.time() >= deadline: + return None + with contextlib.suppress(ConnectionError): + await response.write(b": ping\n\n") + def create_app(self) -> web.Application: app = web.Application(client_max_size=16 * 1024 * 1024) app.router.add_post("/api/event", self.handle_event) diff --git a/src/mercury_composable/trace.py b/src/mercury_composable/trace.py index 379ee25..7d23c55 100644 --- a/src/mercury_composable/trace.py +++ b/src/mercury_composable/trace.py @@ -16,13 +16,44 @@ from dataclasses import dataclass, field from typing import Any +# The business correlation-id rides an engine-managed envelope tag - never an +# envelope header - and is injected into the receiving function's input header +# copy as a read-only view at delivery (the engines' WorkerHandler contract). +MY_CID_TAG = "my_cid" +MY_CORRELATION_ID = "my_correlation_id" +# The engines' RPC round-trip marker tag: an RPC leg emits no trace dataset +# (its metrics fold into the caller's view), so the clients stamp it on +# request() calls and the bus honors it at delivery. +RPC_TAG = "rpc" +# Reserved application log-context tokens (the engines' LogContext contract): +# resolved live per log line; a developer cannot override them via +# update_context. The output key names in app-log-context.yaml are the +# operator's choice - this set governs the template tokens and developer API. +RESERVED_CONTEXT_TOKENS = frozenset( + {"cid", "traceId", "tracePath", "spanId", "parentSpanId", "service", "utc"}) + @dataclass class TraceInfo: + # the executing function's route - outbound calls fill their sender ("from") + # with it, the engines' PostOffice.touch parity + route: str | None = None trace_id: str | None = None trace_path: str | None = None cid: str | None = None + my_correlation_id: str | None = None + # span lineage (the engines' model): span_id is THIS execution's span, + # stamped onto outbound events so the receiver stores it as its + # parent_span_id; parent_span_id is the caller's span from the inbound + # envelope. 16-hex (W3C-shaped), so traceparent stamping fires when the + # trace id is 32-hex. + span_id: str | None = None + parent_span_id: str | None = None annotations: dict[str, Any] = field(default_factory=dict) + # developer-supplied application log-context key-values (update_context) - + # a logging-only sink, rendered into the "context" block of structured log + # lines; distinct from annotations, which feed the trace telemetry + custom_context: dict[str, Any] = field(default_factory=dict) _current: contextvars.ContextVar[TraceInfo | None] = contextvars.ContextVar( @@ -36,15 +67,20 @@ def get_trace() -> TraceInfo | None: @contextmanager -def trace_context(trace_id: str, trace_path: str, - cid: str | None = None) -> Iterator[TraceInfo]: +def trace_context(trace_id: str, trace_path: str, cid: str | None = None, + my_correlation_id: str | None = None, + span_id: str | None = None) -> Iterator[TraceInfo]: """Establish a trace context around a block - the node runWithTrace twin. Useful for callers outside a hosted function (batch jobs, tests) whose PostOffice calls should carry a trace: the client inherits the context - into the outbound envelope. + into the outbound envelope, including the business correlation-id as the + engine-managed my_cid tag. ``span_id`` declares the caller's CURRENT span + (e.g. an edge span from an external OpenTelemetry context) so the next + hop's span parents onto it. """ - info = TraceInfo(trace_id=trace_id, trace_path=trace_path, cid=cid) + info = TraceInfo(trace_id=trace_id, trace_path=trace_path, cid=cid, + my_correlation_id=my_correlation_id, span_id=span_id) token = _set_trace(info) try: yield info @@ -59,6 +95,29 @@ def annotate_trace(key: str, value: Any) -> None: info.annotations[str(key)] = value +def update_context(key: str, value: Any) -> None: + """Add (or remove, when value is None) a custom key-value in the + application log context - the engines' PostOffice.updateContext twin. + + The key-value is rendered into the "context" block of structured log + output (log.format json/compact) when the app-log-context feature is + enabled. Unlike annotate_trace (which feeds the distributed-trace + telemetry), this is a logging-only sink. No-op outside a hosted request. + + :raises ValueError: if key is one of the reserved context tokens + """ + if key in RESERVED_CONTEXT_TOKENS: + raise ValueError(f"Cannot override reserved log context key '{key}'" + f" - reserved keys are {sorted(RESERVED_CONTEXT_TOKENS)}") + info = _current.get() + if info is None: + return + if value is None: + info.custom_context.pop(key, None) + else: + info.custom_context[key] = value + + def _set_trace(info: TraceInfo | None) -> contextvars.Token[TraceInfo | None]: return _current.set(info) diff --git a/src/mercury_composable/version.py b/src/mercury_composable/version.py index 58fc848..2956cc5 100644 --- a/src/mercury_composable/version.py +++ b/src/mercury_composable/version.py @@ -1,3 +1,3 @@ """Package version - the single Python-side source (pyproject.toml mirrors it).""" -__version__ = "0.1.0" +__version__ = "4.12.0" diff --git a/tests/test_event_stream.py b/tests/test_event_stream.py new file mode 100644 index 0000000..539722d --- /dev/null +++ b/tests/test_event_stream.py @@ -0,0 +1,484 @@ +"""Event streaming tests: the multi-shot reply contract and the envelope-mode +SSE dialect over real HTTP - the wrapper half of the engines' Phase 2/3 matrix +(Java EventOverHttpStreamTest / Rust event_over_http_stream twins).""" + +import asyncio +import base64 +import re +from collections.abc import AsyncIterator +from typing import Any + +import aiohttp +import pytest +import pytest_asyncio +from aiohttp import web + +from mercury_composable import ( + AppException, + Body, + EventEnvelope, + EventStreamWriter, + FunctionRegistry, + PostOffice, + event_stream, + trace_context, +) +from mercury_composable.server import EventApiServer + +OCTET = "application/octet-stream" +SSE = "text/event-stream" +REFUSAL = "Streaming function requires a caller that accepts text/event-stream" + +# the relay fixture learns its own host URL after the server binds +_relay_target: dict[str, str] = {} + + +def build_registry() -> FunctionRegistry: + registry = FunctionRegistry() + + async def tokens(headers: dict[str, str], event: EventEnvelope): + out = EventStreamWriter.from_request(event, registry=registry) + mode = headers.get("mode", "tokens") + if mode == "tokens": + out.first(200, SSE) + out.write("alpha") + await asyncio.sleep(0.25) + out.write("beta") + await asyncio.sleep(0.25) + out.close({"segments": 2}) + elif mode == "typed": + # every escape-hatch trigger: a dict body, text with a carriage + # return, a user event name colliding with the reserved word, a + # binary body - plus one plain token that rides a raw frame + out.first(200, SSE) + out.write({"n": 1}) + out.write_named("crlf", "line1\r\nline2") + out.write_named("envelope", "reserved-name") + out.write(b"\x01\x02\x03\x04") + out.write("plain token") + out.close({"done": True}) + elif mode == "error-mid": + out.first(200, SSE) + out.write("partial") + out.fail(AppException(503, "backend on fire")) + elif mode == "error-first": + out.fail(AppException(503, "no backend")) + elif mode == "stall": + # one-second declared idle allowance, then silence - the host must + # fail the stream in-band + out.first(200, SSE, ttl_seconds=1) + out.write("one") + elif mode == "crash-before": + raise RuntimeError("kaboom before head") + elif mode == "crash-mid": + out.first(200, SSE) + out.write("early") + raise RuntimeError("kaboom mid-stream") + elif mode == "manual": + # a single-shot manual answer from an interceptor + reply = EventEnvelope(to=event.reply_to, body={"manual": True}) + if event.cid: + reply.set_correlation_id(event.cid) + registry.send_event(reply) + elif mode == "biz": + # echo the injected business correlation-id view and the span + # lineage of this execution (continuity proof) + from mercury_composable import get_trace + info = get_trace() + out.first(200, SSE) + out.close({"my_correlation_id": headers.get("my_correlation_id"), + "span_id": info.span_id if info else None, + "parent_span_id": info.parent_span_id if info else None}) + + async def relay(_headers: dict[str, str], event: EventEnvelope): + # the composition: forward MY caller's reply address into a call + # against a remote streaming function - segments flow through verbatim + po = PostOffice(registry=registry) + try: + await po.stream_to("unit.tokens", None, reply_to=event.reply_to or "", + endpoint=_relay_target.get("url"), + timeout_ms=10000, cid=event.cid) + finally: + await po.close() + + async def echo(_headers: dict[str, str], body: Body): + return {"echo": body} + + async def biz(headers: dict[str, str], _body: Body): + return {"my_correlation_id": headers.get("my_correlation_id")} + + registry.register("unit.tokens", tokens, interceptor=True) + registry.register("unit.relay", relay, interceptor=True) + registry.register("unit.echo", echo) + registry.register("unit.biz", biz) + return registry + + +@pytest_asyncio.fixture +async def stream_host() -> AsyncIterator[tuple[str, FunctionRegistry]]: + registry = build_registry() + server = EventApiServer(registry) + runner = web.AppRunner(server.create_app()) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + port = runner.addresses[0][1] + url = f"http://127.0.0.1:{port}/api/event" + _relay_target["url"] = url + yield url, registry + await runner.cleanup() + await registry.bus.close() + + +async def collect(po: PostOffice, route: str, url: str | None, *, + mode: str | None = None, timeout_ms: int = 10000, + cid: str = "cid-100") -> list[EventEnvelope]: + headers = {"mode": mode} if mode else None + events = [] + async for event in po.stream(route, None, headers=headers, timeout_ms=timeout_ms, + endpoint=url, cid=cid): + events.append(event) + return events + + +def marker(event: EventEnvelope) -> str | None: + return event_stream.stream_signal(event) + + +# ---- the host produces the envelope-mode dialect ---- + + +async def test_streaming_target_relays_progressively(stream_host: tuple[str, FunctionRegistry]): + url, _ = stream_host + async with PostOffice() as po: + events = await collect(po, "unit.tokens", url) + assert len(events) == 3, "2 data envelopes + eof" + head = events[0] + assert marker(head) == "data" + assert head.get_status() == 200 + assert head.headers.get("content-type") == SSE + assert head.body == "alpha" + assert head.cid == "cid-100", "original correlation id restored" + assert events[1].body == "beta" + eof = events[2] + assert marker(eof) == "eof" + assert eof.body == {"segments": 2} + + +async def test_wire_is_the_hybrid_dialect(stream_host: tuple[str, FunctionRegistry]): + url, _ = stream_host + # raw wire pin: the head and the terminal ride envelope frames; the plain + # text token rides a raw frame + event = EventEnvelope(to="unit.tokens").set_header("mode", "tokens") + headers = {"content-type": OCTET, "x-ttl": "10000", "accept": SSE} + async with ( + aiohttp.ClientSession() as session, + session.post(url, data=event.to_bytes(), headers=headers) as response, + ): + assert response.status == 200 + assert response.headers["content-type"].startswith(SSE) + assert response.headers["cache-control"] == "no-cache" + text = (await response.read()).decode("utf-8") + frames = [f for f in text.split("\n\n") if f.strip()] + assert frames[0].startswith("event: envelope\n"), "head control rides an envelope frame" + assert "data: beta" in frames, "a plain token rides a raw frame" + assert frames[-1].startswith("event: envelope\n"), "the terminal is an envelope frame" + # the terminal decodes to the eof envelope with its exact metadata + encoded = frames[-1].split("data: ", 1)[1] + terminal = EventEnvelope.from_bytes(base64.b64decode(encoded)) + assert marker(terminal) == "eof" + assert terminal.body == {"segments": 2} + # host-internal addressing never leaks to the wire + assert terminal.to is None + assert terminal.reply_to is None + + +async def test_typed_segments_round_trip_exactly(stream_host: tuple[str, FunctionRegistry]): + url, _ = stream_host + async with PostOffice() as po: + events = await collect(po, "unit.tokens", url, mode="typed") + assert len(events) == 6, "5 data envelopes + eof" + assert events[0].body == {"n": 1} + crlf = events[1] + assert event_stream.stream_event_name(crlf) == "crlf" + assert crlf.body == "line1\r\nline2", "carriage return preserved" + reserved = events[2] + assert event_stream.stream_event_name(reserved) == "envelope", \ + "a user event name colliding with the reserved word survives" + assert reserved.body == "reserved-name" + assert events[3].body == b"\x01\x02\x03\x04", "binary body preserved" + assert events[4].body == "plain token" + assert marker(events[5]) == "eof" + assert events[5].body == {"done": True} + + +async def test_single_shot_over_capable_path_is_classic(stream_host: tuple[str, FunctionRegistry]): + url, _ = stream_host + async with PostOffice() as po: + # an interceptor's manual single-shot answer + events = await collect(po, "unit.tokens", url, mode="manual") + assert len(events) == 1 + assert marker(events[0]) is None + assert events[0].body == {"manual": True} + assert events[0].cid == "cid-100" + # a plain (non-interceptor) function - opting in is always safe + events = await collect(po, "unit.echo", url) + assert len(events) == 1 + assert events[0].body == {"echo": None} + + +async def test_business_cid_rides_the_streaming_hop(stream_host: tuple[str, FunctionRegistry]): + # the caller's business correlation-id (the my_cid tag) crosses the HTTP + # hop and is injected as the my_correlation_id header view at delivery + url, _ = stream_host + async with PostOffice() as po: + with trace_context("biz-trace-1", "TEST /stream", my_correlation_id="biz-42"): + events = await collect(po, "unit.tokens", url, mode="biz") + assert marker(events[-1]) == "eof" + assert events[-1].body["my_correlation_id"] == "biz-42" + + +async def test_span_lineage_continues_across_the_hop(stream_host: tuple[str, FunctionRegistry]): + # the engines' span model: the caller's span rides the outbound envelope; + # the receiving execution mints its own span with the caller's as parent + url, _ = stream_host + caller_span = "ab" * 8 + async with PostOffice() as po: + with trace_context("4bf92f3577b34da6a3ce929d0e0e4746", "TEST /lineage", + span_id=caller_span): + events = await collect(po, "unit.tokens", url, mode="biz") + body = events[-1].body + assert body["parent_span_id"] == caller_span + assert body["span_id"] != caller_span + assert re.fullmatch(r"[0-9a-f]{16}", body["span_id"]), "16-hex W3C-shaped span" + + +async def test_trace_dataset_emitted_with_engine_shape( + stream_host: tuple[str, FunctionRegistry], + caplog: pytest.LogCaptureFixture): + # non-RPC executions emit the engines' distributed-trace dataset record; + # RPC round-trips are suppressed (their metrics fold into the caller) + url, registry = stream_host + caller_span = "cd" * 8 + with caplog.at_level("INFO", logger="distributed.tracing"): + async with PostOffice() as po: + with trace_context("4bf92f3577b34da6a3ce929d0e0e4747", "TEST /telemetry", + span_id=caller_span): + await collect(po, "unit.tokens", url, mode="biz") + rpc = await PostOffice(registry=registry).request( + "unit.biz", None, timeout_ms=5000) + assert rpc.get_status() == 200 + def trace_of(message: object) -> dict[str, Any] | None: + # a dataset record's message is {"trace": {...}[, "annotations": ...]} + section = message.get("trace") if isinstance(message, dict) else None + return section if isinstance(section, dict) else None + + sections = (trace_of(r.msg) for r in caplog.records + if r.name == "distributed.tracing") + traces = [t for t in sections if t is not None] + services = [t["service"] for t in traces] + assert "unit.biz" not in services, "RPC legs emit no dataset (engine parity)" + tokens = [t for t in traces if t["service"] == "unit.tokens"] + assert len(tokens) == 1 + trace = tokens[0] + assert trace["id"] == "4bf92f3577b34da6a3ce929d0e0e4747" + assert trace["path"] == "TEST /telemetry" + assert trace["parent_span_id"] == caller_span + assert trace["success"] is True + assert trace["status"] == 200 + # an anonymous /api/event caller: the host fills the sender with its own + # identity, the engines' EventApiService parity + assert trace["from"] == "event.api.service" + for key in ("origin", "start", "exec_time", "span_id"): + assert key in trace, f"engine dataset key {key}" + + +async def test_business_cid_injected_on_local_delivery(stream_host: tuple[str, FunctionRegistry]): + # engine WorkerHandler parity: local bus deliveries inject the read-only + # view too, and no context means no injection + _, registry = stream_host + po = PostOffice(registry=registry) + with trace_context("biz-trace-2", "TEST /local", my_correlation_id="biz-7"): + reply = await po.request("unit.biz", None, timeout_ms=5000) + assert reply.body == {"my_correlation_id": "biz-7"} + plain = await po.request("unit.biz", None, timeout_ms=5000) + assert plain.body == {"my_correlation_id": None} + + +async def test_streaming_target_without_accept_is_refused_406(stream_host: tuple[str, FunctionRegistry]): + url, _ = stream_host + async with PostOffice() as po: + reply = await po.request("unit.tokens", None, headers={"mode": "tokens"}, + endpoint=url, timeout_ms=5000) + assert reply.get_status() == 406 + assert reply.body == REFUSAL + + +async def test_local_rpc_to_streaming_target_is_refused_406(stream_host: tuple[str, FunctionRegistry]): + _, registry = stream_host + po = PostOffice(registry=registry) + reply = await po.request("unit.tokens", None, headers={"mode": "tokens"}, + timeout_ms=5000) + assert reply.get_status() == 406 + assert reply.body == REFUSAL + + +async def test_mid_stream_failure_propagates_exact_status(stream_host: tuple[str, FunctionRegistry]): + url, _ = stream_host + async with PostOffice() as po: + events = await collect(po, "unit.tokens", url, mode="error-mid") + assert events[0].body == "partial" + error = events[-1] + assert marker(error) == "exception" + assert error.get_status() == 503 + # the standard error key-values: '{"type": "error", "status": n, "message": text}' + assert error.body == {"type": "error", "status": 503, "message": "backend on fire"} + + +async def test_failure_before_first_segment_arrives_as_exception(stream_host: tuple[str, FunctionRegistry]): + url, _ = stream_host + async with PostOffice() as po: + events = await collect(po, "unit.tokens", url, mode="error-first") + assert len(events) == 1 + error = events[0] + assert marker(error) == "exception" + assert error.get_status() == 503 + assert error.body["message"] == "no backend" + + +async def test_interceptor_crash_before_head_is_the_classic_error(stream_host: tuple[str, FunctionRegistry]): + url, _ = stream_host + async with PostOffice() as po: + events = await collect(po, "unit.tokens", url, mode="crash-before") + assert len(events) == 1 + assert marker(events[0]) is None, "an unstarted stream fails single-shot" + assert events[0].get_status() == 500 + assert events[0].body == "kaboom before head" + + +async def test_interceptor_crash_mid_stream_fails_in_band(stream_host: tuple[str, FunctionRegistry]): + url, _ = stream_host + async with PostOffice() as po: + events = await collect(po, "unit.tokens", url, mode="crash-mid") + assert events[0].body == "early" + error = events[-1] + assert marker(error) == "exception" + assert error.get_status() == 500 + assert error.body["message"] == "kaboom mid-stream" + + +async def test_idle_stall_fails_in_band_408(stream_host: tuple[str, FunctionRegistry]): + url, _ = stream_host + started = asyncio.get_running_loop().time() + async with PostOffice() as po: + events = await collect(po, "unit.tokens", url, mode="stall", timeout_ms=10000) + elapsed = asyncio.get_running_loop().time() - started + assert events[0].body == "one" + error = events[-1] + assert marker(error) == "exception" + assert error.get_status() == 408 + assert error.body["message"] == "Timeout for 1 seconds" + assert elapsed < 8, f"the producer's 1s idle allowance governs, took {elapsed:.1f}s" + + +async def test_relay_composition_streams_through(stream_host: tuple[str, FunctionRegistry]): + url, _ = stream_host + # the flagship: unit.relay forwards its caller's reply address into a call + # against the remote streaming function - engine-parity composition + async with PostOffice() as po: + events = await collect(po, "unit.relay", url, cid="cid-relay") + assert [e.body for e in events] == ["alpha", "beta", {"segments": 2}] + assert marker(events[-1]) == "eof" + assert events[0].cid == "cid-relay", "the original correlation id rides the chain" + + +async def test_local_stream_uses_the_same_contract(stream_host: tuple[str, FunctionRegistry]): + _, registry = stream_host + po = PostOffice(registry=registry) + events = [] + async for event in po.stream("unit.tokens", None, timeout_ms=10000, cid="cid-local"): + events.append(event) + assert [e.body for e in events] == ["alpha", "beta", {"segments": 2}] + assert marker(events[0]) == "data" + assert marker(events[-1]) == "eof" + + +# ---- the client guards the dialect against a misbehaving peer ---- + + +@pytest_asyncio.fixture +async def misbehaving_peer() -> AsyncIterator[str]: + def envelope_frame_text(event: EventEnvelope) -> str: + encoded = base64.b64encode(event.to_bytes()).decode("ascii") + return f"event: envelope\ndata: {encoded}\n\n" + + head = envelope_frame_text( + EventEnvelope(body="mock-head") + .set_header("x-event-stream", "data").set_header("content-type", SSE) + .set_status(200)) + eof = envelope_frame_text( + EventEnvelope(body={"done": True}).set_header("x-event-stream", "eof")) + + async def handle(request: web.Request) -> web.StreamResponse: + response = web.StreamResponse() + response.headers["content-type"] = SSE + await response.prepare(request) + if request.path == "/mock/raw-first": + await response.write(b"data: hello\n\n") + elif request.path == "/mock/no-terminal": + await response.write(head.encode()) + elif request.path == "/mock/foreign-dialect": + payload = head + "data: mock-token\n\n" + eof + "data: trailing-noise\n\n" + await response.write(payload.encode()) + elif request.path == "/mock/silent": + await response.write(head.encode()) + await asyncio.sleep(5) + await response.write_eof() + return response + + app = web.Application() + app.router.add_post("/mock/{tail:.*}", handle) + runner = web.AppRunner(app, shutdown_timeout=0.5) + await runner.setup() + site = web.TCPSite(runner, "127.0.0.1", 0) + await site.start() + yield f"http://127.0.0.1:{runner.addresses[0][1]}" + await runner.cleanup() + + +async def test_raw_first_frame_is_rejected(misbehaving_peer: str): + async with PostOffice() as po: + events = await collect(po, "any.route", f"{misbehaving_peer}/mock/raw-first") + assert len(events) == 1 + assert marker(events[0]) == "exception" + assert events[0].get_status() == 500 + assert events[0].body["message"] == "Invalid event stream - missing envelope head" + + +async def test_transport_end_without_terminal_is_truncation(misbehaving_peer: str): + async with PostOffice() as po: + events = await collect(po, "any.route", f"{misbehaving_peer}/mock/no-terminal") + assert events[0].body == "mock-head" + error = events[-1] + assert marker(error) == "exception" + assert error.get_status() == 500 + assert error.body["message"] == "Event stream ended without eof" + + +async def test_foreign_dialect_works_and_trailing_frames_drop(misbehaving_peer: str): + async with PostOffice() as po: + events = await collect(po, "any.route", f"{misbehaving_peer}/mock/foreign-dialect") + assert [e.body for e in events] == ["mock-head", "mock-token", {"done": True}] + assert marker(events[1]) == "data", "a raw token after the head is a data segment" + assert marker(events[2]) == "eof" + + +async def test_client_idle_guard_fails_in_band(misbehaving_peer: str): + async with PostOffice() as po: + events = await collect(po, "any.route", f"{misbehaving_peer}/mock/silent", + timeout_ms=2000) + assert events[0].body == "mock-head" + error = events[-1] + assert marker(error) == "exception" + assert error.get_status() == 408 + assert error.body["message"] == "Timeout for 2 seconds" diff --git a/tests/test_log_context.py b/tests/test_log_context.py new file mode 100644 index 0000000..31580e8 --- /dev/null +++ b/tests/test_log_context.py @@ -0,0 +1,106 @@ +"""Application log context tests - the engines' app-log-context twin: +default template rendering, the update_context developer API, feature +gating and the structured-formatter integration.""" + +import json +import logging +from collections.abc import Iterator + +import pytest + +from mercury_composable import trace_context, update_context +from mercury_composable.config import app_config +from mercury_composable.log import EngineJsonFormatter +from mercury_composable.log_context import ( + LogContextConfig, + default_template, + log_context_config, + reset_for_test, +) +from mercury_composable.trace import TraceInfo + + +@pytest.fixture(autouse=True) +def fresh_template() -> Iterator[None]: + reset_for_test() + yield + app_config().set("app.log.context", "true") + reset_for_test() + + +def info_under_test() -> TraceInfo: + return TraceInfo(route="llm.chat", trace_id="4bf92f3577b34da6a3ce929d0e0e4736", + trace_path="POST /api/agent/run", my_correlation_id="biz-7788", + span_id="82d8a6ccd03638fe", parent_span_id="00f067aa0ba902b7") + + +def test_default_template_renders_the_standard_trace_context(): + # the default ships as a packaged YAML resource, the engines' twin + template = default_template() + assert template is not None, "packaged default-log-context.yaml must load" + context = LogContextConfig(template).render(info_under_test()) + assert context == { + "cid": "biz-7788", + "traceId": "4bf92f3577b34da6a3ce929d0e0e4736", + "tracePath": "POST /api/agent/run", + "spanId": "82d8a6ccd03638fe", + "parentSpanId": "00f067aa0ba902b7", + "service": "llm.chat", + "timestamp": context["timestamp"], + } + # absent values are omitted, never rendered as "null" - and cid is the + # BUSINESS correlation-id only + bare = LogContextConfig(default_template()).render( + TraceInfo(route="llm.chat", trace_id="t-1", cid="internal-routing-id")) + assert "cid" not in bare + assert "spanId" not in bare + + +def test_custom_template_tokens_constants_and_validation(): + config = LogContextConfig({"trace": "$traceId", "deployment": "blue"}) + context = config.render(info_under_test()) + assert context == {"trace": "4bf92f3577b34da6a3ce929d0e0e4736", + "deployment": "blue"} + with pytest.raises(ValueError, match="Invalid log context token"): + LogContextConfig({"x": "$bogus"}) + + +def test_update_context_merges_and_guards_reserved_keys(): + config = LogContextConfig(default_template()) + with trace_context("trace-ctx-1", "TEST /ctx") as info: + update_context("tenant", "acme") + assert config.render(info)["tenant"] == "acme" + update_context("tenant", None) + assert "tenant" not in config.render(info) + with pytest.raises(ValueError, match="reserved"): + update_context("cid", "nope") + # outside a hosted request: a silent no-op, the engines' semantics + update_context("tenant", "ignored") + + +def test_structured_formatter_emits_the_context_block(): + record = logging.LogRecord(name="app", level=logging.INFO, pathname=__file__, + lineno=7, msg="charging the order", args=None, + exc_info=None) + formatter = EngineJsonFormatter(compact=True) + with trace_context("4bf92f3577b34da6a3ce929d0e0e4736", "POST /api/agent/run", + my_correlation_id="biz-7788"): + entry = json.loads(formatter.format(record)) + assert entry["message"] == "charging the order" + context = entry["context"] + assert context["cid"] == "biz-7788" + assert context["traceId"] == "4bf92f3577b34da6a3ce929d0e0e4736" + assert context["tracePath"] == "POST /api/agent/run" + # outside a trace: no context block at all + assert "context" not in json.loads(formatter.format(record)) + + +def test_feature_flag_disables_the_context_block(): + app_config().set("app.log.context", "false") + reset_for_test() + assert log_context_config().enabled is False + record = logging.LogRecord(name="app", level=logging.INFO, pathname=__file__, + lineno=7, msg="quiet", args=None, exc_info=None) + with trace_context("trace-off-1", "TEST /off"): + entry = json.loads(EngineJsonFormatter(compact=True).format(record)) + assert "context" not in entry