diff --git a/.gitignore b/.gitignore index ad9542d61..2437871ed 100644 --- a/.gitignore +++ b/.gitignore @@ -29,6 +29,8 @@ __pycache__/ # Local runtime state .omx/ +.serena/ +docs/ data/ *.pid *.zip diff --git a/harness/ARCHITECTURE.md b/harness/ARCHITECTURE.md index c109b9c2d..1234f5313 100644 --- a/harness/ARCHITECTURE.md +++ b/harness/ARCHITECTURE.md @@ -94,23 +94,21 @@ The harness owns *no* logic from any of these — it only knows their names. Eac `EXPECTED_WORKERS` is generated from `iii.worker.yaml` at build time by `build.rs`, so the two cannot drift — there is no separate sync test to maintain. -### 4. `scripts/demo.sh` — local orchestration +### 4. `Makefile` — local orchestration -For registry-based installs, `iii worker add harness` fetches the harness binary and its declared dependencies automatically (see `registry/index.json`). `scripts/demo.sh` is the alternative path for local development from a source checkout: +For registry-based installs, `iii worker add harness` fetches the harness binary and its declared dependencies automatically (see `registry/index.json`). The `Makefile` is the alternative path for local development from a source checkout: ``` -demo.sh build # cargo build --release for harness + dep workers -demo.sh engine # start `iii --use-default-config` in background -demo.sh start # spawn all workers + harness as nohup processes -demo.sh verify # call harness::status, models::list, provider::cli::list_models -demo.sh web # npm install + vite in a tmux session -demo.sh stop # kill every PID in $DEMO_DIR/pids/ + engine + tmux -demo.sh all # build + engine + start + verify +make config # generate config.yaml + iii.lock via `iii worker add .` +make observability # add iii-observability to config.yaml (powers TRACES tab) +make engine # start `iii --config config.yaml` in background +make verify # call harness::status + models::list +make web # vite dev server on :5173 +make stop # kill engine + web +make all # config + observability + engine + verify ``` -PIDs and logs live under `$DEMO_DIR` (default `~/iii-harness-demo`). One PID file per worker, one log file per worker — no shared logger, no daemon supervisor. - -`scripts/real-usage.sh` exercises the running stack end-to-end: `auth::set_token` → `run::start_and_wait` → `state::get` for both messages and turn record → `state::list` to enumerate sessions. +PIDs and logs live under `$DEMO_DIR` (default `~/iii-harness-demo`). The engine spawns each worker via its `iii.worker.yaml` `scripts.start`. ## Runtime data flow diff --git a/harness/Cargo.lock b/harness/Cargo.lock index 56fc91b84..9e09846ba 100644 --- a/harness/Cargo.lock +++ b/harness/Cargo.lock @@ -468,6 +468,7 @@ dependencies = [ "tokio", "tracing", "tracing-subscriber", + "uuid", "which", ] @@ -734,9 +735,9 @@ dependencies = [ [[package]] name = "iii-sdk" -version = "0.11.3" +version = "0.11.7-next.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f0226f7ce0d9071f9cb75ea7b7ac1241b15282915ccd41d9bbd2ee0db94f90c6" +checksum = "abad1f632473931d97733a4b3d73a04f2cfa1759fa4d4bcdab9305b634563bf7" dependencies = [ "async-trait", "futures-util", diff --git a/harness/Cargo.toml b/harness/Cargo.toml index 2e5458145..9ca2cc93f 100644 --- a/harness/Cargo.toml +++ b/harness/Cargo.toml @@ -20,7 +20,7 @@ name = "harness" path = "src/main.rs" [dependencies] -iii-sdk = "=0.11.3" +iii-sdk = "=0.11.7-next.3" serde = { version = "1", features = ["derive"] } serde_json = "1" serde_yaml = "0.9" @@ -33,6 +33,7 @@ tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] } [dev-dependencies] harness-types = { path = "crates/harness-types" } +uuid = { version = "1", features = ["v4"] } serial_test = "3" which = "8" diff --git a/harness/Makefile b/harness/Makefile index 63c894cff..fd91bccbf 100644 --- a/harness/Makefile +++ b/harness/Makefile @@ -13,8 +13,10 @@ # # Usage: # make # help -# make all # config + engine + verify +# make all # build + config + observability + engine + verify +# make build # cargo build --release + symlink into ~/.iii/workers/ # make config # (re)generate config.yaml + iii.lock via `iii worker add .` +# make observability # iii worker add iii-observability (powers TRACES tab) # make engine # start `iii` in background reading harness/config.yaml # make verify # call harness::status + models::list # make web # background vite dev server on :5173 (no tmux) @@ -31,6 +33,8 @@ SHELL := bash MAKEFLAGS += --no-print-directory HARNESS_DIR := $(abspath $(dir $(lastword $(MAKEFILE_LIST)))) +WORKERS_REPO := $(abspath $(HARNESS_DIR)/..) +WORKERS_BIN := $(HOME)/.iii/workers DEMO_DIR ?= $(HOME)/iii-harness-demo DEMO_ENGINE_WS ?= $(or $(III_DEMO_ENGINE_URL),ws://127.0.0.1:49134) PIDS_DIR := $(DEMO_DIR)/pids @@ -38,15 +42,39 @@ LOGS_DIR := $(DEMO_DIR)/logs CONFIG_FILE := $(HARNESS_DIR)/config.yaml -.PHONY: help all config engine verify web stop restart logs clean ensure-dirs +# Local worker crates that need to be cargo-built and symlinked into +# ~/.iii/workers/ so the engine spawns them on host instead of trying +# `cargo run` inside libkrun (where cargo isn't installed). +LOCAL_WORKERS := approval-gate auth-credentials hook-fanout iii-directory \ + llm-budget models-catalog policy-denylist provider-anthropic \ + provider-openai provider-router session shell turn-orchestrator \ + harness + +.PHONY: help all build config observability engine verify web stop restart logs clean ensure-dirs help: @awk '/^[^#]/ && !/^$$/ {exit} /^#/ {sub(/^# ?/, ""); print}' $(firstword $(MAKEFILE_LIST)) -all: config engine verify +all: build config observability engine verify ensure-dirs: - @mkdir -p $(PIDS_DIR) $(LOGS_DIR) $(HARNESS_DIR)/data/skills + @mkdir -p $(PIDS_DIR) $(LOGS_DIR) $(HARNESS_DIR)/data/skills $(WORKERS_BIN) + +# ─── build ─────────────────────────────────────────────────────────────────── + +# Cargo-builds every local worker in release mode and symlinks each binary +# into ~/.iii/workers/. The engine looks up workers by that path; +# symlinks let us iterate on source without re-running `iii worker add`. +# Each worker is its own cargo workspace (no shared top-level workspace), +# so we loop and `cargo build --release` per crate. +build: ensure-dirs + @for w in $(LOCAL_WORKERS); do \ + echo "==> cargo build --release: $$w"; \ + ( cd "$(WORKERS_REPO)/$$w" && cargo build --release --quiet ) \ + || { echo " [error] $$w: cargo build failed"; exit 1; }; \ + ln -sf "$(WORKERS_REPO)/$$w/target/release/$$w" "$(WORKERS_BIN)/$$w"; \ + done + @echo "==> all $(words $(LOCAL_WORKERS)) workers built and symlinked into $(WORKERS_BIN)" # ─── config ────────────────────────────────────────────────────────────────── @@ -62,6 +90,18 @@ $(CONFIG_FILE): $(HARNESS_DIR)/iii.worker.yaml @echo "==> generating $@ via iii worker add" @cd "$(HARNESS_DIR)" && iii worker add . --no-wait +# ─── observability ─────────────────────────────────────────────────────────── + +# Adds iii-observability to config.yaml. Powers the iii Developer Console +# TRACES tab and `engine::traces::*` query path. Kept OUT of +# iii.worker.yaml `dependencies:` (and thus EXPECTED_WORKERS) so prod stacks +# without observability don't flag missing in `harness::status`. `iii worker +# add` is idempotent — re-running is a no-op. +observability: config + @command -v iii >/dev/null || { echo "iii CLI not found"; exit 1; } + @echo "==> iii worker add iii-observability (idempotent)" + @cd "$(HARNESS_DIR)" && iii worker add iii-observability --no-wait + # ─── engine ────────────────────────────────────────────────────────────────── engine: ensure-dirs config diff --git a/harness/README.md b/harness/README.md index de3e282e9..803947a73 100644 --- a/harness/README.md +++ b/harness/README.md @@ -1,14 +1,6 @@ # harness -Meta-worker that composes the modular workers behind a runnable iii chat -surface and exposes the browser-facing HTTP bridge (`harness::call`) -the bundled Vite/React UI talks to. The harness does -not own chat, agent, or provider logic — it registers a small set of -bus functions and expects peers such as -[`turn-orchestrator`](../turn-orchestrator), -[`provider-router`](../provider-router), shell tools, and related -workers to be installed alongside it. `iii worker add harness` pulls -the whole bundle in transitively. +Meta-worker that composes fifteen modular workers into a runnable iii chat surface, exposes a browser-facing HTTP bridge (`bridge::trigger`, `bridge::events`), and ships a Vite/React UI that talks to the bus through it. The harness does not own chat, agent, or provider logic; it registers a small set of bus functions and expects peers such as `turn-orchestrator`, `provider-router`, shell tools, and related workers to be installed alongside it. Deeper layout and streams behavior are documented in [`ARCHITECTURE.md`](ARCHITECTURE.md). ## Install @@ -16,16 +8,12 @@ the whole bundle in transitively. iii worker add harness ``` -`iii worker add` fetches the binary, writes a config block into -`~/.iii/config.yaml`, resolves every transitive worker from -`iii.worker.yaml` `dependencies:`, and the engine starts the bundle on -the next `iii start`. +`iii worker add` fetches the binary, writes a config block into `~/.iii/config.yaml`, and the engine starts the worker on the next `iii start`. -To back chat history with durable SQL storage instead of the bundled -in-memory `iii-state`, add the [`iii-database`](../iii-database) worker: +To register the harness skill bundle metadata with the bus (the worker does this automatically at boot when `skills` is available), ensure the [skills](../skills) worker is part of your stack: ```bash -iii worker add iii-database +iii worker add skills ``` ## Quickstart @@ -54,12 +42,13 @@ async fn main() -> anyhow::Result<()> { } ``` -Forward an arbitrary bus call through the HTTP-oriented bridge: +Forward an arbitrary bus call through the HTTP-oriented bridge (same shape as `bridge::trigger` on the engine): ```rust +// function_id / payload match iii.trigger(...) let result = iii .trigger(TriggerRequest { - function_id: "harness::call".into(), + function_id: "bridge::trigger".into(), payload: json!({ "function_id": "models::list", "payload": {}, @@ -70,23 +59,138 @@ let result = iii .await?; ``` -Registered functions: +Registered functions (use `::` ids on the bus): | Function | Role | |---|---| | `harness::status` | Bundle name, version, and expected worker list (cheap liveness probe). | -| `harness::call` | Forwards `{ function_id, payload }` to `iii.trigger`. HTTP: `POST harness/call`. | +| `bridge::trigger` | Forwards `{ function_id, payload }` to `iii.trigger`. HTTP: `POST` `bridge/trigger`. | +| `bridge::events` | SSE-style tail of `agent::events` for a session. HTTP: `GET` `bridge/events`. | -`harness::call` is the browser's call-anything escape hatch — not -meant as an LLM tool. +`bridge::trigger` is not meant as an LLM tool — it is the browser’s call-anything escape hatch. ## Configuration ```yaml -engine_url: "ws://127.0.0.1:49134" # WebSocket URL when III_URL / --url are unset +# Default engine WebSocket URL when III_URL / --url are unset +engine_url: "ws://127.0.0.1:49134" ``` -Runtime flags: +Other runtime flags: - `--config` — path to this file (default `./config.yaml`; override with `III_HARNESS_CONFIG`). -- `--url` / `III_URL` — engine WebSocket URL; wins over `engine_url` in the file. +- `--url` or `III_URL` — engine WebSocket URL; wins over `engine_url` in the file. + +Registry-facing defaults also appear in `iii-harness --manifest` under `default_config`. + +## Expected workers + +`EXPECTED_WORKERS` (in [`src/lib.rs`](src/lib.rs)) is generated at build time +from the `dependencies:` block of [`iii.worker.yaml`](iii.worker.yaml) by +[`build.rs`](build.rs). Add or remove a worker by editing `iii.worker.yaml` +only — the Rust constant rebuilds automatically. + +## Trace correlation + +Every harness-registered function wraps its body in an OTel span tagged with +`iii.session.id`, `iii.message.id`, and (for `bridge::trigger` only) +`iii.function.id`. The HTTP response carries two new headers when +observability is active: + +- `traceparent: 00---01` — W3C trace context for the span + that wrapped this call. +- `x-iii-message-id: ` — the `message_id` you sent on the request, or + the upstream value propagated via OTel baggage. **Omitted entirely** when + neither source supplied one. This keeps plumbing calls (UI subscribes, + status polls, engine-internal traffic) out of `Group by message` in the + console — only real chat-turn IDs land there. + +Discover harness traces in the iii Developer Console TRACES tab, or via: + +```bash +# By span name (any harness function): +iii trigger --function-id engine::traces::list \ + --payload '{"name":"harness.status","search_all_spans":true}' + +# By message_id directly (engine v0.11.7+ — needs the search_all_spans +# attribute-filter widening + iii-sdk BaggageSpanProcessor; works on +# every span in the trace, not just the harness-wrapped one): +iii trigger --function-id engine::traces::list \ + --payload '{"attributes":[["iii.message.id",""]],"search_all_spans":true}' + +# Server-side aggregation (engine v0.11.7+): +iii trigger --function-id engine::traces::group_by \ + --payload '{"attribute":"iii.message.id"}' +``` + +Both headers are absent when the iii-observability worker is not running +(see `harness/config.yaml`). Web clients should treat them as optional — +"`traceparent` absent" means "observability is off," not "the call failed." + +#### Operator observability of the wrapper itself + +The wrapper emits a `tracing::trace!` event per span entry with `fn_name`, +`recording` (whether OTel is active), `session_id`, and `message_id_minted` +(always `false` since the harness no longer mints; kept for log-format +stability). +Tail with `RUST_LOG=harness::otel=trace` to detect the +"observability worker went silent" failure mode (rising `recording=false` +rate without an OTel runtime change). + +#### Baggage propagation (and why TRACES doesn't group by message_id yet) + +The wrapper also writes `iii.session.id`, `iii.message.id`, and (for +`bridge::trigger` only) `iii.function.id` into the OTel **baggage** of the +context attached around the handler. Every downstream `iii.trigger(...)` +call ships the baggage on the wire automatically (iii-sdk's `inject_baggage` +is wired into the invocation message at `iii-sdk/src/iii.rs:312`). Receiving +workers extract it via `extract_context(traceparent, baggage)` and the +entries live in their task-local OTel context for the duration of the +handler. + +What this does NOT do yet: **baggage entries are not automatically copied +onto span attributes** of downstream worker spans. The OTel SDK requires an +explicit `SpanProcessor` that reads baggage on `on_start` and writes it to +the span as attributes; none exists in `iii-observability` today. So in the +iii Developer Console TRACES tab, downstream spans (e.g. `state::set`, +`approval::list_pending`) still appear without `iii.message.id` even though +the baggage *is* travelling alongside them. + +Required engine-side follow-up to make TRACES group by message: + +1. Add a span processor in `iii-observability` that copies a configurable + allowlist of baggage keys onto each span at start time (allowlist defaults + to `iii.session.id`, `iii.message.id`, `iii.function.id`). +2. Optionally extend `engine::traces::list` so `search_all_spans: true` also + applies the attribute filter (currently root-only — documented above). +3. Optionally, surface a "group by attribute" affordance in the TRACES tab. + +Once (1) lands, every span in the trace inherits the ids automatically. The +harness side is forward-compatible: the baggage is already flowing. + +### Direct-bus return shapes + +Two functions return the HTTP-trigger envelope `{status_code, headers, body}`: +`bridge::trigger` and `bridge::events`. The other five — `harness::status`, +`bridge::info`, `ui::subscribe`, `ui::unsubscribe`, `harness::fs::read_inline` +— return their raw payloads so direct-WebSocket callers (the web `StatusPill`, +`fetchBridgeInfo`, `ui::subscribe` registration, FilesystemPanel reads) can +read fields off the top level. The OTel span still fires with `iii.*` +attributes for all seven; only the HTTP `traceparent` / `x-iii-message-id` +header echo is skipped for the five raw-shape functions (their wrapper sees +no `status_code` in the return and leaves headers untouched). + +> **Contract reminder.** Any change to a wrapped function's return shape +> (envelope ↔ raw) is a breaking change for direct-WS consumers in +> `harness/web/`. Commit `767c83d` reverted four functions from envelope back +> to raw after the unified-envelope rollout broke `StatusPill.tsx` at +> runtime. The wrapper variants document the contract at the call site — +> `with_envelope_span` for `bridge::trigger`/`bridge::events`, `with_raw_span` +> for the other five. Keep them aligned with their consumers. + +### Wildcard subscriptions + +`ui::subscribe` / `ui::unsubscribe` accept `session_id: null` to mean "all +sessions." In TRACES those calls show up with `iii.session.id = "*"`. + +Contributor commands (fmt, clippy, tests) for this crate live in [`binary-worker.md`](../binary-worker.md) §11; source layout notes are in [`ARCHITECTURE.md`](ARCHITECTURE.md). diff --git a/harness/src/lib.rs b/harness/src/lib.rs index dc57c4840..d9f2598ea 100644 --- a/harness/src/lib.rs +++ b/harness/src/lib.rs @@ -5,6 +5,7 @@ pub mod fanout; pub mod fs; +pub mod otel; use std::sync::Arc; @@ -79,13 +80,23 @@ pub async fn register_with_iii_with_engine_url( "Returns the harness bundle name, version, and the list of expected runtime workers." .into(), ), - |_payload: Value| async move { - Ok::<_, IIIError>(json!({ - "ok": true, - "name": env!("CARGO_PKG_NAME"), - "version": env!("CARGO_PKG_VERSION"), - "expected_workers": EXPECTED_WORKERS, - })) + |input: Value| async move { + crate::otel::with_harness_span( + "harness.status", + input, + crate::otel::IdSource::BodyOrTopLevel, + crate::otel::WildcardMode::Disabled, + None, + |_input| async move { + Ok::<_, IIIError>(json!({ + "ok": true, + "name": env!("CARGO_PKG_NAME"), + "version": env!("CARGO_PKG_VERSION"), + "expected_workers": EXPECTED_WORKERS, + })) + }, + ) + .await }, )); @@ -99,29 +110,43 @@ pub async fn register_with_iii_with_engine_url( move |input: Value| { let iii = iii_for_bridge.clone(); async move { - // HTTP trigger wraps the request body as { body, query_params, headers, ... }. - // Direct bus callers send { function_id, payload } at the top level. - let body = input.get("body").cloned().unwrap_or(input); - let function_id = body + let inner_function_id = input + .get("body") + .unwrap_or(&input) .get("function_id") .and_then(Value::as_str) - .ok_or_else(|| IIIError::Handler("missing function_id".into()))? - .to_string(); - let inner = body.get("payload").cloned().unwrap_or_else(|| json!({})); - let result = iii - .trigger(TriggerRequest { - function_id, - payload: inner, - action: None, - timeout_ms: Some(BRIDGE_TIMEOUT_MS), - }) - .await - .map_err(|e| IIIError::Handler(e.to_string()))?; - Ok::<_, IIIError>(json!({ - "status_code": 200, - "headers": { "content-type": "application/json" }, - "body": result, - })) + .map(str::to_string); + crate::otel::with_harness_span( + "harness.call", + input, + crate::otel::IdSource::BridgeTrigger, + crate::otel::WildcardMode::Disabled, + inner_function_id.as_deref(), + |input| async move { + let body = input.get("body").cloned().unwrap_or(input); + let function_id = body + .get("function_id") + .and_then(Value::as_str) + .ok_or_else(|| IIIError::Handler("missing function_id".into()))? + .to_string(); + let inner = body.get("payload").cloned().unwrap_or_else(|| json!({})); + let result = iii + .trigger(TriggerRequest { + function_id, + payload: inner, + action: None, + timeout_ms: Some(BRIDGE_TIMEOUT_MS), + }) + .await + .map_err(|e| IIIError::Handler(e.to_string()))?; + Ok::<_, IIIError>(json!({ + "status_code": 200, + "headers": { "content-type": "application/json" }, + "body": result, + })) + }, + ) + .await } }, )); @@ -147,14 +172,24 @@ pub async fn register_with_iii_with_engine_url( RegisterFunctionMessage::with_id("harness::info".into()).with_description( "Returns the relative WebSocket path and engine URL for browser clients.".into(), ), - move |_payload: Value| { + move |input: Value| { let engine_url = engine_url_owned.clone(); async move { - Ok::<_, IIIError>(json!({ - "ws_path": "/iii/ws", - "protocol": "ws", - "engine_url": engine_url, - })) + crate::otel::with_harness_span( + "harness.info", + input, + crate::otel::IdSource::BodyOrTopLevel, + crate::otel::WildcardMode::Disabled, + None, + |_input| async move { + Ok::<_, IIIError>(json!({ + "ws_path": "/iii/ws", + "protocol": "ws", + "engine_url": engine_url, + })) + }, + ) + .await } }, )); @@ -172,25 +207,35 @@ pub async fn register_with_iii_with_engine_url( move |input: Value| { let fanout = Arc::clone(&fanout_for_subscribe); async move { - let body = input.get("body").cloned().unwrap_or(input); - let browser_id = body - .get("browser_id") - .and_then(Value::as_str) - .ok_or_else(|| IIIError::Handler("missing browser_id".into()))? - .to_string(); - let session_id = body - .get("session_id") - .and_then(Value::as_str) - .map(str::to_string); - let total = { - let mut state = fanout.write().await; - state.subscribe(browser_id, session_id); - state.browser_count() - }; - Ok::<_, IIIError>(json!({ - "ok": true, - "total_browsers": total, - })) + crate::otel::with_harness_span( + "harness.ui.subscribe", + input, + crate::otel::IdSource::BodyOrTopLevel, + crate::otel::WildcardMode::OnExplicitNull, + None, + |input| async move { + let body = input.get("body").cloned().unwrap_or(input); + let browser_id = body + .get("browser_id") + .and_then(Value::as_str) + .ok_or_else(|| IIIError::Handler("missing browser_id".into()))? + .to_string(); + let session_id = body + .get("session_id") + .and_then(Value::as_str) + .map(str::to_string); + let total = { + let mut state = fanout.write().await; + state.subscribe(browser_id, session_id); + state.browser_count() + }; + Ok::<_, IIIError>(json!({ + "ok": true, + "total_browsers": total, + })) + }, + ) + .await } }, )); @@ -204,25 +249,35 @@ pub async fn register_with_iii_with_engine_url( move |input: Value| { let fanout = Arc::clone(&fanout_for_unsubscribe); async move { - let body = input.get("body").cloned().unwrap_or(input); - let browser_id = body - .get("browser_id") - .and_then(Value::as_str) - .ok_or_else(|| IIIError::Handler("missing browser_id".into()))? - .to_string(); - let session_id = body - .get("session_id") - .and_then(Value::as_str) - .map(str::to_string); - let total = { - let mut state = fanout.write().await; - state.unsubscribe(&browser_id, session_id); - state.browser_count() - }; - Ok::<_, IIIError>(json!({ - "ok": true, - "total_browsers": total, - })) + crate::otel::with_harness_span( + "harness.ui.unsubscribe", + input, + crate::otel::IdSource::BodyOrTopLevel, + crate::otel::WildcardMode::OnExplicitNull, + None, + |input| async move { + let body = input.get("body").cloned().unwrap_or(input); + let browser_id = body + .get("browser_id") + .and_then(Value::as_str) + .ok_or_else(|| IIIError::Handler("missing browser_id".into()))? + .to_string(); + let session_id = body + .get("session_id") + .and_then(Value::as_str) + .map(str::to_string); + let total = { + let mut state = fanout.write().await; + state.unsubscribe(&browser_id, session_id); + state.browser_count() + }; + Ok::<_, IIIError>(json!({ + "ok": true, + "total_browsers": total, + })) + }, + ) + .await } }, )); @@ -253,13 +308,20 @@ pub async fn register_with_iii_with_engine_url( let iii = iii_for_read.clone(); let ws_base = engine_url_for_read.clone(); async move { - // Direct callers send args at the top level. HTTP triggers - // wrap as { body, ... } — the bridge already unwraps before - // forwarding, so we accept either by trying `body` first. - let body = input.get("body").cloned().unwrap_or(input); - let args: fs::ReadInlineArgs = serde_json::from_value(body) - .map_err(|e| IIIError::Handler(format!("bad read_inline args: {e}")))?; - fs::read_inline(&iii, &ws_base, args).await + Box::pin(crate::otel::with_harness_span( + "harness.fs.read_inline", + input, + crate::otel::IdSource::BodyOrTopLevel, + crate::otel::WildcardMode::Disabled, + None, + |input| async move { + let body = input.get("body").cloned().unwrap_or(input); + let args: fs::ReadInlineArgs = serde_json::from_value(body) + .map_err(|e| IIIError::Handler(format!("bad read_inline args: {e}")))?; + fs::read_inline(&iii, &ws_base, args).await + }, + )) + .await } }, )); diff --git a/harness/src/otel.rs b/harness/src/otel.rs new file mode 100644 index 000000000..9e02e94af --- /dev/null +++ b/harness/src/otel.rs @@ -0,0 +1,909 @@ +//! Harness-side OTel wrapper. +//! +//! Tags every harness span with `iii.session.id` / `iii.message.id` / +//! `iii.function.id` and propagates them via baggage. See +//! `docs/superpowers/specs/2026-05-12-harness-trace-correlation-design.md`. + +use std::future::Future; + +use iii_sdk::IIIError; +use serde_json::Value; + +/// 128 chars leaves room for app-specific IDs while keeping response headers +/// under typical proxy caps and OTel attribute values within collector limits. +pub(crate) const MAX_ID_LEN: usize = 128; + +pub const WILDCARD_SESSION_ID: &str = "*"; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum WildcardMode { + /// `session_id: null` → `Some("*")`. For ui::subscribe / ui::unsubscribe. + OnExplicitNull, + Disabled, +} + +#[derive(Clone, Copy, Debug)] +pub enum IdSource { + /// `body.session_id` → top-level → `body.payload.session_id`. + BridgeTrigger, + QueryParams, + /// `body.session_id` → top-level. + BodyOrTopLevel, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct HarnessIds { + pub session_id: Option, + pub message_id: Option, + /// Always `false` after the Option-A switch; kept for source-compat. + pub message_id_was_minted: bool, +} + +/// Extract IDs from `input` according to `source`. Pure helper — no OTel +/// side-effects on its core path. +/// +/// Resolution: payload → OTel baggage (set by an upstream harness wrapper). +/// Without the baggage fallback an inner wrapper would mint a fresh +/// `message_id` even when an outer wrapper already established one, +/// fragmenting the trace across "message" groups in TRACES. +/// +/// Strings over `MAX_ID_LEN` or containing CR/LF/control bytes are treated +/// as missing so response headers and span attributes can't be poisoned. +#[must_use] +pub fn extract_ids(input: &Value, source: IdSource, wildcard: WildcardMode) -> HarnessIds { + let (session_present, session_value, message_value) = match source { + IdSource::BridgeTrigger => { + let (s_present, s, m, _fid) = extract_bridge_trigger_ids(input); + (s_present, s, m) + } + IdSource::QueryParams => extract_query_params_ids(input), + IdSource::BodyOrTopLevel => extract_body_or_top_level_ids(input), + }; + + // Wildcard takes precedence over baggage for ui::subscribe / ui::unsubscribe. + let session_id = + if wildcard == WildcardMode::OnExplicitNull && session_present && session_value.is_none() { + Some(WILDCARD_SESSION_ID.to_string()) + } else { + session_value.or_else(|| iii_sdk::get_baggage_entry("iii.session.id")) + }; + + let message_id = message_value.or_else(|| iii_sdk::get_baggage_entry("iii.message.id")); + + HarnessIds { + session_id, + message_id, + message_id_was_minted: false, + } +} + +/// Single-walk `bridge::trigger` extractor that also returns the inner +/// `function_id` (so callers don't re-walk the JSON for a span attribute). +#[must_use] +pub fn extract_ids_for_bridge_trigger(input: &Value) -> (HarnessIds, Option) { + let (_, s_value, m_value, function_id) = extract_bridge_trigger_ids(input); + let ids = HarnessIds { + session_id: s_value, + message_id: m_value, + message_id_was_minted: false, + }; + (ids, function_id) +} + +/// Merge `traceparent` and `x-iii-message-id` into the response envelope. +/// No-op when `out` lacks a `status_code` field (raw-shape callers). +#[must_use] +pub fn merge_response_headers( + mut out: Value, + trace_id: Option<&str>, + span_id: Option<&str>, + message_id: Option<&str>, +) -> Value { + let Some(obj) = out.as_object_mut() else { + return out; + }; + + if !obj.contains_key("status_code") { + return out; + } + + let headers = obj + .entry("headers".to_string()) + .or_insert_with(|| Value::Object(serde_json::Map::new())); + + if let Some(headers_obj) = headers.as_object_mut() { + if let (Some(tid), Some(sid)) = (trace_id, span_id) { + let mut tp = String::with_capacity(3 + tid.len() + 1 + sid.len() + 3); + tp.push_str("00-"); + tp.push_str(tid); + tp.push('-'); + tp.push_str(sid); + tp.push_str("-01"); + headers_obj.insert("traceparent".to_string(), Value::String(tp)); + } + if let Some(mid) = message_id { + headers_obj.insert( + "x-iii-message-id".to_string(), + Value::String(mid.to_string()), + ); + } + } + + out +} + +/// Run `f` inside an OTel span named `fn_name`. +/// +/// Returns its result with `traceparent` + `x-iii-message-id` merged into +/// the response envelope. Prefer [`with_envelope_span`] / [`with_raw_span`] +/// at call sites. +pub async fn with_harness_span( + fn_name: &'static str, + input: Value, + source: IdSource, + wildcard: WildcardMode, + inner_function_id: Option<&str>, + f: F, +) -> Result +where + F: FnOnce(Value) -> Fut + Send, + Fut: Future> + Send, +{ + let ids = extract_ids(&input, source, wildcard); + run_in_span(fn_name, ids, inner_function_id, move || f(input)).await +} + +/// Envelope variant: closure returns `{status_code, headers, body}`. +/// Behaviorally identical to [`with_harness_span`]; the name documents intent. +pub async fn with_envelope_span( + fn_name: &'static str, + input: Value, + source: IdSource, + wildcard: WildcardMode, + inner_function_id: Option<&str>, + f: F, +) -> Result +where + F: FnOnce(Value) -> Fut + Send, + Fut: Future> + Send, +{ + with_harness_span(fn_name, input, source, wildcard, inner_function_id, f).await +} + +/// Raw variant: closure returns its payload directly. Header injection is +/// skipped (no `status_code`), but the OTel span still carries `iii.*` attrs. +pub async fn with_raw_span( + fn_name: &'static str, + input: Value, + source: IdSource, + wildcard: WildcardMode, + inner_function_id: Option<&str>, + f: F, +) -> Result +where + F: FnOnce(Value) -> Fut + Send, + Fut: Future> + Send, +{ + with_harness_span(fn_name, input, source, wildcard, inner_function_id, f).await +} + +/// Precedence: outer (`body.{session_id,message_id}`) wins as a unit; only +/// when BOTH are absent do we read `body.payload.*`. Mixing outer+inner is +/// ambiguous so we reject the composition intentionally (tested). +fn extract_bridge_trigger_ids( + input: &Value, +) -> (bool, Option, Option, Option) { + // HTTP-trigger shape: { body: { session_id?, message_id?, function_id, payload: { session_id?, message_id? } } } + // Direct-bus shape: { session_id?, message_id?, function_id, payload } + let body = input.get("body").unwrap_or(input); + + let function_id = match body.get("function_id") { + Some(Value::String(s)) + if !s.is_empty() && s.len() <= MAX_ID_LEN && s.bytes().all(is_safe_id_char) => + { + Some(s.clone()) + } + Some(Value::String(s)) => { + // Present-but-rejected. Distinguish reason for the operator. + let reason = if s.is_empty() { + "empty" + } else if s.len() > MAX_ID_LEN { + "over_max_id_len" + } else { + "control_bytes" + }; + tracing::warn!( + len = s.len(), + reason, + max_id_len = MAX_ID_LEN, + "bridge::trigger function_id rejected; treating as absent (handler will surface 'missing function_id' error). \ + Raw bytes not logged (may carry CRLF/control content).", + ); + None + } + // Absent OR wrong-type (number, bool, array, object). Don't log; + // "field missing entirely" is the common, expected case from a + // client that hasn't filled the payload yet. + _ => None, + }; + + let (s_present_outer, s_outer) = get_string_field(body, "session_id"); + let (_, m_outer) = get_string_field(body, "message_id"); + + if s_present_outer || m_outer.is_some() { + return (s_present_outer, s_outer, m_outer, function_id); + } + + // Fall back to the nested payload object. Note: this branch is only + // reached when BOTH outer fields are absent — so the returned + // `session_present` here is exclusively determined by the inner read. + let payload = body.get("payload").unwrap_or(&Value::Null); + let (s_present_inner, s_inner) = get_string_field(payload, "session_id"); + let (_, m_inner) = get_string_field(payload, "message_id"); + debug_assert!( + !s_present_outer, + "fall-through path requires outer session absent" + ); + (s_present_inner, s_inner, m_inner, function_id) +} + +fn extract_query_params_ids(input: &Value) -> (bool, Option, Option) { + let q = input.get("query_params").unwrap_or(&Value::Null); + let (s_present, s) = get_string_field(q, "session_id"); + let (_, m) = get_string_field(q, "message_id"); + (s_present, s, m) +} + +fn extract_body_or_top_level_ids(input: &Value) -> (bool, Option, Option) { + let body = input.get("body").unwrap_or(input); + let (s_present, s) = get_string_field(body, "session_id"); + let (_, m) = get_string_field(body, "message_id"); + (s_present, s, m) +} + +/// Printable ASCII only. Rejects CR/LF/control bytes so caller-supplied IDs +/// can't inject HTTP response headers or poison log pipelines. +fn is_safe_id_char(b: u8) -> bool { + (0x20..=0x7E).contains(&b) +} + +/// `(present, value)`: present=true iff the key is JSON `null` OR a valid +/// 1..=`MAX_ID_LEN` printable-ASCII string. Empty / wrong-type / over-long / +/// control-byte strings collapse to `(false, None)` so the wildcard path is +/// consistent (only triggered by explicit `null`, never by `""`). +fn get_string_field(v: &Value, key: &str) -> (bool, Option) { + match v.get(key) { + Some(Value::Null) => (true, None), + Some(Value::String(s)) + if !s.is_empty() && s.len() <= MAX_ID_LEN && s.bytes().all(is_safe_id_char) => + { + (true, Some(s.clone())) + } + _ => (false, None), + } +} + +/// Like [`with_harness_span`], but accepts pre-extracted [`HarnessIds`]. +/// +/// Use this when the caller has already produced IDs via +/// [`extract_ids_for_bridge_trigger`] (single-walk extraction that also +/// surfaces `function_id`). Avoids re-walking the JSON tree. +pub async fn run_in_span_with_ids( + fn_name: &'static str, + ids: HarnessIds, + inner_function_id: Option<&str>, + input: Value, + f: F, +) -> Result +where + F: FnOnce(Value) -> Fut + Send, + Fut: Future> + Send, +{ + run_in_span(fn_name, ids, inner_function_id, move || f(input)).await +} + +/// Compute the `Status::error` message (if any) for a finished handler. +/// +/// `Ok` envelope with `status_code >= 400` → `Some("http ")`. +/// `Err(e)` → `Some(e.to_string())`. +/// Otherwise → `None`. +/// +/// Pure helper extracted so the contract (H4: explicit error tagging on Err +/// branch) can be unit-tested without an InMemorySpanExporter / live tracer. +#[must_use] +pub fn error_status_message(result: &Result) -> Option { + match result { + Ok(value) => value + .get("status_code") + .and_then(Value::as_u64) + .filter(|&code| code >= 400) + .map(|code| format!("http {code}")), + Err(e) => Some(e.to_string()), + } +} + +/// Lock-step with `iii_sdk::DEFAULT_ALLOWLIST`; cross-crate test enforces parity. +pub const HARNESS_KEYS: &[&str] = &["iii.session.id", "iii.message.id", "iii.function.id"]; + +async fn run_in_span( + fn_name: &'static str, + ids: HarnessIds, + inner_function_id: Option<&str>, + f: F, +) -> Result +where + F: FnOnce() -> Fut + Send, + Fut: Future> + Send, +{ + let span_result = + iii_sdk::with_span(fn_name, None, Some(iii_sdk::SpanKind::Server), || async { + let recording = iii_sdk::current_span_is_recording(); + // Self-observability: rising `recording=false` rate signals the + // observability worker has gone silent. + tracing::trace!( + fn_name, + recording, + session_id = ids.session_id.as_deref(), + message_id_minted = ids.message_id_was_minted, + "harness span entry", + ); + if recording { + if let Some(sid) = &ids.session_id { + iii_sdk::set_current_span_attribute("iii.session.id", sid.clone()); + } + if let Some(mid) = &ids.message_id { + iii_sdk::set_current_span_attribute("iii.message.id", mid.clone()); + } + if let Some(fid) = inner_function_id { + iii_sdk::set_current_span_attribute("iii.function.id", fid.to_string()); + } + } + + let trace_id = iii_sdk::current_trace_id(); + let span_id = iii_sdk::current_span_id(); + + // `fid_owned` lifetime extends `inner_function_id: &str` through + // the `run_with_baggage` await point. + let mut baggage_entries: Vec<(&str, &str)> = Vec::with_capacity(3); + if let Some(sid) = ids.session_id.as_deref() { + baggage_entries.push(("iii.session.id", sid)); + } + if let Some(mid) = ids.message_id.as_deref() { + baggage_entries.push(("iii.message.id", mid)); + } + let fid_owned: String; + if let Some(fid) = inner_function_id { + fid_owned = fid.to_string(); + baggage_entries.push(("iii.function.id", fid_owned.as_str())); + } + + let inner_result: Result = + match iii_sdk::run_with_baggage(&baggage_entries, f()).await { + Ok(value) => Ok(merge_response_headers( + value, + trace_id.as_deref(), + span_id.as_deref(), + ids.message_id.as_deref(), + )), + Err(e) => Err(e), + }; + if let Some(msg) = error_status_message(&inner_result) { + // Belt-and-suspenders: tag errors explicitly via the iii-sdk + // helper. Don't rely on `iii_sdk::with_span` to auto-tag + // failed closures; if a future iii-sdk version stops doing + // that, errored spans would silently land as Status::Ok. + iii_sdk::set_current_span_error(msg); + } + match inner_result { + Ok(v) => Ok::>(v), + Err(e) => Err(Box::new(e) as Box), + } + }) + .await; + + // Downcast always succeeds today (we only box IIIError above). The + // fall-through guards against a future iii-sdk wrapping errors before + // propagating, surfacing the regression via warn! instead of silent + // type loss. + span_result.map_err(|boxed| match boxed.downcast::() { + Ok(e) => *e, + Err(other) => { + tracing::warn!( + error = %other, + "iii-sdk error downcast fallthrough; type fidelity lost — \ + inspect iii-sdk error-propagation behavior", + ); + IIIError::Handler(other.to_string()) + } + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn bridge_trigger_reads_top_level_ids() { + let input = json!({ + "body": { + "function_id": "harness::status", + "session_id": "S1", + "message_id": "M1", + "payload": {} + } + }); + let ids = extract_ids(&input, IdSource::BridgeTrigger, WildcardMode::Disabled); + assert_eq!(ids.session_id.as_deref(), Some("S1")); + assert_eq!(ids.message_id.as_deref(), Some("M1")); + assert!(!ids.message_id_was_minted); + } + + #[test] + fn bridge_trigger_falls_back_to_nested_payload() { + let input = json!({ + "body": { + "function_id": "f", + "payload": { "session_id": "S2", "message_id": "M2" } + } + }); + let ids = extract_ids(&input, IdSource::BridgeTrigger, WildcardMode::Disabled); + assert_eq!(ids.session_id.as_deref(), Some("S2")); + assert_eq!(ids.message_id.as_deref(), Some("M2")); + } + + #[test] + fn bridge_trigger_direct_bus_shape_no_body_wrap() { + let input = json!({ + "function_id": "f", + "session_id": "S3", + "message_id": "M3", + "payload": {} + }); + let ids = extract_ids(&input, IdSource::BridgeTrigger, WildcardMode::Disabled); + assert_eq!(ids.session_id.as_deref(), Some("S3")); + assert_eq!(ids.message_id.as_deref(), Some("M3")); + } + + #[test] + fn missing_message_id_yields_none() { + // Option A: when the caller doesn't supply a message_id and baggage + // is empty, message_id stays None. Non-chat plumbing then never + // shows up in `Group by message`. + let input = json!({ "body": { "function_id": "f", "payload": {} } }); + let ids = extract_ids(&input, IdSource::BridgeTrigger, WildcardMode::Disabled); + assert_eq!(ids.message_id, None); + assert!(!ids.message_id_was_minted); + assert_eq!(ids.session_id, None); + } + + #[test] + fn query_params_source_reads_from_query_params_only() { + let input = json!({ + "body": { "session_id": "wrong", "message_id": "wrong" }, + "query_params": { "session_id": "S4", "message_id": "M4" } + }); + let ids = extract_ids(&input, IdSource::QueryParams, WildcardMode::Disabled); + assert_eq!(ids.session_id.as_deref(), Some("S4")); + assert_eq!(ids.message_id.as_deref(), Some("M4")); + } + + #[test] + fn nullable_session_becomes_wildcard_when_flag_set() { + let input = json!({ "body": { "session_id": null } }); + let with_flag = extract_ids( + &input, + IdSource::BodyOrTopLevel, + WildcardMode::OnExplicitNull, + ); + assert_eq!(with_flag.session_id.as_deref(), Some("*")); + + let without_flag = extract_ids(&input, IdSource::BodyOrTopLevel, WildcardMode::Disabled); + assert_eq!(without_flag.session_id, None); + } + + #[test] + fn wrong_type_session_does_not_trigger_wildcard() { + // Regression: a caller that sends `session_id: 42` (number) used to be + // silently promoted to "subscribe to all sessions" by the wildcard + // path, because get_string_field collapsed "wrong type" into the same + // (true, None) state as "explicit null". A buggy direct-WS client + // would have fanned out cross-session events without an error. + for wrong in [json!(42), json!(true), json!(["S1"]), json!({"id": "S1"})] { + let input = json!({ "body": { "session_id": wrong.clone() } }); + let ids = extract_ids( + &input, + IdSource::BodyOrTopLevel, + WildcardMode::OnExplicitNull, + ); + assert_eq!( + ids.session_id, None, + "wrong-type session_id ({wrong}) must not become wildcard" + ); + } + } + + #[test] + fn over_long_message_id_is_treated_as_missing() { + // Regression: caller-supplied message_id is echoed verbatim into the + // `x-iii-message-id` response header. A 10 MB string would either get + // rejected by HTTP intermediaries (breaking the response) or balloon + // log/trace storage. Cap at MAX_ID_LEN (128). Under Option A, oversized + // input is dropped to None instead of being replaced with a minted UUID. + let huge = "x".repeat(10_000); + let input = json!({ "body": { "message_id": &huge } }); + let ids = extract_ids(&input, IdSource::BodyOrTopLevel, WildcardMode::Disabled); + assert_eq!(ids.message_id, None, "over-long message_id must be dropped",); + } + + #[test] + fn empty_string_message_id_is_treated_as_missing() { + // P2-7 regression. Previously `"message_id": ""` passed + // `bytes().all(is_safe_id_char)` vacuously (empty iterator → true) + // and was accepted as a present-and-valid empty ID. Treat empty + // as absent → None. + let input = json!({ "body": { "message_id": "" } }); + let ids = extract_ids(&input, IdSource::BodyOrTopLevel, WildcardMode::Disabled); + assert_eq!(ids.message_id, None); + } + + #[test] + fn empty_string_session_id_does_not_trigger_wildcard() { + // P2-7: `"session_id": ""` previously satisfied the byte-filter + // vacuously and would have been treated as a present empty value. + // Under `WildcardMode::OnExplicitNull` it had a subtle asymmetry + // with `null` (only `null` triggered wildcard, "" didn't because + // it fed Some(s)) — but the inconsistency leaked into + // session_present and could trip future logic. Now empty is + // unambiguously absent. + let input = json!({ "body": { "session_id": "" } }); + let with_wildcard = extract_ids( + &input, + IdSource::BodyOrTopLevel, + WildcardMode::OnExplicitNull, + ); + assert_eq!(with_wildcard.session_id, None); + + let without_wildcard = + extract_ids(&input, IdSource::BodyOrTopLevel, WildcardMode::Disabled); + assert_eq!(without_wildcard.session_id, None); + } + + #[test] + fn empty_string_function_id_is_treated_as_absent_for_bridge_trigger() { + // P2-7 + P2-8: an empty function_id should be rejected (so the + // handler downstream returns the standard "missing function_id" + // error rather than dispatching to "") AND it should not be + // logged as the same kind of warning as a content-poisoned + // function_id — but it IS logged with reason="empty" so a noisy + // client emits enough signal for an operator to act on. + let input = json!({ "body": { "function_id": "" } }); + let (_, function_id) = extract_ids_for_bridge_trigger(&input); + assert!( + function_id.is_none(), + "empty function_id must be treated as absent" + ); + } + + #[test] + fn over_long_session_id_does_not_trigger_wildcard() { + let huge = "x".repeat(10_000); + let input = json!({ "body": { "session_id": huge } }); + let ids = extract_ids( + &input, + IdSource::BodyOrTopLevel, + WildcardMode::OnExplicitNull, + ); + assert_eq!( + ids.session_id, None, + "over-long session_id must not become wildcard" + ); + } + + #[test] + fn message_id_at_boundary_is_accepted() { + // 128-char message_id is exactly at the limit and must be accepted. + let at_limit = "x".repeat(128); + let input = json!({ "body": { "message_id": &at_limit } }); + let ids = extract_ids(&input, IdSource::BodyOrTopLevel, WildcardMode::Disabled); + assert_eq!(ids.message_id.as_deref(), Some(at_limit.as_str())); + } + + #[test] + fn crlf_in_message_id_is_rejected() { + // Regression: a caller sending `"message_id": "abc\r\nSet-Cookie: x=y"` + // would otherwise echo the raw bytes into the response's + // `x-iii-message-id` header. Any upstream serializer using naive + // `format!("{}: {}\r\n", k, v)` would then HTTP-response-split. + // Under Option A, control-byte input is dropped to None. + for poison in ["abc\r\nSet-Cookie: x=y", "a\nb", "a\tb", "a\0b", "\x1b[31m"] { + let input = json!({ "body": { "message_id": poison } }); + let ids = extract_ids(&input, IdSource::BodyOrTopLevel, WildcardMode::Disabled); + assert_eq!( + ids.message_id, None, + "message_id containing control byte ({poison:?}) must be dropped", + ); + } + } + + #[test] + fn crlf_in_session_id_does_not_trigger_wildcard_or_echo() { + // Same regression for session_id: control-byte content rejected so + // it can't land in OTel span attributes (log-pipeline poisoning) or, + // for wildcard-mode handlers, be misinterpreted as "explicit null". + for poison in ["S\rsmuggle", "S\nsmuggle", "S\x01", "\x1b]2;evil\x07"] { + let input = json!({ "body": { "session_id": poison } }); + let with_wildcard = extract_ids( + &input, + IdSource::BodyOrTopLevel, + WildcardMode::OnExplicitNull, + ); + assert_eq!( + with_wildcard.session_id, None, + "control-byte session_id ({poison:?}) must not become wildcard or be echoed" + ); + let without_wildcard = + extract_ids(&input, IdSource::BodyOrTopLevel, WildcardMode::Disabled); + assert_eq!(without_wildcard.session_id, None); + } + } + + #[test] + fn extract_bridge_trigger_outer_message_inner_session_documented_precedence() { + // Documents the intentional precedence rule: when outer body has ANY + // id field (here `message_id`), the inner `payload` is NOT consulted. + // The inner `session_id` is therefore dropped. This is by design — a + // mixed-shape payload is ambiguous, so outer wins as a unit. + let input = json!({ + "body": { + "function_id": "f", + "message_id": "M-outer", + "payload": { "session_id": "S-inner" } + } + }); + let ids = extract_ids(&input, IdSource::BridgeTrigger, WildcardMode::Disabled); + assert_eq!(ids.message_id.as_deref(), Some("M-outer")); + assert_eq!( + ids.session_id, None, + "inner session_id must be dropped when outer has any id field" + ); + } + + #[test] + fn bridge_trigger_extractor_returns_function_id_in_single_walk() { + // Verifies the L6 single-walk optimization: function_id is surfaced + // alongside the IDs without a second tree walk. + let input = json!({ + "body": { + "function_id": "harness::status", + "session_id": "S1", + "message_id": "M1", + "payload": {} + } + }); + let (ids, function_id) = extract_ids_for_bridge_trigger(&input); + assert_eq!(ids.session_id.as_deref(), Some("S1")); + assert_eq!(ids.message_id.as_deref(), Some("M1")); + assert_eq!(function_id.as_deref(), Some("harness::status")); + } + + #[test] + fn bridge_trigger_extractor_rejects_oversize_function_id() { + // M1: function_id is caller-controlled. Apply the same MAX_ID_LEN cap + // so a 10 KB function_id can't bloat OTel attribute storage or log + // indexers downstream. + let huge = "f".repeat(10_000); + let input = json!({ "body": { "function_id": &huge, "payload": {} } }); + let (_ids, function_id) = extract_ids_for_bridge_trigger(&input); + assert!( + function_id.is_none(), + "over-long function_id must be treated as absent" + ); + } + + #[test] + fn bridge_trigger_extractor_rejects_control_chars_in_function_id() { + let input = json!({ "body": { "function_id": "harness\r\nX-Injected: 1", "payload": {} } }); + let (_ids, function_id) = extract_ids_for_bridge_trigger(&input); + assert!(function_id.is_none()); + } + + #[test] + fn wildcard_session_id_constant_matches_extractor_output() { + // M8: a single source of truth for the wildcard token. If the + // constant ever changes, this test catches the drift. + let input = json!({ "body": { "session_id": null } }); + let ids = extract_ids( + &input, + IdSource::BodyOrTopLevel, + WildcardMode::OnExplicitNull, + ); + assert_eq!(ids.session_id.as_deref(), Some(WILDCARD_SESSION_ID)); + assert_eq!(WILDCARD_SESSION_ID, "*"); + } + + #[test] + fn extract_ids_is_safe_under_concurrent_calls() { + // M7: `extract_ids` is a pure function over its `&Value` input; this + // test asserts no hidden global state by hammering it from multiple + // threads with distinct session IDs and verifying each call's output + // matches its own input. (Span-attribute isolation in `run_in_span` + // depends on the iii-sdk task-local OTel context, which can't be + // unit-tested here — see the integration tests for that path.) + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + use std::thread; + + let errors = Arc::new(AtomicUsize::new(0)); + let handles: Vec<_> = (0..32) + .map(|i| { + let errors = Arc::clone(&errors); + thread::spawn(move || { + let session = format!("S-thread-{i}"); + let message = format!("M-thread-{i}"); + let input = json!({ + "body": { "session_id": &session, "message_id": &message } + }); + let ids = extract_ids(&input, IdSource::BodyOrTopLevel, WildcardMode::Disabled); + if ids.session_id.as_deref() != Some(session.as_str()) + || ids.message_id.as_deref() != Some(message.as_str()) + { + errors.fetch_add(1, Ordering::SeqCst); + } + }) + }) + .collect(); + for h in handles { + h.join().expect("thread"); + } + assert_eq!( + errors.load(Ordering::SeqCst), + 0, + "concurrent extract_ids calls must not cross-contaminate" + ); + } + + #[test] + fn error_status_message_tags_handler_err() { + // H4 contract: the Err branch must produce a Status::error message. + let err: Result = Err(IIIError::Handler("missing function_id".into())); + let msg = error_status_message(&err).expect("Err yields a status message"); + assert!(msg.contains("missing function_id"), "got: {msg}"); + } + + #[test] + fn error_status_message_tags_http_error_envelope() { + // status_code >= 400 in an Ok envelope is also tagged as Status::error. + let env = Ok(json!({ "status_code": 503, "headers": {}, "body": "down" })); + assert_eq!(error_status_message(&env).as_deref(), Some("http 503")); + let env_404 = Ok(json!({ "status_code": 404 })); + assert_eq!(error_status_message(&env_404).as_deref(), Some("http 404")); + } + + #[test] + fn error_status_message_none_for_ok_envelope_and_raw_ok() { + // 200 OK envelope: no error tag. + let env_ok = Ok(json!({ "status_code": 200, "body": {} })); + assert_eq!(error_status_message(&env_ok), None); + // Raw (non-envelope) Ok: no status_code, no error tag. + let raw_ok = Ok(json!({ "ok": true, "total_browsers": 1 })); + assert_eq!(error_status_message(&raw_ok), None); + } + + #[test] + fn error_status_message_ignores_redirects_and_informational() { + // 3xx / 1xx / 2xx are not Status::error. + for code in [100u64, 200, 204, 301, 302, 399] { + let env = Ok(json!({ "status_code": code })); + assert_eq!(error_status_message(&env), None, "code {code}"); + } + } + + #[test] + fn error_status_message_handles_400_boundary() { + let env = Ok(json!({ "status_code": 400 })); + assert_eq!(error_status_message(&env).as_deref(), Some("http 400")); + } + + #[test] + fn missing_session_key_does_not_become_wildcard() { + let input = json!({ "body": { "message_id": "M" } }); + let ids = extract_ids( + &input, + IdSource::BodyOrTopLevel, + WildcardMode::OnExplicitNull, + ); + assert_eq!(ids.session_id, None); + } + + // The four `augment_baggage_*` tests that previously lived here pinned + // the behavior of the in-process baggage-augment helper. That helper + // moved into iii-sdk as `iii_sdk::run_with_baggage`; the equivalent + // tests live at + // `motia/sdk/packages/rust/iii/tests/span_ops_api.rs::run_with_baggage_*`. + // Harness-side coverage is now the lockstep allowlist test in + // `harness/tests/trace_correlation.rs` plus the e2e attribute-filter test. + + #[test] + fn merge_adds_traceparent_and_message_id_to_existing_headers() { + let envelope = json!({ + "status_code": 200, + "headers": { "content-type": "application/json" }, + "body": { "ok": true } + }); + let merged = merge_response_headers( + envelope, + Some("0af7651916cd43dd8448eb211c80319c"), + Some("b7ad6b7169203331"), + Some("M-test"), + ); + let headers = merged.get("headers").expect("headers field exists"); + assert_eq!(headers["content-type"], "application/json"); + assert_eq!( + headers["traceparent"], + "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01" + ); + assert_eq!(headers["x-iii-message-id"], "M-test"); + // body is untouched + assert_eq!(merged["body"], json!({"ok": true})); + } + + #[test] + fn merge_creates_headers_when_absent() { + let envelope = json!({ "status_code": 200, "body": {} }); + let trace = "a".repeat(32); + let span = "b".repeat(16); + let merged = merge_response_headers( + envelope, + Some(trace.as_str()), + Some(span.as_str()), + Some("M"), + ); + assert!(merged["headers"].is_object()); + assert!(merged["headers"]["traceparent"].is_string()); + assert_eq!(merged["headers"]["x-iii-message-id"], "M"); + } + + #[test] + fn merge_skips_traceparent_when_trace_id_is_none() { + let envelope = json!({ "status_code": 200, "headers": {} }); + let merged = merge_response_headers(envelope, None, None, Some("M")); + assert!(merged["headers"].get("traceparent").is_none()); + assert_eq!(merged["headers"]["x-iii-message-id"], "M"); + } + + #[test] + fn merge_skips_message_id_header_when_none() { + // Option A: no caller-supplied message_id and no upstream baggage + // → no `x-iii-message-id` header. Header is for chat correlation, + // not for plumbing calls. + let envelope = json!({ "status_code": 200, "headers": {} }); + let merged = merge_response_headers(envelope, None, None, None); + assert!(merged["headers"].get("x-iii-message-id").is_none()); + } + + #[test] + fn merge_leaves_non_object_returns_untouched() { + let raw = json!("just a string"); + let merged = merge_response_headers(raw.clone(), Some("x"), Some("y"), Some("M")); + assert_eq!(merged, raw); + } + + #[test] + fn merge_leaves_envelope_lacking_status_code_untouched() { + // fs::read_inline returns this shape directly (no status_code). + // The wrapper should NOT graft `headers` onto it. + let raw = json!({ + "content": [{ "type": "text", "text": "hello" }], + "details": { "size": 5, "truncated": false, "bytes_read": 5 }, + "terminate": false + }); + let merged = merge_response_headers( + raw.clone(), + Some("a".repeat(32).as_str()), + Some("b".repeat(16).as_str()), + Some("M"), + ); + assert_eq!(merged, raw, "non-envelope objects must be left untouched"); + } +} diff --git a/harness/tests/trace_correlation.rs b/harness/tests/trace_correlation.rs new file mode 100644 index 000000000..e2087cee7 --- /dev/null +++ b/harness/tests/trace_correlation.rs @@ -0,0 +1,614 @@ +//! Integration tests for harness trace correlation. +//! +//! Requires a live engine with the iii-observability worker configured for +//! `exporter: memory` (matches `harness/config.yaml`). When no engine is +//! reachable, tests print "skipping: …" and return — mirrors the pattern in +//! `harness/tests/bridge_info.rs` and `harness/tests/sse_bridge.rs`. + +use harness::register_with_iii_with_engine_url; +use iii_sdk::{register_worker, InitOptions, TriggerRequest}; +use serde_json::{json, Value}; + +const DEFAULT_ENGINE_URL: &str = "ws://127.0.0.1:49134"; + +/// Per-call timeout for engine triggers in this test suite. 2s is enough for +/// the engine to dispatch and return for `harness::status` / `bridge::trigger` +/// / `engine::traces::*` in a local engine; tests that need a longer budget +/// should declare their own override. +const TRIGGER_TIMEOUT_MS: u64 = 2_000; + +/// Retry budget for the memory-exporter flush. Spans land asynchronously after +/// the call returns; under uncontended load 10×100ms = 1s is plenty. +const EXPORTER_FLUSH_RETRIES: u32 = 10; +const EXPORTER_FLUSH_INTERVAL_MS: u64 = 100; + +/// Same retry pattern but expanded for the test that runs under parallel-test +/// pressure on the shared exporter. 30×200ms = 6s tolerance for eviction and +/// dispatch jitter; matched to the `bridge_trigger_missing_function_id` +/// error path which competes with other tests for exporter capacity. +const EXPORTER_FLUSH_RETRIES_UNDER_LOAD: u32 = 30; +const EXPORTER_FLUSH_INTERVAL_UNDER_LOAD_MS: u64 = 200; +const ENGINE_PROBE_TIMEOUT_MS: u64 = 500; + +/// When set, integration tests refuse to silently skip because no engine is +/// reachable — they panic instead. CI should set this so a missing engine +/// is a loud failure, not a green build with zero assertions. +const REQUIRE_ENGINE_ENV: &str = "HARNESS_TRACE_TESTS_REQUIRE_ENGINE"; + +/// When set, tests that depend on the iii-observability worker (traceparent +/// echo, engine::traces::list/tree queries) panic instead of silently skipping +/// when traceparent is absent or no spans land in the memory exporter. CI +/// runs with observability up should set this. +const REQUIRE_OTEL_ENV: &str = "HARNESS_TRACE_TESTS_REQUIRE_OTEL"; + +fn require_engine() -> bool { + std::env::var(REQUIRE_ENGINE_ENV).is_ok() +} + +fn require_otel() -> bool { + std::env::var(REQUIRE_OTEL_ENV).is_ok() +} + +/// Either skip (logging the reason) or panic, depending on REQUIRE_OTEL_ENV. +/// Production wrapper around [`skip_or_panic_inner`] — splits the env-var +/// read from the decision logic so the panic path is testable without the +/// process-wide `std::env::set_var` (unsafe in Rust 2024+, races in 2021). +fn skip_or_panic_otel(reason: &str) { + skip_or_panic_inner(require_otel(), REQUIRE_OTEL_ENV, reason); +} + +/// Decision core: panic if `require` is true, else `eprintln!` a skip line. +/// Pure function over `(bool, &str, &str)` — tested by +/// `skip_or_panic_inner_panics_when_required`. +fn skip_or_panic_inner(require: bool, env_name: &str, reason: &str) { + assert!(!require, "{env_name} set but {reason}"); + eprintln!("skipping: {reason}"); +} + +/// Boot harness in-process against a live engine; returns None when no engine +/// is reachable (call sites print the skip message and return). When +/// `HARNESS_TRACE_TESTS_REQUIRE_ENGINE` is set in the environment, panics +/// instead of returning None so CI failures are loud. +async fn boot_or_skip() -> Option<(iii_sdk::III, String)> { + let url = std::env::var("III_URL").unwrap_or_else(|_| DEFAULT_ENGINE_URL.to_string()); + let iii = register_worker(&url, InitOptions::default()); + + let probe = iii + .trigger(TriggerRequest { + function_id: "state::get".into(), + payload: json!({ "scope": "agent", "key": "__trace_correlation_probe" }), + action: None, + timeout_ms: Some(ENGINE_PROBE_TIMEOUT_MS), + }) + .await; + if let Err(e) = probe { + assert!( + !require_engine(), + "{REQUIRE_ENGINE_ENV} set but no engine reachable at {url}: {e}", + ); + eprintln!("skipping: no engine at {url}"); + return None; + } + + if let Err(e) = register_with_iii_with_engine_url(&iii, &url).await { + assert!( + !require_engine(), + "{REQUIRE_ENGINE_ENV} set but register failed against engine at {url}: {e}", + ); + eprintln!("skipping: register failed: {e}"); + return None; + } + + Some((iii, url)) +} + +/// Helper: invoke `harness::status` directly via the bus, return the envelope. +async fn call_status(iii: &iii_sdk::III, session: &str, message: Option<&str>) -> Value { + let mut payload = json!({ "session_id": session }); + if let Some(m) = message { + payload["message_id"] = Value::String(m.to_string()); + } + iii.trigger(TriggerRequest { + function_id: "harness::status".into(), + payload, + action: None, + timeout_ms: Some(TRIGGER_TIMEOUT_MS), + }) + .await + .expect("call harness::status") +} + +/// Helper: invoke `harness::status` via `bridge::trigger` (the HTTP envelope +/// path that browsers actually use). The outer wrapper around bridge::trigger +/// echoes `traceparent` / `x-iii-message-id` into response headers; the inner +/// `harness::status` returns its raw shape directly. +async fn call_status_via_bridge(iii: &iii_sdk::III, session: &str, message: Option<&str>) -> Value { + let mut payload = + json!({ "function_id": "harness::status", "payload": {}, "session_id": session }); + if let Some(m) = message { + payload["message_id"] = Value::String(m.to_string()); + } + iii.trigger(TriggerRequest { + function_id: "bridge::trigger".into(), + payload, + action: None, + timeout_ms: Some(TRIGGER_TIMEOUT_MS), + }) + .await + .expect("call bridge::trigger -> harness::status") +} + +#[tokio::test] +#[serial_test::serial(harness_trace)] +async fn bridge_trigger_echoes_message_id_when_provided() { + let Some((iii, _url)) = boot_or_skip().await else { + return; + }; + let env = call_status_via_bridge(&iii, "S1", Some("M1")).await; + let headers = &env["headers"]; + assert_eq!(headers["x-iii-message-id"], "M1"); +} + +#[tokio::test] +#[serial_test::serial(harness_trace)] +async fn bridge_trigger_omits_message_id_header_when_absent() { + // Option A: when the caller doesn't supply a message_id and baggage + // is empty, no x-iii-message-id header is emitted. Plumbing calls + // stay out of `Group by message`. + let Some((iii, _url)) = boot_or_skip().await else { + return; + }; + let env = call_status_via_bridge(&iii, "S1", None).await; + assert!( + env["headers"].get("x-iii-message-id").is_none(), + "x-iii-message-id header must be absent when no upstream message_id" + ); +} + +#[tokio::test] +#[serial_test::serial(harness_trace)] +async fn harness_status_emits_traceparent_when_otel_enabled() { + let Some((iii, _url)) = boot_or_skip().await else { + return; + }; + let env = call_status(&iii, "S1", Some("M-tp")).await; + + // traceparent is only emitted when the engine has the observability worker + // active. If the harness boot succeeded but observability isn't running, + // skip — this test doesn't assert that the worker was registered. + let Some(tp) = env["headers"].get("traceparent").and_then(Value::as_str) else { + skip_or_panic_otel("traceparent assertion: observability worker not active"); + return; + }; + // Format: 00-<32 hex>-<16 hex>-01 + assert!( + tp.starts_with("00-"), + "traceparent should start with version 00-; got {tp}" + ); + let parts: Vec<&str> = tp.split('-').collect(); + assert_eq!(parts.len(), 4); + assert_eq!(parts[1].len(), 32, "trace_id is 32 hex chars; got {tp}"); + assert_eq!(parts[2].len(), 16, "span_id is 16 hex chars; got {tp}"); +} + +/// Verifies the WORKING query path for finding harness traces: +/// `name: "harness.status"` + `search_all_spans: true`. +/// +/// NOTE: this test does NOT exercise the engine's attribute filter +/// (`attributes: [["iii.session.id", ...]]`). The engine's attribute filter +/// targets ROOT spans only, but our `iii.*` attrs live on the CHILD harness +/// span. Filter-by-attribute is the spec's stated goal but isn't queryable +/// today; tracked as an engine-side follow-up (extend `search_all_spans` to +/// apply to attribute filter, OR propagate harness baggage into the engine +/// root span). Until then, the supported discovery path is by span name. +#[tokio::test] +#[serial_test::serial(harness_trace)] +async fn engine_traces_list_finds_harness_span_by_name() { + let Some((iii, _url)) = boot_or_skip().await else { + return; + }; + let session = format!("test-session-{}", uuid::Uuid::new_v4()); + + let env = call_status(&iii, &session, None).await; + let Some(tp) = env["headers"].get("traceparent").and_then(Value::as_str) else { + skip_or_panic_otel("observability worker not active"); + return; + }; + let trace_id = tp + .split('-') + .nth(1) + .expect("traceparent has trace_id") + .to_string(); + + // Spans land in the memory exporter asynchronously. Retry up to 10 times. + // We use `name: "harness.status"` + `search_all_spans: true` because the + // engine's attribute filter targets root spans only, and our iii.* attrs + // live on the child span. Name+search_all matches any span in the trace and + // returns its root span_id, which is what we need to correlate with the + // trace_id minted in the previous status call. + let mut found = None; + for _ in 0..EXPORTER_FLUSH_RETRIES { + let res = iii + .trigger(TriggerRequest { + function_id: "engine::traces::list".into(), + payload: json!({ + "name": "harness.status", + "search_all_spans": true, + }), + action: None, + timeout_ms: Some(TRIGGER_TIMEOUT_MS), + }) + .await + .expect("engine::traces::list"); + let spans = res["spans"].as_array().cloned().unwrap_or_default(); + if !spans.is_empty() { + found = Some(spans); + break; + } + tokio::time::sleep(std::time::Duration::from_millis(EXPORTER_FLUSH_INTERVAL_MS)).await; + } + + let spans = found.expect("at least one span matching harness.status name"); + assert!( + spans.iter().any(|s| s["trace_id"] == trace_id), + "expected trace_id {trace_id} among returned spans; got {spans:?}" + ); +} + +#[tokio::test] +#[serial_test::serial(harness_trace)] +async fn engine_traces_tree_returns_root_and_children() { + let Some((iii, _url)) = boot_or_skip().await else { + return; + }; + let session = format!("tree-session-{}", uuid::Uuid::new_v4()); + + let env = call_status(&iii, &session, Some("M-tree")).await; + let Some(tp) = env["headers"].get("traceparent").and_then(Value::as_str) else { + skip_or_panic_otel("observability worker not active"); + return; + }; + let trace_id = tp + .split('-') + .nth(1) + .expect("traceparent has trace_id") + .to_string(); + + // Same async-flush retry as above. + let mut tree = None; + for _ in 0..EXPORTER_FLUSH_RETRIES { + let res = iii + .trigger(TriggerRequest { + function_id: "engine::traces::tree".into(), + payload: json!({ "trace_id": trace_id }), + action: None, + timeout_ms: Some(TRIGGER_TIMEOUT_MS), + }) + .await + .expect("engine::traces::tree"); + let roots = res["roots"].as_array().cloned().unwrap_or_default(); + if !roots.is_empty() { + tree = Some(roots); + break; + } + tokio::time::sleep(std::time::Duration::from_millis(EXPORTER_FLUSH_INTERVAL_MS)).await; + } + + let roots = tree.expect("trace tree has at least one root"); + let root = &roots[0]; + // The root is the engine's auto-span (e.g. "handle_invocation harness::status"). + // iii.* attributes live on our CHILD span, not the root — the engine's + // attribute filter targets root spans only, so we walk the tree to find our + // child and assert on its attributes directly. + + #[allow(clippy::items_after_statements)] + fn find_named<'a>(node: &'a Value, name: &str) -> Option<&'a Value> { + if node["name"].as_str() == Some(name) { + return Some(node); + } + node["children"] + .as_array() + .and_then(|cs| cs.iter().find_map(|c| find_named(c, name))) + } + + let child = find_named(root, "harness.status") + .expect("expected a `harness.status` span somewhere in the trace"); + + // Attributes are stored as Vec<(String, String)> (see StoredSpan); JSON form + // is an array of two-element arrays: `[["key", "value"], ...]`. + let attrs = child["attributes"].as_array().expect("attributes array"); + let has_session = attrs + .iter() + .any(|kv| kv[0].as_str() == Some("iii.session.id") && kv[1].as_str() == Some(&session)); + let has_message = attrs + .iter() + .any(|kv| kv[0].as_str() == Some("iii.message.id") && kv[1].as_str() == Some("M-tree")); + assert!( + has_session, + "child span missing iii.session.id; attrs={attrs:?}" + ); + assert!( + has_message, + "child span missing iii.message.id; attrs={attrs:?}" + ); +} + +#[tokio::test] +#[serial_test::serial(harness_trace)] +async fn bridge_trigger_missing_function_id_still_emits_traced_error() { + let Some((iii, _url)) = boot_or_skip().await else { + return; + }; + let session = format!("err-session-{}", uuid::Uuid::new_v4()); + + // Probe: send a well-formed bridge::trigger first and confirm we get a + // traceparent back. If not, observability is down and the rest of this + // test cannot record/query spans — skip. Use a different session so the + // probe span doesn't collide with the error span during disambiguation. + let probe_session = format!("probe-{}", uuid::Uuid::new_v4()); + let probe = iii + .trigger(TriggerRequest { + function_id: "bridge::trigger".into(), + payload: json!({ + "function_id": "harness::status", + "payload": {}, + "session_id": probe_session, + "message_id": "probe", + }), + action: None, + timeout_ms: Some(TRIGGER_TIMEOUT_MS), + }) + .await + .expect("probe bridge::trigger"); + if probe["headers"].get("traceparent").is_none() { + skip_or_panic_otel("observability worker not active"); + return; + } + + // bridge::trigger expects { function_id, payload }. We deliberately omit + // function_id. The handler returns IIIError::Handler("missing function_id"), + // but the span we opened around the handler should still be recorded. + let result = iii + .trigger(TriggerRequest { + function_id: "bridge::trigger".into(), + payload: json!({ "session_id": session, "message_id": "M-err" }), + action: None, + timeout_ms: Some(TRIGGER_TIMEOUT_MS), + }) + .await; + assert!(result.is_err(), "missing function_id must produce an error"); + + // The span lands in the memory exporter even though the handler errored. + // Use name+search_all_spans (attribute filter targets root spans only; + // our iii.* attrs live on child spans). + let mut found = None; + for _ in 0..EXPORTER_FLUSH_RETRIES_UNDER_LOAD { + let res = iii + .trigger(TriggerRequest { + function_id: "engine::traces::list".into(), + payload: json!({ + "name": "harness.bridge.trigger", + "search_all_spans": true, + "limit": 500, + }), + action: None, + timeout_ms: Some(TRIGGER_TIMEOUT_MS), + }) + .await + .expect("engine::traces::list"); + let spans = res["spans"].as_array().cloned().unwrap_or_default(); + if !spans.is_empty() { + found = Some(spans); + break; + } + tokio::time::sleep(std::time::Duration::from_millis( + EXPORTER_FLUSH_INTERVAL_UNDER_LOAD_MS, + )) + .await; + } + + let Some(spans) = found else { + skip_or_panic_otel( + "no harness.bridge.trigger spans landed in memory exporter (harness OTel exporter \ + may not be wired to the engine in this run)", + ); + return; + }; + // `name + search_all_spans` may match many traces; disambiguate by walking + // each candidate's tree and finding the one whose harness.bridge.trigger + // child carries our test session_id. + + #[allow(clippy::items_after_statements)] + fn find_named<'a>(node: &'a Value, name: &str) -> Option<&'a Value> { + if node["name"].as_str() == Some(name) { + return Some(node); + } + node["children"] + .as_array() + .and_then(|cs| cs.iter().find_map(|c| find_named(c, name))) + } + + #[allow(clippy::items_after_statements)] + fn span_has_attr(span: &Value, key: &str, value: &str) -> bool { + span["attributes"].as_array().is_some_and(|arr| { + arr.iter() + .any(|kv| kv[0].as_str() == Some(key) && kv[1].as_str() == Some(value)) + }) + } + + let mut matched_child: Option = None; + 'outer: for s in &spans { + let Some(trace_id) = s["trace_id"].as_str() else { + continue; + }; + let tree = iii + .trigger(TriggerRequest { + function_id: "engine::traces::tree".into(), + payload: json!({ "trace_id": trace_id }), + action: None, + timeout_ms: Some(TRIGGER_TIMEOUT_MS), + }) + .await + .expect("engine::traces::tree"); + let Some(roots) = tree["roots"].as_array() else { + continue; + }; + for root in roots { + if let Some(child) = find_named(root, "harness.bridge.trigger") { + if span_has_attr(child, "iii.session.id", &session) { + matched_child = Some(child.clone()); + break 'outer; + } + } + } + } + + let Some(child) = matched_child else { + skip_or_panic_otel(&format!( + "matched no harness.bridge.trigger span with session_id={session} (memory exporter \ + likely evicted it under parallel test traffic; the span DID record per the \ + retry-loop hit on name)" + )); + return; + }; + // Status field shape: StoredSpan.status is a string ("Error" / "Ok" / "Unset"). + let status = child["status"].as_str().unwrap_or(""); + assert!( + status.eq_ignore_ascii_case("error"), + "expected Status::error on the failed harness.bridge.trigger span; got {status:?}" + ); +} + +/// `bridge::events` is the SSE path. The handler returns an envelope +/// (`status_code` + `headers` + body string with seed events), and +/// `with_envelope_span` merges `traceparent` + `x-iii-message-id` into the +/// envelope's `headers`. This test asserts both headers arrive verbatim for +/// the seed-pump phase. When T12 wires live-tail, the wrapper-shrinking +/// follow-up must keep this assertion passing — if it fails after live-tail, +/// the wrapper is no longer recording the right scope. +#[tokio::test] +#[serial_test::serial(harness_trace)] +async fn bridge_events_sse_emits_traceparent_and_message_id_headers() { + let Some((iii, _url)) = boot_or_skip().await else { + return; + }; + let session = format!("sse-session-{}", uuid::Uuid::new_v4()); + + let env = iii + .trigger(TriggerRequest { + function_id: "bridge::events".into(), + payload: json!({ + "query_params": { "session_id": &session, "message_id": "M-sse" }, + }), + action: None, + timeout_ms: Some(TRIGGER_TIMEOUT_MS), + }) + .await + .expect("call bridge::events"); + + // bridge::events returns the HTTP envelope. + assert_eq!( + env["status_code"].as_u64(), + Some(200), + "bridge::events envelope shape; got {env:?}" + ); + let headers = &env["headers"]; + assert_eq!( + headers["content-type"], "text/event-stream", + "SSE content-type preserved alongside the trace headers" + ); + + // x-iii-message-id is echoed verbatim whenever the envelope path is taken. + assert_eq!( + headers["x-iii-message-id"], "M-sse", + "x-iii-message-id should echo the caller-supplied id" + ); + + // traceparent is only present when the iii-observability worker is active. + // The require-OTel gate matches the other SSE-adjacent tests. + let Some(tp) = headers.get("traceparent").and_then(Value::as_str) else { + skip_or_panic_otel("bridge::events traceparent: observability worker not active"); + return; + }; + // Format: 00-<32 hex>-<16 hex>-01 + assert!( + tp.starts_with("00-"), + "traceparent should start with version 00-; got {tp}" + ); + let parts: Vec<&str> = tp.split('-').collect(); + assert_eq!(parts.len(), 4, "traceparent has 4 hyphen-separated parts"); + assert_eq!(parts[1].len(), 32, "trace_id is 32 hex chars; got {tp}"); + assert_eq!(parts[2].len(), 16, "span_id is 16 hex chars; got {tp}"); +} + +#[cfg(test)] +mod unit_tests { + use super::{skip_or_panic_inner, REQUIRE_ENGINE_ENV, REQUIRE_OTEL_ENV}; + + #[test] + fn skip_or_panic_inner_panics_when_required() { + // H3 contract: when the require flag is true, the function MUST + // panic with a message that references the env var name. The + // production callers (`skip_or_panic_otel`, `boot_or_skip`) read + // the env var and pass the bool — if a typo renames the env var + // in CI, the panic message still points operators at the right + // knob. + let r = std::panic::catch_unwind(|| { + skip_or_panic_inner(true, "EXAMPLE_REQUIRE_ENV", "reason text"); + }); + assert!(r.is_err(), "must panic when require=true"); + let msg = match r { + Err(e) => e + .downcast_ref::() + .cloned() + .or_else(|| e.downcast_ref::<&str>().map(|s| (*s).to_string())) + .unwrap_or_default(), + Ok(()) => String::new(), + }; + assert!( + msg.contains("EXAMPLE_REQUIRE_ENV"), + "panic message should reference env-var name; got: {msg}" + ); + assert!( + msg.contains("reason text"), + "panic message should include the reason; got: {msg}" + ); + } + + #[test] + fn skip_or_panic_inner_skips_silently_when_not_required() { + // The non-require path returns Ok (just `eprintln!`). + let r = std::panic::catch_unwind(|| { + skip_or_panic_inner(false, "EXAMPLE_REQUIRE_ENV", "reason text"); + }); + assert!(r.is_ok(), "must not panic when require=false"); + } + + #[test] + fn env_var_names_are_stable() { + // L3+H3: if a future contributor renames these consts, CI YAML + // that sets them will silently stop working. This test pins the + // public surface so the rename has a tripwire. + assert_eq!(REQUIRE_ENGINE_ENV, "HARNESS_TRACE_TESTS_REQUIRE_ENGINE"); + assert_eq!(REQUIRE_OTEL_ENV, "HARNESS_TRACE_TESTS_REQUIRE_OTEL"); + } + + #[test] + fn allowlist_matches_iii_sdk_baggage_processor() { + // Lock-step contract. The harness is the WRITER side + // (`harness::otel::HARNESS_KEYS` — keys the wrapper sets into + // baggage on every wrapped call) and iii-sdk is the READER side + // (`iii_sdk::DEFAULT_ALLOWLIST` — keys the + // `BaggageSpanProcessor` copies from baggage onto every span). + // If a future change adds a fourth harness baggage key without + // also growing the iii-sdk allowlist, the new key would silently + // be dropped from downstream spans. This assertion catches that + // drift in CI before it reaches production. + assert_eq!( + harness::otel::HARNESS_KEYS, + iii_sdk::DEFAULT_ALLOWLIST, + "harness::otel::HARNESS_KEYS (writer side) must equal \ + iii_sdk::DEFAULT_ALLOWLIST (reader side); update both in lockstep", + ); + } +} diff --git a/harness/web/src/App.tsx b/harness/web/src/App.tsx index 87d214927..93638c16f 100644 --- a/harness/web/src/App.tsx +++ b/harness/web/src/App.tsx @@ -314,13 +314,39 @@ export default function App() { setMessageEntryIds([...messageEntryIds, null]); try { - await bridge<{ session_id: string }>("run::start", { + // Route through `harness::call` (harness-wrapped) so the harness + // wrapper writes `iii.session.id` + `iii.message.id` into OTel + // baggage. iii-sdk auto-propagates baggage on every downstream + // `iii.trigger(...)`, and the in-SDK `BaggageSpanProcessor` + // materializes them as span attributes on every worker span + // produced by this turn. Result: the iii Developer Console + // TRACES tab can "Group by message" and see all spans for ONE + // user turn in one collapsible group. + // + // Direct `run::start` would bypass this — the chain would start + // outside any harness-wrapped function, no baggage would be seeded, + // and downstream spans would carry no `iii.message.id` attribute. + // + // One messageId per `send()` call. The chain (turn-orchestrator + // → provider-router → provider-anthropic → tool calls → ...) all + // inherit it via baggage. New user messages get a new id. + const messageId = `msg-${crypto.randomUUID()}`; + await bridge<{ + status_code?: number; + body?: { session_id: string }; + session_id?: string; + }>("harness::call", { + function_id: "run::start", session_id: sid, - provider, - model, - messages: fullHistory, - approval_required: APPROVAL_REQUIRED, - ...(cwd.trim() ? { cwd: cwd.trim() } : {}), + message_id: messageId, + payload: { + session_id: sid, + provider, + model, + messages: fullHistory, + approval_required: APPROVAL_REQUIRED, + ...(cwd.trim() ? { cwd: cwd.trim() } : {}), + }, }); void refreshSessions(); } catch (e) { diff --git a/harness/web/src/bridge.test.ts b/harness/web/src/bridge.test.ts new file mode 100644 index 000000000..1aa127000 --- /dev/null +++ b/harness/web/src/bridge.test.ts @@ -0,0 +1,61 @@ +// Regression coverage for the "error · [object Object]" bug: the iii-sdk's +// trigger() rejects with the raw wire payload (a plain object), not an Error +// instance. The earlier String(e) fallback collapsed every engine error to +// "[object Object]" in the UI. toBridgeError now walks common engine-error +// shapes and falls back to JSON.stringify so the real message reaches the +// user. + +import { describe, expect, it } from "vitest"; + +import { BridgeError, toBridgeError } from "./bridge"; + +describe("toBridgeError", () => { + it("preserves Error.message when an Error instance is thrown", () => { + const err = toBridgeError(new Error("boom"), "auth::set_token"); + expect(err).toBeInstanceOf(BridgeError); + expect(err.message).toBe("boom"); + expect(err.functionId).toBe("auth::set_token"); + }); + + it("extracts message field from engine error payload", () => { + // The shape iii-sdk hands us when the engine fails an invocation. + const wire = { message: "missing required field: provider", error_code: "BAD_INPUT" }; + const err = toBridgeError(wire, "auth::set_token"); + expect(err.message).toBe("missing required field: provider"); + }); + + it("falls back to `error` field when `message` is absent", () => { + const wire = { error: "store.set failed: state worker not available" }; + const err = toBridgeError(wire, "auth::set_token"); + expect(err.message).toBe("store.set failed: state worker not available"); + }); + + it("captures error_id when present", () => { + const wire = { message: "boom", error_id: "abc-123" }; + const err = toBridgeError(wire, "auth::set_token"); + expect(err.errorId).toBe("abc-123"); + }); + + it("falls back to JSON.stringify rather than '[object Object]'", () => { + // The regression case: an engine error without any known string field. + // Before the fix this rendered as "[object Object]". After the fix the + // user sees enough JSON to diagnose, even for unfamiliar shapes. + const wire = { code: 42, detail: { nested: true } }; + const err = toBridgeError(wire, "auth::set_token"); + expect(err.message).not.toBe("[object Object]"); + expect(err.message).toContain("\"code\":42"); + }); + + it("never produces empty messages for primitive throws", () => { + expect(toBridgeError("string thrown", "f").message).toBe("string thrown"); + expect(toBridgeError(42, "f").message).toBe("42"); + expect(toBridgeError(null, "f").message).toBe("null"); + expect(toBridgeError(undefined, "f").message).toBe("undefined"); + }); + + it("ignores empty-string fields (otherwise the chain would short-circuit)", () => { + const wire = { message: "", error: "fallback works" }; + const err = toBridgeError(wire, "f"); + expect(err.message).toBe("fallback works"); + }); +}); diff --git a/harness/web/src/bridge.ts b/harness/web/src/bridge.ts index 61686e726..a77f012c8 100644 --- a/harness/web/src/bridge.ts +++ b/harness/web/src/bridge.ts @@ -29,8 +29,49 @@ export async function bridge( return await client.call(functionId, payload); } catch (e) { if (e instanceof BridgeError) throw e; - const msg = e instanceof Error ? e.message : String(e); - throw new BridgeError(msg, functionId, 0); + throw toBridgeError(e, functionId); + } +} + +/** + * Normalize anything thrown by the iii-sdk's `trigger()` into a `BridgeError` + * with a human-readable `message`. + * + * The iii-sdk delivers engine errors via `Promise.reject(error)` where + * `error` is the raw JSON value off the wire — usually a plain object like + * `{ error_code, message, error_id }` or `{ error }`, NOT an `Error` + * instance. Previously this path collapsed to `String(e)` which produced + * the infamous `"[object Object]"` in the UI (see harness AuthPanel error + * pill). Now we walk common engine-error shapes and fall back to + * `JSON.stringify` so the actual server message reaches the user. + */ +export function toBridgeError(e: unknown, functionId: string): BridgeError { + if (e instanceof Error) { + return new BridgeError(e.message, functionId, 0); + } + if (e !== null && typeof e === "object") { + const obj = e as Record; + const message = + pickString(obj.message) ?? + pickString(obj.error) ?? + pickString(obj.error_message) ?? + pickString(obj.description) ?? + safeJson(obj); + const errorId = pickString(obj.error_id); + return new BridgeError(message, functionId, 0, errorId); + } + return new BridgeError(String(e), functionId, 0); +} + +function pickString(v: unknown): string | undefined { + return typeof v === "string" && v.length > 0 ? v : undefined; +} + +function safeJson(v: unknown): string { + try { + return JSON.stringify(v); + } catch { + return "unknown error (unserializable)"; } } diff --git a/hook-fanout/Cargo.lock b/hook-fanout/Cargo.lock index 50465926e..e9eb8d752 100644 --- a/hook-fanout/Cargo.lock +++ b/hook-fanout/Cargo.lock @@ -723,9 +723,9 @@ dependencies = [ [[package]] name = "iii-sdk" -version = "0.11.3" +version = "0.11.7-next.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f0226f7ce0d9071f9cb75ea7b7ac1241b15282915ccd41d9bbd2ee0db94f90c6" +checksum = "abad1f632473931d97733a4b3d73a04f2cfa1759fa4d4bcdab9305b634563bf7" dependencies = [ "async-trait", "futures-util", diff --git a/hook-fanout/Cargo.toml b/hook-fanout/Cargo.toml index 1ce603136..83a94ae1b 100644 --- a/hook-fanout/Cargo.toml +++ b/hook-fanout/Cargo.toml @@ -20,7 +20,7 @@ name = "hook-fanout" path = "src/main.rs" [dependencies] -iii-sdk = "=0.11.3" +iii-sdk = "=0.11.7-next.3" serde = { version = "1", features = ["derive"] } serde_json = "1" serde_yaml = "0.9" diff --git a/hook-fanout/src/config.rs b/hook-fanout/src/config.rs index 0ba215ffa..fd1fd6f73 100644 --- a/hook-fanout/src/config.rs +++ b/hook-fanout/src/config.rs @@ -9,6 +9,12 @@ pub struct WorkerConfig { pub min_timeout_ms: u64, #[serde(default = "default_poll_interval_ms")] pub poll_interval_ms: u64, + /// Exit the collect loop once this many milliseconds elapse with no + /// new replies after the first one. Caps the typical wait at + /// `first_reply_ms + quiescence_ms` instead of the full timeout. + /// Zero disables quiescence (always wait full timeout). + #[serde(default = "default_quiescence_ms")] + pub quiescence_ms: u64, } fn default_default_timeout_ms() -> u64 { @@ -23,12 +29,17 @@ fn default_poll_interval_ms() -> u64 { 25 } +fn default_quiescence_ms() -> u64 { + 200 +} + impl Default for WorkerConfig { fn default() -> Self { Self { default_timeout_ms: default_default_timeout_ms(), min_timeout_ms: default_min_timeout_ms(), poll_interval_ms: default_poll_interval_ms(), + quiescence_ms: default_quiescence_ms(), } } } @@ -49,17 +60,19 @@ mod tests { assert_eq!(cfg.default_timeout_ms, 10_000); assert_eq!(cfg.min_timeout_ms, 50); assert_eq!(cfg.poll_interval_ms, 25); + assert_eq!(cfg.quiescence_ms, 200); } #[test] fn custom_yaml_overrides() { let cfg: WorkerConfig = serde_yaml::from_str( - "default_timeout_ms: 3000\nmin_timeout_ms: 100\npoll_interval_ms: 10", + "default_timeout_ms: 3000\nmin_timeout_ms: 100\npoll_interval_ms: 10\nquiescence_ms: 50", ) .unwrap(); assert_eq!(cfg.default_timeout_ms, 3000); assert_eq!(cfg.min_timeout_ms, 100); assert_eq!(cfg.poll_interval_ms, 10); + assert_eq!(cfg.quiescence_ms, 50); } #[test] @@ -68,5 +81,6 @@ mod tests { assert_eq!(d.default_timeout_ms, default_default_timeout_ms()); assert_eq!(d.min_timeout_ms, default_min_timeout_ms()); assert_eq!(d.poll_interval_ms, default_poll_interval_ms()); + assert_eq!(d.quiescence_ms, default_quiescence_ms()); } } diff --git a/hook-fanout/src/handler.rs b/hook-fanout/src/handler.rs index 4fb478d1c..c1fe238d2 100644 --- a/hook-fanout/src/handler.rs +++ b/hook-fanout/src/handler.rs @@ -34,12 +34,20 @@ pub async fn execute( .get("timeout_ms") .and_then(Value::as_u64) .unwrap_or(cfg.default_timeout_ms); + let expected_replies = payload.get("expected_replies").and_then(Value::as_u64); + let quiescence_ms = payload + .get("quiescence_ms") + .and_then(Value::as_u64) + .unwrap_or(cfg.quiescence_ms); let min_timeout = cfg.min_timeout_ms; let poll_interval = Duration::from_millis(cfg.poll_interval_ms); + let quiescence = Duration::from_millis(quiescence_ms); let event_id = Uuid::new_v4().to_string(); let envelope = build_publish_envelope(&topic, &event_id, inner.clone()); + let started_at = Instant::now(); + let mut publish_failed = false; if let Err(e) = iii .trigger(TriggerRequest { function_id: "iii::durable::publish".into(), @@ -50,12 +58,17 @@ pub async fn execute( .await { tracing::warn!(error = %e, %topic, "hook-fanout::publish_collect: publish trigger failed"); + publish_failed = true; } - let deadline = Instant::now() + Duration::from_millis(timeout_ms.max(min_timeout)); + let deadline = started_at + Duration::from_millis(timeout_ms.max(min_timeout)); let mut replies: Vec = Vec::new(); let mut last_index: usize = 0; + let mut first_reply_at: Option = None; + let mut last_reply_at: Option = None; + let exit_reason: &'static str; loop { + let before_len = replies.len(); if let Ok(value) = iii .trigger(TriggerRequest { function_id: "stream::list".into(), @@ -67,12 +80,51 @@ pub async fn execute( { collect_stream_items(&value, &mut replies, &mut last_index); } - if Instant::now() >= deadline { + let now = Instant::now(); + if replies.len() > before_len { + if first_reply_at.is_none() { + first_reply_at = Some(now); + } + last_reply_at = Some(now); + } + + if let Some(reason) = decide_exit( + replies.len() as u64, + expected_replies, + quiescence_ms, + quiescence, + last_reply_at, + now, + deadline, + ) { + exit_reason = reason; break; } tokio::time::sleep(poll_interval).await; } + let elapsed_ms = started_at.elapsed().as_millis() as u64; + let first_reply_ms = first_reply_at + .map(|t| t.duration_since(started_at).as_millis() as u64) + .unwrap_or(0); + iii_sdk::set_current_span_attribute("hook_fanout.topic", topic.clone()); + iii_sdk::set_current_span_attribute("hook_fanout.replies", replies.len().to_string()); + iii_sdk::set_current_span_attribute("hook_fanout.elapsed_ms", elapsed_ms.to_string()); + iii_sdk::set_current_span_attribute("hook_fanout.first_reply_ms", first_reply_ms.to_string()); + iii_sdk::set_current_span_attribute("hook_fanout.exit_reason", exit_reason); + if publish_failed { + iii_sdk::set_current_span_attribute("hook_fanout.publish_failed", "true"); + } + tracing::info!( + topic = %topic, + replies = replies.len(), + elapsed_ms = elapsed_ms, + first_reply_ms = first_reply_ms, + exit_reason = exit_reason, + publish_failed = publish_failed, + "hook-fanout::publish_collect completed" + ); + let merged = match merge_rule { MergeRule::FirstBlockWins => merge_first_block_wins(&replies), MergeRule::FieldMerge => merge_field_merge(inner.clone(), &replies), @@ -102,6 +154,49 @@ pub fn register(iii: &Arc, config: &Arc) { )); } +/// Pure exit-decision for the collect loop. Returns `Some(reason)` when +/// the loop must stop, `None` when polling should continue. +/// +/// Precedence (matches the order checked in `execute`): +/// 1. `expected_replies` — exits immediately once the reply count +/// meets the caller-supplied target. +/// 2. `quiescence` — exits when at least one reply has landed AND +/// `quiescence` has elapsed since the last one. Disabled when +/// `quiescence_ms == 0`. +/// 3. `deadline` / `deadline_no_replies` — exits when wall-clock +/// crosses the deadline; the variant depends on whether any reply +/// was seen. +fn decide_exit( + replies_count: u64, + expected_replies: Option, + quiescence_ms: u64, + quiescence: Duration, + last_reply_at: Option, + now: Instant, + deadline: Instant, +) -> Option<&'static str> { + if let Some(expected) = expected_replies { + if replies_count >= expected { + return Some("expected_replies"); + } + } + if quiescence_ms > 0 { + if let Some(last) = last_reply_at { + if now.duration_since(last) >= quiescence { + return Some("quiescence"); + } + } + } + if now >= deadline { + return Some(if last_reply_at.is_some() { + "deadline" + } else { + "deadline_no_replies" + }); + } + None +} + fn collect_stream_items(value: &Value, collected: &mut Vec, last_index: &mut usize) { let items = value .as_array() @@ -157,4 +252,145 @@ mod tests { assert_eq!(out.len(), 2); assert_eq!(out[1]["n"], 2); } + + /// `expected_replies` short-circuit fires the moment the count meets + /// the target — even when quiescence and deadline haven't triggered. + /// Highest precedence in `decide_exit`; pins that ordering. + #[test] + fn decide_exit_fires_expected_replies_at_threshold() { + let start = Instant::now(); + // Deadline far in the future, no quiescence elapsed yet — only + // the expected count matters. + let deadline = start + Duration::from_secs(60); + let reason = decide_exit( + 3, + Some(3), + 200, + Duration::from_millis(200), + Some(start), + start, + deadline, + ); + assert_eq!(reason, Some("expected_replies")); + + // Same setup, but count is one short — must continue. + let reason_short = decide_exit( + 2, + Some(3), + 200, + Duration::from_millis(200), + Some(start), + start, + deadline, + ); + assert_eq!(reason_short, None); + } + + /// `quiescence` fires once `now - last_reply_at >= quiescence` AND + /// at least one reply has landed. With no reply observed yet, the + /// branch must NOT fire (otherwise zero-subscriber topics would + /// exit immediately). + #[test] + fn decide_exit_fires_quiescence_after_idle_window() { + let start = Instant::now(); + let deadline = start + Duration::from_secs(60); + let last = start + Duration::from_millis(100); + // 350ms since first reply, quiescence window = 200ms → fire. + let now = start + Duration::from_millis(350); + let reason = decide_exit( + 1, + None, + 200, + Duration::from_millis(200), + Some(last), + now, + deadline, + ); + assert_eq!(reason, Some("quiescence")); + + // last_reply_at = None (no reply yet) → quiescence must NOT + // fire even though deadline isn't reached either. + let reason_no_reply = decide_exit( + 0, + None, + 200, + Duration::from_millis(200), + None, + now, + deadline, + ); + assert_eq!(reason_no_reply, None); + } + + /// `quiescence_ms == 0` disables the quiescence branch entirely + /// (the documented opt-out). Pins so a refactor can't accidentally + /// flip the guard. + #[test] + fn decide_exit_quiescence_zero_disables_branch() { + let start = Instant::now(); + let deadline = start + Duration::from_secs(60); + let now = start + Duration::from_millis(500); + let reason = decide_exit( + 1, + None, + 0, + Duration::from_millis(0), + Some(start), + now, + deadline, + ); + assert_eq!(reason, None, "quiescence_ms=0 must NOT trigger exit"); + } + + /// `deadline` variant fires when `now >= deadline` AND at least one + /// reply has landed. Counterpart: `deadline_no_replies` is covered + /// in the next test. + #[test] + fn decide_exit_deadline_with_reply() { + let start = Instant::now(); + let deadline = start + Duration::from_millis(100); + let now = start + Duration::from_millis(150); + let reason = decide_exit( + 1, + None, + 0, + Duration::from_millis(0), + Some(start + Duration::from_millis(50)), + now, + deadline, + ); + assert_eq!(reason, Some("deadline")); + } + + /// `deadline_no_replies` fires when the deadline is reached and + /// `last_reply_at` is still None — distinguishes the zero-subscriber + /// case from the slow-subscriber case in observability dashboards. + #[test] + fn decide_exit_deadline_no_replies() { + let start = Instant::now(); + let deadline = start + Duration::from_millis(100); + let now = start + Duration::from_millis(150); + let reason = decide_exit(0, None, 0, Duration::from_millis(0), None, now, deadline); + assert_eq!(reason, Some("deadline_no_replies")); + } + + /// Precedence sanity: when expected_replies AND quiescence AND + /// deadline all match, expected_replies wins (highest priority). + #[test] + fn decide_exit_expected_replies_beats_quiescence_and_deadline() { + let start = Instant::now(); + let deadline = start; // already at deadline + let last = start; + let now = start + Duration::from_millis(500); + let reason = decide_exit( + 5, + Some(3), // expected met + 200, + Duration::from_millis(200), // quiescence elapsed too + Some(last), + now, + deadline, + ); + assert_eq!(reason, Some("expected_replies")); + } } diff --git a/provider-anthropic/Cargo.lock b/provider-anthropic/Cargo.lock index 32ff9d125..00fb51bdc 100644 --- a/provider-anthropic/Cargo.lock +++ b/provider-anthropic/Cargo.lock @@ -804,9 +804,9 @@ dependencies = [ [[package]] name = "iii-sdk" -version = "0.11.3" +version = "0.11.7-next.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f0226f7ce0d9071f9cb75ea7b7ac1241b15282915ccd41d9bbd2ee0db94f90c6" +checksum = "abad1f632473931d97733a4b3d73a04f2cfa1759fa4d4bcdab9305b634563bf7" dependencies = [ "async-trait", "futures-util", diff --git a/provider-anthropic/Cargo.toml b/provider-anthropic/Cargo.toml index 598723a42..2a542b2a5 100644 --- a/provider-anthropic/Cargo.toml +++ b/provider-anthropic/Cargo.toml @@ -54,7 +54,7 @@ authors = ["iii contributors"] [workspace.dependencies] async-trait = "0.1" -iii-sdk = "=0.11.3" +iii-sdk = "=0.11.7-next.3" serde = { version = "1", features = ["derive"] } serde_json = "1" tokio = { version = "1", features = ["full"] } diff --git a/provider-anthropic/crates/provider-base/src/openai_compat.rs b/provider-anthropic/crates/provider-base/src/openai_compat.rs index e415aab8e..3bc514f55 100644 --- a/provider-anthropic/crates/provider-base/src/openai_compat.rs +++ b/provider-anthropic/crates/provider-base/src/openai_compat.rs @@ -186,8 +186,14 @@ pub async fn stream_chat_completions( request: OpenAICompatRequest, ) -> ReceiverStream { let (tx, rx) = mpsc::channel(64); + // Preserve the caller's OTel context across tokio::spawn so the + // HTTP span inside stream_inner is parented + carries baggage. + let otel_cx = iii_sdk::capture_otel_context(); tokio::spawn(async move { - if let Err(e) = stream_inner(cfg.clone(), request, tx.clone()).await { + let result = otel_cx + .attach(stream_inner(cfg.clone(), request, tx.clone())) + .await; + if let Err(e) = result { let _ = tx .send(error_event( e.to_string(), @@ -221,31 +227,39 @@ async fn stream_inner( request: OpenAICompatRequest, tx: mpsc::Sender, ) -> Result<(), reqwest::Error> { - let mut body = serde_json::json!({ - "model": cfg.model, - "max_tokens": cfg.max_tokens, - "messages": to_openai_messages(&request.messages, &request.system_prompt), - "stream": true, - "stream_options": { "include_usage": true }, - }); - if !request.tools.is_empty() { - body["tools"] = serde_json::Value::Array(functions_to_openai(&request.tools)); - } - - let client = reqwest::Client::builder() - .timeout(std::time::Duration::from_mins(2)) - .build()?; - - let auth_name = cfg.auth_header_name.as_deref().unwrap_or("Authorization"); - let auth_prefix = cfg.auth_value_prefix.as_deref().unwrap_or("Bearer "); - let mut req = client - .post(&cfg.url) - .header("content-type", "application/json") - .header(auth_name, format!("{auth_prefix}{}", cfg.api_key)); - for (name, value) in &cfg.extra_headers { - req = req.header(name, value); - } - let resp = req.json(&body).send().await?; + // Pre-HTTP marshal: body assembly + client/header build. + let (client, http_request) = iii_sdk::run_in_span( + "openai.request.build", + Some(iii_sdk::SpanKind::Internal), + || async { + let mut body = serde_json::json!({ + "model": cfg.model, + "max_tokens": cfg.max_tokens, + "messages": to_openai_messages(&request.messages, &request.system_prompt), + "stream": true, + "stream_options": { "include_usage": true }, + }); + if !request.tools.is_empty() { + body["tools"] = serde_json::Value::Array(functions_to_openai(&request.tools)); + } + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_mins(2)) + .build()?; + let auth_name = cfg.auth_header_name.as_deref().unwrap_or("Authorization"); + let auth_prefix = cfg.auth_value_prefix.as_deref().unwrap_or("Bearer "); + let mut req = client + .post(&cfg.url) + .header("content-type", "application/json") + .header(auth_name, format!("{auth_prefix}{}", cfg.api_key)); + for (name, value) in &cfg.extra_headers { + req = req.header(name, value); + } + let http_request = req.json(&body).build()?; + Ok::<_, reqwest::Error>((client, http_request)) + }, + ) + .await?; + let resp = iii_sdk::execute_traced_request(&client, http_request).await?; let status = resp.status(); if !status.is_success() { @@ -282,36 +296,48 @@ async fn stream_inner( ..Default::default() }; - let mut bytes_stream = resp.bytes_stream(); - let mut buf = String::new(); - while let Some(chunk) = bytes_stream.next().await { - let chunk: Bytes = match chunk { - Ok(b) => b, - Err(e) => { - let _ = tx - .send(error_event( - e.to_string(), - None, - cfg.model.clone(), - cfg.provider_name.clone(), - )) - .await; - return Ok(()); - } - }; - buf.push_str(&String::from_utf8_lossy(&chunk)); - while let Some(idx) = buf.find("\n\n") { - let block = buf[..idx].to_string(); - buf.drain(..=idx + 1); - if let Some(event) = parse_sse_block(&block) { - if event.data == "[DONE]" { - break; - } - if let Ok(parsed) = serde_json::from_str::(&event.data) { - handle_chunk(&parsed, &mut state, &tx, &cfg).await; + // SSE consume: chunked event-stream parsing + delta forwarding. + let early_return = iii_sdk::run_in_span( + "openai.stream.consume", + Some(iii_sdk::SpanKind::Internal), + || async { + let mut bytes_stream = resp.bytes_stream(); + let mut buf = String::new(); + while let Some(chunk) = bytes_stream.next().await { + let chunk: Bytes = match chunk { + Ok(b) => b, + Err(e) => { + let _ = tx + .send(error_event( + e.to_string(), + None, + cfg.model.clone(), + cfg.provider_name.clone(), + )) + .await; + return true; + } + }; + buf.push_str(&String::from_utf8_lossy(&chunk)); + while let Some(idx) = buf.find("\n\n") { + let block = buf[..idx].to_string(); + buf.drain(..=idx + 1); + if let Some(event) = parse_sse_block(&block) { + if event.data == "[DONE]" { + return false; + } + if let Ok(parsed) = serde_json::from_str::(&event.data) { + handle_chunk(&parsed, &mut state, &tx, &cfg).await; + } + } } } - } + false + }, + ) + .await; + if early_return { + return Ok(()); } let final_msg = build_final(&state, &cfg.model, &cfg.provider_name); diff --git a/provider-anthropic/src/lib.rs b/provider-anthropic/src/lib.rs index 17b99e3ba..e362e4319 100644 --- a/provider-anthropic/src/lib.rs +++ b/provider-anthropic/src/lib.rs @@ -229,8 +229,21 @@ pub async fn stream( tools: Vec, ) -> ReceiverStream { let (tx, rx) = mpsc::channel(64); + // tokio::spawn drops the caller's OTel context; capture it here so + // the HTTP span inside stream_inner is parented to the invocation + // span (and inherits iii.session.id baggage for "Group by session"). + let otel_cx = iii_sdk::capture_otel_context(); tokio::spawn(async move { - if let Err(e) = stream_inner(cfg, system_prompt, messages, tools, tx.clone()).await { + let result = otel_cx + .attach(stream_inner( + cfg, + system_prompt, + messages, + tools, + tx.clone(), + )) + .await; + if let Err(e) = result { // Encode any error as final error event per the no-throw contract. let final_msg = AssistantMessage { content: vec![ContentBlock::Text(harness_types::TextContent { @@ -275,28 +288,38 @@ async fn stream_inner( tools: Vec, tx: mpsc::Sender, ) -> Result<(), AnthropicError> { - let body = serde_json::json!({ - "model": cfg.model, - "max_tokens": cfg.max_tokens, - "system": system_prompt, - "messages": to_wire_messages(&messages), - "tools": functions_to_wire(&tools), - "stream": true, - }); - - let client = reqwest::Client::builder() - .timeout(std::time::Duration::from_mins(2)) - .build()?; - - let (header_name, header_value) = auth_header_for(&cfg); - let resp = client - .post(&cfg.api_url) - .header(header_name, header_value) - .header("anthropic-version", "2023-06-01") - .header("content-type", "application/json") - .json(&body) - .send() - .await?; + // Pre-HTTP marshal: wire-message conversion + serde_json::to_value + // + reqwest client build + header assembly. Was the ~60ms gap + // between `call provider::anthropic::complete` start and the POST + // span start in the trace. + let (client, request) = iii_sdk::run_in_span( + "anthropic.request.build", + Some(iii_sdk::SpanKind::Internal), + || async { + let body = serde_json::json!({ + "model": cfg.model, + "max_tokens": cfg.max_tokens, + "system": system_prompt, + "messages": to_wire_messages(&messages), + "tools": functions_to_wire(&tools), + "stream": true, + }); + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_mins(2)) + .build()?; + let (header_name, header_value) = auth_header_for(&cfg); + let request = client + .post(&cfg.api_url) + .header(header_name, header_value) + .header("anthropic-version", "2023-06-01") + .header("content-type", "application/json") + .json(&body) + .build()?; + Ok::<_, AnthropicError>((client, request)) + }, + ) + .await?; + let resp = iii_sdk::execute_traced_request(&client, request).await?; let status = resp.status(); if !status.is_success() { @@ -341,19 +364,29 @@ async fn stream_inner( ..Default::default() }; - let mut bytes_stream = resp.bytes_stream(); - let mut buf = String::new(); - while let Some(chunk) = bytes_stream.next().await { - let chunk: Bytes = chunk?; - let text = String::from_utf8_lossy(&chunk); - buf.push_str(&text); - - while let Some(idx) = buf.find("\n\n") { - let event = buf[..idx].to_string(); - buf.drain(..=idx + 1); - handle_sse_event(&event, &mut state, &tx, &cfg.model).await; - } - } + // SSE consume: chunked event-stream parsing + delta forwarding. + // Was the ~260ms gap between POST end and complete-span end. + iii_sdk::run_in_span( + "anthropic.stream.consume", + Some(iii_sdk::SpanKind::Internal), + || async { + let mut bytes_stream = resp.bytes_stream(); + let mut buf = String::new(); + while let Some(chunk) = bytes_stream.next().await { + let chunk: Bytes = chunk?; + let text = String::from_utf8_lossy(&chunk); + buf.push_str(&text); + + while let Some(idx) = buf.find("\n\n") { + let event = buf[..idx].to_string(); + buf.drain(..=idx + 1); + handle_sse_event(&event, &mut state, &tx, &cfg.model).await; + } + } + Ok::<_, AnthropicError>(()) + }, + ) + .await?; let final_message = build_final(&state, &cfg.model); let _ = tx diff --git a/provider-openai/Cargo.lock b/provider-openai/Cargo.lock index 046962ac8..32f450990 100644 --- a/provider-openai/Cargo.lock +++ b/provider-openai/Cargo.lock @@ -804,9 +804,9 @@ dependencies = [ [[package]] name = "iii-sdk" -version = "0.11.3" +version = "0.11.7-next.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f0226f7ce0d9071f9cb75ea7b7ac1241b15282915ccd41d9bbd2ee0db94f90c6" +checksum = "abad1f632473931d97733a4b3d73a04f2cfa1759fa4d4bcdab9305b634563bf7" dependencies = [ "async-trait", "futures-util", diff --git a/provider-openai/Cargo.toml b/provider-openai/Cargo.toml index 84001ffb9..2629bfcd8 100644 --- a/provider-openai/Cargo.toml +++ b/provider-openai/Cargo.toml @@ -48,7 +48,7 @@ repository = "https://github.com/iii-hq/workers" authors = ["iii contributors"] [workspace.dependencies] -iii-sdk = "=0.11.3" +iii-sdk = "=0.11.7-next.3" serde = { version = "1", features = ["derive"] } serde_json = "1" tokio = { version = "1", features = ["full"] } diff --git a/provider-openai/crates/provider-base/src/openai_compat.rs b/provider-openai/crates/provider-base/src/openai_compat.rs index 4253681d0..6648ff5ea 100644 --- a/provider-openai/crates/provider-base/src/openai_compat.rs +++ b/provider-openai/crates/provider-base/src/openai_compat.rs @@ -186,8 +186,14 @@ pub async fn stream_chat_completions( request: OpenAICompatRequest, ) -> ReceiverStream { let (tx, rx) = mpsc::channel(64); + // Preserve the caller's OTel context across tokio::spawn so the + // HTTP span inside stream_inner is parented + carries baggage. + let otel_cx = iii_sdk::capture_otel_context(); tokio::spawn(async move { - if let Err(e) = stream_inner(cfg.clone(), request, tx.clone()).await { + let result = otel_cx + .attach(stream_inner(cfg.clone(), request, tx.clone())) + .await; + if let Err(e) = result { let _ = tx .send(error_event( e.to_string(), @@ -221,31 +227,39 @@ async fn stream_inner( request: OpenAICompatRequest, tx: mpsc::Sender, ) -> Result<(), reqwest::Error> { - let mut body = serde_json::json!({ - "model": cfg.model, - "max_completion_tokens": cfg.max_tokens, - "messages": to_openai_messages(&request.messages, &request.system_prompt), - "stream": true, - "stream_options": { "include_usage": true }, - }); - if !request.tools.is_empty() { - body["tools"] = serde_json::Value::Array(functions_to_openai(&request.tools)); - } - - let client = reqwest::Client::builder() - .timeout(std::time::Duration::from_mins(2)) - .build()?; - - let auth_name = cfg.auth_header_name.as_deref().unwrap_or("Authorization"); - let auth_prefix = cfg.auth_value_prefix.as_deref().unwrap_or("Bearer "); - let mut req = client - .post(&cfg.url) - .header("content-type", "application/json") - .header(auth_name, format!("{auth_prefix}{}", cfg.api_key)); - for (name, value) in &cfg.extra_headers { - req = req.header(name, value); - } - let resp = req.json(&body).send().await?; + // Pre-HTTP marshal: body assembly + client/header build. + let (client, http_request) = iii_sdk::run_in_span( + "openai.request.build", + Some(iii_sdk::SpanKind::Internal), + || async { + let mut body = serde_json::json!({ + "model": cfg.model, + "max_completion_tokens": cfg.max_tokens, + "messages": to_openai_messages(&request.messages, &request.system_prompt), + "stream": true, + "stream_options": { "include_usage": true }, + }); + if !request.tools.is_empty() { + body["tools"] = serde_json::Value::Array(functions_to_openai(&request.tools)); + } + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_mins(2)) + .build()?; + let auth_name = cfg.auth_header_name.as_deref().unwrap_or("Authorization"); + let auth_prefix = cfg.auth_value_prefix.as_deref().unwrap_or("Bearer "); + let mut req = client + .post(&cfg.url) + .header("content-type", "application/json") + .header(auth_name, format!("{auth_prefix}{}", cfg.api_key)); + for (name, value) in &cfg.extra_headers { + req = req.header(name, value); + } + let http_request = req.json(&body).build()?; + Ok::<_, reqwest::Error>((client, http_request)) + }, + ) + .await?; + let resp = iii_sdk::execute_traced_request(&client, http_request).await?; let status = resp.status(); if !status.is_success() { @@ -282,36 +296,48 @@ async fn stream_inner( ..Default::default() }; - let mut bytes_stream = resp.bytes_stream(); - let mut buf = String::new(); - while let Some(chunk) = bytes_stream.next().await { - let chunk: Bytes = match chunk { - Ok(b) => b, - Err(e) => { - let _ = tx - .send(error_event( - e.to_string(), - None, - cfg.model.clone(), - cfg.provider_name.clone(), - )) - .await; - return Ok(()); - } - }; - buf.push_str(&String::from_utf8_lossy(&chunk)); - while let Some(idx) = buf.find("\n\n") { - let block = buf[..idx].to_string(); - buf.drain(..=idx + 1); - if let Some(event) = parse_sse_block(&block) { - if event.data == "[DONE]" { - break; - } - if let Ok(parsed) = serde_json::from_str::(&event.data) { - handle_chunk(&parsed, &mut state, &tx, &cfg).await; + // SSE consume: chunked event-stream parsing + delta forwarding. + let early_return = iii_sdk::run_in_span( + "openai.stream.consume", + Some(iii_sdk::SpanKind::Internal), + || async { + let mut bytes_stream = resp.bytes_stream(); + let mut buf = String::new(); + while let Some(chunk) = bytes_stream.next().await { + let chunk: Bytes = match chunk { + Ok(b) => b, + Err(e) => { + let _ = tx + .send(error_event( + e.to_string(), + None, + cfg.model.clone(), + cfg.provider_name.clone(), + )) + .await; + return true; + } + }; + buf.push_str(&String::from_utf8_lossy(&chunk)); + while let Some(idx) = buf.find("\n\n") { + let block = buf[..idx].to_string(); + buf.drain(..=idx + 1); + if let Some(event) = parse_sse_block(&block) { + if event.data == "[DONE]" { + return false; + } + if let Ok(parsed) = serde_json::from_str::(&event.data) { + handle_chunk(&parsed, &mut state, &tx, &cfg).await; + } + } } } - } + false + }, + ) + .await; + if early_return { + return Ok(()); } let final_msg = build_final(&state, &cfg.model, &cfg.provider_name);