diff --git a/CHANGELOG.md b/CHANGELOG.md index 14bdb81..8646771 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ adheres to [Semantic Versioning](https://semver.org/). ## Unreleased +### GASP + +- **Plugin logs in the run's record (opt-in).** `GaspRecorder::extension()` (install on the agent: it learns each run's id) and `GaspRecorder::plugin_log_layer()` (add to the `tracing` subscriber) record a rutis plugin's diagnostics — the yoagent-rutis bridge's `log`, target `yoagent_rutis::plugin` — as `observation.created` on the run they belong to, through the summarizer, with `{source: "plugin", level}` metadata. Only lines whose `run_id` is one of the recorder's runs are recorded. The `gasp` feature now also enables `tracing-subscriber` (registry only). Test: `tests/gasp_plugin_log_test.rs`. + ### yoagent-rutis - **pi extensions** (`plugins/pi/pi-extensions-adapter.ts`): the tools and tool policies of [pi](https://github.com/earendil-works/pi) coding-agent extensions, loaded unchanged with pi's own loader, as one yoagent handler following pi 1.1.0's semantics. `registerTool` → tools (pi's activation and first-registration-wins, `prepareArguments` and validation before the policies, only the judged arguments run), `setActiveTools` → an enforced allowlist, `tool_call` → `before_tool` (yoagent's built-ins under pi's names, arguments translated both ways, relative paths resolved where the policies look; an override of a built-in is enforced; `terminate` stops the run), `tool_result` → `after_tool` (a failing handler withholds the result), `input` → `on_input`, `before_agent_start` additions → a turn note. Fails closed: a load error, a failing `session_start`, an unfired deciding event (`context`, `message_end`, ...; unless `allowUnmapped`) or a pi tool named like a yoagent built-in (unless `withoutBuiltins`) refuses the load; other unmapped API (observer and session events, commands, renderers, providers, MCP servers) is reported, or refuses with `strict`. Example `pi_extensions` (any extension files; `--live` with DeepSeek; `--without`), test `pi_test`. `ctx.executeTool` runs pi tools as nested calls (pi's pipeline; never rejects); a deciding event registered mid-run stops that run; adapter warnings reach the host's logs. `pi.appendEntry` / `pi.sendMessage` are recorded in the adapter's in-memory session instead of throwing (pi tools that record their result this way no longer fail after the work). CI: a weekly `pi latest` job runs the adapter's tests against the newest pi. diff --git a/CLAUDE.md b/CLAUDE.md index f2cec45..c0a30c2 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -112,7 +112,7 @@ Behind the `openapi` Cargo feature. `OpenApiToolAdapter` parses an OpenAPI 3.0 s ### GASP Bridge (`gasp.rs`, feature-gated) -Behind the `gasp` Cargo feature (dep: `yoagent-state`, the GASP reference runtime). `GaspRecorder` consumes the `AgentEvent` stream (via `recording_sender`, which also tees to a forward sender) and maps it onto `yoagent_state`'s `YoAgentStateSink`: AgentStart→run.started, assistant MessageEnd→model.called/finished pair, ToolExecutionStart/End→tool.called/finished, AgentEnd→run.finished + one git commit per run at stream close (scaffolding committed at init so clones restore). Stale open runs are closed as "interrupted" on open; a dropped sender finishes the run with the derived outcome; InputRejected → "rejected". Recording failures stop recording but the forward tee keeps flowing (error surfaces only via the returned handle — await it); events are forwarded BEFORE recording. `Ok(None)` when no AgentStart arrived. Zero loop changes. CI job `gasp-conformance` emits a repo via `examples/gasp_emit.rs` and runs the gasp conformance checker (7 checks) against it. +Behind the `gasp` Cargo feature (dep: `yoagent-state`, the GASP reference runtime). `GaspRecorder` consumes the `AgentEvent` stream (via `recording_sender`, which also tees to a forward sender) and maps it onto `yoagent_state`'s `YoAgentStateSink`: AgentStart→run.started, assistant MessageEnd→model.called/finished pair, ToolExecutionStart/End→tool.called/finished, AgentEnd→run.finished + one git commit per run at stream close (scaffolding committed at init so clones restore). Stale open runs are closed as "interrupted" on open; a dropped sender finishes the run with the derived outcome; InputRejected → "rejected". Recording failures stop recording but the forward tee keeps flowing (error surfaces only via the returned handle — await it); events are forwarded BEFORE recording. `Ok(None)` when no AgentStart arrived. Zero loop changes. **Plugin logs (opt-in):** `recorder.extension()` (`GaspRunLink`, an `Extension` whose `start_run` adds `run.run_id` to the open recording's `OpenRoute.runs`; no hooks) + `recorder.plugin_log_layer()` (`PluginLogLayer`, a `tracing_subscriber::Layer` — the `gasp` feature enables optional `tracing-subscriber` with `registry` only — that forwards events of target `PLUGIN_LOG_TARGET` = `yoagent_rutis::plugin` whose `run_id` field is in `runs` over an unbounded channel); `recording_sender` opens the route, `consume` `select!`s events and log lines, buffers lines until `AgentStart`, writes each as `record_observation` (`observed_in` the GASP run, title `plugin `, summary through the summarizer, metadata `{source: "plugin", level}`), records lines already queued before recording `AgentEnd` (so they stay inside the run), and at stream end clears the route **only if it is still its own** (`OpenRoute.tx.same_channel(&Logs.own)`; a next `recording_sender` may already have replaced it) and drains queued lines. Lines with another or no `run_id` are never recorded. Test: `tests/gasp_plugin_log_test.rs` (own binary, global subscriber). CI job `gasp-conformance` emits a repo via `examples/gasp_emit.rs` and runs the gasp conformance checker (7 checks) against it. ### Decision Models (`decision/`, feature-gated) diff --git a/Cargo.toml b/Cargo.toml index fb593dc..bcecfb8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -60,6 +60,8 @@ openapiv3 = { version = "2", optional = true } serde_yaml_ng = { version = "0.10", optional = true } # 0.4.2 floor: 0.4.1 declared MSRV 1.85 but required 1.88 (let-chain) yoagent-state = { version = "0.5", optional = true } +# The GASP recorder's plugin-log layer (`gasp::GaspRecorder::plugin_log_layer`). +tracing-subscriber = { version = "0.3", optional = true, default-features = false, features = ["registry"] } [features] default = ["native"] @@ -70,7 +72,7 @@ default = ["native"] # (e.g. Cloudflare Workers); see docs for what remains available there. native = ["tokio/process", "tokio/fs", "reqwest/default", "reqwest/socks"] openapi = ["dep:openapiv3", "dep:serde_yaml_ng", "reqwest/query"] -gasp = ["dep:yoagent-state"] +gasp = ["dep:yoagent-state", "dep:tracing-subscriber"] # Decision models (typed questions -> calibrated answers), e.g. TypeSafe Jev. # Off by default; adds no dependencies. Nothing is sent until you pick a model. decision = [] diff --git a/docs/concepts/gasp.md b/docs/concepts/gasp.md index a500425..66cd0b9 100644 --- a/docs/concepts/gasp.md +++ b/docs/concepts/gasp.md @@ -41,6 +41,7 @@ let run_id = record_handle.await??; // Option: None if no AgentStart ar | loop starts | `run.started` (the goal is stamped in the run commit's `Goal:` trailer) | | each assistant turn | `model.called` / `model.finished` paired nodes (a retried provider attempt — an error `MessageEnd` followed by `ProviderRetry` — is not recorded) | | each tool execution | `tool.called` / `tool.finished` (with success flag) | +| a plugin logs (opt-in, below) | `observation.created`, linked to the run, `{source: "plugin", level}` metadata | | loop ends | `run.finished` with the outcome: `completed` / `truncated` / `error` / `aborted` (a cancelled run) / `refused`, `loop_aborted:` when loop detection stopped it, `rejected` when an input filter refused the prompt, `interrupted` for a run a crashed process left open | The semantic log stores bounded one-line summaries — the **task string @@ -62,6 +63,21 @@ handle — **always await it** — while event forwarding to your UI continues uninterrupted. GASP repos are single-writer: don't share one repo between live workers, and record one run at a time. +## Plugin logs (opt-in) + +A rutis plugin's diagnostics — the [yoagent-rutis](https://github.com/yologdev/yoagent/tree/main/integrations/yoagent-rutis) bridge's `log`, such as a pi extension's "command `/todos` not available" or a DSH image that could not be read — can be part of the run's record. Install two things from the recorder: + +```rust,ignore +use tracing_subscriber::layer::SubscriberExt; + +tracing::subscriber::set_global_default( + tracing_subscriber::registry().with(recorder.plugin_log_layer()), +)?; +let agent = agent.with_extension(recorder.extension()); // learns each run's id +``` + +`tracing-subscriber` 0.3 must be a dependency of your own crate (the layer implements its `Layer` trait). The layer forwards lines of the target `yoagent_rutis::plugin` (`gasp::PLUGIN_LOG_TARGET`) whose `run_id` is one of this recorder's runs — another agent's plugins in the same process never land here, and a line without a run stays in your logs only. Each line goes through the summarizer, like tool output, and is recorded as an `observation.created` on the run. Installed with `with_tree_extension`, delegated runs' lines are recorded on the same run. Record every run of an agent that has the extension: a run started *without* a recording while one is open is attributed to that recording. + ## Tested conformance yoagent's CI emits an agent repo with a mock provider and runs the GASP diff --git a/integrations/yoagent-rutis/README.md b/integrations/yoagent-rutis/README.md index 6129166..21a7120 100644 --- a/integrations/yoagent-rutis/README.md +++ b/integrations/yoagent-rutis/README.md @@ -235,7 +235,9 @@ def apply(ctx, config): the host's `tracing` output under the target `yoagent_rutis::plugin`, where a terminal or service host shows it — a runtime process's own stderr may go nowhere. Pass `{ run_id }`, or the hook argument itself, to - attribute the line to its run (the event's `run_id` field). It never + attribute the line to its run (the event's `run_id` field) — a host + recording runs with yoagent's GASP recorder can then keep the line in the + run's record (`GaspRecorder::plugin_log_layer`). It never rejects (any message, extra context fields ignored). Fire-and-forget in JavaScript; in Python `await yoagent.log(...)` — an un-awaited coroutine is never sent. On a host without it (yoagent-rutis 0.1.0) rutis's diff --git a/src/gasp.rs b/src/gasp.rs index 6ce9276..8895205 100644 --- a/src/gasp.rs +++ b/src/gasp.rs @@ -43,8 +43,26 @@ //! //! Recording requires a git identity (`user.name`/`user.email`), like any //! git workflow. - +//! +//! # Plugin logs +//! +//! A rutis plugin's diagnostics (the `yoagent-rutis` bridge's `log`, written +//! to `tracing` under [`PLUGIN_LOG_TARGET`] with the run's `run_id`) can be +//! recorded as observations on the run they belong to — a plugin's "your +//! policy did not apply" is part of the run's record. Install two things +//! from the recorder: [`GaspRecorder::extension`] on the agent (it learns each +//! run's id) and [`GaspRecorder::plugin_log_layer`] on your `tracing` +//! subscriber (it forwards matching lines). Only lines whose `run_id` is one +//! of this recorder's runs are recorded, so another agent's plugins in the +//! same process never land here; lines without a run stay in `tracing` only. +//! Each line is summarized like tool output (so a redacting summarizer +//! applies) and recorded as `observation.created`, linked to the run, with +//! `{source: "plugin", level}` metadata. + +use crate::extension::{Extension, ExtensionError, RunContext, RunHooks}; use crate::types::*; +use std::collections::HashSet; +use std::sync::{Arc, Mutex}; use tokio::sync::mpsc; use tokio::task::JoinHandle; use yoagent_state::{ @@ -104,6 +122,108 @@ pub struct GaspRecorder { actor: ActorRef, goal: GoalId, summarize: Summarizer, + route: Route, +} + +/// The `tracing` target the `yoagent-rutis` bridge writes plugin diagnostics +/// to (its `log` method), with the run as a `run_id` field. +pub const PLUGIN_LOG_TARGET: &str = "yoagent_rutis::plugin"; + +/// Where plugin log lines go while a recording is open: the yoagent run ids +/// it covers (learned by [`GaspRecorder::extension`]) and its channel. +type Route = Arc>>; + +struct OpenRoute { + runs: HashSet, + tx: mpsc::UnboundedSender, +} + +/// One plugin log line on its way to the recording. +struct PluginLog { + level: tracing::Level, + message: String, +} + +/// The extension that links yoagent's run ids to the open recording — see +/// [`GaspRecorder::extension`]. +#[derive(Clone)] +pub struct GaspRunLink { + route: Route, +} + +#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))] +#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)] +impl Extension for GaspRunLink { + fn name(&self) -> &str { + "gasp-run-link" + } + + async fn start_run(&self, run: &RunContext<'_>) -> Result, ExtensionError> { + if let Some(open) = self + .route + .lock() + .unwrap_or_else(|e| e.into_inner()) + .as_mut() + { + open.runs.insert(run.run_id.to_string()); + } + Ok(Box::new(LinkedRun)) + } +} + +/// No hooks: the link only needs the run's start. +struct LinkedRun; + +#[cfg_attr(target_arch = "wasm32", async_trait::async_trait(?Send))] +#[cfg_attr(not(target_arch = "wasm32"), async_trait::async_trait)] +impl RunHooks for LinkedRun {} + +/// The `tracing` layer that forwards plugin log lines to the open recording — +/// see [`GaspRecorder::plugin_log_layer`]. +#[derive(Clone)] +pub struct PluginLogLayer { + route: Route, +} + +impl tracing_subscriber::Layer for PluginLogLayer { + fn on_event(&self, event: &tracing::Event<'_>, _: tracing_subscriber::layer::Context<'_, S>) { + let meta = event.metadata(); + if meta.target() != PLUGIN_LOG_TARGET { + return; + } + #[derive(Default)] + struct Fields { + run_id: String, + message: String, + } + impl tracing::field::Visit for Fields { + fn record_str(&mut self, field: &tracing::field::Field, value: &str) { + match field.name() { + "run_id" => self.run_id = value.to_string(), + "message" => self.message = value.to_string(), + _ => {} + } + } + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + match field.name() { + "run_id" => self.run_id = format!("{value:?}"), + "message" => self.message = format!("{value:?}"), + _ => {} + } + } + } + let mut fields = Fields::default(); + event.record(&mut fields); + let route = self.route.lock().unwrap_or_else(|e| e.into_inner()); + if let Some(open) = route.as_ref() { + if open.runs.contains(&fields.run_id) { + let _ = open.tx.send(PluginLog { + level: *meta.level(), + message: fields.message, + }); + } + } + } } impl GaspRecorder { @@ -207,6 +327,7 @@ impl GaspRecorder { actor, goal, summarize: std::sync::Arc::new(|text: &str| summarize(text)), + route: Route::default(), }) } @@ -265,6 +386,28 @@ impl GaspRecorder { &self.goal } + /// The extension that lets this recorder attribute plugin log lines to + /// the run they belong to: install it on the agent + /// ([`Agent::with_extension`](crate::Agent::with_extension); as a tree + /// extension, delegated runs' lines are recorded on the same run). It + /// only learns each run's id; it has no other effect. + pub fn extension(&self) -> GaspRunLink { + GaspRunLink { + route: self.route.clone(), + } + } + + /// The `tracing` layer that forwards plugin log lines + /// ([`PLUGIN_LOG_TARGET`]) of this recorder's runs to the recording, as + /// observations. Add it to your subscriber + /// (`tracing_subscriber::registry().with(recorder.plugin_log_layer())`) + /// and install [`extension`](Self::extension) on the agent. + pub fn plugin_log_layer(&self) -> PluginLogLayer { + PluginLogLayer { + route: self.route.clone(), + } + } + /// Returns a sender to pass to /// [`Agent::prompt_with_sender`](crate::Agent::prompt_with_sender) (or /// the raw loop) and a handle resolving to the recorded [`RunId`] — @@ -290,11 +433,62 @@ impl GaspRecorder { let goal = self.goal.clone(); let task = task.into(); let summarize = self.summarize.clone(); - let handle = tokio::spawn(consume(sink, store, goal, task, summarize, rx, forward)); + // Plugin log lines of this recording's runs (see `extension`). + let (log_tx, log_rx) = mpsc::unbounded_channel(); + *self.route.lock().unwrap_or_else(|e| e.into_inner()) = Some(OpenRoute { + runs: HashSet::new(), + tx: log_tx.clone(), + }); + let logs = Logs { + state: self.state.clone(), + actor: self.actor.clone(), + rx: log_rx, + route: self.route.clone(), + own: log_tx, + }; + let handle = tokio::spawn(consume( + sink, store, goal, task, summarize, rx, forward, logs, + )); (tx, handle) } } +/// The plugin log lines a recording receives, and where to write them. +struct Logs { + state: YoAgentState, + actor: ActorRef, + rx: mpsc::UnboundedReceiver, + route: Route, + /// This recording's own sender: the route is cleared only while it is + /// still this recording's (a next recording may have opened its own). + own: mpsc::UnboundedSender, +} + +impl Logs { + /// Record one line as an observation on the run. + async fn record( + &self, + summarize: &Summarizer, + run_id: &RunId, + line: PluginLog, + ) -> Result<(), StateError> { + let level = line.level.as_str().to_ascii_lowercase(); + self.state + .record_observation( + self.actor.clone(), + Observation { + id: ObservationId::generate(), + title: format!("plugin {level}"), + summary: summarize(&line.message), + observed_in: Some(run_id.clone()), + metadata: serde_json::json!({ "source": "plugin", "level": level }), + }, + ) + .await + .map(|_| ()) + } +} + /// Per-run bookkeeping threaded through event recording. struct RunTracking { run_id: RunId, @@ -309,6 +503,7 @@ struct RunTracking { /// the sender side is dropped (the loop finished, or the caller dropped the /// `*_with_sender` future mid-run / the loop task panicked — `AgentEnd` is /// otherwise sent unconditionally). +#[allow(clippy::too_many_arguments)] // each argument is one piece of a recording async fn consume( sink: YoAgentStateAdapter, store: GitEventStore, @@ -317,6 +512,7 @@ async fn consume( summarize: Summarizer, mut rx: mpsc::UnboundedReceiver, forward: Option>, + mut logs: Logs, ) -> Result, StateError> { let mut tracking = RunTracking { run_id: RunId::generate(), @@ -331,8 +527,28 @@ async fn consume( // after it means the attempt was retried, not the turn's result, so it is // not recorded as a model call (and does not advance the turn count). let mut held: Option = None; - - while let Some(event) = rx.recv().await { + // Plugin log lines that arrive before the run is recorded as started. + let mut early: Vec = Vec::new(); + + loop { + let event = tokio::select! { + event = rx.recv() => match event { + Some(event) => event, + None => break, + }, + Some(line) = logs.rx.recv() => { + if recording_error.is_none() { + if tracking.started { + if let Err(e) = logs.record(&summarize, &tracking.run_id, line).await { + recording_error = Some(e); + } + } else { + early.push(line); + } + } + continue; + } + }; // Forward FIRST: the tee observes the loop, not the recorder's disk. // It must neither lag behind per-event fsyncs nor die when recording // fails. @@ -353,6 +569,20 @@ async fn consume( } else { due.push(event); } + let starts = due.iter().any(|e| matches!(e, AgentEvent::AgentStart)); + // Lines already queued when the run ends are recorded before its end, + // so they stay inside the run (correlated with it) in the log. + if tracking.started + && recording_error.is_none() + && due.iter().any(|e| matches!(e, AgentEvent::AgentEnd { .. })) + { + while let Ok(line) = logs.rx.try_recv() { + if let Err(e) = logs.record(&summarize, &tracking.run_id, line).await { + recording_error = Some(e); + break; + } + } + } for event in &due { if let Err(e) = record_event(&sink, &summarize, &mut tracking, event).await { tracing::error!( @@ -364,6 +594,33 @@ async fn consume( break; } } + if starts && recording_error.is_none() { + for line in early.drain(..) { + if let Err(e) = logs.record(&summarize, &tracking.run_id, line).await { + recording_error = Some(e); + break; + } + } + } + } + // The recording is over: stop routing — unless a next recording already + // replaced the route with its own — then record lines already queued. + { + let mut route = logs.route.lock().unwrap_or_else(|e| e.into_inner()); + if route + .as_ref() + .is_some_and(|open| open.tx.same_channel(&logs.own)) + { + *route = None; + } + } + while let Ok(line) = logs.rx.try_recv() { + if recording_error.is_some() || !tracking.started { + break; + } + if let Err(e) = logs.record(&summarize, &tracking.run_id, line).await { + recording_error = Some(e); + } } if let (Some(event), None) = (held.take(), &recording_error) { // The stream ended on the error: it was the final one. diff --git a/tests/gasp_plugin_log_test.rs b/tests/gasp_plugin_log_test.rs new file mode 100644 index 0000000..884d6d6 --- /dev/null +++ b/tests/gasp_plugin_log_test.rs @@ -0,0 +1,123 @@ +//! Plugin log lines land in the GASP record of the run they belong to. +//! +//! A stand-in for a rutis plugin writes `tracing` events the way the +//! yoagent-rutis bridge's `log` does (target `yoagent_rutis::plugin`, a +//! `run_id` field). With the recorder's extension on the agent and its layer +//! on the subscriber, the line of this run is recorded as an observation on +//! it; a line of another run, and one without a run, are not. +//! +//! Its own test binary: it installs a global subscriber, once. +#![cfg(feature = "gasp")] + +use std::sync::Arc; +use tracing_subscriber::layer::SubscriberExt; +use yoagent::extension::{Extension, ExtensionError, RunContext, RunHooks, TurnDecision}; +use yoagent::gasp::{GaspRecorder, GoalRef, PLUGIN_LOG_TARGET}; +use yoagent::provider::mock::MockResponse; +use yoagent::provider::{MockProvider, ModelConfig}; +use yoagent::types::TurnContext; +use yoagent::Agent; + +/// Logs like a plugin through the bridge, once per run, at its first request. +struct PluginStandIn; + +struct StandInRun { + run_id: String, +} + +#[async_trait::async_trait] +impl Extension for PluginStandIn { + fn name(&self) -> &str { + "plugin-stand-in" + } + async fn start_run(&self, run: &RunContext<'_>) -> Result, ExtensionError> { + Ok(Box::new(StandInRun { + run_id: run.run_id.to_string(), + })) + } +} + +#[async_trait::async_trait] +impl RunHooks for StandInRun { + async fn before_model(&mut self, _turn: &TurnContext<'_>) -> TurnDecision { + let run_id = self.run_id.as_str(); + tracing::warn!(target: PLUGIN_LOG_TARGET, run_id, "pi command /todos not available"); + tracing::error!(target: PLUGIN_LOG_TARGET, run_id = "someone-else", "another agent's line"); + tracing::warn!(target: PLUGIN_LOG_TARGET, run_id = "", "a line with no run"); + TurnDecision::Continue + } +} + +fn events(dir: &std::path::Path) -> Vec { + std::fs::read_to_string(dir.join("state/events.jsonl")) + .unwrap() + .lines() + .map(|l| serde_json::from_str(l).unwrap()) + .collect() +} + +#[tokio::test(flavor = "multi_thread")] +async fn plugin_log_lines_are_recorded_on_their_runs() { + for (k, v) in [ + ("GIT_AUTHOR_NAME", "yoagent-test"), + ("GIT_AUTHOR_EMAIL", "test@yolog.dev"), + ("GIT_COMMITTER_NAME", "yoagent-test"), + ("GIT_COMMITTER_EMAIL", "test@yolog.dev"), + ] { + std::env::set_var(k, v); + } + let dir = tempfile::tempdir().unwrap(); + let recorder = GaspRecorder::init( + dir.path(), + "test-agent", + "w1", + GoalRef::New { + title: "plugin logs".into(), + }, + ) + .await + .unwrap(); + tracing::subscriber::set_global_default( + tracing_subscriber::registry().with(recorder.plugin_log_layer()), + ) + .unwrap(); + + let mut agent = Agent::from_provider( + MockProvider::new(vec![ + MockResponse::Text("one".into()), + MockResponse::Text("two".into()), + ]), + ModelConfig::mock(), + ) + .with_extension(Arc::new(PluginStandIn)) + .with_extension(recorder.extension()); + + // Run 1: its own line recorded on it; another run's and a run-less one not. + let (tx1, handle1) = recorder.recording_sender("first", None); + // Keep the first recording open while the second starts: its end must not + // switch off the second's routing. + let keep_open = tx1.clone(); + agent.prompt_with_sender("first", tx1).await; + let (tx2, handle2) = recorder.recording_sender("second", None); + drop(keep_open); + let run1 = handle1.await.unwrap().unwrap().expect("run 1 recorded"); + agent.prompt_with_sender("second", tx2).await; + let run2 = handle2.await.unwrap().unwrap().expect("run 2 recorded"); + + let all = events(dir.path()); + let observations: Vec<&serde_json::Value> = all + .iter() + .filter(|e| e["kind"] == "observation.created") + .map(|e| &e["payload"]) + .collect(); + assert_eq!(observations.len(), 2, "{observations:#?}"); + for (payload, run) in observations.iter().zip([&run1, &run2]) { + assert_eq!( + payload["summary"], "pi command /todos not available", + "{payload}" + ); + assert_eq!(payload["observed_in"], run.0.as_str(), "{payload}"); + assert_eq!(payload["metadata"]["source"], "plugin"); + assert_eq!(payload["metadata"]["level"], "warn"); + } +}