diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 53b8291..6860fa5 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -344,6 +344,10 @@ pub struct AgentLoop { context_window: usize, input_types: Vec, media_registry: MediaHandlerRegistry, + /// Optional sink receiving a clone of every message appended to + /// `emitted_messages` during `process_inner`. Sub-agent runs use it to + /// persist an incremental transcript; ordinary Turns leave it `None`. + transcript_sink: Option>, } #[derive(Debug, Clone)] pub struct AgentProcessResult { @@ -402,6 +406,7 @@ impl AgentLoop { model_name, input_types, media_registry: MediaHandlerRegistry::with_defaults(), + transcript_sink: None, }) } @@ -427,6 +432,7 @@ impl AgentLoop { model_name, input_types, media_registry: MediaHandlerRegistry::with_defaults(), + transcript_sink: None, }) } @@ -448,6 +454,7 @@ impl AgentLoop { model_name, input_types, media_registry: MediaHandlerRegistry::with_defaults(), + transcript_sink: None, } } @@ -470,6 +477,7 @@ impl AgentLoop { model_name, input_types, media_registry: MediaHandlerRegistry::with_defaults(), + transcript_sink: None, } } @@ -491,6 +499,26 @@ impl AgentLoop { self } + /// Attach a transcript sink that receives a clone of every message this + /// loop appends to `emitted_messages`. Used by sub-agent runs to persist + /// an incremental transcript; ordinary Turns leave it unset. + pub fn with_transcript_sink( + mut self, + sink: tokio::sync::mpsc::UnboundedSender, + ) -> Self { + self.transcript_sink = Some(sink); + self + } + + /// Forward a message to the transcript sink, if one is installed. The + /// sink is unbounded and the receiver outlives this loop, so send failures + /// are impossible in practice; ignore them defensively. + fn forward_to_transcript_sink(&self, message: &ChatMessage) { + if let Some(sink) = &self.transcript_sink { + let _ = sink.send(message.clone()); + } + } + /// Preemptive trim: truncate old tool results in-place when history is /// approaching the context window limit. Old results (outside of `keep_recent` /// zone) are replaced with a short placeholder; recent results are truncated @@ -676,6 +704,7 @@ impl AgentLoop { /// user messages: the client renders the durable Signal projection, not /// a user bubble, while the model still sees the envelope. fn append_steering_messages( + &self, messages: &mut Vec, emitted_messages: &mut Vec, consumed_steering: &mut Vec, @@ -689,7 +718,8 @@ impl AgentLoop { .into_chat_message(turn.turn_id.clone(), iteration); consumed_steering.push(input); messages.push(message.clone()); - emitted_messages.push(message); + emitted_messages.push(message.clone()); + self.forward_to_transcript_sink(&message); } } @@ -953,8 +983,9 @@ impl AgentLoop { }; Self::annotate_message(&mut assistant_message, turn.as_ref(), iteration, false); messages.push(assistant_message.clone()); - emitted_messages.push(assistant_message); - Self::append_steering_messages( + emitted_messages.push(assistant_message.clone()); + self.forward_to_transcript_sink(&assistant_message); + self.append_steering_messages( &mut messages, &mut emitted_messages, &mut consumed_steering, @@ -968,6 +999,7 @@ impl AgentLoop { attach_reply_media(&mut assistant_message, &reply_media_refs); Self::annotate_message(&mut assistant_message, turn.as_ref(), iteration, true); emitted_messages.push(assistant_message.clone()); + self.forward_to_transcript_sink(&assistant_message); crate::observability::metrics::global_metrics().record_turn( Some(&accumulated_usage), turn_start.elapsed().as_millis() as u64, @@ -1014,7 +1046,8 @@ impl AgentLoop { assistant_message.provider_state = response.provider_state; Self::annotate_message(&mut assistant_message, turn.as_ref(), iteration, false); messages.push(assistant_message.clone()); - emitted_messages.push(assistant_message); + emitted_messages.push(assistant_message.clone()); + self.forward_to_transcript_sink(&assistant_message); // Execute tools and add results to messages let tool_results = match self @@ -1069,7 +1102,8 @@ impl AgentLoop { ); Self::annotate_message(&mut tool_message, turn.as_ref(), iteration, false); messages.push(tool_message.clone()); - emitted_messages.push(tool_message); + emitted_messages.push(tool_message.clone()); + self.forward_to_transcript_sink(&tool_message); } LoopDetectionResult::Ok => { let mut tool_message = ChatMessage::tool_with_media( @@ -1080,7 +1114,8 @@ impl AgentLoop { ); Self::annotate_message(&mut tool_message, turn.as_ref(), iteration, false); messages.push(tool_message.clone()); - emitted_messages.push(tool_message); + emitted_messages.push(tool_message.clone()); + self.forward_to_transcript_sink(&tool_message); } } } @@ -1098,7 +1133,7 @@ impl AgentLoop { let Some(turn_context) = turn.as_ref() else { unreachable!("steering messages require a turn context"); }; - Self::append_steering_messages( + self.append_steering_messages( &mut messages, &mut emitted_messages, &mut consumed_steering, @@ -1172,6 +1207,7 @@ impl AgentLoop { true, ); emitted_messages.push(assistant_message.clone()); + self.forward_to_transcript_sink(&assistant_message); crate::observability::metrics::global_metrics().record_turn( Some(&accumulated_usage), turn_start.elapsed().as_millis() as u64, @@ -1211,6 +1247,7 @@ impl AgentLoop { attach_reply_media(&mut final_message, &reply_media_refs); Self::annotate_message(&mut final_message, turn.as_ref(), summary_iteration, true); emitted_messages.push(final_message.clone()); + self.forward_to_transcript_sink(&final_message); let turn_usage = (accumulated_usage.total_tokens > 0).then_some(&accumulated_usage); crate::observability::metrics::global_metrics() .record_turn(turn_usage, turn_start.elapsed().as_millis() as u64); diff --git a/src/agent/sub_agent.rs b/src/agent/sub_agent.rs index a271ca7..500b825 100644 --- a/src/agent/sub_agent.rs +++ b/src/agent/sub_agent.rs @@ -431,9 +431,12 @@ impl SubAgentManager { let mut effective_config = config.clone(); effective_config.max_iterations = Some(resolved.max_iterations); let max_result_chars = resolved.max_result_chars; + + let (transcript_tx, transcript_rx) = tokio::sync::mpsc::unbounded_channel(); let agent = self .build_sub_agent_with_provider(&effective_config, tools, &resolved.provider_config) - .map_err(|e| SubAgentError::ProviderCreation(e.to_string()))?; + .map_err(|e| SubAgentError::ProviderCreation(e.to_string()))? + .with_transcript_sink(transcript_tx); let history = vec![ ChatMessage::system(system_prompt), @@ -443,65 +446,69 @@ impl SubAgentManager { let start = Instant::now(); let tool_context = resolved.tool_context; - let result = tokio::select! { + let writer = self.spawn_transcript_writer(task_id, transcript_rx); + + let outcome = tokio::select! { result = tokio::time::timeout( std::time::Duration::from_secs(timeout_secs), agent.process_with_context(history, tool_context.clone()), - ) => result, - _ = tool_context.cancellation.cancelled() => { - return Ok(SubAgentResult { - task_id: task_id.to_string(), - content: String::new(), - content_truncated: false, - full_content: String::new(), - status: TaskStatus::Cancelled, - tool_calls_count: 0, - iterations: 0, - duration_ms: start.elapsed().as_millis() as u64, - }); - } + ) => match result { + Ok(inner) => ExecutionOutcome::Finished(Box::new(inner)), + Err(_elapsed) => ExecutionOutcome::TimedOut, + }, + _ = tool_context.cancellation.cancelled() => ExecutionOutcome::Cancelled, }; let duration_ms = start.elapsed().as_millis() as u64; - Ok(match result { - Ok(Ok(agent_result)) => { - let (content, truncated) = truncate_sub_agent_result_at( - &agent_result.final_response.content, - max_result_chars, - ); - let tool_calls_count = agent_result - .emitted_messages - .iter() - .filter(|m| m.tool_calls.is_some()) - .count(); - let iterations = agent_result - .emitted_messages - .iter() - .filter(|m| m.role == "assistant" && m.tool_calls.is_some()) - .count(); - SubAgentResult { - task_id: task_id.to_string(), - content, - content_truncated: truncated, - full_content: agent_result.final_response.content, - status: TaskStatus::Completed, - tool_calls_count, - iterations, - duration_ms, + // Drop the agent (which owns the transcript sender) so the writer can + // drain, then await the writer before the caller's terminal commit so + // the persisted transcript is complete first. + drop(agent); + if let Err(error) = writer.await { + tracing::warn!(run_id = task_id, error = %error, "transcript writer failed"); + } + + Ok(match outcome { + ExecutionOutcome::Finished(result) => match *result { + Ok(agent_result) => { + let (content, truncated) = truncate_sub_agent_result_at( + &agent_result.final_response.content, + max_result_chars, + ); + let tool_calls_count = agent_result + .emitted_messages + .iter() + .filter(|m| m.tool_calls.is_some()) + .count(); + let iterations = agent_result + .emitted_messages + .iter() + .filter(|m| m.role == "assistant" && m.tool_calls.is_some()) + .count(); + SubAgentResult { + task_id: task_id.to_string(), + content, + content_truncated: truncated, + full_content: agent_result.final_response.content, + status: TaskStatus::Completed, + tool_calls_count, + iterations, + duration_ms, + } } - } - Ok(Err(error)) => SubAgentResult { - task_id: task_id.to_string(), - content: String::new(), - content_truncated: false, - full_content: String::new(), - status: terminal_status_from_error(error), - tool_calls_count: 0, - iterations: 0, - duration_ms, + Err(error) => SubAgentResult { + task_id: task_id.to_string(), + content: String::new(), + content_truncated: false, + full_content: String::new(), + status: terminal_status_from_error(error), + tool_calls_count: 0, + iterations: 0, + duration_ms, + }, }, - Err(_elapsed) => SubAgentResult { + ExecutionOutcome::TimedOut => SubAgentResult { task_id: task_id.to_string(), content: String::new(), content_truncated: false, @@ -511,8 +518,64 @@ impl SubAgentManager { iterations: 0, duration_ms, }, + ExecutionOutcome::Cancelled => SubAgentResult { + task_id: task_id.to_string(), + content: String::new(), + content_truncated: false, + full_content: String::new(), + status: TaskStatus::Cancelled, + tool_calls_count: 0, + iterations: 0, + duration_ms, + }, }) } + + /// Spawn a task that drains the transcript channel into + /// `agent_run_messages`, assigning a monotonically increasing `seq` and + /// stripping `provider_state` (which must never be persisted or exposed). + /// With no storage the writer becomes a drain-and-discard no-op. + fn spawn_transcript_writer( + &self, + run_id: &str, + receiver: tokio::sync::mpsc::UnboundedReceiver, + ) -> tokio::task::JoinHandle<()> { + let storage = self.storage.clone(); + let run_id = run_id.to_string(); + tokio::spawn(async move { + let Some(storage) = storage else { + let mut receiver = receiver; + while receiver.recv().await.is_some() {} + return; + }; + let mut seq = 0i64; + let mut receiver = receiver; + while let Some(mut message) = receiver.recv().await { + message.provider_state = None; + let now = chrono::Utc::now().timestamp_millis(); + if let Err(error) = storage + .append_agent_run_message(&run_id, seq, &message, now) + .await + { + tracing::warn!( + run_id = %run_id, + seq, + error = %error, + "failed to append transcript message" + ); + } + seq += 1; + } + }) + } +} + +/// Intermediate outcome of a resolved run, unified so the transcript writer +/// is awaited on every exit path before `execute_resolved` returns. +enum ExecutionOutcome { + Finished(Box>), + TimedOut, + Cancelled, } fn terminal_status_from_error(error: AgentError) -> TaskStatus { diff --git a/src/gateway/http.rs b/src/gateway/http.rs index bf3a60f..d46e23c 100644 --- a/src/gateway/http.rs +++ b/src/gateway/http.rs @@ -1215,9 +1215,20 @@ pub async fn get_agent_run( let Some(run) = run else { return Err(ApiError::not_found(format!("run {id} not found"))); }; - Ok(Json( - json!({ "run": crate::protocol::AgentRunView::from_record(&run, 100_000) }), - )) + let session_id = run.root_session_id.clone(); + let transcript = state + .storage + .list_agent_run_messages(&id, 10_000) + .await + .map_err(ApiError::internal)? + .into_iter() + .map(crate::protocol::AgentTranscriptMessage::from) + .collect::>(); + Ok(Json(json!({ + "run": crate::protocol::AgentRunView::from_record(&run, 100_000), + "session_id": session_id, + "transcript": transcript, + }))) } pub async fn get_agent_run_events( diff --git a/src/protocol.rs b/src/protocol.rs index db77fdd..c8c76f1 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -98,6 +98,48 @@ pub struct AgentEventView { pub created_at: i64, } +/// Serialized transcript message for a single Agent run, exposed through the +/// HTTP detail endpoint. `reasoning_content` is client-visible here but +/// `provider_state` is never included. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AgentTranscriptMessage { + pub id: String, + pub run_id: String, + pub seq: i64, + pub role: String, + pub content: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reasoning_content: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tool_call_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tool_name: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tool_calls: Option>, + pub created_at: i64, +} + +impl From for AgentTranscriptMessage { + fn from(record: crate::storage::agent_run::AgentRunMessageRecord) -> Self { + let tool_calls = record + .tool_calls_json + .as_deref() + .and_then(|json| serde_json::from_str(json).ok()); + Self { + id: record.id, + run_id: record.run_id, + seq: record.seq, + role: record.role, + content: record.content, + reasoning_content: record.reasoning_content, + tool_call_id: record.tool_call_id, + tool_name: record.tool_name, + tool_calls, + created_at: record.created_at, + } + } +} + impl AgentRunView { pub fn from_record( record: &crate::storage::agent_run::AgentRunRecord, diff --git a/src/storage/agent_run.rs b/src/storage/agent_run.rs index 3bb8245..2624cf2 100644 --- a/src/storage/agent_run.rs +++ b/src/storage/agent_run.rs @@ -61,6 +61,22 @@ pub const AGENT_SCHEMA_STATEMENTS: &[&str] = &[ "CREATE INDEX IF NOT EXISTS idx_agent_runs_parent ON agent_runs(parent_run_id, created_at)", "CREATE INDEX IF NOT EXISTS idx_agent_runs_recovery ON agent_runs(runtime_generation, status, deadline_at)", r#" + CREATE TABLE IF NOT EXISTS agent_run_messages ( + id TEXT PRIMARY KEY, + run_id TEXT NOT NULL, + seq INTEGER NOT NULL, + role TEXT NOT NULL, + content TEXT NOT NULL, + reasoning_content TEXT, + tool_call_id TEXT, + tool_name TEXT, + tool_calls_json TEXT, + created_at INTEGER NOT NULL, + FOREIGN KEY (run_id) REFERENCES agent_runs(id) ON DELETE CASCADE + ) + "#, + "CREATE INDEX IF NOT EXISTS idx_agent_run_messages_run_seq ON agent_run_messages(run_id, seq)", + r#" CREATE TABLE IF NOT EXISTS agent_session_state ( root_session_id TEXT PRIMARY KEY, revision INTEGER NOT NULL DEFAULT 0, @@ -226,6 +242,23 @@ pub struct AgentRunRecord { pub updated_at: i64, } +/// Raw persisted transcript row for an Agent run. Incrementally appended by +/// the run's transcript writer; `tool_calls_json` is stored verbatim and only +/// parsed into `providers::ToolCall` at the protocol boundary. +#[derive(Debug, Clone)] +pub struct AgentRunMessageRecord { + pub id: String, + pub run_id: String, + pub seq: i64, + pub role: String, + pub content: String, + pub reasoning_content: Option, + pub tool_call_id: Option, + pub tool_name: Option, + pub tool_calls_json: Option, + pub created_at: i64, +} + /// One run to admit inside `accept_agent_runs`. #[derive(Debug, Clone)] pub struct NewAgentRun { @@ -377,6 +410,23 @@ fn run_record_from_row(row: &sqlx::sqlite::SqliteRow) -> Result Result { + Ok(AgentRunMessageRecord { + id: row.get("id"), + run_id: row.get("run_id"), + seq: row.get("seq"), + role: row.get("role"), + content: row.get("content"), + reasoning_content: row.get("reasoning_content"), + tool_call_id: row.get("tool_call_id"), + tool_name: row.get("tool_name"), + tool_calls_json: row.get("tool_calls_json"), + created_at: row.get("created_at"), + }) +} + impl super::Storage { /// Admit a batch of runs in one transaction, claiming any referenced /// plan items atomically. If any plan item was already taken the whole @@ -497,6 +547,62 @@ impl super::Storage { } } + /// Append one transcript message for an Agent run. The writer owns the + /// monotonically increasing `seq`; `provider_state` is expected to have + /// been stripped by the caller before this is called. + pub async fn append_agent_run_message( + &self, + run_id: &str, + seq: i64, + message: &crate::bus::ChatMessage, + now: i64, + ) -> Result<(), StorageError> { + let tool_calls_json = message + .tool_calls + .as_ref() + .map(serde_json::to_string) + .transpose() + .map_err(|error| StorageError::Migration(format!("serialize tool_calls: {error}")))?; + sqlx::query( + "INSERT INTO agent_run_messages (id, run_id, seq, role, content, \ + reasoning_content, tool_call_id, tool_name, tool_calls_json, created_at) \ + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + ) + .bind(uuid::Uuid::new_v4().to_string()) + .bind(run_id) + .bind(seq) + .bind(&message.role) + .bind(&message.content) + .bind(&message.reasoning_content) + .bind(&message.tool_call_id) + .bind(&message.tool_name) + .bind(tool_calls_json) + .bind(now) + .execute(&self.pool) + .await?; + Ok(()) + } + + /// List the persisted transcript for a run ordered by `seq`. The + /// transcript is naturally bounded by the run's iteration budget; the + /// default `limit` is a generous ceiling, not a pagination contract. + pub async fn list_agent_run_messages( + &self, + run_id: &str, + limit: i64, + ) -> Result, StorageError> { + let rows = sqlx::query( + "SELECT id, run_id, seq, role, content, reasoning_content, tool_call_id, \ + tool_name, tool_calls_json, created_at \ + FROM agent_run_messages WHERE run_id = ? ORDER BY seq ASC LIMIT ?", + ) + .bind(run_id) + .bind(limit) + .fetch_all(&self.pool) + .await?; + rows.iter().map(agent_run_message_record_from_row).collect() + } + /// List runs for a session ordered by `(created_at DESC, id DESC)`. /// The cursor is the pair of the last row the client has seen. pub async fn list_agent_runs( @@ -1061,17 +1167,18 @@ mod tests { } #[tokio::test] - async fn fresh_database_creates_schema_v8_agent_tables() { + async fn fresh_database_creates_schema_v9_agent_tables() { let (storage, _dir) = create_test_storage().await; let version: i64 = sqlx::query_scalar("PRAGMA user_version") .fetch_one(storage.pool()) .await .unwrap(); - assert_eq!(version, 8); + assert_eq!(version, 9); for table in [ "agent_runs", "agent_session_state", "agent_inbox_events", + "agent_run_messages", ] { let exists: i64 = sqlx::query_scalar( "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?", @@ -1371,4 +1478,47 @@ mod tests { .unwrap() ); } + + #[tokio::test] + async fn transcript_messages_round_trip_in_seq_order() { + let (storage, _dir) = create_test_storage().await; + storage + .accept_agent_runs(AcceptAgentRequest { + runs: vec![new_run("run-1", "exec-1", "cli:test:d1")], + now: 100, + }) + .await + .unwrap(); + + let mut assistant = crate::bus::ChatMessage::assistant_with_tool_calls( + "calling".to_string(), + vec![crate::providers::ToolCall { + id: "call-1".to_string(), + name: "bash".to_string(), + arguments: serde_json::json!({}), + }], + ); + assistant.reasoning_content = Some("thinking".to_string()); + let tool = crate::bus::ChatMessage::tool("call-1", "bash", "output"); + + storage + .append_agent_run_message("run-1", 0, &assistant, 200) + .await + .unwrap(); + storage + .append_agent_run_message("run-1", 1, &tool, 201) + .await + .unwrap(); + + let messages = storage.list_agent_run_messages("run-1", 10_000).await.unwrap(); + assert_eq!(messages.len(), 2); + assert_eq!(messages[0].seq, 0); + assert_eq!(messages[0].role, "assistant"); + assert_eq!(messages[0].reasoning_content.as_deref(), Some("thinking")); + assert!(messages[0].tool_calls_json.is_some()); + assert_eq!(messages[1].seq, 1); + assert_eq!(messages[1].role, "tool"); + assert_eq!(messages[1].tool_call_id.as_deref(), Some("call-1")); + assert_eq!(messages[1].tool_name.as_deref(), Some("bash")); + } } diff --git a/src/storage/mod.rs b/src/storage/mod.rs index e583d68..d050965 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -18,7 +18,7 @@ use sqlx::{Pool, Row, Sqlite}; use std::path::Path; use tokio::time::{Duration, sleep}; -const SCHEMA_VERSION: i64 = 8; +const SCHEMA_VERSION: i64 = 9; const INSERT_MESSAGE_SQL: &str = r#" INSERT INTO messages ( id, session_id, seq, role, content, reasoning_content, provider_state, @@ -395,25 +395,35 @@ impl Storage { } let mut tx = self.pool.begin().await?; - // Legacy table removed in schema v7; drop it so old databases do not - // keep dead rows around. - sqlx::query("DROP TABLE IF EXISTS background_tasks") - .execute(&mut *tx) - .await?; - // Schema v8 removes the batch "group" concept entirely: the - // `agent_run_groups` table is gone, and the run/inbox tables are - // rebuilt without their `group_id`/`scope_kind`/`scope_id` columns. - // Drop in dependency order (inbox -> runs -> groups) so foreign-key - // enforcement never blocks the implicit row delete. - sqlx::query("DROP TABLE IF EXISTS agent_inbox_events") - .execute(&mut *tx) - .await?; - sqlx::query("DROP TABLE IF EXISTS agent_runs") - .execute(&mut *tx) - .await?; - sqlx::query("DROP TABLE IF EXISTS agent_run_groups") - .execute(&mut *tx) - .await?; + // The legacy drops below are a pre-v8 rebuild concern: the batch + // "group" concept was removed in v8 and the old `background_tasks` + // table in v7. Gate them on `current < 8` so a v8 -> v9 upgrade only + // adds the new transcript table and preserves existing run history. + if current < 8 { + // Legacy table removed in schema v7; drop it so old databases do + // not keep dead rows around. + sqlx::query("DROP TABLE IF EXISTS background_tasks") + .execute(&mut *tx) + .await?; + // Schema v8 removes the batch "group" concept entirely: the + // `agent_run_groups` table is gone, and the run/inbox tables are + // rebuilt without their `group_id`/`scope_kind`/`scope_id` columns. + // Drop the transcript table before runs and the remaining tables in + // dependency order (messages -> inbox -> runs -> groups) so + // foreign-key enforcement never blocks the implicit row delete. + sqlx::query("DROP TABLE IF EXISTS agent_run_messages") + .execute(&mut *tx) + .await?; + sqlx::query("DROP TABLE IF EXISTS agent_inbox_events") + .execute(&mut *tx) + .await?; + sqlx::query("DROP TABLE IF EXISTS agent_runs") + .execute(&mut *tx) + .await?; + sqlx::query("DROP TABLE IF EXISTS agent_run_groups") + .execute(&mut *tx) + .await?; + } for (table, column, definition) in [ ("messages", "source", "source TEXT"), ("messages", "reasoning_content", "reasoning_content TEXT"), @@ -1885,6 +1895,65 @@ mod tests { ); } + #[tokio::test] + async fn v8_migration_preserves_agent_runs_and_adds_transcript_table() { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("v8.db"); + let pool = SqlitePoolOptions::new() + .connect_with( + SqliteConnectOptions::new() + .filename(&db_path) + .create_if_missing(true), + ) + .await + .unwrap(); + // The v8 `agent_runs` shape is unchanged in v9: v9 only adds the + // transcript table. Build a v8 database holding a durable run so the + // upgrade must preserve it rather than dropping the table. + sqlx::query(agent_run::AGENT_SCHEMA_STATEMENTS[0]) + .execute(&pool) + .await + .unwrap(); + sqlx::query( + "INSERT INTO agent_runs (id, root_session_id, caller_agent_id, caller_scope_id, \ + agent_id, definition_hash, provider_profile, provider_name, model_id, mode, \ + depth, execution_id, task, budget_json, status, runtime_generation, attempt, \ + completion_slot_reserved, deadline_at, revision, created_at, updated_at) \ + VALUES ('run-1', 'cli:c:d', 'ROOT', 'turn-1', 'researcher', 'hash', 'profile', \ + 'test', 'model', 'foreground', 1, 'exec-1', 'task', '{}', 'completed', 1, 1, \ + 0, 1000, 0, 1, 1)", + ) + .execute(&pool) + .await + .unwrap(); + sqlx::query("PRAGMA user_version = 8") + .execute(&pool) + .await + .unwrap(); + drop(pool); + + let storage = Storage::new(&db_path).await.unwrap(); + let run = storage.get_agent_run("run-1").await.unwrap(); + assert!( + run.is_some(), + "v8 agent run must survive the v9 upgrade without a rebuild" + ); + assert_eq!(run.unwrap().status.as_str(), "completed"); + + let version: i64 = sqlx::query_scalar("PRAGMA user_version") + .fetch_one(storage.pool()) + .await + .unwrap(); + assert_eq!(version, 9); + let exists: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'agent_run_messages'", + ) + .fetch_one(storage.pool()) + .await + .unwrap(); + assert_eq!(exists, 1, "agent_run_messages table must be created"); + } + #[tokio::test] async fn test_upsert_and_get_session() { let (storage, _dir) = create_test_storage().await; diff --git a/webui/src/App.svelte b/webui/src/App.svelte index 22d3e6b..ab5fc9a 100644 --- a/webui/src/App.svelte +++ b/webui/src/App.svelte @@ -21,10 +21,10 @@ { name: "chat", label: "聊天", description: "与 PicoBot 协作并管理会话" }, { name: "overview", label: "概览", description: "查看运行状态与系统容量" }, { name: "tools", label: "工具", description: "浏览工具、Skills 与 MCP 连接" }, - { name: "agents", label: "子代理", description: "管理具名子代理定义" }, + { name: "agents", label: "子代理", description: "查看活动中的子代理与历史运行" }, { name: "logs", label: "日志", description: "检查实时事件与运行记录" }, { name: "memory", label: "记忆", description: "查找和维护长期记忆" }, - { name: "tasks", label: "任务", description: "跟踪定时任务与后台工作" }, + { name: "tasks", label: "任务", description: "管理定时任务" }, { name: "settings", label: "配置", description: "管理 Gateway 与 Agent 配置" } ]; const icons = { @@ -153,7 +153,7 @@ {:else if current === "settings"} toast.show(text, error)} /> {:else if current === "overview"} {:else if current === "tools"} - {:else if current === "agents"} toast.show(text, error)} /> + {:else if current === "agents"} {:else}
即将上线
{/if} diff --git a/webui/src/lib/Icon.svelte b/webui/src/lib/Icon.svelte index 862fe7f..0e4fb57 100644 --- a/webui/src/lib/Icon.svelte +++ b/webui/src/lib/Icon.svelte @@ -43,6 +43,8 @@ {:else if name === "panel"} + {:else if name === "back"} + {/if} diff --git a/webui/src/lib/components/SubAgentDefinitions.svelte b/webui/src/lib/components/SubAgentDefinitions.svelte new file mode 100644 index 0000000..06479b5 --- /dev/null +++ b/webui/src/lib/components/SubAgentDefinitions.svelte @@ -0,0 +1,339 @@ + + +
+
+
+

具名子代理

+

+ 子代理由 ~/.picobot/agents/*.md 定义；工具、Skill、Provider 与模型在此直接指定。主 Agent 可委托给任意子代理。改动需热重载后生效。 +

+
+ +
+ + {#if loading} +
加载中…
+ {:else if error} +
{error}
+ {:else if agents.length === 0} +
暂无子代理定义
+ {:else} +
+ {#each agents as agent (agent.id)} +
+
+
+

{agent.id}

+

{agent.description}

+
+ provider: {agent.provider || agent.llm_profile || "—"} + model: {agent.model || "—"} + {#if agent.tools?.length}{agent.tools.length} 个工具{/if} + {#if agent.skills?.length}{agent.skills.length} 个 Skill{/if} + {delegateLabel(agent)} +
+ {#if agent.tools?.length} +
+ {#each agent.tools as tool (tool)}{tool}{/each} +
+ {/if} +
+
+ + + +
+
+
+ {/each} +
+ {/if} + + {#if editing} + + + {/if} +
+ + diff --git a/webui/src/pages/AgentsPage.svelte b/webui/src/pages/AgentsPage.svelte index 0d3b644..9196a91 100644 --- a/webui/src/pages/AgentsPage.svelte +++ b/webui/src/pages/AgentsPage.svelte @@ -1,41 +1,31 @@
-
-
-

具名子代理

-

- 子代理由 ~/.picobot/agents/*.md 定义；工具、Skill、Provider 与模型在此直接指定。改动需热重载后生效。 -

+ {#if selected} +
+ + {#if detail} +
+ {detail.run.agent_id} + +
+ {/if}
- -
- {#if loading} -
加载中…
- {:else if error} -
{error}
- {:else if agents.length === 0} -
暂无子代理定义
- {:else} -
- {#each agents as agent (agent.id)} + {#if !detail && !detailError} +
加载详情…
+ {:else if detailError} +
{detailError}
+ {:else if detail} +
-
-
-

{agent.id}

-

{agent.description}

-
- provider: {agent.provider || agent.llm_profile || "—"} - model: {agent.model || "—"} - {#if agent.tools?.length}{agent.tools.length} 个工具{/if} - {#if agent.skills?.length}{agent.skills.length} 个 Skill{/if} - {#if agent.delegates?.length}委托: {agent.delegates.join(", ")}{/if} -
- {#if agent.tools?.length} -
- {#each agent.tools as tool (tool)}{tool}{/each} -
- {/if} -
-
- - - -
+
+ agent: {detail.run.agent_id} + provider: {detail.run.provider_name} + model: {detail.run.model_id} + mode: {detail.run.mode} + depth: {detail.run.depth} + {detail.run.tool_calls_count} 次工具调用 · {detail.run.iterations} 轮 + {#if detail.session_id}session: {detail.session_id}{/if} + {#if detail.run.parent_run_id}parent: {detail.run.parent_run_id}{/if} + 开始 {formatTime(detail.run.started_at)} + 结束 {formatTime(detail.run.finished_at)} + 耗时 {durationBetween(detail.run.started_at, detail.run.finished_at)}
- {/each} + + {#if detail.events.length} +
+ {#each detail.events as event (event.id)} +
+ +
+
+ {event.event_type === "signal" ? `信号 · ${event.severity || "info"}` : `完成 · ${event.status}`} + {event.delivery === "steer" ? "steer" : "queue"} +
+ {#if event.event_type === "signal"} + {#if event.payload_json}

{signalSummary(event)}

{/if} + {:else} +

{event.status}{event.last_error ? ` · ${event.last_error}` : ""}

+ {/if} +
+
+ {/each} +
+ {/if} + +
+ {#if detail.run.task} +
+
任务
+
+
+
+
+ {/if} + + {#each detail.transcript.filter((m) => m.role !== "tool") as message (message.id)} +
+
{message.role === "assistant" ? "" : message.role}
+
+ {#if message.reasoning_content} +
+ 思考过程 +
+
+ {/if} + {#if message.content}
{/if} + {#if message.tool_calls?.length} +
+ {#each message.tool_calls as call (call.id)} + + {/each} +
+ {/if} +
+
+ {/each} + + {#if detail.transcript.length === 0 && !detail.run.task} +
无转录
+ {/if} +
+ + {#if detail.run.error} +
{detail.run.error}
+ {/if} +
+ {/if} + {:else} +
+
+

子代理活动

+

+ 查看活动中的子代理与历史运行；子代理定义在「配置 → 子代理定义」中管理。 +

+
+
- {/if} - {#if editing} - -
diff --git a/webui/src/pages/SettingsPage.svelte b/webui/src/pages/SettingsPage.svelte index 88134c3..7e6ccc6 100644 --- a/webui/src/pages/SettingsPage.svelte +++ b/webui/src/pages/SettingsPage.svelte @@ -3,6 +3,7 @@ import { Tabs } from "bits-ui"; import { api } from "../lib/api.js"; import AppearanceSettings from "../lib/components/AppearanceSettings.svelte"; + import SubAgentDefinitions from "../lib/components/SubAgentDefinitions.svelte"; let { notify } = $props(); let tab = $state("appearance"); @@ -16,6 +17,7 @@ const isConfig = $derived(tab === "config"); const isAppearance = $derived(tab === "appearance"); + const isSubagents = $derived(tab === "subagents"); const title = $derived(tab === "config" ? "运行配置" : tab === "user" ? "用户档案" : tab === "agents" ? "Agent 行为准则" : "页面外观"); const isDirty = $derived(content !== original); @@ -34,7 +36,7 @@ async function load() { loading = true; - if (isAppearance) { loading = false; return; } + if (isAppearance || isSubagents) { loading = false; return; } try { if (isConfig) { const result = await api("/api/config"); @@ -119,12 +121,16 @@
- 外观config.jsonUSER.mdAGENTS.md + 外观config.jsonUSER.mdAGENTS.md子代理定义 {#if isAppearance} + {:else if isSubagents} +
+ +
{:else if isConfig}
@@ -188,6 +194,7 @@