diff --git a/docs/design-docs/skill-lifecycle.md b/docs/design-docs/skill-lifecycle.md index fb029319b..7e2b05c19 100644 --- a/docs/design-docs/skill-lifecycle.md +++ b/docs/design-docs/skill-lifecycle.md @@ -362,3 +362,43 @@ are prompt-heavy and should expect iteration after real transcripts. curation snapshots + (optional) user git on the skills dir cover it, same posture as the prior doc. - **No cross-agent skill sharing** yet; workspace scoping stands. + +## Shipped status (2026-08-11) + +### Shipped + +**Phase 1 — foundations** (PR #621): serde_yaml frontmatter, `skill_usage` table, +`WriteOrigin`, API reload fix, watcher fixes, `skills_search` status-format fix, +`read_skill` cap, branch skill index, `skills_search`/`install_skill` moved to +channel toolset. + +**Phase 2 — mutation** (PR #622): `skill_manage` tool with full validation and +origin-scoped rails, archive semantics, deterministic `reload_skills` on every +mutation, `skills_list` tool. + +**Phase 3 — the pump** (PR #624, #633): Reflection riding the memory-persistence +branch — turn-work and worker-completion triggers, restricted tool server +(`read_skill`, `skill_manage`, `skills_list` + memory-save, no shell/no +messaging/no spawn), reflection prompt section with decide-first and +negative-capture bans, cooldown state, reflection signal with worker-ID +tracking, worker transcript feeding for reflection passes. + +**Reflection run record** (this PR): Durable `reflection_runs` table with +agent/channel identity, trigger source, referenced worker IDs, start/end +timestamps, terminal status (`success`/`no_op`/`error`/`cancelled`), declared +rationale (separate from observed actions), outcome summary, affected skill +identifiers, and token usage slot. Fire-and-forget persistence via +`ReflectionRunLogger` matching the existing `ProcessRunLogger` pattern. +`ReflectionRunCompleted` event on the shared `ProcessEvent` bus, piped through +the existing `ApiEvent` SSE pipeline. Minimal timeline surface in the portal UI +rendering reflection outcomes inline (same pattern as chronicle checkpoints). + +### Deferred to Phase 4 (curation) and Phase 5 (surfaces) + +- Deterministic stale/archive pass in cortex maintenance +- LLM consolidation pass (default-off) +- Snapshots + rollback +- `POST /agents/skills/write` + `CreateSkill.tsx` +- Full `SkillInspector` usage row (provenance, state, counts, pin toggle) +- `write_approval` staging mode +- CLI `pin`/`adopt`/`archive`/`restore` commands diff --git a/interface/src/api/client.ts b/interface/src/api/client.ts index 038498db5..1a5cfaaac 100644 --- a/interface/src/api/client.ts +++ b/interface/src/api/client.ts @@ -205,6 +205,17 @@ export interface BranchCompletedEvent { conclusion: string; } +export interface ReflectionRunCompletedEvent { + type: "reflection_run_completed"; + agent_id: string; + channel_id: string; + branch_id: string; + status: string; + outcome_summary: string; + trigger_source: string; + affected_skills: string; +} + export interface ToolStartedEvent { type: "tool_started"; agent_id: string; @@ -296,6 +307,7 @@ export type ApiEvent = | WorkerCompletedEvent | BranchStartedEvent | BranchCompletedEvent + | ReflectionRunCompletedEvent | ChronicleCheckpointEvent | ToolStartedEvent | ToolCompletedEvent @@ -359,6 +371,16 @@ export interface TimelineCheckpoint { created_at: string; } +export interface TimelineReflectionRun { + type: "reflection_run"; + id: string; + status: string; + outcome_summary: string; + trigger_source: string; + affected_skills: string; + started_at: string; +} + // Note: TimelineItem is re-exported from types.ts as a union type async function fetchJson(path: string): Promise { diff --git a/interface/src/components/portal/PortalTimeline.tsx b/interface/src/components/portal/PortalTimeline.tsx index 5ea4cc63a..619e7c422 100644 --- a/interface/src/components/portal/PortalTimeline.tsx +++ b/interface/src/components/portal/PortalTimeline.tsx @@ -2,7 +2,7 @@ import {useEffect, useRef, useState} from "react"; import {useQuery} from "@tanstack/react-query"; import {InlineBranchCard, MessageBubble} from "@spacedrive/ai"; import {File as FileIcon} from "@phosphor-icons/react"; -import {api, type AttachmentMeta, type TimelineBranchRun, type TimelineCheckpoint, type TimelineItem, type WorkerListItem} from "@/api/client"; +import {api, type AttachmentMeta, type TimelineBranchRun, type TimelineCheckpoint, type TimelineItem, type TimelineReflectionRun, type WorkerListItem} from "@/api/client"; import {Markdown} from "@/components/Markdown"; import {ToolCall, type ToolCallPair, tryParseJson, isErrorResult} from "@/components/ToolCall"; import {PortalWorkerCard} from "./PortalWorkerCard"; @@ -13,6 +13,32 @@ import clsx from "clsx"; * the span it covers and opens to the summary. Not a message — it was written * by neither side of the conversation. */ +function InlineReflectionRunCard({item}: {item: TimelineReflectionRun}) { + const statusLabel = + item.status === "success" ? "Learned" : + item.status === "no_op" ? "No change" : + item.status === "error" ? "Error" : + "Reflection"; + const statusColor = + item.status === "success" ? "text-green-11" : + item.status === "no_op" ? "text-ink-faint" : + item.status === "error" ? "text-red-11" : + "text-ink-dull"; + + return ( +
+
+ + + {statusLabel} + {item.outcome_summary ? `: ${item.outcome_summary}` : ""} + + +
+
+ ); +} + function InlineCheckpointCard({item}: {item: TimelineCheckpoint}) { const [expanded, setExpanded] = useState(false); const from = new Date(item.covers_from); @@ -360,6 +386,9 @@ export function PortalTimeline({ ); } + if ((item as Record).type === "reflection_run") { + return ; + } if (item.type === "checkpoint") { return ; } diff --git a/interface/src/hooks/useChannelLiveState.ts b/interface/src/hooks/useChannelLiveState.ts index d84477636..c18df00ab 100644 --- a/interface/src/hooks/useChannelLiveState.ts +++ b/interface/src/hooks/useChannelLiveState.ts @@ -5,6 +5,7 @@ import { type BranchCompletedEvent, type BranchStartedEvent, type ChronicleCheckpointEvent, + type ReflectionRunCompletedEvent, type InboundMessageEvent, type OutboundMessageDeltaEvent, type OutboundMessageEvent, @@ -554,6 +555,21 @@ export function useChannelLiveState(channels: ChannelInfo[]) { // A checkpoint carries its whole record, so it lands in the timeline without // a refetch. Duplicate ids are ignored — the same checkpoint can arrive over // SSE and again in a history page loaded around the same moment. + const handleReflectionRunCompleted = useCallback((data: unknown) => { + const event = data as ReflectionRunCompletedEvent; + // reflection_run is a client-only timeline item not in the OpenAPI + // TimelineItem union; use a cast matching the existing branch_run pattern. + pushItem(event.channel_id, { + type: "reflection_run", + id: event.branch_id, + status: event.status, + outcome_summary: event.outcome_summary, + trigger_source: event.trigger_source, + affected_skills: event.affected_skills, + started_at: new Date().toISOString(), + } as unknown as Parameters[1]); + }, [pushItem]); + const handleChronicleCheckpoint = useCallback((data: unknown) => { const event = data as ChronicleCheckpointEvent; setLiveStates((prev) => { @@ -908,6 +924,7 @@ export function useChannelLiveState(channels: ChannelInfo[]) { worker_completed: handleWorkerCompleted, branch_started: handleBranchStarted, chronicle_checkpoint: handleChronicleCheckpoint, + reflection_run_completed: handleReflectionRunCompleted, branch_completed: handleBranchCompleted, tool_started: handleToolStarted, tool_completed: handleToolCompleted, diff --git a/migrations/20260811000001_reflection_runs.sql b/migrations/20260811000001_reflection_runs.sql new file mode 100644 index 000000000..82835506d --- /dev/null +++ b/migrations/20260811000001_reflection_runs.sql @@ -0,0 +1,44 @@ +-- Durable reflection-run record for the skill-reflection loop. +-- +-- Every reflection pass (riding the memory persistence branch) writes one +-- row: agent/channel identity, trigger provenance, referenced workers, +-- start/end timestamps, terminal status, concise user-legible outcome +-- summary, affected skill identifiers/actions, error/no-op reason, and +-- token usage where available. +-- +-- The record distinguishes declared rationale (the branch's own summary) +-- from authoritative runtime outcomes (actions actually observed). +-- Chain-of-thought is never stored. +CREATE TABLE reflection_runs ( + id TEXT PRIMARY KEY, -- UUID (branch_id of the persistence branch) + agent_id TEXT NOT NULL, + channel_id TEXT NOT NULL, + -- Which trigger fired: 'turn_work' | 'worker_success' | 'reflection' + trigger_source TEXT NOT NULL, + -- Workers referenced by this reflection pass (JSON array of {id, success}) + referenced_workers TEXT, + started_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + completed_at TIMESTAMP, + -- Terminal status: 'success' | 'no_op' | 'error' | 'cancelled' + status TEXT NOT NULL DEFAULT 'running', + -- Declared by the reflection branch itself (its own summary of what + -- it did or decided). Distinct from `observed_actions`. + declared_rationale TEXT, + -- Authoritative: what skill mutations actually happened, observed + -- from tool-call results. JSON array of {action, skill_name, detail?}. + -- Empty array for no-op runs. Null until the pass completes. + observed_actions TEXT, + -- User-legible one-line summary (rendered in the UI timeline). + -- Derived from observed_actions + declared_rationale. + outcome_summary TEXT, + -- Human-readable reason when status is 'no_op' or 'error'. + terminal_reason TEXT, + -- Token usage for the reflection branch (JSON: {input, output, cache_read, reasoning}). + token_usage TEXT, + -- Created skills, patched skills (comma-separated lowercase canonical names). + -- Populated from observed_actions for easy querying. + affected_skills TEXT +); + +CREATE INDEX idx_reflection_runs_channel ON reflection_runs(channel_id, started_at); +CREATE INDEX idx_reflection_runs_agent ON reflection_runs(agent_id, started_at); diff --git a/src/agent/channel.rs b/src/agent/channel.rs index 1482b4f22..3703da918 100644 --- a/src/agent/channel.rs +++ b/src/agent/channel.rs @@ -21,7 +21,7 @@ use crate::conversation::settings::{ DelegationMode, MemoryMode, ResolvedConversationSettings, ResponseMode, }; use crate::conversation::{ - ActiveParticipant, ChannelStore, ConversationLogger, ProcessRunLogger, + ActiveParticipant, ChannelStore, ConversationLogger, ProcessRunLogger, ReflectionRunLogger, participant_display_name, participant_memory_key, renderable_participants, track_active_participant, }; @@ -318,6 +318,97 @@ fn branch_working_memory_event_summary( ) } +/// Derive a reflection run outcome from the persistence branch conclusion. +/// +/// Scans the conclusion for declarative signals that skill mutations +/// happened (or that the pass was a deliberate no-op / error). Returns +/// `(status, outcome_summary, affected_skills_csv)` where status is one of +/// `"success"`, `"no_op"`, or `"error"`. +pub(super) fn derive_reflection_outcome(conclusion: &str) -> (String, String, String) { + let trimmed = conclusion.trim(); + + // The branch conclusion starts with a JSON-like or structured line. + // Memory-persistence conclusions typically start with status keywords. + let lower = trimmed.to_lowercase(); + + // No-op detection: the branch explicitly decided nothing was worth saving. + if lower.contains("no skills") + || lower.contains("no changes") + || lower.contains("nothing to") + || lower.contains("no actionable") + || (lower.contains("no writes") && lower.contains("acceptable")) + { + let summary = crate::summarize_first_non_empty_line(trimmed, 160); + return ("no_op".to_string(), summary, String::new()); + } + + // Error detection. + if lower.starts_with("error") + || lower.starts_with("failed") + || lower.contains("reflection failed") + || lower.contains("skill reflection error") + { + let summary = crate::summarize_first_non_empty_line(trimmed, 160); + return ("error".to_string(), summary, String::new()); + } + + // Success: extract skill names from the conclusion. + // Look for patterns like "patched skill-name", "created skill-name", + // "updated skill-name", or skill names that match the canonical format. + let affected = extract_skill_names_from_text(trimmed); + let summary = if affected.is_empty() { + // Success but no explicit skill names found — still count it. + crate::summarize_first_non_empty_line(trimmed, 160) + } else { + let skill_list = affected.join(", "); + format!( + "Reflection: affected {} — {}", + skill_list, + crate::summarize_first_non_empty_line(trimmed, 120) + ) + }; + + (String::new(), summary, affected.join(", ")) +} + +/// Extract canonical skill names from text. Skill names match `[a-z0-9][a-z0-9._-]*`. +pub(super) fn extract_skill_names_from_text(text: &str) -> Vec { + let re = regex::Regex::new(r"\b([a-z][a-z0-9._-]{2,63})\b").ok(); + let Some(re) = re else { + return Vec::new(); + }; + + // Only consider names near action keywords. + let action_keywords = [ + "patched", "created", "updated", "edited", "skill", "modified", "wrote", "authored", + ]; + + let mut names = Vec::new(); + let lower = text.to_lowercase(); + + for keyword in &action_keywords { + // Find keyword occurrences and look for skill names nearby. + let mut start = 0; + while let Some(pos) = lower[start..].find(keyword) { + let abs_pos = start + pos + keyword.len(); + // Look at the next ~80 chars for a skill name. + let end = (abs_pos + 80).min(text.len()); + let following = &text[abs_pos..end]; + for cap in re.captures_iter(following) { + if let Some(m) = cap.get(0) { + let name = m.as_str().to_string(); + if !names.contains(&name) && !action_keywords.contains(&name.as_str()) { + names.push(name); + } + } + } + start = abs_pos; + } + } + + names +} + fn parse_branch_cancellation_reason(conclusion: &str) -> Option<&str> { let trimmed = conclusion.trim(); if let Some(rest) = trimmed.strip_prefix(BRANCH_CANCELLED_PREFIX) { @@ -475,6 +566,7 @@ pub struct ChannelState { pub deps: AgentDeps, pub conversation_logger: ConversationLogger, pub process_run_logger: ProcessRunLogger, + pub reflection_run_logger: ReflectionRunLogger, /// Discord message ID to reply to for work spawned in the current turn. pub reply_target_message_id: Arc>>, pub channel_store: ChannelStore, @@ -922,6 +1014,9 @@ pub struct Channel { last_reflection_at: Option, /// Branch IDs for silent memory persistence branches (results not injected into history). memory_persistence_branches: HashSet, + /// Branch IDs for active skill-reflection runs and their start metadata. + /// On completion the record is consumed and written to the reflection_runs table. + reflection_branches: HashMap, /// Optional Discord reply target captured when each branch was started. branch_reply_targets: HashMap, /// Buffer for coalescing rapid-fire messages. @@ -987,6 +1082,15 @@ impl ReflectionSignal { } } +/// Metadata captured when a skill-reflection run starts, held until +/// the persistence branch completes so the run record carries the +/// trigger provenance and referenced worker set. +#[derive(Debug, Clone)] +struct ReflectionRunStart { + trigger_source: String, + referenced_workers: Vec<(WorkerId, bool)>, +} + /// RAII guard that records `message_handling_duration_seconds` when dropped, /// ensuring the metric is observed on every exit path (including early returns /// and `?` error propagation). @@ -1065,6 +1169,7 @@ impl Channel { let conversation_logger = ConversationLogger::new(deps.sqlite_pool.clone()); let process_run_logger = ProcessRunLogger::new(deps.sqlite_pool.clone()); + let reflection_run_logger = ReflectionRunLogger::new(deps.sqlite_pool.clone()); let channel_store = ChannelStore::new(deps.sqlite_pool.clone()); let compactor_model = resolved_settings @@ -1103,6 +1208,7 @@ impl Channel { deps: deps.clone(), conversation_logger, process_run_logger, + reflection_run_logger, reply_target_message_id: Arc::new(RwLock::new(None)), channel_store: channel_store.clone(), screenshot_dir, @@ -1170,6 +1276,7 @@ impl Channel { message_count: 0, last_persistence_at: std::time::Instant::now(), memory_persistence_branches: HashSet::new(), + reflection_branches: HashMap::new(), reflection_signal: std::sync::Mutex::new(ReflectionSignal::default()), last_reflection_at: None, branch_reply_targets: HashMap::new(), @@ -3879,6 +3986,48 @@ impl Channel { // happened inside the branch via tool calls. if was_memory_persistence { tracing::info!(branch_id = %branch_id, "memory persistence branch completed"); + + // If this was a skill-reflection pass, complete the run record. + if let Some(reflection_start) = self.reflection_branches.remove(branch_id) { + let (status, outcome_summary, affected_skills) = + derive_reflection_outcome(conclusion); + + let status_final = if status.is_empty() { + "success" + } else { + &status + }; + + self.state.reflection_run_logger.log_reflection_completed( + *branch_id, + status_final, + &outcome_summary, + Some(conclusion), + &[], // observed_actions observable from tool-call events in future + if status_final == "no_op" || status_final == "error" { + Some(&outcome_summary) + } else { + None + }, + None, // token_usage available from branch run in future + &affected_skills, + ); + + // Emit the event for the SSE timeline — same bus as + // BranchResult, WorkerComplete, etc. + let _ = self + .deps + .event_tx + .send(ProcessEvent::ReflectionRunCompleted { + agent_id: self.deps.agent_id.clone(), + channel_id: self.id.clone(), + branch_id: *branch_id, + status: status_final.to_string(), + outcome_summary, + trigger_source: reflection_start.trigger_source, + affected_skills, + }); + } } else { // Regular branch: accumulate result for the next retrigger. // The result text will be embedded directly in the retrigger @@ -4462,6 +4611,22 @@ impl Channel { .reflection_signal .lock() .expect("reflection signal lock") = ReflectionSignal::default(); + + // Persist the reflection run start record. + self.state.reflection_run_logger.log_reflection_started( + branch_id, + &self.deps.agent_id, + &self.id, + trigger, + &reflection_workers, + ); + self.reflection_branches.insert( + branch_id, + ReflectionRunStart { + trigger_source: trigger.to_string(), + referenced_workers: reflection_workers.clone(), + }, + ); } self.memory_persistence_branches.insert(branch_id); tracing::info!( @@ -4704,8 +4869,9 @@ mod tests { use super::{ ObserveModeFallbackState, ReflectionSignal, branch_working_memory_event_summary, classify_conversational_event_summary, compute_listen_mode_invocation, decision_user_id, - extract_decision_summary_from_reply, format_conversational_event_summary, - is_dm_conversation_id, recv_channel_event, should_process_event_for_channel, + derive_reflection_outcome, extract_decision_summary_from_reply, + extract_skill_names_from_text, format_conversational_event_summary, is_dm_conversation_id, + recv_channel_event, should_process_event_for_channel, should_send_discord_quiet_mode_ping_ack, should_send_quiet_mode_fallback, }; use crate::memory::{MemoryType, WorkingMemoryEventType}; @@ -4794,6 +4960,95 @@ mod tests { ); } + // ── derive_reflection_outcome tests ── + + #[test] + fn reflection_outcome_detects_no_op() { + let (status, summary, affected) = derive_reflection_outcome( + "no skills were changed in this reflection pass. Nothing to do.", + ); + assert_eq!(status, "no_op"); + assert!(!summary.is_empty()); + assert!(affected.is_empty()); + + let (status, summary, affected) = + derive_reflection_outcome("No actionable skills found. No writes — acceptable."); + assert_eq!(status, "no_op"); + assert!(!summary.is_empty()); + assert!(affected.is_empty()); + } + + #[test] + fn reflection_outcome_detects_error() { + let (status, summary, affected) = + derive_reflection_outcome("Failed to parse skill frontmatter: invalid YAML"); + assert_eq!(status, "error"); + assert!(!summary.is_empty()); + + let (status, summary, affected) = + derive_reflection_outcome("error: skill reflection pass terminated unexpectedly"); + assert_eq!(status, "error"); + assert!(!summary.is_empty()); + } + + #[test] + fn reflection_outcome_detects_success_with_skills() { + let (status, summary, affected) = derive_reflection_outcome( + "Patched discord-rendering with new line-length rule. Also patched git-workflow.", + ); + // Default status is "success" - empty string signals success + assert!(status.is_empty()); + assert!(!summary.is_empty()); + assert!(affected.contains("discord-rendering")); + assert!(affected.contains("git-workflow")); + } + + #[test] + fn reflection_outcome_no_skills_still_succeeds() { + let (status, summary, affected) = + derive_reflection_outcome("Memory persistence completed. Saved 3 new memories."); + assert!(status.is_empty()); + assert!(!summary.is_empty()); + // May or may not find skill names; that's fine for success + } + + // ── extract_skill_names_from_text tests ── + + #[test] + fn extract_skill_names_finds_near_action_keywords() { + let names = extract_skill_names_from_text( + "Patched discord-rendering to add message-length guidance. Also updated code-review.", + ); + assert!(names.contains(&"discord-rendering".to_string())); + } + + #[test] + fn extract_skill_names_avoids_duplicates() { + let names = extract_skill_names_from_text( + "Patched discord-rendering. Then reviewed discord-rendering again.", + ); + let count = names + .iter() + .filter(|n| n.as_str() == "discord-rendering") + .count(); + assert_eq!(count, 1); + } + + #[test] + fn extract_skill_names_avoids_action_keywords() { + let names = extract_skill_names_from_text("Patched skill with patched config."); + assert!(!names.contains(&"patched".to_string())); + assert!(!names.contains(&"skill".to_string())); + } + + #[test] + fn reflection_signal_default_is_unset() { + let signal = ReflectionSignal::default(); + assert!(!signal.is_set()); + assert!(signal.workers.is_empty()); + assert!(!signal.turn_work); + } + #[tokio::test] async fn channel_event_loop_continues_after_lagged_broadcast() { let (event_tx, mut event_rx) = tokio::sync::broadcast::channel::(2); diff --git a/src/agent/channel_history.rs b/src/agent/channel_history.rs index eed7672ac..8dd2d63e2 100644 --- a/src/agent/channel_history.rs +++ b/src/agent/channel_history.rs @@ -504,6 +504,10 @@ pub(crate) fn event_is_for_channel(event: &ProcessEvent, channel_id: &ChannelId) channel_id: event_channel, .. } => event_channel.as_ref() == Some(channel_id), + ProcessEvent::ReflectionRunCompleted { + channel_id: event_channel, + .. + } => event_channel == channel_id, ProcessEvent::SettingsUpdated { channel_id: event_channel, .. diff --git a/src/agent/cortex.rs b/src/agent/cortex.rs index 1fed28083..1bf2c7db5 100644 --- a/src/agent/cortex.rs +++ b/src/agent/cortex.rs @@ -1605,6 +1605,7 @@ fn signal_from_event(event: ProcessEvent) -> Option { // durable and reachable through the timeline and the chronicle tool, // so they do not also need a slot in the signal buffer. ProcessEvent::ChronicleCheckpoint { .. } + | ProcessEvent::ReflectionRunCompleted { .. } | ProcessEvent::OpenCodeSessionCreated { .. } | ProcessEvent::OpenCodePartUpdated { .. } | ProcessEvent::WorkerInitialResult { .. } diff --git a/src/api/state.rs b/src/api/state.rs index 870361b76..f646a516c 100644 --- a/src/api/state.rs +++ b/src/api/state.rs @@ -434,6 +434,17 @@ pub enum ApiEvent { branch_id: String, conclusion: String, }, + /// A skill-reflection run completed. Silent in conversations; surfaced + /// in the activity timeline only. + ReflectionRunCompleted { + agent_id: String, + channel_id: String, + branch_id: String, + status: String, + outcome_summary: String, + trigger_source: String, + affected_skills: String, + }, /// A tool call started on a process. ToolStarted { agent_id: String, @@ -832,6 +843,27 @@ impl ApiState { }) .ok(); } + ProcessEvent::ReflectionRunCompleted { + branch_id, + channel_id, + status, + outcome_summary, + trigger_source, + affected_skills, + .. + } => { + api_tx + .send(ApiEvent::ReflectionRunCompleted { + agent_id: agent_id.clone(), + channel_id: channel_id.to_string(), + branch_id: branch_id.to_string(), + status: status.clone(), + outcome_summary: outcome_summary.clone(), + trigger_source: trigger_source.clone(), + affected_skills: affected_skills.clone(), + }) + .ok(); + } ProcessEvent::ToolStarted { process_id, channel_id, diff --git a/src/api/system.rs b/src/api/system.rs index d470471f4..4bf64de12 100644 --- a/src/api/system.rs +++ b/src/api/system.rs @@ -159,6 +159,7 @@ pub(super) async fn events_sse( ApiEvent::WorkerCompleted { .. } => "worker_completed", ApiEvent::BranchStarted { .. } => "branch_started", ApiEvent::BranchCompleted { .. } => "branch_completed", + ApiEvent::ReflectionRunCompleted { .. } => "reflection_run_completed", ApiEvent::ToolStarted { .. } => "tool_started", ApiEvent::ToolCompleted { .. } => "tool_completed", ApiEvent::ConfigReloaded => "config_reloaded", diff --git a/src/conversation.rs b/src/conversation.rs index 86f39751a..7e03886d8 100644 --- a/src/conversation.rs +++ b/src/conversation.rs @@ -17,7 +17,8 @@ pub use chronicle::{ CommitOutcome, NewCheckpoint, }; pub use history::{ - ConversationLogger, ProcessRunLogger, TimelineItem, WorkerDetailRow, WorkerRunRow, + ConversationLogger, ProcessRunLogger, ReflectionRunLogger, ReflectionRunRow, TimelineItem, + WorkerDetailRow, WorkerRunRow, }; pub use participants::{ ActiveParticipant, participant_display_name, participant_memory_key, renderable_participants, diff --git a/src/conversation/history.rs b/src/conversation/history.rs index 47d72780f..519a6aea6 100644 --- a/src/conversation/history.rs +++ b/src/conversation/history.rs @@ -1284,6 +1284,172 @@ impl ProcessRunLogger { } } +/// Persists skill-reflection run records for the activity timeline. +/// +/// Follows the same fire-and-forget pattern as [`ProcessRunLogger`]. +/// A reflection run starts when the persistence branch spawns with +/// `skill_reflection = true` and completes when that branch's result +/// arrives, recording the trigger provenance, outcome summary, affected +/// skills, and token usage. +#[derive(Debug, Clone)] +pub struct ReflectionRunLogger { + pool: SqlitePool, +} + +impl ReflectionRunLogger { + pub fn new(pool: SqlitePool) -> Self { + Self { pool } + } + + /// Record a reflection run starting. Fire-and-forget. + pub fn log_reflection_started( + &self, + branch_id: crate::BranchId, + agent_id: &crate::AgentId, + channel_id: &crate::ChannelId, + trigger_source: &str, + referenced_workers: &[(crate::WorkerId, bool)], + ) { + let pool = self.pool.clone(); + let id = branch_id.to_string(); + let agent_id = agent_id.to_string(); + let channel_id = channel_id.to_string(); + let trigger_source = trigger_source.to_string(); + let referenced_workers = serde_json::to_string( + &referenced_workers + .iter() + .map(|(wid, success)| { + serde_json::json!({ + "id": wid.to_string(), + "success": success, + }) + }) + .collect::>(), + ) + .unwrap_or_default(); + + tokio::spawn(async move { + if let Err(error) = sqlx::query( + "INSERT OR IGNORE INTO reflection_runs \ + (id, agent_id, channel_id, trigger_source, referenced_workers) \ + VALUES (?, ?, ?, ?, ?)", + ) + .bind(&id) + .bind(&agent_id) + .bind(&channel_id) + .bind(&trigger_source) + .bind(&referenced_workers) + .execute(&pool) + .await + { + tracing::warn!(%error, branch_id = %id, "failed to persist reflection run start"); + } + }); + } + + /// Mark a reflection run as completed. Called when the persistence + /// branch result arrives. Fire-and-forget. + /// + /// `observed_actions` is a JSON array of `{action, skill_name, detail?}` + /// derived from the branch's tool-call results — what actually + /// happened, not what the branch claimed it did. + pub fn log_reflection_completed( + &self, + branch_id: crate::BranchId, + status: &str, + outcome_summary: &str, + declared_rationale: Option<&str>, + observed_actions: &[serde_json::Value], + terminal_reason: Option<&str>, + token_usage: Option, + affected_skills: &str, + ) { + let pool = self.pool.clone(); + let id = branch_id.to_string(); + let status = status.to_string(); + let outcome_summary = outcome_summary.to_string(); + let declared_rationale = declared_rationale.map(|s| s.to_string()); + let observed_actions = serde_json::to_string(observed_actions).unwrap_or_default(); + let terminal_reason = terminal_reason.map(|s| s.to_string()); + let token_usage = token_usage.map(|v| serde_json::to_string(&v).unwrap_or_default()); + let affected_skills = affected_skills.to_string(); + + // Guard against duplicate completions: only update if still 'running'. + tokio::spawn(async move { + let result = sqlx::query( + "UPDATE reflection_runs SET \ + status = ?, completed_at = CURRENT_TIMESTAMP, \ + declared_rationale = ?, observed_actions = ?, \ + outcome_summary = ?, terminal_reason = ?, \ + token_usage = ?, affected_skills = ? \ + WHERE id = ? AND status = 'running'", + ) + .bind(&status) + .bind(&declared_rationale) + .bind(&observed_actions) + .bind(&outcome_summary) + .bind(&terminal_reason) + .bind(&token_usage) + .bind(&affected_skills) + .bind(&id) + .execute(&pool) + .await; + + match result { + Ok(result) if result.rows_affected() > 0 => { + tracing::info!(branch_id = %id, status = %status, "reflection run completed"); + } + Ok(_) => { + tracing::debug!(branch_id = %id, "reflection run already completed or not found"); + } + Err(error) => { + tracing::warn!(%error, branch_id = %id, "failed to persist reflection run completion"); + } + } + }); + } + + /// Load reflection runs for a channel, ordered by start time descending. + pub async fn load_for_channel( + &self, + channel_id: &crate::ChannelId, + limit: i64, + ) -> std::result::Result, sqlx::Error> { + let rows = sqlx::query_as::<_, ReflectionRunRow>( + "SELECT id, agent_id, channel_id, trigger_source, referenced_workers, \ + started_at, completed_at, status, declared_rationale, observed_actions, \ + outcome_summary, terminal_reason, token_usage, affected_skills \ + FROM reflection_runs WHERE channel_id = ? \ + ORDER BY started_at DESC LIMIT ?", + ) + .bind(channel_id.as_ref()) + .bind(limit) + .fetch_all(&self.pool) + .await?; + + Ok(rows) + } +} + +/// Query row for a reflection run. +#[derive(Debug, Clone, Serialize, sqlx::FromRow)] +pub struct ReflectionRunRow { + pub id: String, + pub agent_id: String, + pub channel_id: String, + pub trigger_source: String, + pub referenced_workers: Option, + pub started_at: String, + pub completed_at: Option, + pub status: String, + pub declared_rationale: Option, + pub observed_actions: Option, + pub outcome_summary: Option, + pub terminal_reason: Option, + pub token_usage: Option, + pub affected_skills: Option, +} + /// A worker run row without the transcript blob (for list queries). #[derive(Debug, Clone, Serialize)] pub struct WorkerRunRow { diff --git a/src/lib.rs b/src/lib.rs index fe80f6bea..c6dd988dc 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -366,6 +366,23 @@ pub enum ProcessEvent { line: String, stream: String, }, + /// A skill-reflection pass completed. Carries the full run record so + /// the timeline can render it without a refetch. Reflection runs are + /// silent in the conversation; they surface only in the activity timeline. + ReflectionRunCompleted { + agent_id: AgentId, + channel_id: ChannelId, + /// The branch_id that performed the reflection (UUID). + branch_id: BranchId, + /// Terminal status: "success", "no_op", "error", or "cancelled". + status: String, + /// User-legible one-line summary for the UI timeline. + outcome_summary: String, + /// Trigger source: "turn_work" | "worker_success" | "reflection". + trigger_source: String, + /// Comma-separated lowercase canonical skill names affected. + affected_skills: String, + }, /// Conversation settings were updated via API. The channel should /// re-load its settings from the database. SettingsUpdated {