From 3e70488bdbd24de966884ee4935af45cfd22db9c Mon Sep 17 00:00:00 2001 From: jamiepine <32987599+jamiepine@users.noreply.github.com> Date: Tue, 11 Aug 2026 01:26:28 +0000 Subject: [PATCH 1/2] feat(skills): durable reflection-run record with activity timeline surface MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a persisted reflection_run lifecycle table and event for the skill-reflection pump. Every reflection pass (riding the memory- persistence branch) now records agent/channel identity, trigger provenance, referenced worker IDs, start/end timestamps, terminal status (success/no_op/error), declared rationale distinct from observed actions, outcome summary, and affected skill identifiers. ReflectionRunCompleted flows through the shared ProcessEvent bus, the existing ApiEvent SSE pipeline, and renders as an inline card in the portal timeline — same pattern as chronicle checkpoints, no parallel event system. Deferred curation (snapshots, stale/archive passes, consolidation) and full SkillInspector surface remain for follow-up PRs as documented in the updated skill-lifecycle.md shipped-status section. --- docs/design-docs/skill-lifecycle.md | 40 +++ interface/src/api/client.ts | 22 ++ .../src/components/portal/PortalTimeline.tsx | 31 ++- interface/src/hooks/useChannelLiveState.ts | 15 + migrations/20260811000001_reflection_runs.sql | 44 +++ src/agent/channel.rs | 261 +++++++++++++++++- src/agent/channel_history.rs | 4 + src/agent/cortex.rs | 1 + src/api/state.rs | 32 +++ src/api/system.rs | 1 + src/conversation.rs | 3 +- src/conversation/history.rs | 166 +++++++++++ src/lib.rs | 17 ++ 13 files changed, 632 insertions(+), 5 deletions(-) create mode 100644 migrations/20260811000001_reflection_runs.sql 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..3ecdc93f5 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.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..96dde7a64 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,19 @@ 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; + 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(), + }); + }, [pushItem]); + const handleChronicleCheckpoint = useCallback((data: unknown) => { const event = data as ChronicleCheckpointEvent; setLiveStates((prev) => { @@ -908,6 +922,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 { From c9b31d8fe8e89ee5526b68862e76193465f33175 Mon Sep 17 00:00:00 2001 From: jamiepine <32987599+jamiepine@users.noreply.github.com> Date: Tue, 11 Aug 2026 01:29:56 +0000 Subject: [PATCH 2/2] fix(interface): widen timeline types for reflection_run item The reflection_run timeline type is client-only (not in the OpenAPI TimelineItem union). Use double-cast through unknown for pushItem and Record for the type discriminator in PortalTimeline, matching the existing pattern used by checkpoint rendering. --- interface/src/components/portal/PortalTimeline.tsx | 4 ++-- interface/src/hooks/useChannelLiveState.ts | 4 +++- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/interface/src/components/portal/PortalTimeline.tsx b/interface/src/components/portal/PortalTimeline.tsx index 3ecdc93f5..619e7c422 100644 --- a/interface/src/components/portal/PortalTimeline.tsx +++ b/interface/src/components/portal/PortalTimeline.tsx @@ -386,8 +386,8 @@ export function PortalTimeline({ ); } - if (item.type === "reflection_run") { - return ; + 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 96dde7a64..c18df00ab 100644 --- a/interface/src/hooks/useChannelLiveState.ts +++ b/interface/src/hooks/useChannelLiveState.ts @@ -557,6 +557,8 @@ export function useChannelLiveState(channels: ChannelInfo[]) { // 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, @@ -565,7 +567,7 @@ export function useChannelLiveState(channels: ChannelInfo[]) { 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) => {