From 4405f2c25a4fad3051e18413b1e695e0fd5d07ca Mon Sep 17 00:00:00 2001 From: Yuanhao Li Date: Sat, 11 Jul 2026 01:11:26 +0200 Subject: [PATCH 1/3] =?UTF-8?q?feat:=20GASP=20bridge=20=E2=80=94=20yoagent?= =?UTF-8?q?=20is=20a=20tested=20GASP-conformant=20runtime?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase C2(b) of the roadmap: record agent runs into a GASP agent repo ("the repo is the agent"), with the protocol's conformance checker running in CI. - New `gasp` Cargo feature (optional dep: yoagent-state 0.4, the GASP reference runtime — feather-light: serde/tokio/uuid/chrono, git via shell). Placement decision: the bridge lives in yoagent because the reverse direction would force yoagent-state to depend on the whole agent runtime (reqwest et al.) just for the AgentEvent enum. - `gasp::GaspRecorder`: consumes the AgentEvent stream via `recording_sender(task, forward)` — a sender for prompt_with_sender plus a handle resolving to the RunId; events are teed to an optional forward sender so the UI keeps streaming. Zero agent-loop changes. - Mapping: AgentStart→run.started · assistant MessageEnd→model.called/ finished pair · ToolExecutionStart/End→tool.called/finished · AgentEnd→run.finished + one git commit per run (append-only history). Summaries are bounded one-liners — full transcripts belong in GASP's transcripts/ tier (= Session::to_jsonl). - Robustness: stale open runs are closed as "interrupted" on open (crashed process), and a dropped sender without AgentEnd finishes the run as "interrupted" before committing. - `GoalRef::{Existing, New}` — runs chain to a caller-managed goal. - New CI job `gasp-conformance`: emits a repo via examples/gasp_emit.rs (MockProvider — deterministic, no network) and runs the gasp conformance checker; all 7 checks verified passing locally ("conformant: all checks passed"). Tests (4): full-run event skeleton in order + commit assertion; tee to forward sender; goal reuse across recorder reopen (single goal.created); dropped-sender interrupted-run closure. Docs: concepts/gasp.md (+ SUMMARY, session-trees cross-link), README integrations bullet, lib.rs bullet, CLAUDE.md section, CHANGELOG. Co-Authored-By: Claude Fable 5 --- .github/workflows/ci.yml | 25 +++ CHANGELOG.md | 12 ++ CLAUDE.md | 4 + Cargo.toml | 6 + README.md | 1 + docs/SUMMARY.md | 1 + docs/concepts/gasp.md | 66 +++++++ docs/concepts/session-trees.md | 4 +- examples/gasp_emit.rs | 83 +++++++++ src/gasp.rs | 306 +++++++++++++++++++++++++++++++++ src/lib.rs | 6 + tests/gasp_test.rs | 223 ++++++++++++++++++++++++ 12 files changed, 735 insertions(+), 2 deletions(-) create mode 100644 docs/concepts/gasp.md create mode 100644 examples/gasp_emit.rs create mode 100644 src/gasp.rs create mode 100644 tests/gasp_test.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6eecff5..e229b22 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -59,3 +59,28 @@ jobs: - uses: Swatinem/rust-cache@v2 - name: Check run: cargo check --all-features + + # yoagent claims to be a tested GASP-conformant runtime — this job is the + # test: emit an agent repo with the gasp bridge (mock provider, no network) + # and run the protocol's conformance checker against it. + gasp-conformance: + name: GASP conformance + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@stable + - uses: Swatinem/rust-cache@v2 + - name: Emit agent repo via the gasp bridge + run: | + git config --global user.email "ci@yolog.dev" + git config --global user.name "yoagent-ci" + cargo run --example gasp_emit --features gasp -- /tmp/emitted-repo + - name: Checkout gasp (conformance checker) + uses: actions/checkout@v4 + with: + repository: yologdev/gasp + path: gasp-checker + - name: Run conformance checker + run: | + cd gasp-checker/conformance-check + cargo run -q -- /tmp/emitted-repo diff --git a/CHANGELOG.md b/CHANGELOG.md index 0f60173..d6c5e63 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,18 @@ All notable changes to `yoagent` are documented here. The format loosely follows [Keep a Changelog](https://keepachangelog.com/), and the project adheres to [Semantic Versioning](https://semver.org/). +## Unreleased + +### Added + +- **GASP bridge** (feature `gasp`) — `gasp::GaspRecorder` records agent runs + into a [GASP](https://github.com/yologdev/gasp) agent repo via + `yoagent-state`: append-only `state/events.jsonl` (goal/run/model/tool + events), one git commit per run, stale/interrupted runs closed safely, + events teed to your UI. yoagent is now a **tested** GASP-conformant + runtime: CI emits a repo and runs the protocol's 7-check conformance suite + against it. New `gasp_emit` example and docs page. + ## 0.11.0 ### Fixed (pre-release review of the items below) diff --git a/CLAUDE.md b/CLAUDE.md index ca8e88c..a5fed3d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -84,6 +84,10 @@ Behind the `openapi` Cargo feature. `OpenApiToolAdapter` parses an OpenAPI 3.0 s `McpClient` communicates via `McpTransport` trait (stdio or HTTP). `McpToolAdapter` wraps MCP tools to implement `AgentTool`, making them transparent to the agent loop. Added via `Agent::with_mcp_server_stdio()` / `with_mcp_server_http()`. +### 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. Stale open runs are closed as "interrupted" on open; a dropped sender finishes the run as "interrupted". 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. + ### Session Trees (`session.rs`) `Session` stores history as an id/parent_id tree: `append` advances the head, `seek`/`seek_checkpoint` move it, appending after a seek forks a new branch (never overwrites). `path_messages()` feeds a branch into `Agent::with_messages`; `append_new(agent.messages())` is the post-run sync. JSONL persistence (`to_jsonl`/`from_jsonl`, head = last line). Freestanding — no loop changes; maps to GASP's `transcripts/` tier. diff --git a/Cargo.toml b/Cargo.toml index bb06bb1..aebd90e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,12 +31,18 @@ rand = "0.10.0" base64 = "0.22" openapiv3 = { version = "2", optional = true } serde_yaml_ng = { version = "0.10", optional = true } +yoagent-state = { version = "0.4.1", optional = true } [features] openapi = ["dep:openapiv3", "dep:serde_yaml_ng", "reqwest/query"] +gasp = ["dep:yoagent-state"] [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } tracing-subscriber = { version = "0.3", features = ["registry", "fmt"] } tempfile = "3" wiremock = "0.6" + +[[example]] +name = "gasp_emit" +required-features = ["gasp"] diff --git a/README.md b/README.md index e864468..e3fbda7 100644 --- a/README.md +++ b/README.md @@ -65,6 +65,7 @@ Everything is observable via events. Supports 7 API protocols covering 20+ LLM p **Integrations** - OpenAPI tool adapter — auto-generate tools from any OpenAPI 3.0 spec (`features = ["openapi"]`) - MCP (Model Context Protocol) — connect to MCP tool servers via stdio or HTTP +- GASP (`features = ["gasp"]`) — record runs into a [GASP](https://github.com/yologdev/gasp) agent repo (append-only semantic event log; restore = clone + replay); yoagent is a **tested** GASP-conformant runtime — the protocol's 7-check conformance suite runs in CI **Context Management** - Context overflow detection across all major providers (Anthropic, OpenAI, Google, Bedrock, xAI, Groq, OpenRouter, llama.cpp, and more) diff --git a/docs/SUMMARY.md b/docs/SUMMARY.md index 384a479..006fad6 100644 --- a/docs/SUMMARY.md +++ b/docs/SUMMARY.md @@ -20,6 +20,7 @@ - [Sub-Agents](concepts/sub-agents.md) - [State Persistence](concepts/persistence.md) - [Session Trees](concepts/session-trees.md) +- [GASP: Your Agent Is a Git Repo](concepts/gasp.md) - [Lifecycle Callbacks](concepts/callbacks.md) - [Telemetry](concepts/telemetry.md) diff --git a/docs/concepts/gasp.md b/docs/concepts/gasp.md new file mode 100644 index 0000000..f012437 --- /dev/null +++ b/docs/concepts/gasp.md @@ -0,0 +1,66 @@ +# GASP: Your Agent Is a Git Repo + +[GASP](https://github.com/yologdev/gasp) — the Git Agent State Protocol — +keeps an agent's durable self in a git repository: an **append-only semantic +event log** (`state/events.jsonl`) that folds into a typed +goal/run/model/tool graph, alongside identity, skills, and memory tiers. +Restore = `git clone` + replay. Clone your agent onto a new machine and it +remembers everything, with lineage. + +yoagent bridges to GASP through the `gasp` feature (backed by +[`yoagent-state`](https://crates.io/crates/yoagent-state), the reference +runtime). The bridge is a consumer of the [`AgentEvent`] stream — **zero +agent-loop changes**: + +```toml +yoagent = { version = "0.11", features = ["gasp"] } +``` + +```rust +use yoagent::gasp::{GaspRecorder, GoalRef}; + +let recorder = GaspRecorder::init( + "./my-agent-repo", "my-agent", "worker-1", + GoalRef::New { title: "ship the feature".into() }, +).await?; + +let (tx, record_handle) = recorder.recording_sender("implement the parser", None); +agent.prompt_with_sender("implement the parser", tx).await; +let run_id = record_handle.await??; // events appended + committed +``` + +## What gets recorded + +| Agent activity | GASP events | +|---|---| +| loop starts | `run.started` (chained to your goal) | +| each assistant turn | `model.called` / `model.finished` paired nodes | +| each tool execution | `tool.called` / `tool.finished` (with success flag) | +| loop ends | `run.finished` with the outcome (`completed` / `error` / `aborted` / ...) | + +The semantic log carries **summaries** — task, model, tool names, outcomes — +never full conversations. Full transcripts belong in GASP's cold +`transcripts/` tier, which is exactly the shape of +[`Session::to_jsonl`](session-trees.md): drop your session file there and the +two tiers compose. + +Robustness: a crashed process leaves no open run — the recorder closes stale +runs as `interrupted` on open, and a dropped event stream finishes the run as +`interrupted` before committing. One git commit per run keeps the history +append-only. + +## Tested conformance + +yoagent's CI emits an agent repo with a mock provider and runs the GASP +conformance checker against it — all seven mechanical checks (envelope +round-trip, replay, vocabulary, append-only git history, causation integrity, +restore, domain↔ops consistency) must pass on every commit. Try it yourself: + +```bash +cargo run --example gasp_emit --features gasp -- /tmp/my-agent +git clone https://github.com/yologdev/gasp && cd gasp/conformance-check +cargo run -q -- /tmp/my-agent +# conformant: all checks passed +``` + +[`AgentEvent`]: https://docs.rs/yoagent/latest/yoagent/enum.AgentEvent.html diff --git a/docs/concepts/session-trees.md b/docs/concepts/session-trees.md index 16f8c99..6524b6a 100644 --- a/docs/concepts/session-trees.md +++ b/docs/concepts/session-trees.md @@ -73,5 +73,5 @@ remains for single-branch persistence. This tree is the shape of the [GASP](https://github.com/yologdev/gasp) `transcripts/` tier — the raw-conversation cold tier. The semantic event log -(goals/runs/evals living in a git repo) is a separate adapter layered on the -`AgentEvent` stream. +lives in the [`gasp` feature](gasp.md): a recorder over the `AgentEvent` +stream that emits a conformance-checked agent repo. diff --git a/examples/gasp_emit.rs b/examples/gasp_emit.rs new file mode 100644 index 0000000..8ea061e --- /dev/null +++ b/examples/gasp_emit.rs @@ -0,0 +1,83 @@ +//! GASP emission example: run a (mock) agent and record it into a GASP agent +//! repo — the repo IS the agent's durable state. +//! +//! Run with: cargo run --example gasp_emit --features gasp -- [repo-path] +//! +//! The emitted repo passes the GASP conformance checker +//! (github.com/yologdev/gasp) — yoagent's CI verifies exactly that. Point it +//! at a path of your choice and inspect `state/events.jsonl` and `git log` +//! afterwards. + +use yoagent::gasp::{GaspRecorder, GoalRef}; +use yoagent::provider::mock::*; +use yoagent::provider::{MockProvider, ModelConfig}; +use yoagent::*; + +/// A tiny tool so the log shows tool-call pairs, not just model calls. +struct TouchTool; + +#[async_trait::async_trait] +impl AgentTool for TouchTool { + fn name(&self) -> &str { + "touch" + } + fn label(&self) -> &str { + "Touch" + } + fn description(&self) -> &str { + "pretends to touch a file" + } + fn parameters_schema(&self) -> serde_json::Value { + serde_json::json!({"type": "object", "properties": {"path": {"type": "string"}}}) + } + async fn execute( + &self, + params: serde_json::Value, + _ctx: ToolContext, + ) -> Result { + Ok(ToolResult { + content: vec![Content::Text { + text: format!("touched {}", params["path"]), + }], + details: serde_json::Value::Null, + }) + } +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let repo = std::env::args() + .nth(1) + .unwrap_or_else(|| "/tmp/yoagent-gasp-demo".into()); + + let recorder = GaspRecorder::init( + &repo, + "demo-agent", + "worker-1", + GoalRef::New { + title: "demonstrate GASP emission".into(), + }, + ) + .await?; + + // Mock agent: one tool call, then a final answer — deterministic, no key. + let provider = MockProvider::new(vec![ + MockResponse::ToolCalls(vec![MockToolCall { + provider_metadata: None, + name: "touch".into(), + arguments: serde_json::json!({"path": "hello.txt"}), + }]), + MockResponse::Text("Done: touched hello.txt".into()), + ]); + let mut agent = + Agent::from_provider(provider, ModelConfig::mock()).with_tools(vec![Box::new(TouchTool)]); + + let (tx, record_handle) = recorder.recording_sender("touch hello.txt", None); + agent.prompt_with_sender("touch hello.txt", tx).await; + let run_id = record_handle.await??; + + println!("recorded run {run_id} into {repo}"); + println!("inspect: cat {repo}/state/events.jsonl"); + println!("verify: conformance-check {repo}"); + Ok(()) +} diff --git a/src/gasp.rs b/src/gasp.rs new file mode 100644 index 0000000..5099966 --- /dev/null +++ b/src/gasp.rs @@ -0,0 +1,306 @@ +//! GASP bridge — record agent runs into a [GASP](https://github.com/yologdev/gasp) +//! agent repo (feature `gasp`). +//! +//! GASP ("the repo is the agent") keeps an agent's durable self in a git +//! repository: an append-only semantic event log (`state/events.jsonl`) that +//! folds into a typed goal/run/model/tool graph, with restore = `clone + +//! replay`. This module is the bridge between yoagent's [`AgentEvent`] stream +//! and the [`yoagent_state`] reference runtime — **zero agent-loop changes**; +//! the recorder is just another consumer of the event stream. +//! +//! ```no_run +//! use yoagent::{Agent, gasp::{GaspRecorder, GoalRef}, provider::ModelConfig}; +//! +//! # #[tokio::main] +//! # async fn main() -> Result<(), Box> { +//! let recorder = GaspRecorder::init( +//! "./my-agent-repo", +//! "my-agent", +//! "worker-1", +//! GoalRef::New { title: "ship the feature".into() }, +//! ) +//! .await?; +//! +//! let mut agent = Agent::from_config(ModelConfig::anthropic("claude-sonnet-5", "Sonnet 5")); +//! let (tx, record_handle) = recorder.recording_sender("implement the parser", None); +//! agent.prompt_with_sender("implement the parser", tx).await; +//! let run_id = record_handle.await??; // events appended + committed +//! # let _ = run_id; Ok(()) +//! # } +//! ``` +//! +//! The semantic log records **summaries** (task, model, tool names, outcome) +//! — full conversations belong in GASP's cold `transcripts/` tier, which is +//! exactly the shape of [`Session::to_jsonl`](crate::Session::to_jsonl). + +use crate::types::*; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; +use yoagent_state::{ + ActorRef, GitEventStore, Goal, YoAgentModelCalled, YoAgentModelFinished, YoAgentRunFinished, + YoAgentRunStarted, YoAgentState, YoAgentStateAdapter, YoAgentStateSink, YoAgentToolCalled, + YoAgentToolFinished, +}; +pub use yoagent_state::{GoalId, RunId, StateError}; + +/// Which GASP goal recorded runs chain to. +#[derive(Debug, Clone)] +pub enum GoalRef { + /// Use an existing goal (persist the id in your app's config). + Existing(GoalId), + /// Create a new goal with this title when the recorder opens. + New { title: String }, +} + +/// Records an agent's [`AgentEvent`] stream into a GASP agent repo. +/// +/// One recorder = one writer (`worker_id` names it in the repo lease) and one +/// goal; each `recording_sender` call records one run. Events are appended to +/// `state/events.jsonl` as they arrive and committed when the run closes, so +/// the git history stays append-only (GASP conformance check 4). +pub struct GaspRecorder { + state: YoAgentState, + store: GitEventStore, + actor: ActorRef, + goal: GoalId, +} + +impl GaspRecorder { + /// Initialize a fresh agent repo at `root` (git init + minimal manifest) + /// and open a recorder on it. + pub async fn init( + root: impl AsRef, + agent_id: &str, + worker_id: &str, + goal: GoalRef, + ) -> Result { + let store = yoagent_state::init_agent_repo(root, agent_id, worker_id)?; + Self::with_store(store, agent_id, goal).await + } + + /// Open a recorder on an existing GASP agent repo. + pub async fn open( + root: impl Into, + agent_id: &str, + worker_id: &str, + goal: GoalRef, + ) -> Result { + let store = GitEventStore::open(root, worker_id)?; + Self::with_store(store, agent_id, goal).await + } + + async fn with_store( + store: GitEventStore, + agent_id: &str, + goal: GoalRef, + ) -> Result { + let actor = ActorRef::agent(agent_id); + let state = YoAgentState::load(store.clone()).await?; + + // A run left open by a crashed/killed process would make the next + // record_run_started fail — close it explicitly instead. + if let Some(stale) = state.resume_open_run().await? { + tracing::warn!(run = %stale, "closing stale open run as interrupted"); + state + .record_run_finished(actor.clone(), stale, "interrupted") + .await?; + } + + let goal = match goal { + GoalRef::Existing(id) => id, + GoalRef::New { title } => { + let id = GoalId::generate(); + state + .record_goal(Goal::new(id.clone(), title.clone(), title, actor.clone())) + .await?; + id + } + }; + + Ok(Self { + state, + store, + actor, + goal, + }) + } + + /// The goal this recorder chains runs to (persist it to reuse across + /// processes via [`GoalRef::Existing`]). + pub fn goal(&self) -> &GoalId { + &self.goal + } + + /// Returns a sender to pass to + /// [`Agent::prompt_with_sender`](crate::Agent::prompt_with_sender) (or + /// the raw loop) and a handle that resolves to the recorded [`RunId`] + /// once the run closes and is committed. + /// + /// Every event is optionally forwarded to `forward` (your UI) — the + /// recorder tees, it doesn't consume exclusively. + pub fn recording_sender( + &self, + task: impl Into, + forward: Option>, + ) -> ( + mpsc::UnboundedSender, + JoinHandle>, + ) { + let (tx, rx) = mpsc::unbounded_channel(); + let sink = YoAgentStateAdapter::new(self.state.clone(), self.actor.clone()); + let store = self.store.clone(); + let goal = self.goal.clone(); + let task = task.into(); + let handle = tokio::spawn(consume(sink, store, goal, task, rx, forward)); + (tx, handle) + } +} + +/// Map the event stream onto the GASP sink. Runs in its own task; ends when +/// the sender side is dropped (the loop finished). +async fn consume( + sink: YoAgentStateAdapter, + store: GitEventStore, + goal: GoalId, + task: String, + mut rx: mpsc::UnboundedReceiver, + forward: Option>, +) -> Result { + let run_id = RunId::generate(); + let mut started = false; + let mut finished = false; + let mut turn: usize = 0; + let mut outcome = "interrupted".to_string(); + + while let Some(event) = rx.recv().await { + match &event { + AgentEvent::AgentStart => { + sink.on_run_started(YoAgentRunStarted { + run_id: run_id.clone(), + task: task.clone(), + metadata: serde_json::json!({}), + }) + .await?; + started = true; + } + AgentEvent::MessageEnd { + message: + AgentMessage::Llm(Message::Assistant { + content, + model, + stop_reason, + .. + }), + } if started => { + turn += 1; + sink.on_model_called(YoAgentModelCalled { + run_id: run_id.clone(), + model: model.clone(), + prompt_summary: if turn == 1 { + summarize(&task) + } else { + format!("turn {turn}") + }, + }) + .await?; + let text = content + .iter() + .find_map(|c| match c { + Content::Text { text } if !text.is_empty() => Some(text.as_str()), + _ => None, + }) + .unwrap_or("(no text)"); + sink.on_model_finished(YoAgentModelFinished { + run_id: run_id.clone(), + model: model.clone(), + output_summary: summarize(text), + }) + .await?; + outcome = outcome_for(stop_reason).to_string(); + } + AgentEvent::ToolExecutionStart { + tool_name, args, .. + } if started => { + sink.on_tool_called(YoAgentToolCalled { + run_id: run_id.clone(), + tool: tool_name.clone(), + input_summary: summarize(&args.to_string()), + }) + .await?; + } + AgentEvent::ToolExecutionEnd { + tool_name, + result, + is_error, + .. + } if started => { + let text = result + .content + .iter() + .find_map(|c| match c { + Content::Text { text } => Some(text.as_str()), + _ => None, + }) + .unwrap_or("(no output)"); + sink.on_tool_finished(YoAgentToolFinished { + run_id: run_id.clone(), + tool: tool_name.clone(), + output_summary: summarize(text), + success: !is_error, + }) + .await?; + } + AgentEvent::AgentEnd { .. } if started && !finished => { + sink.on_run_finished(YoAgentRunFinished { + run_id: run_id.clone(), + outcome: outcome.clone(), + metadata: serde_json::json!({}), + }) + .await?; + finished = true; + } + _ => {} + } + if let Some(fwd) = &forward { + let _ = fwd.send(event); + } + } + + // Loop dropped the sender without AgentEnd (abort, filter reject, panic): + // close the run so the log never carries an open run across processes. + if started && !finished { + sink.on_run_finished(YoAgentRunFinished { + run_id: run_id.clone(), + outcome: "interrupted".into(), + metadata: serde_json::json!({}), + }) + .await?; + } + + if started { + store.commit_run(&run_id, &goal, &outcome, &[])?; + } + Ok(run_id) +} + +/// One-line, bounded summary for semantic events — full content belongs in +/// the transcripts tier, not the event log. +fn summarize(text: &str) -> String { + let one_line = text.split_whitespace().collect::>().join(" "); + if one_line.chars().count() <= 200 { + one_line + } else { + let truncated: String = one_line.chars().take(200).collect(); + format!("{truncated}…") + } +} + +fn outcome_for(stop_reason: &StopReason) -> &'static str { + match stop_reason { + StopReason::Stop | StopReason::ToolUse => "completed", + StopReason::Length => "truncated", + StopReason::Error => "error", + StopReason::Aborted => "aborted", + StopReason::Refusal => "refused", + } +} diff --git a/src/lib.rs b/src/lib.rs index 2a2cdaa..fd16aaf 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -48,6 +48,9 @@ //! policy engines (yoagent ships no policy — you install it). //! - **Sub-agents** ([`SubAgentTool`]) — delegation with per-sub-agent models //! and [`SharedState`] for passing artifacts by reference. +//! - **GASP** (feature `gasp`) — record runs into a +//! [GASP](https://github.com/yologdev/gasp) agent repo: append-only +//! semantic event log, restore = clone + replay, conformance-checked in CI. //! - **Session trees** ([`Session`]) — branching conversation history with //! fork, checkpoints, and JSONL persistence; edit an earlier turn and //! re-run without losing the original branch. @@ -77,6 +80,9 @@ pub mod types; #[cfg(feature = "openapi")] pub mod openapi; +#[cfg(feature = "gasp")] +pub mod gasp; + pub use agent::{Agent, AgentBuildError, StructuredPromptError}; pub use agent_loop::{agent_loop, agent_loop_continue}; pub use context::{CompactionStrategy, DefaultCompaction}; diff --git a/tests/gasp_test.rs b/tests/gasp_test.rs new file mode 100644 index 0000000..69618fa --- /dev/null +++ b/tests/gasp_test.rs @@ -0,0 +1,223 @@ +//! Tests for the GASP bridge (feature `gasp`): event mapping, commit +//! behavior, goal reuse, and interrupted-run recovery. +#![cfg(feature = "gasp")] + +use std::process::Command; +use yoagent::gasp::{GaspRecorder, GoalRef}; +use yoagent::provider::mock::*; +use yoagent::provider::{MockProvider, ModelConfig}; +use yoagent::*; + +struct NoopTool; + +#[async_trait::async_trait] +impl AgentTool for NoopTool { + fn name(&self) -> &str { + "noop" + } + fn label(&self) -> &str { + "Noop" + } + fn description(&self) -> &str { + "does nothing" + } + fn parameters_schema(&self) -> serde_json::Value { + serde_json::json!({"type": "object"}) + } + async fn execute( + &self, + _params: serde_json::Value, + _ctx: ToolContext, + ) -> Result { + Ok(ToolResult { + content: vec![Content::Text { text: "ok".into() }], + details: serde_json::Value::Null, + }) + } +} + +fn tool_then_text_provider() -> MockProvider { + MockProvider::new(vec![ + MockResponse::ToolCalls(vec![MockToolCall { + provider_metadata: None, + name: "noop".into(), + arguments: serde_json::json!({}), + }]), + MockResponse::Text("done".into()), + ]) +} + +fn event_kinds(repo: &std::path::Path) -> Vec { + std::fs::read_to_string(repo.join("state/events.jsonl")) + .expect("events.jsonl exists") + .lines() + .map(|l| { + serde_json::from_str::(l).unwrap()["kind"] + .as_str() + .unwrap() + .to_string() + }) + .collect() +} + +#[tokio::test] +async fn records_a_full_run_with_expected_kinds_and_commit() { + let dir = tempfile::tempdir().unwrap(); + let recorder = GaspRecorder::init( + dir.path(), + "test-agent", + "w1", + GoalRef::New { + title: "test goal".into(), + }, + ) + .await + .unwrap(); + + let mut agent = Agent::from_provider(tool_then_text_provider(), ModelConfig::mock()) + .with_tools(vec![Box::new(NoopTool)]); + let (tx, handle) = recorder.recording_sender("do the thing", None); + agent.prompt_with_sender("do the thing", tx).await; + let run_id = handle.await.unwrap().unwrap(); + + let kinds = event_kinds(dir.path()); + // Semantic skeleton, in order (ops_applied lines interleave freely). + let semantic: Vec<&str> = kinds + .iter() + .map(|s| s.as_str()) + .filter(|k| *k != "state.ops_applied") + .collect(); + assert_eq!( + semantic, + vec![ + "goal.created", + "run.started", + "model.called", + "model.finished", + "tool.called", + "tool.finished", + "model.called", + "model.finished", + "run.finished", + ], + "full log: {kinds:?}" + ); + + // The run is committed (append-only history is what conformance walks). + let log = Command::new("git") + .args(["log", "--oneline"]) + .current_dir(dir.path()) + .output() + .unwrap(); + let log = String::from_utf8_lossy(&log.stdout); + assert!( + log.contains(&format!("run {run_id}")), + "commit missing: {log}" + ); + assert!(log.contains("completed")); +} + +#[tokio::test] +async fn events_are_teed_to_the_forward_sender() { + let dir = tempfile::tempdir().unwrap(); + let recorder = GaspRecorder::init( + dir.path(), + "test-agent", + "w1", + GoalRef::New { + title: "tee".into(), + }, + ) + .await + .unwrap(); + + let (ui_tx, mut ui_rx) = tokio::sync::mpsc::unbounded_channel(); + let mut agent = Agent::from_provider(MockProvider::text("hi"), ModelConfig::mock()); + let (tx, handle) = recorder.recording_sender("t", Some(ui_tx)); + agent.prompt_with_sender("t", tx).await; + handle.await.unwrap().unwrap(); + + let mut forwarded = 0; + while ui_rx.try_recv().is_ok() { + forwarded += 1; + } + assert!(forwarded > 0, "UI sender must receive the teed events"); +} + +#[tokio::test] +async fn goal_is_reused_across_runs_and_recorder_reopens() { + let dir = tempfile::tempdir().unwrap(); + let recorder = GaspRecorder::init( + dir.path(), + "test-agent", + "w1", + GoalRef::New { + title: "persistent goal".into(), + }, + ) + .await + .unwrap(); + let goal = recorder.goal().clone(); + + // Run 1. + let mut agent = Agent::from_provider(MockProvider::text("one"), ModelConfig::mock()); + let (tx, handle) = recorder.recording_sender("run one", None); + agent.prompt_with_sender("one", tx).await; + handle.await.unwrap().unwrap(); + drop(recorder); + + // Reopen with the SAME goal — no new goal.created may appear. + let recorder = GaspRecorder::open( + dir.path().to_path_buf(), + "test-agent", + "w1", + GoalRef::Existing(goal.clone()), + ) + .await + .unwrap(); + assert_eq!(recorder.goal(), &goal); + + let mut agent = Agent::from_provider(MockProvider::text("two"), ModelConfig::mock()); + let (tx, handle) = recorder.recording_sender("run two", None); + agent.prompt_with_sender("two", tx).await; + handle.await.unwrap().unwrap(); + + let kinds = event_kinds(dir.path()); + assert_eq!( + kinds.iter().filter(|k| *k == "goal.created").count(), + 1, + "existing goal must be reused" + ); + assert_eq!(kinds.iter().filter(|k| *k == "run.finished").count(), 2); +} + +#[tokio::test] +async fn dropped_sender_without_agent_end_closes_run_as_interrupted() { + let dir = tempfile::tempdir().unwrap(); + let recorder = GaspRecorder::init( + dir.path(), + "test-agent", + "w1", + GoalRef::New { + title: "crash".into(), + }, + ) + .await + .unwrap(); + + // Simulate a crashed loop: send AgentStart, then drop the sender. + let (tx, handle) = recorder.recording_sender("doomed", None); + tx.send(AgentEvent::AgentStart).unwrap(); + drop(tx); + handle.await.unwrap().unwrap(); + + let kinds = event_kinds(dir.path()); + assert!(kinds.iter().any(|k| k == "run.finished")); + let last_finish = std::fs::read_to_string(dir.path().join("state/events.jsonl")) + .unwrap() + .lines() + .rfind(|l| l.contains("run.finished")) + .unwrap() + .to_string(); + assert!(last_finish.contains("interrupted"), "got: {last_finish}"); +} From 96c0446edc7e11cf8ac108d493ceaac254791e1f Mon Sep 17 00:00:00 2001 From: Yuanhao Li Date: Sat, 11 Jul 2026 01:18:03 +0200 Subject: [PATCH 2/3] fix: gasp tests need a git identity on CI runners; scope MSRV job MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Tests set GIT_AUTHOR/COMMITTER env vars (commit_run shells out to plain git commit; runners have no identity — macos images happened to). Verified with GIT_CONFIG_GLOBAL=/dev/null locally. - MSRV job excludes the gasp feature: yoagent-state uses let-chains (stable 1.88) despite declaring rust-version 1.85 — follow-up upstream. Documented in the gasp docs page. Co-Authored-By: Claude Fable 5 --- .github/workflows/ci.yml | 4 +++- docs/concepts/gasp.md | 4 ++++ tests/gasp_test.rs | 17 +++++++++++++++++ 3 files changed, 24 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e229b22..ba8ad2e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -58,7 +58,9 @@ jobs: - uses: dtolnay/rust-toolchain@1.86 - uses: Swatinem/rust-cache@v2 - name: Check - run: cargo check --all-features + # The optional `gasp` feature is excluded: yoagent-state currently + # requires a newer rustc (let-chains) than the crate's 1.86 MSRV. + run: cargo check --features openapi # yoagent claims to be a tested GASP-conformant runtime — this job is the # test: emit an agent repo with the gasp bridge (mock provider, no network) diff --git a/docs/concepts/gasp.md b/docs/concepts/gasp.md index f012437..cb30fef 100644 --- a/docs/concepts/gasp.md +++ b/docs/concepts/gasp.md @@ -16,6 +16,10 @@ agent-loop changes**: yoagent = { version = "0.11", features = ["gasp"] } ``` +> The `gasp` feature currently requires a newer Rust than the crate's 1.86 +> MSRV (its `yoagent-state` dependency uses let-chains, stable since 1.88). + + ```rust use yoagent::gasp::{GaspRecorder, GoalRef}; diff --git a/tests/gasp_test.rs b/tests/gasp_test.rs index 69618fa..165095d 100644 --- a/tests/gasp_test.rs +++ b/tests/gasp_test.rs @@ -36,6 +36,19 @@ impl AgentTool for NoopTool { } } +/// CI runners have no git identity configured; commit_run shells out to +/// plain `git commit`, so provide one via env (idempotent across tests). +fn ensure_git_identity() { + 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); + } +} + fn tool_then_text_provider() -> MockProvider { MockProvider::new(vec![ MockResponse::ToolCalls(vec![MockToolCall { @@ -62,6 +75,7 @@ fn event_kinds(repo: &std::path::Path) -> Vec { #[tokio::test] async fn records_a_full_run_with_expected_kinds_and_commit() { + ensure_git_identity(); let dir = tempfile::tempdir().unwrap(); let recorder = GaspRecorder::init( dir.path(), @@ -119,6 +133,7 @@ async fn records_a_full_run_with_expected_kinds_and_commit() { #[tokio::test] async fn events_are_teed_to_the_forward_sender() { + ensure_git_identity(); let dir = tempfile::tempdir().unwrap(); let recorder = GaspRecorder::init( dir.path(), @@ -146,6 +161,7 @@ async fn events_are_teed_to_the_forward_sender() { #[tokio::test] async fn goal_is_reused_across_runs_and_recorder_reopens() { + ensure_git_identity(); let dir = tempfile::tempdir().unwrap(); let recorder = GaspRecorder::init( dir.path(), @@ -193,6 +209,7 @@ async fn goal_is_reused_across_runs_and_recorder_reopens() { #[tokio::test] async fn dropped_sender_without_agent_end_closes_run_as_interrupted() { + ensure_git_identity(); let dir = tempfile::tempdir().unwrap(); let recorder = GaspRecorder::init( dir.path(), From 14cdf8f05655dcc24851af1ba0dd723ef18d476c Mon Sep 17 00:00:00 2001 From: Yuanhao Li Date: Sat, 11 Jul 2026 01:46:29 +0200 Subject: [PATCH 3/3] =?UTF-8?q?fix:=20address=20PR=20#68=20review=20findin?= =?UTF-8?q?gs=20=E2=80=94=20restore=20actually=20restores?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four-agent review; the code reviewer cloned the emitted repo and re-ran the conformance checker, proving check 6 passed only on the emitting worktree. Critical: - Restore fixed: init/open now commit the scaffolding (AGENT.md, identity/, .gitignore) plus pre-run events (goal.created, stale-run closure) via commit_scaffolding, so a fresh `git clone` is a complete conformant agent. Verified locally: checker passes against a CLONE. CI now clones the emitted repo and checks the clone (the actual GASP restore), and pins the gasp checker to a rev so upstream changes can't break yoagent CI silently. - Consumer resilience: events are forwarded to the UI tee BEFORE recording; a mid-run StateError logs (tracing::error!), stops recording, and keeps draining + forwarding — a recorder failure never blinds the UI. The error surfaces via the handle (documented as the only error channel). Contract fixes: - recording_sender resolves to Option: Ok(None) when AgentStart never arrived, instead of fabricating an id that exists nowhere in the log. - Dropped-sender fallback closes the run with the DERIVED outcome (was hardcoded "interrupted", disagreeing with the commit trailer). - InputRejected runs record outcome "rejected" (was indistinguishable from a crash in the durable log). - GoalRef::Existing validated at open — dangling goal ids fail loudly. - release_lease after each run (was held for the full 600s TTL, locking out the next worker/process). - with_summarizer redaction hook: summaries of tool inputs/outputs persist to a shareable repo — now redactable; docs warn explicitly. Comment/doc corrections (review-verified): "chained to your goal" → Goal: commit trailer; dropped-sender comment listed impossible triggers (filter reject/abort DO send AgentEnd); stale-run comment described the mechanism backwards (marker is in-memory; resume_open_run restores it); "exactly the shape of transcripts/" → "a natural format for" (spec leaves it open); example's verify command now shows the real clone+cargo-run invocation; sequential-runs + single-writer constraints documented. Tests 4 → 12 (+1 unit): fresh-clone restore (manifest/identity/log); sink-failure keeps tee alive through AgentEnd + surfaces Err (unix, read-only log); crash recovery via handle.abort() then reopen + record again; Ok(None) no-run; all four stop_reason→outcome mappings; InputRejected → "rejected"; dangling goal rejection; tee completeness (kind-sequence equality vs a direct run); summarize char-boundary unit test; skeleton test switched to allowlist filtering. Co-Authored-By: Claude Fable 5 --- .github/workflows/ci.yml | 8 +- CHANGELOG.md | 13 +- CLAUDE.md | 2 +- docs/concepts/gasp.md | 26 +- docs/concepts/session-trees.md | 5 +- examples/gasp_emit.rs | 5 +- src/gasp.rs | 430 +++++++++++++++++++++++---------- tests/gasp_test.rs | 352 ++++++++++++++++++++++++++- 8 files changed, 690 insertions(+), 151 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ba8ad2e..e6a3449 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -81,8 +81,12 @@ jobs: uses: actions/checkout@v4 with: repository: yologdev/gasp + ref: 08a60a2ee30d6bc065ffbadec34a7993dc109fdf # pin: checker changes must not break yoagent CI silently path: gasp-checker - - name: Run conformance checker + - name: Run conformance checker against a fresh clone (true restore) run: | + # GASP restore IS `git clone` — checking the emitting worktree would + # pass on untracked scaffolding that a clone doesn't have. + git clone -q /tmp/emitted-repo /tmp/restored-repo cd gasp-checker/conformance-check - cargo run -q -- /tmp/emitted-repo + cargo run -q -- /tmp/restored-repo diff --git a/CHANGELOG.md b/CHANGELOG.md index d6c5e63..c894027 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,10 +11,15 @@ adheres to [Semantic Versioning](https://semver.org/). - **GASP bridge** (feature `gasp`) — `gasp::GaspRecorder` records agent runs into a [GASP](https://github.com/yologdev/gasp) agent repo via `yoagent-state`: append-only `state/events.jsonl` (goal/run/model/tool - events), one git commit per run, stale/interrupted runs closed safely, - events teed to your UI. yoagent is now a **tested** GASP-conformant - runtime: CI emits a repo and runs the protocol's 7-check conformance suite - against it. New `gasp_emit` example and docs page. + events), one git commit per run (scaffolding committed at init so `git + clone` restores a complete agent), stale/interrupted runs closed safely, + events teed to your UI **before** recording (a recording failure never + blinds the UI; the error surfaces via the returned handle). Redaction hook + via `with_summarizer` — summaries of tool inputs/outputs are persisted to + a shareable repo. yoagent is now a **tested** GASP-conformant runtime: CI + emits a repo and runs the protocol's 7-check suite against a **fresh + clone** (the actual restore operation). New `gasp_emit` example and docs + page. ## 0.11.0 diff --git a/CLAUDE.md b/CLAUDE.md index a5fed3d..f299423 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -86,7 +86,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. Stale open runs are closed as "interrupted" on open; a dropped sender finishes the run as "interrupted". 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. CI job `gasp-conformance` emits a repo via `examples/gasp_emit.rs` and runs the gasp conformance checker (7 checks) against it. ### Session Trees (`session.rs`) diff --git a/docs/concepts/gasp.md b/docs/concepts/gasp.md index cb30fef..4c825f6 100644 --- a/docs/concepts/gasp.md +++ b/docs/concepts/gasp.md @@ -37,21 +37,29 @@ let run_id = record_handle.await??; // events appended + committed | Agent activity | GASP events | |---|---| -| loop starts | `run.started` (chained to your goal) | +| loop starts | `run.started` (the goal is stamped in the run commit's `Goal:` trailer) | | each assistant turn | `model.called` / `model.finished` paired nodes | | each tool execution | `tool.called` / `tool.finished` (with success flag) | | loop ends | `run.finished` with the outcome (`completed` / `error` / `aborted` / ...) | -The semantic log carries **summaries** — task, model, tool names, outcomes — -never full conversations. Full transcripts belong in GASP's cold -`transcripts/` tier, which is exactly the shape of -[`Session::to_jsonl`](session-trees.md): drop your session file there and the -two tiers compose. +The semantic log stores bounded one-line summaries — the **task string +(verbatim)**, model ids, and the **first 200 characters of tool inputs, tool +outputs, and assistant text** — never full transcripts. If secrets can flow +through tool arguments or outputs, install a redacting summarizer +(`recorder.with_summarizer(...)`) before recording: the repo is designed to +be cloned and shared, and committed history is hard to scrub. Full +transcripts belong in GASP's cold `transcripts/` tier — +[`Session::to_jsonl`](session-trees.md) is a natural format for it. Robustness: a crashed process leaves no open run — the recorder closes stale -runs as `interrupted` on open, and a dropped event stream finishes the run as -`interrupted` before committing. One git commit per run keeps the history -append-only. +runs as `interrupted` on open, and a dropped event stream finishes the run +with the outcome derived so far. One git commit per run keeps the history +append-only, and the init scaffolding (manifest, identity) is committed so a +fresh clone is a complete, conformant agent. If recording fails mid-run +(disk, lease, git), recording stops and the error surfaces via the returned +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. ## Tested conformance diff --git a/docs/concepts/session-trees.md b/docs/concepts/session-trees.md index 6524b6a..5c89e9d 100644 --- a/docs/concepts/session-trees.md +++ b/docs/concepts/session-trees.md @@ -71,7 +71,8 @@ remains for single-branch persistence. ## GASP -This tree is the shape of the [GASP](https://github.com/yologdev/gasp) -`transcripts/` tier — the raw-conversation cold tier. The semantic event log +This tree is a natural format for the +[GASP](https://github.com/yologdev/gasp) `transcripts/` tier — the +raw-conversation cold tier (the spec leaves its format open). The semantic event log lives in the [`gasp` feature](gasp.md): a recorder over the `AgentEvent` stream that emits a conformance-checked agent repo. diff --git a/examples/gasp_emit.rs b/examples/gasp_emit.rs index 8ea061e..a4f31e9 100644 --- a/examples/gasp_emit.rs +++ b/examples/gasp_emit.rs @@ -74,10 +74,11 @@ async fn main() -> Result<(), Box> { let (tx, record_handle) = recorder.recording_sender("touch hello.txt", None); agent.prompt_with_sender("touch hello.txt", tx).await; - let run_id = record_handle.await??; + let run_id = record_handle.await??.expect("run recorded"); println!("recorded run {run_id} into {repo}"); println!("inspect: cat {repo}/state/events.jsonl"); - println!("verify: conformance-check {repo}"); + println!("verify: git clone https://github.com/yologdev/gasp"); + println!(" cd gasp/conformance-check && cargo run -q -- {repo}"); Ok(()) } diff --git a/src/gasp.rs b/src/gasp.rs index 5099966..acb20f9 100644 --- a/src/gasp.rs +++ b/src/gasp.rs @@ -24,50 +24,71 @@ //! let mut agent = Agent::from_config(ModelConfig::anthropic("claude-sonnet-5", "Sonnet 5")); //! let (tx, record_handle) = recorder.recording_sender("implement the parser", None); //! agent.prompt_with_sender("implement the parser", tx).await; -//! let run_id = record_handle.await??; // events appended + committed +//! let run_id = record_handle.await??.expect("run recorded"); //! # let _ = run_id; Ok(()) //! # } //! ``` //! -//! The semantic log records **summaries** (task, model, tool names, outcome) -//! — full conversations belong in GASP's cold `transcripts/` tier, which is -//! exactly the shape of [`Session::to_jsonl`](crate::Session::to_jsonl). +//! # What is persisted +//! +//! The semantic log stores bounded one-line summaries — the **task string +//! (verbatim)**, model ids, and the **first 200 characters of tool inputs, +//! tool outputs, and assistant text** — never full transcripts. If secrets +//! can flow through tool arguments or outputs (API keys, connection +//! strings), install a redacting summarizer via +//! [`GaspRecorder::with_summarizer`] **before** recording: the log lives in a +//! git repo designed to be cloned and shared, and committed history is hard +//! to scrub. Full transcripts belong in GASP's cold `transcripts/` tier — +//! [`Session::to_jsonl`](crate::Session::to_jsonl) is a natural format for it. +//! +//! Recording requires a git identity (`user.name`/`user.email`), like any +//! git workflow. use crate::types::*; use tokio::sync::mpsc; use tokio::task::JoinHandle; use yoagent_state::{ - ActorRef, GitEventStore, Goal, YoAgentModelCalled, YoAgentModelFinished, YoAgentRunFinished, - YoAgentRunStarted, YoAgentState, YoAgentStateAdapter, YoAgentStateSink, YoAgentToolCalled, - YoAgentToolFinished, + ActorRef, GitEventStore, Goal, NodeId, YoAgentModelCalled, YoAgentModelFinished, + YoAgentRunFinished, YoAgentRunStarted, YoAgentState, YoAgentStateAdapter, YoAgentStateSink, + YoAgentToolCalled, YoAgentToolFinished, }; pub use yoagent_state::{GoalId, RunId, StateError}; -/// Which GASP goal recorded runs chain to. +/// Which GASP goal recorded runs belong to (stamped into each run-boundary +/// commit's `Goal:` trailer). #[derive(Debug, Clone)] pub enum GoalRef { - /// Use an existing goal (persist the id in your app's config). + /// Use an existing goal. Validated at open: the goal must exist in the + /// repo's graph, so a typo'd persisted id fails loudly instead of + /// chaining runs to a goal that exists nowhere. Existing(GoalId), /// Create a new goal with this title when the recorder opens. New { title: String }, } +type Summarizer = std::sync::Arc String + Send + Sync>; + /// Records an agent's [`AgentEvent`] stream into a GASP agent repo. /// /// One recorder = one writer (`worker_id` names it in the repo lease) and one -/// goal; each `recording_sender` call records one run. Events are appended to -/// `state/events.jsonl` as they arrive and committed when the run closes, so -/// the git history stays append-only (GASP conformance check 4). +/// goal; each [`recording_sender`](Self::recording_sender) call records one +/// run. **Runs are sequential**: a second in-flight `recording_sender` run +/// fails (`run X is already open`) — one run at a time per repo. Events are +/// appended to `state/events.jsonl` as they arrive and committed when the run +/// closes, so the git history stays append-only (GASP conformance check 4). +/// The repo must have a **single writer**: two live workers sharing a repo +/// contend on the lease and can interrupt each other's runs. pub struct GaspRecorder { state: YoAgentState, store: GitEventStore, actor: ActorRef, goal: GoalId, + summarize: Summarizer, } impl GaspRecorder { - /// Initialize a fresh agent repo at `root` (git init + minimal manifest) - /// and open a recorder on it. + /// Initialize a fresh agent repo at `root` (git init + minimal manifest, + /// committed so a clone restores it) and open a recorder on it. pub async fn init( root: impl AsRef, agent_id: &str, @@ -97,17 +118,33 @@ impl GaspRecorder { let actor = ActorRef::agent(agent_id); let state = YoAgentState::load(store.clone()).await?; - // A run left open by a crashed/killed process would make the next - // record_run_started fail — close it explicitly instead. + // The open-run marker is in-memory only; `resume_open_run` restores + // it from the log. A run left open by a crashed process is closed + // here for log hygiene — no unpaired `run.started` may leak across + // process boundaries. if let Some(stale) = state.resume_open_run().await? { tracing::warn!(run = %stale, "closing stale open run as interrupted"); state .record_run_finished(actor.clone(), stale, "interrupted") - .await?; + .await + .map_err(|e| { + StateError::Validation(format!( + "found an open run and could not close it ({e}); if another \ + worker is live on this repo, do not share it — GASP repos \ + are single-writer" + )) + })?; } let goal = match goal { - GoalRef::Existing(id) => id, + GoalRef::Existing(id) => { + if state.get_node(NodeId::new(id.as_str())).await.is_none() { + return Err(StateError::Validation(format!( + "goal {id} does not exist in this repo's graph" + ))); + } + id + } GoalRef::New { title } => { let id = GoalId::generate(); state @@ -117,15 +154,33 @@ impl GaspRecorder { } }; + // Commit the scaffolding (AGENT.md, identity/, .gitignore) plus any + // events recorded above. Without this, a fresh `git clone` — the + // restore operation GASP is built around — has no manifest and fails + // conformance check 6. + commit_scaffolding(&store)?; + Ok(Self { state, store, actor, goal, + summarize: std::sync::Arc::new(|text: &str| summarize(text)), }) } - /// The goal this recorder chains runs to (persist it to reuse across + /// Replace the default summarizer (single line, 200 chars) — e.g. to + /// redact secrets from tool arguments/outputs before they are persisted + /// to the shareable git repo. + pub fn with_summarizer( + mut self, + summarize: impl Fn(&str) -> String + Send + Sync + 'static, + ) -> Self { + self.summarize = std::sync::Arc::new(summarize); + self + } + + /// The goal this recorder's runs belong to (persist it to reuse across /// processes via [`GoalRef::Existing`]). pub fn goal(&self) -> &GoalId { &self.goal @@ -133,154 +188,253 @@ impl GaspRecorder { /// Returns a sender to pass to /// [`Agent::prompt_with_sender`](crate::Agent::prompt_with_sender) (or - /// the raw loop) and a handle that resolves to the recorded [`RunId`] - /// once the run closes and is committed. + /// the raw loop) and a handle resolving to the recorded [`RunId`] — + /// `Ok(None)` when the stream carried no run at all (`AgentStart` never + /// arrived), so callers never receive an id that isn't in the log. /// - /// Every event is optionally forwarded to `forward` (your UI) — the - /// recorder tees, it doesn't consume exclusively. + /// **The handle is the only error channel** — always await it. If + /// recording fails mid-run (disk, lease, git), recording stops and the + /// error is returned by the handle, but **event forwarding continues**: + /// every event is teed to `forward` (your UI) before recording, so a + /// recorder failure never blinds the UI. pub fn recording_sender( &self, task: impl Into, forward: Option>, ) -> ( mpsc::UnboundedSender, - JoinHandle>, + JoinHandle, StateError>>, ) { let (tx, rx) = mpsc::unbounded_channel(); let sink = YoAgentStateAdapter::new(self.state.clone(), self.actor.clone()); let store = self.store.clone(); let goal = self.goal.clone(); let task = task.into(); - let handle = tokio::spawn(consume(sink, store, goal, task, rx, forward)); + let summarize = self.summarize.clone(); + let handle = tokio::spawn(consume(sink, store, goal, task, summarize, rx, forward)); (tx, handle) } } +/// Per-run bookkeeping threaded through event recording. +struct RunTracking { + run_id: RunId, + task: String, + started: bool, + finished: bool, + turn: usize, + outcome: String, +} + /// Map the event stream onto the GASP sink. Runs in its own task; ends when -/// the sender side is dropped (the loop finished). +/// 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). async fn consume( sink: YoAgentStateAdapter, store: GitEventStore, goal: GoalId, task: String, + summarize: Summarizer, mut rx: mpsc::UnboundedReceiver, forward: Option>, -) -> Result { - let run_id = RunId::generate(); - let mut started = false; - let mut finished = false; - let mut turn: usize = 0; - let mut outcome = "interrupted".to_string(); +) -> Result, StateError> { + let mut tracking = RunTracking { + run_id: RunId::generate(), + task, + started: false, + finished: false, + turn: 0, + outcome: "interrupted".to_string(), + }; + let mut recording_error: Option = None; while let Some(event) = rx.recv().await { - match &event { - AgentEvent::AgentStart => { - sink.on_run_started(YoAgentRunStarted { - run_id: run_id.clone(), - task: task.clone(), - metadata: serde_json::json!({}), - }) - .await?; - started = true; - } - AgentEvent::MessageEnd { - message: - AgentMessage::Llm(Message::Assistant { - content, - model, - stop_reason, - .. - }), - } if started => { - turn += 1; - sink.on_model_called(YoAgentModelCalled { - run_id: run_id.clone(), - model: model.clone(), - prompt_summary: if turn == 1 { - summarize(&task) - } else { - format!("turn {turn}") - }, - }) - .await?; - let text = content - .iter() - .find_map(|c| match c { - Content::Text { text } if !text.is_empty() => Some(text.as_str()), - _ => None, - }) - .unwrap_or("(no text)"); - sink.on_model_finished(YoAgentModelFinished { - run_id: run_id.clone(), - model: model.clone(), - output_summary: summarize(text), - }) - .await?; - outcome = outcome_for(stop_reason).to_string(); - } - AgentEvent::ToolExecutionStart { - tool_name, args, .. - } if started => { - sink.on_tool_called(YoAgentToolCalled { - run_id: run_id.clone(), - tool: tool_name.clone(), - input_summary: summarize(&args.to_string()), - }) - .await?; - } - AgentEvent::ToolExecutionEnd { - tool_name, - result, - is_error, - .. - } if started => { - let text = result - .content - .iter() - .find_map(|c| match c { - Content::Text { text } => Some(text.as_str()), - _ => None, - }) - .unwrap_or("(no output)"); - sink.on_tool_finished(YoAgentToolFinished { - run_id: run_id.clone(), - tool: tool_name.clone(), - output_summary: summarize(text), - success: !is_error, - }) - .await?; - } - AgentEvent::AgentEnd { .. } if started && !finished => { - sink.on_run_finished(YoAgentRunFinished { - run_id: run_id.clone(), - outcome: outcome.clone(), - metadata: serde_json::json!({}), - }) - .await?; - finished = true; - } - _ => {} - } + // 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. if let Some(fwd) = &forward { - let _ = fwd.send(event); + let _ = fwd.send(event.clone()); + } + if recording_error.is_some() { + continue; // recording is dead; keep draining + forwarding + } + if let Err(e) = record_event(&sink, &summarize, &mut tracking, &event).await { + tracing::error!( + run = %tracking.run_id, + error = %e, + "GASP recording failed; recording stops but event forwarding continues" + ); + recording_error = Some(e); } } - // Loop dropped the sender without AgentEnd (abort, filter reject, panic): - // close the run so the log never carries an open run across processes. - if started && !finished { + if let Some(e) = recording_error { + let _ = store.release_lease(); + return Err(e); + } + if !tracking.started { + // No AgentStart ever arrived: nothing was recorded — say so instead + // of fabricating a RunId that exists nowhere in the log. + return Ok(None); + } + if !tracking.finished { + // Sender dropped without AgentEnd: close the run with the outcome + // derived so far (matches the commit trailer below). sink.on_run_finished(YoAgentRunFinished { - run_id: run_id.clone(), - outcome: "interrupted".into(), + run_id: tracking.run_id.clone(), + outcome: tracking.outcome.clone(), metadata: serde_json::json!({}), }) .await?; } + store.commit_run(&tracking.run_id, &goal, &tracking.outcome, &[])?; + // Free the lease so another worker (or the next process) can record + // immediately instead of waiting out the TTL. + let _ = store.release_lease(); + Ok(Some(tracking.run_id)) +} + +/// Record a single event. Mutates tracking state; any sink error aborts +/// recording (handled by the caller) without touching the forwarding path. +async fn record_event( + sink: &YoAgentStateAdapter, + summarize: &Summarizer, + tracking: &mut RunTracking, + event: &AgentEvent, +) -> Result<(), StateError> { + match event { + AgentEvent::AgentStart => { + sink.on_run_started(YoAgentRunStarted { + run_id: tracking.run_id.clone(), + task: tracking.task.clone(), + metadata: serde_json::json!({}), + }) + .await?; + tracking.started = true; + } + AgentEvent::MessageEnd { + message: + AgentMessage::Llm(Message::Assistant { + content, + model, + stop_reason, + .. + }), + } if tracking.started => { + tracking.turn += 1; + sink.on_model_called(YoAgentModelCalled { + run_id: tracking.run_id.clone(), + model: model.clone(), + prompt_summary: if tracking.turn == 1 { + summarize(&tracking.task) + } else { + format!("turn {}", tracking.turn) + }, + }) + .await?; + let text = content + .iter() + .find_map(|c| match c { + Content::Text { text } if !text.is_empty() => Some(text.as_str()), + _ => None, + }) + .unwrap_or("(no text)"); + sink.on_model_finished(YoAgentModelFinished { + run_id: tracking.run_id.clone(), + model: model.clone(), + output_summary: summarize(text), + }) + .await?; + tracking.outcome = outcome_for(stop_reason).to_string(); + } + AgentEvent::ToolExecutionStart { + tool_name, args, .. + } if tracking.started => { + sink.on_tool_called(YoAgentToolCalled { + run_id: tracking.run_id.clone(), + tool: tool_name.clone(), + input_summary: summarize(&args.to_string()), + }) + .await?; + } + AgentEvent::ToolExecutionEnd { + tool_name, + result, + is_error, + .. + } if tracking.started => { + let text = result + .content + .iter() + .find_map(|c| match c { + Content::Text { text } => Some(text.as_str()), + _ => None, + }) + .unwrap_or("(no output)"); + sink.on_tool_finished(YoAgentToolFinished { + run_id: tracking.run_id.clone(), + tool: tool_name.clone(), + output_summary: summarize(text), + success: !is_error, + }) + .await?; + } + AgentEvent::InputRejected { .. } if tracking.started => { + // A policy rejection is not a crash — label it distinctly in the + // durable log. + tracking.outcome = "rejected".to_string(); + } + AgentEvent::AgentEnd { .. } if tracking.started && !tracking.finished => { + sink.on_run_finished(YoAgentRunFinished { + run_id: tracking.run_id.clone(), + outcome: tracking.outcome.clone(), + metadata: serde_json::json!({}), + }) + .await?; + tracking.finished = true; + } + _ => {} + } + Ok(()) +} - if started { - store.commit_run(&run_id, &goal, &outcome, &[])?; +/// Commit the repo scaffolding (manifest, identity, gitignore) and any +/// pre-run events (goal creation, stale-run closure) so `git clone` restores +/// a complete agent. No-op when nothing changed. +fn commit_scaffolding(store: &GitEventStore) -> Result<(), StateError> { + let events = store.events_path(); + let root = events + .parent() + .and_then(|p| p.parent()) + .ok_or_else(|| StateError::Store("events path has no repo root".into()))? + .to_path_buf(); + let run = |args: &[&str]| -> Result { + std::process::Command::new("git") + .args(args) + .current_dir(&root) + .output() + .map_err(|e| StateError::Store(format!("git {}: {e}", args.join(" ")))) + }; + run(&[ + "add", + "--", + "AGENT.md", + "identity", + ".gitignore", + "state/events.jsonl", + ])?; + let staged = run(&["diff", "--cached", "--quiet"])?; + if !staged.status.success() { + let out = run(&["commit", "-q", "-m", "gasp: agent scaffolding"])?; + if !out.status.success() { + return Err(StateError::Store(format!( + "scaffolding commit failed: {}", + String::from_utf8_lossy(&out.stderr) + ))); + } } - Ok(run_id) + Ok(()) } /// One-line, bounded summary for semantic events — full content belongs in @@ -304,3 +458,21 @@ fn outcome_for(stop_reason: &StopReason) -> &'static str { StopReason::Refusal => "refused", } } + +#[cfg(test)] +mod tests { + use super::summarize; + + #[test] + fn summarize_collapses_and_truncates_on_char_boundaries() { + assert_eq!(summarize("a\nb\t c"), "a b c"); + // 300 multibyte chars: must truncate at 200 CHARS (not bytes) + '…'. + let long: String = "ö".repeat(300); + let s = summarize(&long); + assert_eq!(s.chars().count(), 201); + assert!(s.ends_with('…')); + // Exactly 200 chars: untouched. + let exact: String = "x".repeat(200); + assert_eq!(summarize(&exact), exact); + } +} diff --git a/tests/gasp_test.rs b/tests/gasp_test.rs index 165095d..073d86a 100644 --- a/tests/gasp_test.rs +++ b/tests/gasp_test.rs @@ -92,14 +92,20 @@ async fn records_a_full_run_with_expected_kinds_and_commit() { .with_tools(vec![Box::new(NoopTool)]); let (tx, handle) = recorder.recording_sender("do the thing", None); agent.prompt_with_sender("do the thing", tx).await; - let run_id = handle.await.unwrap().unwrap(); + let run_id = handle.await.unwrap().unwrap().expect("run recorded"); let kinds = event_kinds(dir.path()); // Semantic skeleton, in order (ops_applied lines interleave freely). + // Allowlist the semantic kinds (bookkeeping kinds like state.ops_applied + // may grow in yoagent-state minors without breaking this test). let semantic: Vec<&str> = kinds .iter() .map(|s| s.as_str()) - .filter(|k| *k != "state.ops_applied") + .filter(|k| { + ["goal.", "run.", "model.", "tool."] + .iter() + .any(|p| k.starts_with(p)) + }) .collect(); assert_eq!( semantic, @@ -238,3 +244,345 @@ async fn dropped_sender_without_agent_end_closes_run_as_interrupted() { .to_string(); assert!(last_finish.contains("interrupted"), "got: {last_finish}"); } + +// --------------------------------------------------------------------------- +// Review batch: restore-from-clone, failure paths, outcomes, validation +// --------------------------------------------------------------------------- + +/// The GASP restore operation IS `git clone` — a clone must contain the +/// manifest, identity, and the committed event log. +#[tokio::test] +async fn fresh_clone_restores_manifest_identity_and_log() { + ensure_git_identity(); + let dir = tempfile::tempdir().unwrap(); + let recorder = GaspRecorder::init( + dir.path(), + "test-agent", + "w1", + GoalRef::New { + title: "clone me".into(), + }, + ) + .await + .unwrap(); + + let mut agent = Agent::from_provider(MockProvider::text("hi"), ModelConfig::mock()); + let (tx, handle) = recorder.recording_sender("t", None); + agent.prompt_with_sender("t", tx).await; + handle.await.unwrap().unwrap().expect("run recorded"); + + let clone_dir = tempfile::tempdir().unwrap(); + let clone_path = clone_dir.path().join("restored"); + let out = Command::new("git") + .args(["clone", "-q", dir.path().to_str().unwrap()]) + .arg(&clone_path) + .output() + .unwrap(); + assert!(out.status.success()); + + assert!( + clone_path.join("AGENT.md").is_file(), + "manifest must restore" + ); + assert!( + clone_path.join("identity").is_dir(), + "identity must restore" + ); + let kinds = event_kinds(&clone_path); + assert!(kinds.iter().any(|k| k == "goal.created")); + assert!(kinds.iter().any(|k| k == "run.finished")); +} + +/// A mid-run recording failure must NOT blind the UI tee: forwarding +/// continues, the error surfaces via the handle, and nothing new commits. +#[tokio::test] +#[cfg(unix)] +async fn sink_failure_keeps_tee_alive_and_surfaces_error() { + use std::os::unix::fs::PermissionsExt; + ensure_git_identity(); + let dir = tempfile::tempdir().unwrap(); + let recorder = GaspRecorder::init( + dir.path(), + "test-agent", + "w1", + GoalRef::New { + title: "doomed".into(), + }, + ) + .await + .unwrap(); + + // Make the log unwritable so the first append fails. + let events = dir.path().join("state/events.jsonl"); + std::fs::set_permissions(&events, std::fs::Permissions::from_mode(0o444)).unwrap(); + + let (ui_tx, mut ui_rx) = tokio::sync::mpsc::unbounded_channel(); + let mut agent = Agent::from_provider(MockProvider::text("hi"), ModelConfig::mock()); + let (tx, handle) = recorder.recording_sender("t", Some(ui_tx)); + agent.prompt_with_sender("t", tx).await; + let result = handle.await.unwrap(); + + // Restore permissions so the tempdir cleans up everywhere. + std::fs::set_permissions(&events, std::fs::Permissions::from_mode(0o644)).unwrap(); + + assert!( + result.is_err(), + "recording failure must surface via the handle" + ); + // The tee survived: events kept flowing, ending with AgentEnd. + let mut last = None; + while let Ok(e) = ui_rx.try_recv() { + last = Some(e); + } + assert!( + matches!(last, Some(AgentEvent::AgentEnd { .. })), + "tee must deliver events through AgentEnd despite recording failure" + ); +} + +/// with_store's crash recovery: an aborted consumer leaves an open run; the +/// next open closes it as interrupted and can record again. +#[tokio::test] +async fn reopen_after_crash_closes_stale_run_and_records_again() { + ensure_git_identity(); + let dir = tempfile::tempdir().unwrap(); + let recorder = GaspRecorder::init( + dir.path(), + "test-agent", + "w1", + GoalRef::New { + title: "crashy".into(), + }, + ) + .await + .unwrap(); + let goal = recorder.goal().clone(); + + // Open a run, then kill the consumer before its drop-fallback can close + // it — a faithful crash simulation. + let (tx, handle) = recorder.recording_sender("doomed", None); + tx.send(AgentEvent::AgentStart).unwrap(); + while !std::fs::read_to_string(dir.path().join("state/events.jsonl")) + .unwrap_or_default() + .contains("run.started") + { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + handle.abort(); + let _ = handle.await; + drop(tx); + drop(recorder); + + // Reopen: the stale run must be closed and a new run must succeed. + let recorder = GaspRecorder::open( + dir.path().to_path_buf(), + "test-agent", + "w1", + GoalRef::Existing(goal), + ) + .await + .expect("reopen after crash"); + + let mut agent = Agent::from_provider(MockProvider::text("recovered"), ModelConfig::mock()); + let (tx, handle) = recorder.recording_sender("recovery run", None); + agent.prompt_with_sender("go", tx).await; + handle.await.unwrap().unwrap().expect("new run recorded"); + + let kinds = event_kinds(dir.path()); + assert_eq!(kinds.iter().filter(|k| *k == "run.started").count(), 2); + assert_eq!(kinds.iter().filter(|k| *k == "run.finished").count(), 2); +} + +/// No AgentStart → Ok(None): callers never get an id that isn't in the log. +#[tokio::test] +async fn no_agent_start_returns_none_and_writes_nothing() { + ensure_git_identity(); + let dir = tempfile::tempdir().unwrap(); + let recorder = GaspRecorder::init( + dir.path(), + "test-agent", + "w1", + GoalRef::New { + title: "nothing".into(), + }, + ) + .await + .unwrap(); + + let (tx, handle) = recorder.recording_sender("never runs", None); + drop(tx); + let outcome = handle.await.unwrap().unwrap(); + assert!(outcome.is_none(), "no run happened — no RunId"); + + let kinds = event_kinds(dir.path()); + assert!(!kinds.iter().any(|k| k.starts_with("run."))); +} + +/// outcome_for mapping, pinned through the durable log via synthetic events. +#[tokio::test] +async fn stop_reasons_map_to_distinct_outcomes() { + ensure_git_identity(); + for (stop, expected) in [ + (StopReason::Length, "truncated"), + (StopReason::Error, "error"), + (StopReason::Aborted, "aborted"), + (StopReason::Refusal, "refused"), + ] { + let dir = tempfile::tempdir().unwrap(); + let recorder = GaspRecorder::init( + dir.path(), + "test-agent", + "w1", + GoalRef::New { + title: "outcomes".into(), + }, + ) + .await + .unwrap(); + let (tx, handle) = recorder.recording_sender("t", None); + tx.send(AgentEvent::AgentStart).unwrap(); + tx.send(AgentEvent::MessageEnd { + message: AgentMessage::Llm(Message::assistant( + vec![Content::Text { text: "x".into() }], + stop.clone(), + "m", + "mock", + Usage::default(), + )), + }) + .unwrap(); + tx.send(AgentEvent::AgentEnd { messages: vec![] }).unwrap(); + drop(tx); + handle.await.unwrap().unwrap().expect("recorded"); + + let log = std::fs::read_to_string(dir.path().join("state/events.jsonl")).unwrap(); + let finish = log + .lines() + .rfind(|l| l.contains("run.finished")) + .unwrap() + .to_string(); + assert!( + finish.contains(expected), + "stop {stop:?} must record outcome {expected}; got {finish}" + ); + } +} + +/// An input-filter rejection is a policy outcome, not a crash. +#[tokio::test] +async fn input_rejected_runs_record_outcome_rejected() { + ensure_git_identity(); + struct RejectAll; + impl InputFilter for RejectAll { + fn filter(&self, _text: &str) -> FilterResult { + FilterResult::Reject("policy".into()) + } + } + + let dir = tempfile::tempdir().unwrap(); + let recorder = GaspRecorder::init( + dir.path(), + "test-agent", + "w1", + GoalRef::New { + title: "reject".into(), + }, + ) + .await + .unwrap(); + + let mut agent = Agent::from_provider(MockProvider::text("unused"), ModelConfig::mock()) + .with_input_filter(RejectAll); + let (tx, handle) = recorder.recording_sender("t", None); + agent.prompt_with_sender("anything", tx).await; + handle.await.unwrap().unwrap().expect("recorded"); + + let log = std::fs::read_to_string(dir.path().join("state/events.jsonl")).unwrap(); + let finish = log.lines().rfind(|l| l.contains("run.finished")).unwrap(); + assert!(finish.contains("rejected"), "got: {finish}"); +} + +/// A dangling GoalRef::Existing must fail loudly at open. +#[tokio::test] +async fn dangling_existing_goal_errors_at_open() { + ensure_git_identity(); + let dir = tempfile::tempdir().unwrap(); + // Create a valid repo first. + let r = GaspRecorder::init( + dir.path(), + "test-agent", + "w1", + GoalRef::New { + title: "real".into(), + }, + ) + .await + .unwrap(); + drop(r); + + let bogus = yoagent::gasp::GoalId::generate(); + let err = GaspRecorder::open( + dir.path().to_path_buf(), + "test-agent", + "w1", + GoalRef::Existing(bogus), + ) + .await; + assert!(err.is_err(), "dangling goal id must be rejected"); +} + +/// The tee must deliver the complete stream: identical event kinds to a +/// direct (untee'd) run of the same deterministic agent. +#[tokio::test] +async fn tee_delivers_the_complete_event_stream() { + ensure_git_identity(); + fn kind_of(e: &AgentEvent) -> &'static str { + match e { + AgentEvent::AgentStart => "AgentStart", + AgentEvent::AgentEnd { .. } => "AgentEnd", + AgentEvent::TurnStart => "TurnStart", + AgentEvent::TurnEnd { .. } => "TurnEnd", + AgentEvent::MessageStart { .. } => "MessageStart", + AgentEvent::MessageUpdate { .. } => "MessageUpdate", + AgentEvent::MessageEnd { .. } => "MessageEnd", + AgentEvent::ToolExecutionStart { .. } => "ToolExecutionStart", + AgentEvent::ToolExecutionUpdate { .. } => "ToolExecutionUpdate", + AgentEvent::ToolExecutionEnd { .. } => "ToolExecutionEnd", + AgentEvent::ProgressMessage { .. } => "ProgressMessage", + AgentEvent::InputRejected { .. } => "InputRejected", + } + } + + // Direct run. + let mut agent = Agent::from_provider(MockProvider::text("same"), ModelConfig::mock()); + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + agent.prompt_with_sender("t", tx).await; + let mut direct = Vec::new(); + while let Ok(e) = rx.try_recv() { + direct.push(kind_of(&e)); + } + + // Teed run of the identical agent config. + let dir = tempfile::tempdir().unwrap(); + let recorder = GaspRecorder::init( + dir.path(), + "test-agent", + "w1", + GoalRef::New { + title: "tee-eq".into(), + }, + ) + .await + .unwrap(); + let mut agent = Agent::from_provider(MockProvider::text("same"), ModelConfig::mock()); + let (ui_tx, mut ui_rx) = tokio::sync::mpsc::unbounded_channel(); + let (tx, handle) = recorder.recording_sender("t", Some(ui_tx)); + agent.prompt_with_sender("t", tx).await; + handle.await.unwrap().unwrap().expect("recorded"); + let mut teed = Vec::new(); + while let Ok(e) = ui_rx.try_recv() { + teed.push(kind_of(&e)); + } + + assert_eq!(direct, teed, "tee must deliver every event, in order"); +}