diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 0c4ac348..6156af44 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -281,11 +281,44 @@ impl Method { /// Presentation category for an Internal Worker exposed through its parent's /// protocol stream. Internal Workers never become independently addressable /// protocol subjects. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[cfg_attr(feature = "typescript", derive(ts_rs::TS))] #[serde(rename_all = "snake_case")] pub enum InternalWorkerKind { SubWorker, + Service { kind: String }, +} + +/// Stable parent-owned lifecycle for one compaction run. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub struct CompactionLifecycle { + pub schema_version: u32, + pub compaction_id: String, + pub revision: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub internal_worker: Option, + pub state: CompactionLifecycleState, + /// Milliseconds since the Unix epoch. + pub started_at_ms: u64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ended_at_ms: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub summary: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub new_segment_id: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(rename_all = "snake_case")] +pub enum CompactionLifecycleState { + Running, + Done, + Failed, + Interrupted, } /// Stable presentation identity for one parent-owned Internal Worker session. @@ -607,23 +640,18 @@ pub enum Event { /// This is not part of LLM history or prompt context; clients may display it /// briefly as operational status. MemoryWorker(MemoryWorkerEvent), - /// Worker has started compacting the current session. - /// - /// Fired immediately before a compaction run. Success is signalled by - /// `CompactDone` (with the new `SegmentId`); failure by `CompactFailed`. - /// Broadcast to all clients; not replayed to late subscribers. - CompactStart, - /// Compaction completed and the session was rotated. - /// - /// `new_segment_id` is the UUID of the freshly created session that - /// replaced the old history. - CompactDone { - #[cfg_attr(feature = "typescript", ts(type = "string"))] - new_segment_id: uuid::Uuid, + /// Worker has started compacting the current session, or bound the run to its + /// observable Internal Worker. Revisions upsert one stable lifecycle item. + CompactStart { + lifecycle: CompactionLifecycle, }, - /// Compaction failed. The session is unchanged. + /// Compaction completed and the session was rotated. + CompactDone { + lifecycle: CompactionLifecycle, + }, + /// Compaction failed or was cancelled. The session is unchanged. CompactFailed { - error: String, + lifecycle: CompactionLifecycle, }, Shutdown, } @@ -1732,45 +1760,74 @@ mod tests { assert_eq!(parsed["data"]["timestamp_ms"], 1_700_000_000_000i64); } + fn test_compaction_lifecycle(state: CompactionLifecycleState) -> CompactionLifecycle { + CompactionLifecycle { + schema_version: 2, + compaction_id: "0192f0e8-4d84-7d6e-a000-000000000000".into(), + revision: 1, + internal_worker: None, + state, + started_at_ms: 1_700_000_000_000, + ended_at_ms: None, + summary: None, + error: None, + new_segment_id: None, + } + } + #[test] fn event_compact_start_roundtrip() { - let event = Event::CompactStart; + let event = Event::CompactStart { + lifecycle: test_compaction_lifecycle(CompactionLifecycleState::Running), + }; let json = serde_json::to_string(&event).unwrap(); - assert_eq!(json, r#"{"event":"compact_start"}"#); + let parsed: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(parsed["event"], "compact_start"); + assert_eq!(parsed["data"]["lifecycle"]["state"], "running"); let decoded: Event = serde_json::from_str(&json).unwrap(); - assert!(matches!(decoded, Event::CompactStart)); + assert!(matches!(decoded, Event::CompactStart { lifecycle } if lifecycle.revision == 1)); } #[test] fn event_compact_done_roundtrip() { let id = uuid::Uuid::parse_str("0192f0e8-4d84-7d6e-a000-000000000001").unwrap(); - let event = Event::CompactDone { new_segment_id: id }; + let mut lifecycle = test_compaction_lifecycle(CompactionLifecycleState::Done); + lifecycle.new_segment_id = Some(id.to_string()); + lifecycle.summary = Some("accepted summary".into()); + let event = Event::CompactDone { lifecycle }; let json = serde_json::to_string(&event).unwrap(); let parsed: serde_json::Value = serde_json::from_str(&json).unwrap(); assert_eq!(parsed["event"], "compact_done"); assert_eq!( - parsed["data"]["new_segment_id"], + parsed["data"]["lifecycle"]["new_segment_id"], "0192f0e8-4d84-7d6e-a000-000000000001" ); let decoded: Event = serde_json::from_str(&json).unwrap(); match decoded { - Event::CompactDone { new_segment_id } => assert_eq!(new_segment_id, id), + Event::CompactDone { lifecycle } => { + assert_eq!( + lifecycle.new_segment_id.as_deref(), + Some(id.to_string().as_str()) + ) + } other => panic!("expected CompactDone, got {other:?}"), } } #[test] fn event_compact_failed_roundtrip() { - let event = Event::CompactFailed { - error: "provider 429".into(), - }; + let mut lifecycle = test_compaction_lifecycle(CompactionLifecycleState::Failed); + lifecycle.error = Some("provider 429".into()); + let event = Event::CompactFailed { lifecycle }; let json = serde_json::to_string(&event).unwrap(); let parsed: serde_json::Value = serde_json::from_str(&json).unwrap(); assert_eq!(parsed["event"], "compact_failed"); - assert_eq!(parsed["data"]["error"], "provider 429"); + assert_eq!(parsed["data"]["lifecycle"]["error"], "provider 429"); let decoded: Event = serde_json::from_str(&json).unwrap(); match decoded { - Event::CompactFailed { error } => assert_eq!(error, "provider 429"), + Event::CompactFailed { lifecycle } => { + assert_eq!(lifecycle.error.as_deref(), Some("provider 429")) + } other => panic!("expected CompactFailed, got {other:?}"), } } diff --git a/crates/protocol/src/typescript.rs b/crates/protocol/src/typescript.rs index 53245d7b..29dffce8 100644 --- a/crates/protocol/src/typescript.rs +++ b/crates/protocol/src/typescript.rs @@ -4,11 +4,11 @@ use ts_rs::{Config, TS}; use crate::{ Alert, AlertLevel, AlertSource, CommandEvent, CommandSnapshot, CommandStatus, CommandStream, - CommandStreamSlice, CompletionEntry, CompletionKind, ErrorCode, Event, Greeting, InFlightBlock, - InFlightSnapshot, InFlightToolCallState, InternalWorkerKind, InternalWorkerRef, - InternalWorkerSnapshot, InvokeKind, MemoryWorkerEvent, Method, Permission, RewindSummary, - RewindTarget, RewindTargetId, RunResult, ScopeRule, Segment, TurnResult, WorkerEvent, - WorkerStatus, + CommandStreamSlice, CompactionLifecycle, CompactionLifecycleState, CompletionEntry, + CompletionKind, ErrorCode, Event, Greeting, InFlightBlock, InFlightSnapshot, + InFlightToolCallState, InternalWorkerKind, InternalWorkerRef, InternalWorkerSnapshot, + InvokeKind, MemoryWorkerEvent, Method, Permission, RewindSummary, RewindTarget, RewindTargetId, + RunResult, ScopeRule, Segment, TurnResult, WorkerEvent, WorkerStatus, subscription::{ EventSubscriptionSelector, SubscriptionEvent, SubscriptionEventPayload, SubscriptionFrame, SubscriptionFramePayload, SubscriptionId, SubscriptionRejectionCode, SubscriptionRequest, @@ -53,6 +53,8 @@ pub fn generated_protocol_types() -> String { push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); diff --git a/crates/tui/src/app.rs b/crates/tui/src/app.rs index 45104695..2a8ee1bb 100644 --- a/crates/tui/src/app.rs +++ b/crates/tui/src/app.rs @@ -1342,13 +1342,20 @@ impl App { } } } - Event::CompactStart => { - self.blocks.push(Block::Compact(CompactEvent::Streaming { - started_at: Instant::now(), - })); + Event::CompactStart { .. } => { + if self.last_streaming_compact_mut().is_none() { + self.blocks.push(Block::Compact(CompactEvent::Streaming { + started_at: Instant::now(), + })); + } } - Event::CompactDone { new_segment_id } => { + Event::CompactDone { lifecycle } => { self.session_context_tokens = 0; + let new_segment_id = lifecycle + .new_segment_id + .as_deref() + .and_then(|value| uuid::Uuid::parse_str(value).ok()) + .unwrap_or_default(); if let Some(evt) = self.last_streaming_compact_mut() { let elapsed_secs = match evt { CompactEvent::Streaming { started_at } => { @@ -1367,7 +1374,10 @@ impl App { })); } } - Event::CompactFailed { error } => { + Event::CompactFailed { lifecycle } => { + let error = lifecycle + .error + .unwrap_or_else(|| "compaction failed".to_string()); if let Some(evt) = self.last_streaming_compact_mut() { let elapsed_secs = match evt { CompactEvent::Streaming { started_at } => { @@ -2486,7 +2496,7 @@ fn event_is_stale_after_rewind(event: &Event) -> bool { event, Event::Alert(_) | Event::MemoryWorker(_) - | Event::CompactStart + | Event::CompactStart { .. } | Event::CompactDone { .. } | Event::CompactFailed { .. } | Event::SegmentRotated { .. } @@ -4076,13 +4086,34 @@ mod completion_flow_tests { } } + fn test_compaction_lifecycle( + state: protocol::CompactionLifecycleState, + ) -> protocol::CompactionLifecycle { + protocol::CompactionLifecycle { + schema_version: 2, + compaction_id: "compaction-test".into(), + revision: 1, + internal_worker: None, + state, + started_at_ms: 1, + ended_at_ms: None, + summary: None, + error: None, + new_segment_id: None, + } + } + #[test] fn compact_done_replaces_live_block() { let mut app = App::new("test".into()); let id = uuid::Uuid::parse_str("12345678-1234-5678-1234-567812345678").unwrap(); - app.handle_worker_event(Event::CompactStart); - app.handle_worker_event(Event::CompactDone { new_segment_id: id }); + app.handle_worker_event(Event::CompactStart { + lifecycle: test_compaction_lifecycle(protocol::CompactionLifecycleState::Running), + }); + let mut lifecycle = test_compaction_lifecycle(protocol::CompactionLifecycleState::Done); + lifecycle.new_segment_id = Some(id.to_string()); + app.handle_worker_event(Event::CompactDone { lifecycle }); assert_eq!(compact_block_count(&app), 1); assert!(matches!( @@ -4098,10 +4129,12 @@ mod completion_flow_tests { fn compact_failed_replaces_live_block() { let mut app = App::new("test".into()); - app.handle_worker_event(Event::CompactStart); - app.handle_worker_event(Event::CompactFailed { - error: "provider 429".into(), + app.handle_worker_event(Event::CompactStart { + lifecycle: test_compaction_lifecycle(protocol::CompactionLifecycleState::Running), }); + let mut lifecycle = test_compaction_lifecycle(protocol::CompactionLifecycleState::Failed); + lifecycle.error = Some("provider 429".into()); + app.handle_worker_event(Event::CompactFailed { lifecycle }); assert_eq!(compact_block_count(&app), 1); assert!(matches!( @@ -4117,7 +4150,9 @@ mod completion_flow_tests { fn shutdown_marks_live_compact_incomplete() { let mut app = App::new("test".into()); - app.handle_worker_event(Event::CompactStart); + app.handle_worker_event(Event::CompactStart { + lifecycle: test_compaction_lifecycle(protocol::CompactionLifecycleState::Running), + }); app.handle_worker_event(Event::Shutdown); assert!(app.quit); @@ -4208,9 +4243,9 @@ mod completion_flow_tests { let mut app = App::new("test".into()); app.session_context_tokens = 42_000; - app.handle_worker_event(Event::CompactDone { - new_segment_id: uuid::Uuid::nil(), - }); + let mut lifecycle = test_compaction_lifecycle(protocol::CompactionLifecycleState::Done); + lifecycle.new_segment_id = Some(uuid::Uuid::nil().to_string()); + app.handle_worker_event(Event::CompactDone { lifecycle }); assert_eq!(app.session_context_tokens, 0); } diff --git a/crates/worker/src/compact/worker.rs b/crates/worker/src/compact/worker.rs index 31eeb840..2f30407e 100644 --- a/crates/worker/src/compact/worker.rs +++ b/crates/worker/src/compact/worker.rs @@ -31,12 +31,11 @@ use workdir::LocalWorkdirSession; use workdir::{ReadRequest, WorkdirPath, WorkdirSessionHandle}; use crate::compact::usage_tracker::UsageTracker; -use crate::fs_view::ReadRequirement; -#[cfg(test)] -use crate::fs_view::slice_lines; -use crate::session_capture::{ - ReadDetail, ReadOptions, ReadSelector, SearchOptions, SessionCapture, ToolPart, +use crate::feature::{ + FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, ToolContribution, + ToolDeclaration, }; +use crate::fs_view::ReadRequirement; /// Aggregated output of a compact worker run. #[derive(Debug, Default, Clone)] @@ -91,248 +90,19 @@ struct SummaryParams { pub text: String, } -/// Input to `search_session_log`. -#[derive(Debug, Deserialize, schemars::JsonSchema)] -struct SearchSessionParams { - /// Case-insensitive substring to search in compact-target history. - pub query: String, - /// 0-based item offset to start searching from. - #[serde(default)] - pub offset: Option, - /// Maximum number of hits to return. - #[serde(default)] - pub limit: Option, -} - -/// Input to `read_session_items`. -#[derive(Debug, Deserialize, schemars::JsonSchema)] -struct ReadSessionParams { - /// 0-based compact-target history item offset. - pub offset: usize, - /// Maximum number of items to return. - pub limit: usize, - /// `compact` omits tool arguments/full results; `full` includes message text and tool result content. - #[serde(default = "default_session_read_mode")] - pub mode: String, -} - -fn default_session_read_mode() -> String { - "compact".to_string() -} - -const SESSION_TOOL_MAX_OUTPUT_TOKENS: u64 = 12_000; -const SESSION_SEARCH_MAX_RESULTS: usize = 50; -const SESSION_READ_MAX_ITEMS: usize = 80; - const MARK_DESCRIPTION: &str = "Inject a file's contents into the compacted context so the \ next session starts with it already read. Use this for files the next task needs in full. \ Optionally specify `offset` (0-based line) and `limit` (line count) to inject only a slice. \ -Counts against `auto_read_budget`; overflow returns an error and the mark is not recorded. \ -Paths must be absolute."; +Counts against `auto_read_budget`; overflow returns an error and the mark is not recorded."; -const REFERENCE_DESCRIPTION: &str = "Record a file path as a named reference in the compacted \ -context without injecting its contents. Use for files that are contextually relevant but \ -whose current content the next session can fetch on demand."; +const REFERENCE_DESCRIPTION: &str = "Record a Workdir-relative file path as a named reference in \ +the compacted context without injecting its contents. Use for files that are contextually \ +relevant but whose current content the next session can fetch on demand."; const SUMMARY_DESCRIPTION: &str = "Provide the final structured summary text. Subsequent calls \ replace the previous content; only the last call is used. Must be called before the compact run \ ends or compaction fails."; -const SEARCH_SESSION_DESCRIPTION: &str = "Search the compact-target session history by \ -case-insensitive substring. Returns item indexes and compact snippets. Use this when the initial \ -overview is not enough to identify which part of the session matters. Results are bounded; narrow \ -the query if important details are omitted."; - -const READ_SESSION_DESCRIPTION: &str = "Read a bounded range of compact-target session history \ -items by 0-based index. mode='compact' omits tool arguments, full tool results, and reasoning \ -bodies; mode='full' includes message text and tool result content but still remains bounded. Use \ -this to verify details before writing the summary."; - -struct SessionLogToolState { - items: Arc>, - view: SessionCapture, -} - -struct SearchSessionLogTool { - state: Arc, -} - -#[async_trait] -impl Tool for SearchSessionLogTool { - async fn execute( - &self, - input_json: &str, - _ctx: agen::tool::ToolExecutionContext, - ) -> Result { - let params: SearchSessionParams = serde_json::from_str(input_json).map_err(|e| { - ToolError::InvalidArgument(format!("invalid search_session_log input: {e}")) - })?; - let query = params.query.trim().to_lowercase(); - if query.is_empty() { - return Err(ToolError::InvalidArgument( - "search_session_log query must not be empty".to_string(), - )); - } - let offset = params.offset.unwrap_or(0); - let limit = params - .limit - .unwrap_or(20) - .clamp(1, SESSION_SEARCH_MAX_RESULTS); - let hits = self.state.view.search(&SearchOptions { - query: params.query.clone(), - kind: None, - tool_part: None, - tool_name: None, - limit: Some(limit), - min_entry_index: Some(offset as u64), - from: None, - through: None, - offset: 0, - }); - let blocks = hits - .iter() - .map(|hit| { - let part = hit - .tool_part - .map(|part| format!(" {part:?}")) - .unwrap_or_default(); - let tool = hit - .tool_name - .as_ref() - .map(|name| format!(" {name}")) - .unwrap_or_default(); - format!( - "[{} {}{}{} {:?}] {}\n{}", - hit.id, - hit.kind.as_str(), - part, - tool, - hit.entry_range, - hit.label, - hit.summary - ) - }) - .collect::>(); - let mut content = blocks.join("\n\n"); - let truncated = truncate_to_token_budget(&mut content, SESSION_TOOL_MAX_OUTPUT_TOKENS); - let summary = if hits.is_empty() { - format!("No session log hits for {query:?} from item offset {offset}.") - } else if truncated { - format!( - "Found {} session log hit(s) for {query:?}; output truncated. Narrow the query.", - hits.len() - ) - } else { - format!("Found {} session log hit(s) for {query:?}.", hits.len()) - }; - Ok(ToolOutput { - summary, - content: (!content.is_empty()).then_some(content), - - attachments: Vec::new(), - }) - } -} - -struct ReadSessionItemsTool { - state: Arc, -} - -#[async_trait] -impl Tool for ReadSessionItemsTool { - async fn execute( - &self, - input_json: &str, - _ctx: agen::tool::ToolExecutionContext, - ) -> Result { - let params: ReadSessionParams = serde_json::from_str(input_json).map_err(|e| { - ToolError::InvalidArgument(format!("invalid read_session_items input: {e}")) - })?; - let mode = SessionReadMode::parse(¶ms.mode)?; - let offset = params.offset.min(self.state.items.len()); - let limit = params.limit.clamp(1, SESSION_READ_MAX_ITEMS); - let end = offset.saturating_add(limit).min(self.state.items.len()); - let detail = match mode { - SessionReadMode::Compact => ReadDetail::Compact, - SessionReadMode::Full => ReadDetail::Full, - }; - let read = if offset >= end { - crate::session_capture::ReadResult { - entries: Vec::new(), - truncated: false, - } - } else { - self.state.view.read( - ReadSelector::EntryRange([offset as u64, end.saturating_sub(1) as u64]), - ReadOptions { - include_tools: true, - tool_part: ToolPart::Both, - detail, - max_items: limit, - max_bytes: 48 * 1024, - }, - ) - }; - let mut content = read - .entries - .iter() - .map(|entry| entry.text.clone()) - .collect::>() - .join("\n\n"); - let token_truncated = - truncate_to_token_budget(&mut content, SESSION_TOOL_MAX_OUTPUT_TOKENS); - let truncated = read.truncated || token_truncated; - let summary = if truncated { - format!( - "Read session items {offset}..{end} in {mode:?} mode; output truncated. Narrow the range." - ) - } else { - format!("Read session items {offset}..{end} in {mode:?} mode.") - }; - Ok(ToolOutput { - summary, - content: (!content.is_empty()).then_some(content), - - attachments: Vec::new(), - }) - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum SessionReadMode { - Compact, - Full, -} - -impl SessionReadMode { - fn parse(value: &str) -> Result { - match value { - "compact" => Ok(Self::Compact), - "full" => Ok(Self::Full), - other => Err(ToolError::InvalidArgument(format!( - "invalid read_session_items mode {other:?}; expected 'compact' or 'full'" - ))), - } - } -} - -fn truncate_to_token_budget(text: &mut String, max_tokens: u64) -> bool { - let max_bytes = max_tokens.saturating_mul(4) as usize; - if text.len() <= max_bytes { - return false; - } - let mut cut = 0; - for (idx, _) in text.char_indices() { - if idx > max_bytes { - break; - } - cut = idx; - } - text.truncate(cut); - text.push_str("\n… [session tool output truncated]"); - true -} - struct MarkReadRequiredTool { session: WorkdirSessionHandle, ctx: Arc>, @@ -508,36 +278,63 @@ pub(crate) fn write_summary_tool(ctx: Arc>) -> ToolD }) } -pub(crate) fn search_session_log_tool(items: Arc>) -> ToolDefinition { - let view = SessionCapture::new("compact-target", (*items).clone()); - let state = Arc::new(SessionLogToolState { items, view }); - Arc::new(move || { - let schema = schemars::schema_for!(SearchSessionParams); - let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({})); - let meta = ToolMeta::new("search_session_log") - .description(SEARCH_SESSION_DESCRIPTION) - .input_schema(schema_value); - let tool: Arc = Arc::new(SearchSessionLogTool { - state: state.clone(), - }); - (meta, tool) - }) +#[derive(Clone)] +pub(crate) struct CompactionOutputFeature { + session: Option, + tracker: tools::Tracker, + context: Arc>, } -pub(crate) fn read_session_items_tool(items: Arc>) -> ToolDefinition { - let view = SessionCapture::new("compact-target", (*items).clone()); - let state = Arc::new(SessionLogToolState { items, view }); - Arc::new(move || { - let schema = schemars::schema_for!(ReadSessionParams); - let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({})); - let meta = ToolMeta::new("read_session_items") - .description(READ_SESSION_DESCRIPTION) - .input_schema(schema_value); - let tool: Arc = Arc::new(ReadSessionItemsTool { - state: state.clone(), - }); - (meta, tool) - }) +impl CompactionOutputFeature { + pub(crate) fn new( + session: Option, + tracker: tools::Tracker, + context: Arc>, + ) -> Self { + Self { + session, + tracker, + context, + } + } +} + +impl FeatureModule for CompactionOutputFeature { + fn descriptor(&self) -> FeatureDescriptor { + let descriptor = FeatureDescriptor::builtin("compaction-output", "Compaction Output") + .with_description("Read-only Workdir access and bounded compaction output decisions.") + .with_tool(ToolDeclaration::new("add_reference", REFERENCE_DESCRIPTION)) + .with_tool(ToolDeclaration::new("write_summary", SUMMARY_DESCRIPTION)); + if self.session.is_some() { + descriptor + .with_tool(ToolDeclaration::new("Read", "Read a Workdir file.")) + .with_tool(ToolDeclaration::new("mark_read_required", MARK_DESCRIPTION)) + } else { + descriptor + } + } + + fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { + if let Some(session) = &self.session { + context.tools().register(ToolContribution::new( + "Read", + tools::read_tool(session.clone(), self.tracker.clone()), + ))?; + context.tools().register(ToolContribution::new( + "mark_read_required", + mark_read_required_tool(session.clone(), self.context.clone()), + ))?; + } + context.tools().register(ToolContribution::new( + "add_reference", + add_reference_tool(self.context.clone()), + ))?; + context.tools().register(ToolContribution::new( + "write_summary", + write_summary_tool(self.context.clone()), + ))?; + Ok(()) + } } /// Interceptor that monitors compact-worker context occupancy. @@ -810,53 +607,4 @@ mod tests { assert_eq!(guard.references.len(), 1); assert_eq!(guard.references[0], PathBuf::from(p)); } - - #[tokio::test] - async fn search_session_log_returns_bounded_hits_without_full_tool_content() { - let items = Arc::new(vec![ - Item::user_message("investigate compact failure"), - Item::tool_result_with_content( - "call-1", - "read trace with compact failure", - "very large raw trace body with secret detail", - ), - ]); - let view = SessionCapture::new("test", (*items).clone()); - let tool: Arc = Arc::new(SearchSessionLogTool { - state: Arc::new(SessionLogToolState { items, view }), - }); - let input = serde_json::json!({ "query": "compact", "limit": 10 }).to_string(); - let out = tool.execute(&input, Default::default()).await.unwrap(); - let content = out.content.unwrap(); - - assert!(content.contains("investigate compact failure")); - assert!(content.contains("read trace with compact failure")); - assert!(!content.contains("secret detail")); - } - - #[tokio::test] - async fn read_session_items_full_mode_can_read_tool_result_content() { - let items = Arc::new(vec![Item::tool_result_with_content( - "call-1", - "read trace", - "raw trace detail", - )]); - let view = SessionCapture::new("test", (*items).clone()); - let tool: Arc = Arc::new(ReadSessionItemsTool { - state: Arc::new(SessionLogToolState { items, view }), - }); - let input = serde_json::json!({ "offset": 0, "limit": 1, "mode": "full" }).to_string(); - let out = tool.execute(&input, Default::default()).await.unwrap(); - let content = out.content.unwrap(); - - assert!(content.contains("raw trace detail")); - } - - #[test] - fn slice_lines_handles_offset_and_limit() { - let text = "a\nb\nc\nd"; - assert_eq!(slice_lines(text, 0, None), "a\nb\nc\nd"); - assert_eq!(slice_lines(text, 1, Some(2)), "b\nc"); - assert_eq!(slice_lines(text, 10, None), ""); - } } diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index e801107e..78b38644 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -383,6 +383,7 @@ impl WorkerController { worker.attach_alerter(alerter.clone()); // Also hand the raw broadcast sender so Worker-internal operations // can emit typed lifecycle `Event`s (currently: compact progress). + worker.attach_internal_worker_registry(spawned_registry.clone()); worker.attach_event_tx(event_tx.clone()); // Bash spills long outputs to a per-worker subdir under the runtime @@ -421,7 +422,6 @@ impl WorkerController { wire_event_bridges_on_engine(&mut worker, &event_tx, &alerter, &in_flight); // === 3. Tool registration (builtin / memory / spawn-orchestration) === - spawned_registry.attach_parent_protocol(event_tx.clone(), worker.session_id().to_string()); let fs_for_view = register_worker_tools( &mut worker, bash_output_dir, diff --git a/crates/worker/src/feature/builtin/flow_transition.rs b/crates/worker/src/feature/builtin/flow_transition.rs index 02b88d14..c307b919 100644 --- a/crates/worker/src/feature/builtin/flow_transition.rs +++ b/crates/worker/src/feature/builtin/flow_transition.rs @@ -560,6 +560,7 @@ where attempt.instance_id, attempt.checked_state_revision )), max_turns: Some(12), + engine_configurator: None, features, required_tools: &[ "ShowOverview", @@ -571,6 +572,7 @@ where workspace: WorkerWorkspaceContext::no_workspace(), filesystem: WorkerFilesystemAuthority::None, scope: Scope::empty(), + workdir_session: None, }, }; match run_internal_worker(spec).await { diff --git a/crates/worker/src/internal_worker.rs b/crates/worker/src/internal_worker.rs index a4b2cba3..f47ef183 100644 --- a/crates/worker/src/internal_worker.rs +++ b/crates/worker/src/internal_worker.rs @@ -43,6 +43,8 @@ pub(crate) struct InternalWorkerAuthority { pub workspace: WorkerWorkspaceContext, pub filesystem: WorkerFilesystemAuthority, pub scope: Scope, + /// Provider-bound session inherited in an attenuated form from the owner. + pub workdir_session: Option, } pub(crate) struct InternalWorkerSpec { @@ -53,6 +55,7 @@ pub(crate) struct InternalWorkerSpec { pub input: String, pub cache_key: Option, pub max_turns: Option, + pub engine_configurator: Option>) + Send>>, pub features: FeatureRegistryBuilder, pub required_tools: &'static [&'static str], pub authority: InternalWorkerAuthority, @@ -102,6 +105,7 @@ where input, cache_key, max_turns, + engine_configurator, features, required_tools, authority, @@ -128,7 +132,11 @@ where }); engine.set_cache_key(cache_key); engine.set_max_turns(max_turns); + if let Some(configure) = engine_configurator { + configure(&mut engine); + } let store = EphemeralSessionStore::default(); + let inherited_workdir_session = authority.workdir_session.clone(); let mut worker = Worker::new( manifest, engine, @@ -144,6 +152,9 @@ where identity: identity.clone(), history_entries: 0, })?; + if let Some(session) = inherited_workdir_session { + worker.bind_workdir_session(Some(session)); + } let install_report = worker.install_features(features); let installed_tools = install_report.installed_tool_names(); @@ -250,7 +261,6 @@ impl InternalWorkerSessionStatus { #[derive(Debug, thiserror::Error)] pub(crate) enum InternalWorkerSessionError { - #[cfg(test)] #[error("failed to build internal Worker session: {message}")] Build { message: String }, #[error("internal Worker session is busy")] @@ -410,7 +420,6 @@ impl InternalWorkerSessionHandle { Ok(()) } - #[cfg(test)] pub(crate) async fn wait_until_idle(&self) -> InternalWorkerSessionStatus { loop { let notified = self.state_changed.notified(); @@ -475,6 +484,7 @@ pub(crate) async fn spawn_internal_worker_session( input, cache_key, max_turns, + engine_configurator, features, required_tools, authority, @@ -492,7 +502,11 @@ pub(crate) async fn spawn_internal_worker_session( }); engine.set_cache_key(cache_key); engine.set_max_turns(max_turns); + if let Some(configure) = engine_configurator { + configure(&mut engine); + } let store = EphemeralSessionStore::default(); + let inherited_workdir_session = authority.workdir_session.clone(); let mut worker = Worker::new( manifest, engine, @@ -505,6 +519,9 @@ pub(crate) async fn spawn_internal_worker_session( .map_err(|source| InternalWorkerSessionError::Build { message: source.to_string(), })?; + if let Some(session) = inherited_workdir_session { + worker.bind_workdir_session(Some(session)); + } let install_report = worker.install_features(features); let installed_tools = install_report.installed_tool_names(); let install_failed = install_report @@ -539,6 +556,102 @@ pub(crate) async fn spawn_internal_worker_session( spawn_prepared_internal_worker_session(worker, store, input, None).await } +/// Prepare an observable Internal Worker from the same bounded spec as one-shot +/// helpers, but do not start its first turn. The owner must register the returned +/// handle before calling `send`, preserving the snapshot/live boundary. +pub(crate) fn prepare_internal_worker_from_spec( + spec: InternalWorkerSpec, + visibility: InternalWorkerVisibility, +) -> std::pin::Pin< + Box< + dyn std::future::Future< + Output = Result, + > + Send, + >, +> { + Box::pin(async move { + let InternalWorkerSpec { + identity, + mut manifest, + client, + system_prompt, + input: _, + cache_key, + max_turns, + engine_configurator, + features, + required_tools, + authority, + } = spec; + manifest.worker.name = format!("internal-{}-{}", identity.kind, identity.run_id); + manifest.feature = Default::default(); + manifest.plugins = Default::default(); + manifest.mcp = Default::default(); + manifest.skills = None; + manifest.compaction = None; + manifest.memory = None; + + let mut engine = Engine::new(client).system_prompt(system_prompt); + engine.set_cache_key(cache_key); + engine.set_max_turns(max_turns); + if let Some(configure) = engine_configurator { + configure(&mut engine); + } + let store = EphemeralSessionStore::default(); + let inherited_workdir_session = authority.workdir_session.clone(); + let mut worker = Worker::new( + manifest, + engine, + store.clone(), + authority.workspace, + authority.filesystem, + authority.scope, + ) + .await + .map_err(|source| InternalWorkerSessionError::Build { + message: source.to_string(), + })?; + if let Some(session) = inherited_workdir_session { + worker.bind_workdir_session(Some(session)); + } + let install_report = worker.install_features(features); + let installed_tools = install_report.installed_tool_names(); + let install_failed = install_report + .reports + .iter() + .any(|report| !report.installed); + let missing = required_tools + .iter() + .filter(|required| { + !installed_tools + .iter() + .any(|installed| installed == **required) + }) + .copied() + .collect::>(); + if install_failed || !missing.is_empty() { + let diagnostics = install_report + .reports + .iter() + .flat_map(|report| report.diagnostics.iter()) + .map(|diagnostic| diagnostic.message.as_str()) + .collect::>() + .join("; "); + return Err(InternalWorkerSessionError::Build { + message: format!( + "internal Worker feature installation failed: {diagnostics}; missing tools: {}", + missing.join(", ") + ), + }); + } + + Box::pin(prepare_internal_worker_session( + worker, store, visibility, None, None, + )) + .await + }) +} + fn spawn_internal_log_event_bridge(sink: SegmentLogSink, event_tx: broadcast::Sender) { let (_, mut log_rx) = sink.subscribe_with_snapshot(); tokio::spawn(async move { @@ -1076,12 +1189,14 @@ permission = "write" input: "input".to_string(), cache_key: Some("internal-test".to_string()), max_turns: Some(1), + engine_configurator: None, features: FeatureRegistryBuilder::new(), required_tools, authority: InternalWorkerAuthority { workspace: WorkerWorkspaceContext::no_workspace(), filesystem: WorkerFilesystemAuthority::None, scope: Scope::empty(), + workdir_session: None, }, } } diff --git a/crates/worker/src/session_capture.rs b/crates/worker/src/session_capture.rs index 10d05a0f..d3ae8ec6 100644 --- a/crates/worker/src/session_capture.rs +++ b/crates/worker/src/session_capture.rs @@ -143,6 +143,7 @@ pub(crate) struct SearchHit { #[derive(Debug, Clone, Copy)] pub(crate) enum ReadSelector<'a> { Id(&'a str), + #[cfg(test)] EntryRange([u64; 2]), } @@ -418,6 +419,7 @@ impl SessionCapture { .iter() .filter(|entry| entry.id.as_str() == id) .collect(), + #[cfg(test)] ReadSelector::EntryRange([start, end]) => self .index .iter() diff --git a/crates/worker/src/spawn/registry.rs b/crates/worker/src/spawn/registry.rs index c800b01d..86ea8bae 100644 --- a/crates/worker/src/spawn/registry.rs +++ b/crates/worker/src/spawn/registry.rs @@ -1,7 +1,8 @@ -//! Parent-owned registry of direct Internal SubWorker sessions. +//! Parent-owned registry of direct Internal Worker sessions. //! -//! `SubWorkerSpawn` inserts typed `InternalWorkerSessionHandle`s; List/Send/Stop and -//! worker-observation use the same in-memory authority. Internal children are not persisted, restored, discovered as +//! `SubWorkerSpawn` inserts controllable SubWorker handles, while host services such as +//! compaction insert parent-visible service handles without joining the model-facing +//! List/Send/Stop surface. Internal children are not persisted, restored, discovered as //! Runtime Workers, or addressed through sockets. Restore consumes any legacy persisted process //! child records only to reclaim their delegated scope and clear obsolete metadata. //! Parent registry drop closes all session handles and synchronously returns delegated Write deny @@ -177,6 +178,52 @@ impl InternalSpawnedWorkerRecord { } } +/// Parent-visible service Internal Worker. Unlike a SubWorker this record has no +/// delegated scope, model-facing control name, or stop-summary authority. +#[derive(Clone)] +pub(crate) struct InternalServiceWorkerRecord { + pub service_kind: String, + pub display_name: String, + pub session: InternalWorkerSessionHandle, + protocol_revision: Arc, + protocol_emit_lock: Arc>, + protocol_terminal: Arc, + forwarding_started: Arc, +} + +impl InternalServiceWorkerRecord { + pub(crate) fn new( + service_kind: impl Into, + display_name: impl Into, + session: InternalWorkerSessionHandle, + ) -> Self { + Self { + service_kind: service_kind.into(), + display_name: display_name.into(), + session, + protocol_revision: Arc::new(AtomicU64::new(0)), + protocol_emit_lock: Arc::new(Mutex::new(())), + protocol_terminal: Arc::new(AtomicBool::new(false)), + forwarding_started: Arc::new(AtomicBool::new(false)), + } + } + + fn protocol_ref(&self, parent_session_id: Option) -> InternalWorkerRef { + InternalWorkerRef { + session_id: self.session.session_id_string(), + name: self.display_name.clone(), + parent_session_id, + kind: InternalWorkerKind::Service { + kind: self.service_kind.clone(), + }, + } + } + + fn protocol_revision(&self) -> u64 { + self.protocol_revision.load(Ordering::Acquire) + } +} + pub(crate) struct InternalSpawnReservation { registry: Arc, worker_name: String, @@ -214,6 +261,7 @@ impl Drop for InternalSpawnReservation { pub struct SpawnedWorkerRegistry { internal_records: std::sync::Mutex>, + service_records: std::sync::Mutex>, internal_names: std::sync::Mutex>, parent_scope: Option, parent_protocol: Mutex, String)>>, @@ -226,10 +274,21 @@ pub struct SpawnedWorkerRegistryLoad { } impl SpawnedWorkerRegistry { + pub(crate) fn new_for_internal_services() -> Arc { + Arc::new(Self { + internal_records: std::sync::Mutex::new(Vec::new()), + service_records: std::sync::Mutex::new(Vec::new()), + internal_names: std::sync::Mutex::new(HashSet::new()), + parent_scope: None, + parent_protocol: Mutex::new(None), + }) + } + /// Empty registry used by tests and non-spawning projections. pub fn new(_runtime_dir: Arc) -> Arc { Arc::new(Self { internal_records: std::sync::Mutex::new(Vec::new()), + service_records: std::sync::Mutex::new(Vec::new()), internal_names: std::sync::Mutex::new(HashSet::new()), parent_scope: None, parent_protocol: Mutex::new(None), @@ -239,6 +298,7 @@ impl SpawnedWorkerRegistry { pub(crate) fn new_internal(_parent_name: String, parent_scope: SharedScope) -> Arc { Arc::new(Self { internal_records: std::sync::Mutex::new(Vec::new()), + service_records: std::sync::Mutex::new(Vec::new()), internal_names: std::sync::Mutex::new(HashSet::new()), parent_scope: Some(parent_scope), parent_protocol: Mutex::new(None), @@ -317,6 +377,7 @@ impl SpawnedWorkerRegistry { Ok(SpawnedWorkerRegistryLoad { registry: Arc::new(Self { internal_records: std::sync::Mutex::new(Vec::new()), + service_records: std::sync::Mutex::new(Vec::new()), internal_names: std::sync::Mutex::new(HashSet::new()), parent_scope, parent_protocol: Mutex::new(None), @@ -356,6 +417,144 @@ impl SpawnedWorkerRegistry { for record in self.internal_records.lock().unwrap().clone() { self.start_protocol_forwarding(record); } + for record in self.service_records.lock().unwrap().clone() { + self.start_service_protocol_forwarding(record); + } + } + + /// Register a parent-visible service Internal Worker before its first turn. + pub(crate) fn attach_service( + &self, + record: InternalServiceWorkerRecord, + ) -> io::Result { + let parent_session_id = self + .parent_protocol + .lock() + .unwrap() + .as_ref() + .map(|(_, id)| id.clone()); + let worker_ref = record.protocol_ref(parent_session_id); + let session_id = record.session.session_id_string(); + let mut records = self + .service_records + .lock() + .map_err(|_| io::Error::other("internal service-worker registry lock poisoned"))?; + if records + .iter() + .any(|candidate| candidate.session.session_id_string() == session_id) + { + return Err(io::Error::new( + io::ErrorKind::AlreadyExists, + "internal service Worker is already registered", + )); + } + records.push(record.clone()); + drop(records); + self.start_service_protocol_forwarding(record); + Ok(worker_ref) + } + + /// Stop and remove one parent-owned service Worker. This is host-only and is + /// intentionally separate from the SubWorker control surface. + pub(crate) async fn stop_service(&self, session_id: &str) -> io::Result { + let record = self + .service_records + .lock() + .map_err(|_| io::Error::other("internal service-worker registry lock poisoned"))? + .iter() + .find(|record| record.session.session_id_string() == session_id) + .cloned(); + let Some(record) = record else { + return Ok(false); + }; + record + .session + .stop() + .await + .map_err(|error| io::Error::other(error.to_string()))?; + self.remove_service(session_id) + } + + /// Remove one service Worker and emit the terminal projection fence. + pub(crate) fn remove_service(&self, session_id: &str) -> io::Result { + let removed = { + let mut records = self + .service_records + .lock() + .map_err(|_| io::Error::other("internal service-worker registry lock poisoned"))?; + records + .iter() + .position(|record| record.session.session_id_string() == session_id) + .map(|index| records.remove(index)) + }; + if let Some(record) = removed { + self.publish_service_removal(&record); + Ok(true) + } else { + Ok(false) + } + } + + fn start_service_protocol_forwarding(&self, record: InternalServiceWorkerRecord) { + if record.session.visibility() != InternalWorkerVisibility::ParentClient + || record.forwarding_started.swap(true, Ordering::AcqRel) + { + return; + } + let Some((parent_tx, parent_session_id)) = self.parent_protocol.lock().unwrap().clone() + else { + record.forwarding_started.store(false, Ordering::Release); + return; + }; + let worker = record.protocol_ref(Some(parent_session_id)); + let protocol_revision = record.protocol_revision.clone(); + let protocol_emit_lock = record.protocol_emit_lock.clone(); + let protocol_terminal = record.protocol_terminal.clone(); + let mut child_rx = record.session.subscribe_events(); + tokio::spawn(async move { + loop { + match child_rx.recv().await { + Ok(event) => { + let shutdown = matches!(event, Event::Shutdown); + let _emit_guard = protocol_emit_lock + .lock() + .unwrap_or_else(|error| error.into_inner()); + if protocol_terminal.load(Ordering::Acquire) { + break; + } + let revision = protocol_revision.fetch_add(1, Ordering::AcqRel) + 1; + let _ = parent_tx.send(Event::InternalWorker { + worker: worker.clone(), + revision, + event: Box::new(event), + }); + if shutdown { + break; + } + } + Err(broadcast::error::RecvError::Lagged(skipped)) => { + let _emit_guard = protocol_emit_lock + .lock() + .unwrap_or_else(|error| error.into_inner()); + if protocol_terminal.load(Ordering::Acquire) { + break; + } + let revision = protocol_revision.fetch_add(1, Ordering::AcqRel) + 1; + let _ = parent_tx.send(Event::InternalWorker { + worker: worker.clone(), + revision, + event: Box::new(Event::Error { + code: protocol::ErrorCode::Internal, + message: format!( + "internal Worker output lagged by {skipped} events; reconnect to resynchronize" + ), + }), + }); + } + Err(broadcast::error::RecvError::Closed) => break, + } + } + }); } fn start_protocol_forwarding(&self, record: InternalSpawnedWorkerRecord) { @@ -427,28 +626,37 @@ impl SpawnedWorkerRegistry { .unwrap() .as_ref() .map(|(_, id)| id.clone()); - self.internal_records + let mut snapshots = self + .internal_records .lock() .unwrap() .iter() .filter(|record| record.session.visibility() == InternalWorkerVisibility::ParentClient) .map(|record| { - let snapshot = record.session.protocol_snapshot(); - InternalWorkerSnapshot { - worker: record.protocol_ref(parent_session_id.clone()), - revision: record.protocol_revision(), - entries: snapshot - .entries - .into_iter() - .filter_map(|entry| serde_json::to_value(entry).ok()) - .collect(), - status: snapshot.status, - error: snapshot.error, - in_flight: snapshot.in_flight, - internal_workers: snapshot.internal_workers, - } + internal_worker_snapshot( + record.protocol_ref(parent_session_id.clone()), + record.protocol_revision(), + &record.session, + ) }) - .collect() + .collect::>(); + snapshots.extend( + self.service_records + .lock() + .unwrap() + .iter() + .filter(|record| { + record.session.visibility() == InternalWorkerVisibility::ParentClient + }) + .map(|record| { + internal_worker_snapshot( + record.protocol_ref(parent_session_id.clone()), + record.protocol_revision(), + &record.session, + ) + }), + ); + snapshots } pub(crate) fn get_internal(&self, worker_name: &str) -> Option { @@ -567,6 +775,47 @@ impl SpawnedWorkerRegistry { revision, }); } + + fn publish_service_removal(&self, record: &InternalServiceWorkerRecord) { + if record.session.visibility() != InternalWorkerVisibility::ParentClient { + return; + } + let Some((parent_tx, parent_session_id)) = self.parent_protocol.lock().unwrap().clone() + else { + return; + }; + let _emit_guard = record + .protocol_emit_lock + .lock() + .unwrap_or_else(|error| error.into_inner()); + record.protocol_terminal.store(true, Ordering::Release); + let revision = record.protocol_revision.fetch_add(1, Ordering::AcqRel) + 1; + let _ = parent_tx.send(Event::InternalWorkerRemoved { + worker: record.protocol_ref(Some(parent_session_id)), + revision, + }); + } +} + +fn internal_worker_snapshot( + worker: InternalWorkerRef, + revision: u64, + session: &InternalWorkerSessionHandle, +) -> InternalWorkerSnapshot { + let snapshot = session.protocol_snapshot(); + InternalWorkerSnapshot { + worker, + revision, + entries: snapshot + .entries + .into_iter() + .filter_map(|entry| serde_json::to_value(entry).ok()) + .collect(), + status: snapshot.status, + error: snapshot.error, + in_flight: snapshot.in_flight, + internal_workers: snapshot.internal_workers, + } } impl Drop for SpawnedWorkerRegistry { @@ -813,6 +1062,63 @@ mod tests { assert_eq!(snapshots[0].in_flight.blocks.len(), 1); } + #[tokio::test] + async fn service_worker_is_parent_visible_but_not_subworker_controllable() { + let registry = registry(); + let (parent_tx, mut parent_rx) = broadcast::channel(16); + registry.attach_parent_protocol(parent_tx, "parent-session".into()); + let (session, child_tx) = + test_internal_worker_session(InternalWorkerVisibility::ParentClient); + let session_id = session.session_id_string(); + + let worker_ref = registry + .attach_service(InternalServiceWorkerRecord::new( + "compaction", + "Compaction", + session, + )) + .unwrap(); + assert!(matches!( + worker_ref.kind, + InternalWorkerKind::Service { ref kind } if kind == "compaction" + )); + assert!(registry.list_internal().is_empty()); + + child_tx + .send(Event::TextDone { + text: "summary candidate".into(), + }) + .unwrap(); + let event = tokio::time::timeout(Duration::from_secs(1), parent_rx.recv()) + .await + .unwrap() + .unwrap(); + assert!(matches!( + event, + Event::InternalWorker { worker, revision: 1, event } + if worker.session_id == session_id + && matches!(worker.kind, InternalWorkerKind::Service { ref kind } if kind == "compaction") + && matches!(*event, Event::TextDone { ref text } if text == "summary candidate") + )); + let snapshots = registry.internal_worker_snapshots(); + assert_eq!(snapshots.len(), 1); + assert!(matches!( + snapshots[0].worker.kind, + InternalWorkerKind::Service { ref kind } if kind == "compaction" + )); + assert!(registry.remove_service(&session_id).unwrap()); + let removed = tokio::time::timeout(Duration::from_secs(1), parent_rx.recv()) + .await + .unwrap() + .unwrap(); + assert!(matches!( + removed, + Event::InternalWorkerRemoved { worker, revision: 2 } + if worker.session_id == session_id + )); + assert!(registry.internal_worker_snapshots().is_empty()); + } + #[tokio::test] async fn service_private_internal_output_is_never_disclosed() { let registry = registry(); diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 47b57ac8..03430c10 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -46,12 +46,11 @@ use crate::hook::{ }; use crate::in_flight::InFlightEvents; use crate::internal_worker::{ - InternalWorkerAuthority, InternalWorkerIdentity, InternalWorkerSpec, run_internal_worker, - run_internal_worker_with_cancel_sender, + InternalWorkerAuthority, InternalWorkerIdentity, InternalWorkerSpec, InternalWorkerVisibility, + prepare_internal_worker_from_spec, run_internal_worker, run_internal_worker_with_cancel_sender, }; const COMPACTION_EXTENSION_DOMAIN: &str = "yoi.compaction"; -const COMPACTION_BLOCK_ID: &str = "compact"; const WORKER_ORCHESTRATION_INSTRUCTION_ID: &str = "worker.orchestration"; const WORKER_ORCHESTRATION_PROMPT_REF: &str = "common.worker_orchestration"; @@ -76,7 +75,8 @@ use crate::skill::{SkillActivationResponse, SkillClientError}; #[cfg(test)] use async_trait::async_trait; use protocol::{ - AlertLevel, AlertSource, Event, RewindSummary, RewindTarget, RewindTargetId, Segment, + AlertLevel, AlertSource, CompactionLifecycle, CompactionLifecycleState, Event, RewindSummary, + RewindTarget, RewindTargetId, Segment, }; use tokio::net::UnixStream; use tokio::sync::broadcast; @@ -971,6 +971,9 @@ pub struct Worker { /// notifications, events sent here are NOT replayed to clients that /// connect after the fact — they are fire-and-forget broadcasts. event_tx: Option>, + /// Parent-owned projection/control boundary for observable Internal service Workers. + /// Service Workers are never exposed through the model-facing SubWorker control surface. + internal_worker_registry: Option>, in_flight: Option, /// Monotonic counter incremented by worker event bridges when an /// assistant-side execution artifact becomes visible to clients before @@ -1115,6 +1118,7 @@ impl Worker feature_instructions: self.feature_instructions.clone(), alerter: self.alerter.clone(), event_tx: self.event_tx.clone(), + internal_worker_registry: self.internal_worker_registry.clone(), in_flight: self.in_flight.clone(), ai_activity_counter: self.ai_activity_counter.clone(), pending_notifies: NotifyBuffer::new(), @@ -1315,6 +1319,7 @@ impl Worker { feature_instructions: Vec::new(), alerter: None, event_tx: None, + internal_worker_registry: None, in_flight: None, ai_activity_counter: Arc::new(AtomicUsize::new(0)), pending_notifies: NotifyBuffer::new(), @@ -1957,9 +1962,21 @@ impl Worker { /// Worker-internal operations (currently: compaction) can surface /// progress to connected clients. pub fn attach_event_tx(&mut self, event_tx: broadcast::Sender) { + let session_id = self.session_id().to_string(); + let registry = self.internal_worker_registry.get_or_insert_with( + crate::spawn::registry::SpawnedWorkerRegistry::new_for_internal_services, + ); + registry.attach_parent_protocol(event_tx.clone(), session_id); self.event_tx = Some(event_tx); } + pub(crate) fn attach_internal_worker_registry( + &mut self, + registry: Arc, + ) { + self.internal_worker_registry = Some(registry); + } + /// Shared activity counter incremented by worker event bridges when any /// assistant-side output is surfaced before history persistence. pub fn ai_activity_counter(&self) -> Arc { @@ -2895,52 +2912,46 @@ impl Worker { .map_err(WorkerError::Engine) } - fn persist_compaction_block( + fn persist_compaction_lifecycle( &mut self, - state: &str, - message: &str, - error: Option<&str>, - new_segment_id: Option, + lifecycle: &CompactionLifecycle, ) -> Result<(), WorkerError> { - let payload = serde_json::json!({ - "kind": "compaction_block", - "schema_version": 1, - "block_id": COMPACTION_BLOCK_ID, - "state": state, - "message": message, - "error": error, - "new_segment_id": new_segment_id.map(|id| id.to_string()), - }); Ok(self.commit_entry(LogEntry::Extension { ts: segment_log::now_millis(), domain: COMPACTION_EXTENSION_DOMAIN.into(), - payload, + payload: serde_json::to_value(lifecycle).map_err(|error| { + WorkerError::InvalidState(format!( + "serialize compaction lifecycle {}: {error}", + lifecycle.compaction_id + )) + })?, })?) } - fn persist_and_send_compact_start(&mut self) -> Result<(), WorkerError> { - self.persist_compaction_block("running", "Compacting…", None, None)?; - self.send_event(Event::CompactStart); + fn persist_and_send_compact_start( + &mut self, + lifecycle: CompactionLifecycle, + ) -> Result<(), WorkerError> { + self.persist_compaction_lifecycle(&lifecycle)?; + self.send_event(Event::CompactStart { lifecycle }); Ok(()) } fn persist_and_send_compact_done( &mut self, - new_segment_id: SegmentId, + lifecycle: CompactionLifecycle, ) -> Result<(), WorkerError> { - self.persist_compaction_block("done", "Compacted.", None, Some(new_segment_id))?; - self.send_event(Event::CompactDone { new_segment_id }); + self.persist_compaction_lifecycle(&lifecycle)?; + self.send_event(Event::CompactDone { lifecycle }); Ok(()) } - fn persist_and_send_compact_failed(&mut self, error: String) -> Result<(), WorkerError> { - self.persist_compaction_block( - "failed", - &format!("Compact failed: {error}"), - Some(error.as_str()), - None, - )?; - self.send_event(Event::CompactFailed { error }); + fn persist_and_send_compact_failed( + &mut self, + lifecycle: CompactionLifecycle, + ) -> Result<(), WorkerError> { + self.persist_compaction_lifecycle(&lifecycle)?; + self.send_event(Event::CompactFailed { lifecycle }); Ok(()) } @@ -2969,14 +2980,12 @@ impl Worker { .map(|s| s.retained_tokens()) .unwrap_or(manifest::defaults::COMPACT_RETAINED_TOKENS); - self.persist_and_send_compact_start()?; match self.compact(retained).await { Ok(new_segment_id) => { info!( new_segment_id = %new_segment_id, "Compaction succeeded, resuming execution" ); - self.persist_and_send_compact_done(new_segment_id)?; if let Some(ref state) = self.compact_state { state.record_compact_success(); } @@ -2984,7 +2993,6 @@ impl Worker { } Err(e) => { warn!(error = %e, "Compaction failed during run"); - self.persist_and_send_compact_failed(e.to_string())?; self.alert( AlertLevel::Error, AlertSource::Compactor, @@ -3017,45 +3025,16 @@ impl Worker { } let retained = state.retained_tokens(); - if let Err(err) = self.persist_and_send_compact_start() { - warn!(error = %err, "failed to persist proactive compact start"); - self.alert( - AlertLevel::Warn, - AlertSource::Compactor, - format!("pre-run compaction not started: failed to persist status block: {err}"), - ); - return; - } match self.compact(retained).await { Ok(new_segment_id) => { info!( new_segment_id = %new_segment_id, "Proactive pre-run compaction succeeded" ); - if let Err(err) = self.persist_and_send_compact_done(new_segment_id) { - warn!(error = %err, "failed to persist proactive compact completion"); - self.alert( - AlertLevel::Warn, - AlertSource::Compactor, - format!( - "pre-run compaction completed but status block was not persisted: {err}" - ), - ); - } state.record_compact_success(); } Err(e) => { warn!(error = %e, "Proactive pre-run compaction failed"); - if let Err(err) = self.persist_and_send_compact_failed(e.to_string()) { - warn!(error = %err, "failed to persist proactive compact failure"); - self.alert( - AlertLevel::Warn, - AlertSource::Compactor, - format!( - "pre-run compaction failed and status block was not persisted: {err}" - ), - ); - } self.alert( AlertLevel::Warn, AlertSource::Compactor, @@ -3113,11 +3092,9 @@ impl Worker { } self.join_memory_task().await; - self.persist_and_send_compact_start()?; match self.compact(retained).await { Ok(new_segment_id) => { info!(new_segment_id = %new_segment_id, "Manual compaction succeeded"); - self.persist_and_send_compact_done(new_segment_id)?; if let Some(ref state) = state { state.record_compact_success(); } @@ -3125,7 +3102,6 @@ impl Worker { } Err(e) => { warn!(error = %e, "Manual compaction failed"); - self.persist_and_send_compact_failed(e.to_string())?; self.alert( AlertLevel::Error, AlertSource::Compactor, @@ -3262,20 +3238,79 @@ impl Worker { Ok(()) } - /// Compact the current session by summarising history via a - /// disposable Engine, then replacing history with - /// `[summary, ...recent_turns]` in a new Segment of the same Session. - /// - /// The summary Engine uses: - /// - `compaction.model` from the manifest if configured, or - /// - a clone of the main LlmClient via `clone_boxed()`. - /// - /// Returns the new Segment ID. The Worker keeps its Session ID. + /// Runs one parent-owned observable compaction service and returns the new + /// Segment ID. Lifecycle revisions are committed before they are broadcast. pub async fn compact(&mut self, retained_tokens: u64) -> Result { + let mut lifecycle = CompactionLifecycle { + schema_version: 2, + compaction_id: uuid::Uuid::now_v7().to_string(), + revision: 1, + internal_worker: None, + state: CompactionLifecycleState::Running, + started_at_ms: segment_log::now_millis(), + ended_at_ms: None, + summary: None, + error: None, + new_segment_id: None, + }; + self.persist_and_send_compact_start(lifecycle.clone())?; + match self.compact_impl(retained_tokens, &mut lifecycle).await { + Ok((new_segment_id, summary)) => { + lifecycle.revision = lifecycle.revision.saturating_add(1); + lifecycle.state = CompactionLifecycleState::Done; + lifecycle.ended_at_ms = Some(segment_log::now_millis()); + lifecycle.summary = Some(summary); + lifecycle.new_segment_id = Some(new_segment_id.to_string()); + let terminal = self.persist_and_send_compact_done(lifecycle.clone()); + self.release_compaction_service(&lifecycle).await; + terminal?; + Ok(new_segment_id) + } + Err(error) => { + lifecycle.revision = lifecycle.revision.saturating_add(1); + lifecycle.state = if matches!(error, WorkerError::CompactCancelled) { + CompactionLifecycleState::Interrupted + } else { + CompactionLifecycleState::Failed + }; + lifecycle.ended_at_ms = Some(segment_log::now_millis()); + lifecycle.error = Some(error.to_string().chars().take(2_000).collect()); + let terminal = self.persist_and_send_compact_failed(lifecycle.clone()); + self.release_compaction_service(&lifecycle).await; + terminal?; + Err(error) + } + } + } + + async fn release_compaction_service(&self, lifecycle: &CompactionLifecycle) { + let Some(session_id) = lifecycle + .internal_worker + .as_ref() + .map(|worker| worker.session_id.as_str()) + else { + return; + }; + let Some(registry) = &self.internal_worker_registry else { + return; + }; + if let Err(error) = registry.stop_service(session_id).await { + warn!( + compaction_id = %lifecycle.compaction_id, + internal_worker_session_id = %session_id, + error = %error, + "failed to release terminal compaction Internal Worker" + ); + } + } + + async fn compact_impl( + &mut self, + retained_tokens: u64, + lifecycle: &mut CompactionLifecycle, + ) -> Result<(SegmentId, String), WorkerError> { use crate::compact::worker::{ - CompactWorkerContext, CompactWorkerInterceptor, add_reference_tool, - mark_read_required_tool, read_session_items_tool, search_session_log_tool, - write_summary_tool, + CompactWorkerContext, CompactWorkerInterceptor, CompactionOutputFeature, }; use crate::fs_view::WorkerFsView; @@ -3385,10 +3420,12 @@ impl Worker { auto_read_budget, ))); - // Build an independent compact worker. It clones the main Worker's - // provider handle, so compact-time reads use the same WorkdirSession instance. - // No-workdir Workers deliberately omit compact-time filesystem tools. + // Build a normal parent-owned Internal Worker over a pinned immutable + // capture. Only SessionExplore and compaction output tools are installed. let workdir = self.workdir_session.clone(); + let read_only_workdir = workdir.clone().map(|session| { + Arc::new(ReadOnlyWorkdirSession::new(session)) as workdir::WorkdirSessionHandle + }); let summary_tracker = tools::Tracker::new(); let summary_client: Box = self.build_compactor_client()?; let summary_system_prompt = self @@ -3396,51 +3433,129 @@ impl Worker { .load_full() .compact_system() .map_err(WorkerError::PromptCatalog)?; - let mut summary_worker = Engine::new(summary_client).system_prompt(summary_system_prompt); - summary_worker.set_cache_key(Some(self.segment_id().to_string())); - - // Occupancy-based input-token meter + interceptor. The tracker pairs - // each pre-request history length with the following UsageEvent, then - // the interceptor projects current prompt occupancy with the same - // UsageRecord counter used by the main Worker thresholds. let summary_usage_tracker = Arc::new(UsageTracker::new()); - { - let tracker = summary_usage_tracker.clone(); - summary_worker.on_usage(move |event| { - tracker.record_usage(event); - }); - } let compactor_warning_cb = self.alerter.clone().map(|alerter| { Arc::new(move |message: String| { alerter.alert(AlertLevel::Warn, AlertSource::Compactor, message); }) as Arc }); - summary_worker.set_interceptor(CompactWorkerInterceptor::new( - summary_usage_tracker, + let interceptor = CompactWorkerInterceptor::new( + summary_usage_tracker.clone(), worker_context_max_tokens, finish_warning_remaining_tokens, final_reserve_tokens, compactor_warning_cb, - )); - summary_worker.set_max_turns(worker_max_turns); + ); + let tracker_for_engine = summary_usage_tracker.clone(); + let features = crate::feature::FeatureRegistryBuilder::new() + .with_module( + crate::feature::builtin::session_explore::SessionExploreFeature::new( + crate::feature::builtin::session_explore::SessionExploreState::new( + crate::session_capture::SessionCapture::new( + self.segment_id().to_string(), + items_to_summarise.clone(), + ), + ), + ), + ) + .with_module(CompactionOutputFeature::new( + read_only_workdir.clone(), + summary_tracker, + ctx.clone(), + )); + let required_tools: &'static [&'static str] = if read_only_workdir.is_some() { + &[ + "ShowOverview", + "SearchEntries", + "ReadEntry", + "Read", + "mark_read_required", + "add_reference", + "write_summary", + ] + } else { + &[ + "ShowOverview", + "SearchEntries", + "ReadEntry", + "add_reference", + "write_summary", + ] + }; + let handle = prepare_internal_worker_from_spec( + InternalWorkerSpec { + identity: InternalWorkerIdentity { + kind: "compaction", + run_id: uuid::Uuid::parse_str(&lifecycle.compaction_id).map_err(|error| { + WorkerError::InvalidState(format!("invalid compaction id: {error}")) + })?, + }, + manifest: self.manifest.clone(), + client: summary_client, + system_prompt: summary_system_prompt, + input: summary_input.text.clone(), + cache_key: Some(self.segment_id().to_string()), + max_turns: worker_max_turns, + engine_configurator: Some(Box::new(move |engine| { + let tracker = tracker_for_engine; + engine.on_usage(move |event| { + tracker.record_usage(event); + }); + engine.set_interceptor(interceptor); + })), + features, + required_tools, + authority: InternalWorkerAuthority { + workspace: WorkerWorkspaceContext::no_workspace(), + filesystem: WorkerFilesystemAuthority::None, + scope: Scope::empty(), + workdir_session: read_only_workdir, + }, + }, + InternalWorkerVisibility::ParentClient, + ) + .await + .map_err(|error| WorkerError::InvalidState(error.to_string()))?; + let registry = self + .internal_worker_registry + .get_or_insert_with( + crate::spawn::registry::SpawnedWorkerRegistry::new_for_internal_services, + ) + .clone(); + let internal_ref = registry + .attach_service(crate::spawn::registry::InternalServiceWorkerRecord::new( + "compaction", + "Compaction", + handle.clone(), + )) + .map_err(|error| WorkerError::InvalidState(error.to_string()))?; + lifecycle.revision = lifecycle.revision.saturating_add(1); + lifecycle.internal_worker = Some(internal_ref); + self.persist_and_send_compact_start(lifecycle.clone())?; - // Tools: read_file (shared scope, fresh tracker), bounded session - // history exploration, and compact-specific tools that populate `ctx`. - let compact_target_items = Arc::new(items_to_summarise.clone()); - if let Some(workdir) = workdir.clone() { - summary_worker.register_tool(tools::read_tool(workdir.clone(), summary_tracker)); - summary_worker.register_tool(mark_read_required_tool(workdir, ctx.clone())); + if let Err(error) = handle.send(summary_input.text).await { + let _ = registry.remove_service(&handle.session_id_string()); + return Err(WorkerError::InvalidState(error.to_string())); + } + match handle.wait_until_idle().await { + crate::internal_worker::InternalWorkerSessionStatus::Idle => {} + crate::internal_worker::InternalWorkerSessionStatus::Stopped => { + return Err(WorkerError::CompactCancelled); + } + crate::internal_worker::InternalWorkerSessionStatus::Failed => { + return Err(WorkerError::InvalidState( + handle + .protocol_snapshot() + .error + .unwrap_or_else(|| "compactor Internal Worker failed".into()), + )); + } + status => { + return Err(WorkerError::InvalidState(format!( + "compactor Internal Worker ended in {status:?}" + ))); + } } - summary_worker.register_tool(search_session_log_tool(compact_target_items.clone())); - summary_worker.register_tool(read_session_items_tool(compact_target_items)); - summary_worker.register_tool(add_reference_tool(ctx.clone())); - summary_worker.register_tool(write_summary_tool(ctx.clone())); - - let out = summary_worker - .run(summary_input.text) - .await - .map_err(WorkerError::Engine)?; - let mut locked_engine = out.engine; // Guard: nudge the worker once more if the expected outputs // (summary, and any auto-read nominations when default refs @@ -3469,10 +3584,24 @@ impl Worker { } }; if let Some(prompt) = nudge { - let _ = locked_engine - .run(prompt) + handle + .send(prompt) .await - .map_err(WorkerError::Engine)?; + .map_err(|error| WorkerError::InvalidState(error.to_string()))?; + match handle.wait_until_idle().await { + crate::internal_worker::InternalWorkerSessionStatus::Idle => {} + crate::internal_worker::InternalWorkerSessionStatus::Stopped => { + return Err(WorkerError::CompactCancelled); + } + _ => { + return Err(WorkerError::InvalidState( + handle + .protocol_snapshot() + .error + .unwrap_or_else(|| "compactor Internal Worker failed".into()), + )); + } + } } let mut final_ctx = ctx.lock().expect("compact ctx poisoned").clone(); @@ -3487,10 +3616,24 @@ impl Worker { {summary_max_tokens}). Rewrite it now with `write_summary`, preserving the \ same five sections but making it concise. Target ≈{summary_target_tokens} tokens." ); - let _ = locked_engine - .run(prompt) + handle + .send(prompt) .await - .map_err(WorkerError::Engine)?; + .map_err(|error| WorkerError::InvalidState(error.to_string()))?; + match handle.wait_until_idle().await { + crate::internal_worker::InternalWorkerSessionStatus::Idle => {} + crate::internal_worker::InternalWorkerSessionStatus::Stopped => { + return Err(WorkerError::CompactCancelled); + } + _ => { + return Err(WorkerError::InvalidState( + handle + .protocol_snapshot() + .error + .unwrap_or_else(|| "compactor Internal Worker failed".into()), + )); + } + } final_ctx = ctx.lock().expect("compact ctx poisoned").clone(); summary_text = final_ctx .summary @@ -3694,7 +3837,7 @@ impl Worker { .lock() .expect("extract_pointer poisoned") = None; - Ok(new_segment_id) + Ok((new_segment_id, summary_text)) } /// Build the LlmClient for the compactor Engine. @@ -4080,6 +4223,7 @@ impl Worker { input: input_text, cache_key: Some(self.segment_id().to_string()), max_turns: extract_worker_max_turns, + engine_configurator: None, features, required_tools: &[ "ShowOverview", @@ -4092,6 +4236,7 @@ impl Worker { workspace: self.workspace_context.clone(), filesystem: WorkerFilesystemAuthority::None, scope: Scope::empty(), + workdir_session: None, }, }; let internal_result = match cancel_observer { @@ -4566,6 +4711,7 @@ where feature_instructions: common.feature_instructions, alerter: None, event_tx: None, + internal_worker_registry: None, in_flight: None, ai_activity_counter: Arc::new(AtomicUsize::new(0)), pending_notifies: NotifyBuffer::new(), @@ -4643,6 +4789,7 @@ where feature_instructions: common.feature_instructions, alerter: None, event_tx: None, + internal_worker_registry: None, in_flight: None, ai_activity_counter: Arc::new(AtomicUsize::new(0)), pending_notifies: NotifyBuffer::new(), @@ -4755,6 +4902,7 @@ where feature_instructions: common.feature_instructions, alerter: None, event_tx: None, + internal_worker_registry: None, in_flight: None, ai_activity_counter: Arc::new(AtomicUsize::new(0)), pending_notifies: NotifyBuffer::new(), @@ -5071,6 +5219,7 @@ where feature_instructions: common.feature_instructions, alerter: None, event_tx: None, + internal_worker_registry: None, in_flight: None, ai_activity_counter: Arc::new(AtomicUsize::new(0)), pending_notifies: NotifyBuffer::new(), @@ -5725,6 +5874,9 @@ pub enum WorkerError { #[error("compact worker did not produce a summary (write_summary was never called)")] CompactSummaryMissing, + #[error("compaction was cancelled")] + CompactCancelled, + #[error("compact summary too large: {tokens} tokens exceeds max {max}")] CompactSummaryTooLarge { tokens: u64, max: u64 }, diff --git a/crates/worker/tests/compact_events_test.rs b/crates/worker/tests/compact_events_test.rs index 973f080f..2323cf03 100644 --- a/crates/worker/tests/compact_events_test.rs +++ b/crates/worker/tests/compact_events_test.rs @@ -431,7 +431,7 @@ async fn pre_run_compact_success_broadcasts_start_and_done() { let kinds: Vec<&str> = events .iter() .map(|e| match e { - Event::CompactStart => "start", + Event::CompactStart { .. } => "start", Event::CompactDone { .. } => "done", Event::CompactFailed { .. } => "failed", _ => "other", @@ -445,10 +445,61 @@ async fn pre_run_compact_success_broadcasts_start_and_done() { !kinds.contains(&"failed"), "unexpected CompactFailed in {kinds:?}" ); + let starts = events + .iter() + .filter_map(|event| match event { + Event::CompactStart { lifecycle } => Some(lifecycle), + _ => None, + }) + .collect::>(); + assert_eq!( + starts.len(), + 2, + "start and Internal Worker binding revisions" + ); + assert_eq!(starts[0].compaction_id, starts[1].compaction_id); + assert_eq!(starts[0].revision, 1); + assert!(starts[0].internal_worker.is_none()); + assert_eq!(starts[1].revision, 2); + assert!(matches!( + starts[1].internal_worker.as_ref().map(|worker| &worker.kind), + Some(protocol::InternalWorkerKind::Service { kind }) if kind == "compaction" + )); + assert!(events.iter().any(|event| matches!( + event, + Event::InternalWorker { worker, .. } + if matches!(&worker.kind, protocol::InternalWorkerKind::Service { kind } if kind == "compaction") + )), "compactor activity must be projected through the parent stream"); + let completed = events + .iter() + .find_map(|event| match event { + Event::CompactDone { lifecycle } => Some(lifecycle), + _ => None, + }) + .expect("completed lifecycle"); + assert_eq!(completed.compaction_id, starts[0].compaction_id); + assert_eq!(completed.revision, 3); + assert_eq!(completed.summary.as_deref(), Some("summary")); + assert_eq!(completed.state, protocol::CompactionLifecycleState::Done); + let done_index = events + .iter() + .position(|event| matches!(event, Event::CompactDone { .. })) + .expect("done event"); + let removed_index = events + .iter() + .position(|event| matches!(event, Event::InternalWorkerRemoved { .. })) + .expect("terminal compactor session must be released"); + assert!( + done_index < removed_index, + "terminal lifecycle precedes release fence" + ); // CompactDone carries the new Segment ID; the Session ID is unchanged. let new_id_in_event = events.iter().find_map(|e| match e { - Event::CompactDone { new_segment_id } => Some(*new_segment_id), + Event::CompactDone { lifecycle } => lifecycle + .new_segment_id + .as_deref() + .and_then(|value| uuid::Uuid::parse_str(value).ok()), _ => None, }); assert!(new_id_in_event.is_some(), "CompactDone missing"); @@ -488,7 +539,7 @@ async fn mid_turn_compact_success_broadcasts_start_and_done() { let kinds: Vec<&str> = events .iter() .map(|e| match e { - Event::CompactStart => "start", + Event::CompactStart { .. } => "start", Event::CompactDone { .. } => "done", Event::CompactFailed { .. } => "failed", _ => "other", @@ -504,7 +555,10 @@ async fn mid_turn_compact_success_broadcasts_start_and_done() { ); let new_id_in_event = events.iter().find_map(|e| match e { - Event::CompactDone { new_segment_id } => Some(*new_segment_id), + Event::CompactDone { lifecycle } => lifecycle + .new_segment_id + .as_deref() + .and_then(|value| uuid::Uuid::parse_str(value).ok()), _ => None, }); assert_eq!(new_id_in_event, Some(worker.segment_id())); @@ -659,7 +713,7 @@ async fn pre_run_compact_failure_broadcasts_start_and_failed() { let kinds: Vec<&str> = events .iter() .map(|e| match e { - Event::CompactStart => "start", + Event::CompactStart { .. } => "start", Event::CompactDone { .. } => "done", Event::CompactFailed { .. } => "failed", _ => "other", @@ -817,11 +871,14 @@ async fn controller_compact_method_emits_start_and_done() { .expect("timeout waiting for compact events") .expect("event") { - Event::CompactStart => saw_start = true, + Event::CompactStart { .. } => saw_start = true, Event::CompactDone { .. } => { break; } - Event::CompactFailed { error } => panic!("manual compact failed: {error}"), + Event::CompactFailed { lifecycle } => panic!( + "manual compact failed: {}", + lifecycle.error.as_deref().unwrap_or("unknown error") + ), _ => {} } } diff --git a/resources/prompts/internal/compact_system.md b/resources/prompts/internal/compact_system.md index abd24515..2d30eb9f 100644 --- a/resources/prompts/internal/compact_system.md +++ b/resources/prompts/internal/compact_system.md @@ -5,8 +5,8 @@ The conversation input is a bounded overview/index, not the full transcript. Tre ## Workflow 1. Read the provided overview/index and current TaskStore snapshot. -2. If the overview does not contain enough detail, use `search_session_log` to find relevant compact-target history items, then `read_session_items` to inspect only the needed range. -3. Use `read_file` to inspect referenced files before deciding what the next session needs. Prefer skimming over blind inclusion. +2. If the provided index is not enough, call `ShowOverview` on the pinned capture. Use `SearchEntries` to locate relevant entries, then `ReadEntry` with the returned stable entry reference to inspect only what is needed. +3. Use `Read` to inspect referenced Workdir files before deciding what the next session needs. Prefer skimming over blind inclusion. 4. For files whose current contents are load-bearing for the active work, call `mark_read_required` to inject them into the next session. These count against the auto-read token budget — spend it deliberately. 5. For files the next session should know about but can fetch on demand, call `add_reference` to record the path without embedding contents. 6. Finish with `write_summary` carrying the final text. You may call it multiple times; only the last call is kept. diff --git a/web/workspace/src/lib/generated/protocol.ts b/web/workspace/src/lib/generated/protocol.ts index 5faf77fa..b7383f87 100644 --- a/web/workspace/src/lib/generated/protocol.ts +++ b/web/workspace/src/lib/generated/protocol.ts @@ -32,6 +32,14 @@ export type CommandSnapshot = { command_id: string, tool_call_id: string | null, export type CommandEvent = { "kind": "started", command_id: string, tool_call_id: string | null, observed_at_ms: number, } | { "kind": "output", command_id: string, stream: CommandStream, start_offset: number, end_offset: number, content: string, observed_at_ms: number, } | { "kind": "terminal", command_id: string, status: CommandStatus, exit_code: number | null, stdout_end_offset: number, stderr_end_offset: number, observed_at_ms: number, }; +export type CompactionLifecycleState = "running" | "done" | "failed" | "interrupted"; + +export type CompactionLifecycle = { schema_version: number, compaction_id: string, revision: number, internal_worker?: InternalWorkerRef | null, state: CompactionLifecycleState, +/** + * Milliseconds since the Unix epoch. + */ +started_at_ms: number, ended_at_ms?: number | null, summary?: string | null, error?: string | null, new_segment_id?: string | null, }; + export type ScopeRule = { /** * Target path. Must be absolute by the time a `Scope` is built from @@ -63,7 +71,7 @@ export type InFlightBlock = { "kind": "text", text: string, finished?: boolean, export type InFlightSnapshot = { blocks?: Array, commands?: Array, }; -export type InternalWorkerKind = "sub_worker"; +export type InternalWorkerKind = "sub_worker" | { "service": { kind: string, } }; export type InternalWorkerRef = { session_id: string, name: string, parent_session_id?: string | null, kind: InternalWorkerKind, }; @@ -193,4 +201,4 @@ in_flight?: InFlightSnapshot, * Parent-owned Internal Worker sessions visible to this client. * Service-private Internal Workers are deliberately excluded. */ -internal_workers?: Array, } } | { "event": "internal_worker", "data": { worker: InternalWorkerRef, revision: number, event: Event, } } | { "event": "internal_worker_removed", "data": { worker: InternalWorkerRef, revision: number, } } | { "event": "segment_rotated", "data": { entry: unknown, } } | { "event": "status", "data": { status: WorkerStatus, } } | { "event": "command", "data": { event: CommandEvent, } } | { "event": "completions", "data": { kind: CompletionKind, entries: Array, } } | { "event": "rewind_targets", "data": { head_entries: number, targets: Array, } } | { "event": "rewind_applied", "data": { entries: Array, input: Array, summary: RewindSummary, } } | { "event": "workers_listed", "data": { workers: unknown, } } | { "event": "worker_restored", "data": { result: unknown, } } | { "event": "peer_registered", "data": { result: unknown, } } | { "event": "alert", "data": Alert } | { "event": "memory_worker", "data": MemoryWorkerEvent } | { "event": "compact_start" } | { "event": "compact_done", "data": { new_segment_id: string, } } | { "event": "compact_failed", "data": { error: string, } } | { "event": "shutdown" }; +internal_workers?: Array, } } | { "event": "internal_worker", "data": { worker: InternalWorkerRef, revision: number, event: Event, } } | { "event": "internal_worker_removed", "data": { worker: InternalWorkerRef, revision: number, } } | { "event": "segment_rotated", "data": { entry: unknown, } } | { "event": "status", "data": { status: WorkerStatus, } } | { "event": "command", "data": { event: CommandEvent, } } | { "event": "completions", "data": { kind: CompletionKind, entries: Array, } } | { "event": "rewind_targets", "data": { head_entries: number, targets: Array, } } | { "event": "rewind_applied", "data": { entries: Array, input: Array, summary: RewindSummary, } } | { "event": "workers_listed", "data": { workers: unknown, } } | { "event": "worker_restored", "data": { result: unknown, } } | { "event": "peer_registered", "data": { result: unknown, } } | { "event": "alert", "data": Alert } | { "event": "memory_worker", "data": MemoryWorkerEvent } | { "event": "compact_start", "data": { lifecycle: CompactionLifecycle, } } | { "event": "compact_done", "data": { lifecycle: CompactionLifecycle, } } | { "event": "compact_failed", "data": { lifecycle: CompactionLifecycle, } } | { "event": "shutdown" }; diff --git a/web/workspace/src/lib/workspace/console/ConsoleLineItem.svelte b/web/workspace/src/lib/workspace/console/ConsoleLineItem.svelte index 65787335..c2e3a640 100644 --- a/web/workspace/src/lib/workspace/console/ConsoleLineItem.svelte +++ b/web/workspace/src/lib/workspace/console/ConsoleLineItem.svelte @@ -8,6 +8,33 @@ }; let { item }: Props = $props(); + let nowMs = $state(Date.now()); + + $effect(() => { + if (item.compaction?.state !== 'running') return; + nowMs = Date.now(); + const timer = window.setInterval(() => { + nowMs = Date.now(); + }, 1000); + return () => window.clearInterval(timer); + }); + + function compactionElapsedMs(line: ConsoleLine): number { + const compaction = line.compaction; + if (!compaction) return 0; + return Math.max(0, (compaction.endedAtMs ?? nowMs) - compaction.startedAtMs); + } + + function formatElapsed(ms: number): string { + const seconds = Math.floor(ms / 1000); + if (seconds < 60) return `${seconds}s`; + const minutes = Math.floor(seconds / 60); + return `${minutes}m ${seconds % 60}s`; + } + + function compactionState(line: ConsoleLine): string { + return line.compaction?.state.replace('_', ' ') ?? ''; + } function lineClass(line: ConsoleLine): string { return line.error ? 'error' : line.kind; @@ -52,7 +79,29 @@ class:error-line={item.error} data-console-line-id={item.id} > - {#if shouldRenderHeading(item)} + {#if item.compaction} +
+ Compaction · {compactionState(item)} + {formatElapsed(compactionElapsedMs(item))} +
+ {#if item.compaction.activity.length > 0} +
    + {#each item.compaction.activity as activity} +
  • {activity}
  • + {/each} +
+ {/if} + {#if item.compaction.summary} +
{item.compaction.summary}
+ {:else if item.compaction.candidate} +
+ candidate + {item.compaction.candidate} +
+ {:else if item.compaction.error} +
{item.compaction.error}
+ {/if} + {:else if shouldRenderHeading(item)}
{item.title}
@@ -63,7 +112,9 @@ {toolSummary(item).suffix} {/if} - {#if item.kind === 'tool'} + {#if item.compaction} + + {:else if item.kind === 'tool'} {#if bodyTextAfterToolSummary(item)}

{#if isBashTool(item)} @@ -141,6 +192,43 @@ white-space: pre-line; } + .compaction-heading { + display: flex; + justify-content: space-between; + gap: var(--space-2); + color: var(--text-muted); + font-family: var(--font-mono); + font-size: 0.78rem; + font-variant-numeric: tabular-nums; + font-weight: 750; + } + + .compaction-activity { + margin: var(--space-1) 0; + padding-left: 1.25rem; + color: var(--tui-dark-gray); + font-family: var(--font-mono); + font-size: 0.75rem; + } + + .compaction-summary, + .compaction-candidate, + .compaction-error { + margin-top: var(--space-1); + white-space: pre-wrap; + } + + .compaction-candidate-label { + display: block; + color: var(--text-muted); + font-family: var(--font-mono); + font-size: 0.7rem; + } + + .compaction-error { + color: var(--tui-error); + } + .activity-summary { font-size: 14px; } diff --git a/web/workspace/src/lib/workspace/console/model.test.ts b/web/workspace/src/lib/workspace/console/model.test.ts index 840ff822..98fb085a 100644 --- a/web/workspace/src/lib/workspace/console/model.test.ts +++ b/web/workspace/src/lib/workspace/console/model.test.ts @@ -647,42 +647,173 @@ Deno.test("projectConsole renders alert events", () => { assertEquals(projection.lines[1].error, true); }); -Deno.test("projectConsole shows compact progress as a status block", () => { +Deno.test("projectConsole upserts compaction lifecycle by stable id", () => { + const running = { + schema_version: 2, + compaction_id: "compaction-1", + revision: 1, + internal_worker: null, + state: "running", + started_at_ms: 1_000, + ended_at_ms: null, + summary: null, + error: null, + new_segment_id: null, + } as const; const projection = projectConsole([ { eventId: "compact-1", - event: { event: "compact_start" } satisfies Event, + event: { event: "compact_start", data: { lifecycle: running } } satisfies Event, }, ]); assertEquals(projection.lines.length, 1); - assertEquals(projection.lines[0].id, "status-compact"); - assertEquals(projection.lines[0].kind, "status"); - assertEquals(projection.lines[0].body, "Compacting…"); + assertEquals(projection.lines[0].id, "compaction-compaction-1"); + assertEquals(projection.lines[0].compaction?.state, "running"); assertEquals(projection.lines[0].streaming, true); const completed = projectConsole([ { eventId: "compact-1", - event: { event: "compact_start" } satisfies Event, + event: { event: "compact_start", data: { lifecycle: running } } satisfies Event, }, { eventId: "compact-2", event: { event: "compact_done", - data: { new_segment_id: "00000000-0000-0000-0000-000000000001" }, + data: { + lifecycle: { + ...running, + revision: 2, + state: "done", + ended_at_ms: 4_000, + summary: "accepted summary", + new_segment_id: "00000000-0000-0000-0000-000000000001", + }, + }, } satisfies Event, }, ]); assertEquals(completed.lines.length, 1); - assertEquals(completed.lines[0].id, "status-compact"); - assertEquals(completed.lines[0].body, "Compacted."); + assertEquals(completed.lines[0].id, "compaction-compaction-1"); + assertEquals(completed.lines[0].compaction?.state, "done"); + assertEquals(completed.lines[0].compaction?.summary, "accepted summary"); assertEquals(completed.lines[0].streaming, false); }); -Deno.test("createConsoleProjector updates only compact status block", () => { +Deno.test("compaction service activity stays nested in one lifecycle item", () => { + const worker = { + session_id: "compactor-session", + name: "Compaction", + parent_session_id: "parent-session", + kind: { service: { kind: "compaction" } }, + } as const; + const lifecycle = { + schema_version: 2, + compaction_id: "compaction-nested", + revision: 2, + internal_worker: worker, + state: "running", + started_at_ms: 1_000, + ended_at_ms: null, + summary: null, + error: null, + new_segment_id: null, + } as const; + const projection = projectConsole([ + { + eventId: "compaction-start", + event: { event: "compact_start", data: { lifecycle } } satisfies Event, + }, + { + eventId: "compaction-tool", + event: { + event: "internal_worker", + data: { + worker, + revision: 1, + event: { + event: "tool_call_start", + data: { id: "call-1", name: "write_summary" }, + }, + }, + } satisfies Event, + }, + { + eventId: "compaction-tool-done", + event: { + event: "internal_worker", + data: { + worker, + revision: 2, + event: { + event: "tool_call_done", + data: { + id: "call-1", + name: "write_summary", + arguments: JSON.stringify({ text: "draft candidate" }), + }, + }, + }, + } satisfies Event, + }, + ]); + + assertEquals(projection.lines.length, 1); + assertEquals(projection.lines[0].id, "compaction-compaction-nested"); + assertEquals(projection.lines[0].compaction?.activity, ["write_summary — running"]); + assertEquals(projection.lines[0].compaction?.candidate, "draft candidate"); + assert( + consoleWorkerViews(projection).length === 1, + "service is not a selectable SubWorker pane", + ); +}); + +Deno.test("snapshot normalizes orphaned running compaction to interrupted", () => { + const projection = projectConsole([{ + eventId: "snapshot", + observedAtMs: 9_000, + event: snapshotEvent("/repo", [{ + kind: "extension", + ts: 1, + domain: "yoi.compaction", + payload: { + schema_version: 2, + compaction_id: "compaction-orphaned", + revision: 2, + internal_worker: { + session_id: "missing-service", + name: "Compaction", + parent_session_id: "parent", + kind: { service: { kind: "compaction" } }, + }, + state: "running", + started_at_ms: 1_000, + }, + }]), + }]); + + assertEquals(projection.lines.length, 1); + assertEquals(projection.lines[0].compaction?.state, "interrupted"); + assertEquals(projection.lines[0].compaction?.endedAtMs, 9_000); + assertEquals(projection.lines[0].streaming, false); +}); + +Deno.test("createConsoleProjector ignores stale compaction revisions", () => { const projector = createConsoleProjector(); + const base = { + schema_version: 2, + compaction_id: "compaction-identity", + revision: 1, + internal_worker: null, + state: "running", + started_at_ms: 1_000, + ended_at_ms: null, + summary: null, + error: null, + new_segment_id: null, + } as const; let projection = projector.append([ { eventId: "compact-identity-1", @@ -693,31 +824,37 @@ Deno.test("createConsoleProjector updates only compact status block", () => { }, { eventId: "compact-identity-2", - event: { event: "compact_start" } satisfies Event, + event: { event: "compact_start", data: { lifecycle: base } } satisfies Event, }, ]); const userLine = projection.lines[0]; - const compactLine = projection.lines[1]; projection = projector.append([ { eventId: "compact-identity-3", event: { event: "compact_done", - data: { new_segment_id: "00000000-0000-0000-0000-000000000001" }, + data: { + lifecycle: { + ...base, + revision: 2, + state: "done", + ended_at_ms: 2_000, + summary: "accepted", + new_segment_id: "00000000-0000-0000-0000-000000000001", + }, + }, } satisfies Event, }, + { + eventId: "compact-identity-stale", + event: { event: "compact_start", data: { lifecycle: base } } satisfies Event, + }, ]); - assert( - projection.lines[0] === userLine, - "unrelated message line should keep object identity", - ); - assert( - projection.lines[1] !== compactLine, - "compact status line should update object identity", - ); - assertEquals(projection.lines[1].body, "Compacted."); + assert(projection.lines[0] === userLine, "unrelated line retains identity"); + assertEquals(projection.lines[1].compaction?.state, "done"); + assertEquals(projection.lines[1].compaction?.summary, "accepted"); }); Deno.test("projectConsole keeps streaming tool call updates in the same Call block", () => { diff --git a/web/workspace/src/lib/workspace/console/model.ts b/web/workspace/src/lib/workspace/console/model.ts index b4f695f8..67a0f29e 100644 --- a/web/workspace/src/lib/workspace/console/model.ts +++ b/web/workspace/src/lib/workspace/console/model.ts @@ -3,6 +3,7 @@ import type { CommandEvent, CommandSnapshot, CommandStreamSlice, + CompactionLifecycle, Event as ProtocolEvent, InFlightBlock, InFlightToolCallState, @@ -67,12 +68,26 @@ export type ConsoleDiffLine = { export type ConsoleViewMode = "overview" | "normal"; +export type ConsoleCompaction = { + id: string; + revision: number; + state: "running" | "done" | "failed" | "interrupted"; + startedAtMs: number; + endedAtMs?: number; + summary?: string; + candidate?: string; + error?: string; + internalWorkerSessionId?: string; + activity: string[]; +}; + export type ConsoleLine = { id: string; kind: ConsoleLineKind; title: string; body: string; detail?: string; + compaction?: ConsoleCompaction; diff?: ConsoleDiffLine[]; eventId?: string | null; source: "event"; @@ -114,16 +129,17 @@ export type ConsoleWorkerView = { export function consoleWorkerViews( projection: ConsoleProjection, ): ConsoleWorkerView[] { - const labels = projection.internalWorkers.map((worker) => - worker.worker.name || "subworker" + const children = projection.internalWorkers.filter((worker) => + worker.worker.kind === "sub_worker" ); + const labels = children.map((worker) => worker.worker.name || "subworker"); const labelCounts = new Map(); for (const label of labels) { labelCounts.set(label, (labelCounts.get(label) ?? 0) + 1); } return [ { sessionId: null, label: "main", console: projection }, - ...projection.internalWorkers.map((worker, index) => { + ...children.map((worker, index) => { const label = labels[index] ?? "subworker"; return { sessionId: worker.worker.session_id, @@ -664,6 +680,105 @@ function projectInternalWorkerSnapshot( return { worker: snapshot.worker, revision: snapshot.revision, console }; } +function compactionCandidate( + projection: ConsoleProjection, + sessionId: string | undefined, +): string | undefined { + if (!sessionId) return undefined; + const worker = projection.internalWorkers.find( + (candidate) => candidate.worker.session_id === sessionId, + ); + const calls = worker?.console.lines + .map((line) => line.toolCall) + .filter((call): call is ToolCallView => call?.name === "write_summary") ?? []; + const latest = calls.at(-1); + const raw = latest?.arguments ?? latest?.argsStream; + if (!raw) return undefined; + try { + const parsed = JSON.parse(raw) as { text?: unknown }; + return typeof parsed.text === "string" ? parsed.text : undefined; + } catch { + return undefined; + } +} + +function compactionActivity( + projection: ConsoleProjection, + sessionId: string | undefined +): string[] { + if (!sessionId) return []; + const worker = projection.internalWorkers.find( + (candidate) => candidate.worker.session_id === sessionId + ); + if (!worker) return []; + return worker.console.lines + .filter((line) => line.kind === "tool" || line.kind === "status" || line.kind === "error") + .slice(-12) + .map((line) => + line.toolCall?.name === "write_summary" + ? `write_summary — ${line.toolCall.state}` + : line.body || line.title + ) + .filter((value, index, values) => value.length > 0 && values.indexOf(value) === index); +} + +function applyCompactionLifecycle( + projection: ConsoleProjection, + lifecycle: CompactionLifecycle +): ConsoleProjection { + const lineId = `compaction-${lifecycle.compaction_id}`; + const existing = projection.lines.find((line) => line.id === lineId)?.compaction; + if (existing && existing.revision >= lifecycle.revision) return projection; + const internalWorkerSessionId = lifecycle.internal_worker?.session_id; + const compaction: ConsoleCompaction = { + id: lifecycle.compaction_id, + revision: lifecycle.revision, + state: lifecycle.state, + startedAtMs: lifecycle.started_at_ms, + endedAtMs: lifecycle.ended_at_ms ?? undefined, + summary: lifecycle.summary ?? undefined, + candidate: compactionCandidate(projection, internalWorkerSessionId), + error: lifecycle.error ?? undefined, + internalWorkerSessionId, + activity: compactionActivity(projection, internalWorkerSessionId) + }; + const line: ConsoleLine = { + id: lineId, + source: "event", + kind: compaction.state === "failed" ? "error" : "status", + title: "Compaction", + body: compaction.summary ?? compaction.error ?? "", + streaming: compaction.state === "running", + error: compaction.state === "failed", + compaction + }; + const index = projection.lines.findIndex((candidate) => candidate.id === lineId); + const lines = [...projection.lines]; + if (index >= 0) lines[index] = line; + else lines.push(line); + return { ...projection, lines }; +} + +function refreshCompactionActivity( + projection: ConsoleProjection, + sessionId: string +): ConsoleProjection { + let changed = false; + const lines = projection.lines.map((line) => { + if (line.compaction?.internalWorkerSessionId !== sessionId) return line; + changed = true; + return { + ...line, + compaction: { + ...line.compaction, + candidate: compactionCandidate(projection, sessionId), + activity: compactionActivity(projection, sessionId) + } + }; + }); + return changed ? { ...projection, lines } : projection; +} + export function applyProtocolEvent( projection: ConsoleProjection, envelope: ConsoleEventInput, @@ -809,6 +924,31 @@ export function applyProtocolEvent( projectInternalWorkerSnapshot(worker, envelope.eventId, next.cwd) ); next.removedInternalWorkers = {}; + for (const line of next.lines) { + const compaction = line.compaction; + if (!compaction) continue; + const sessionId = compaction.internalWorkerSessionId; + if (sessionId) { + line.compaction = { + ...compaction, + candidate: compactionCandidate(next, sessionId), + activity: compactionActivity(next, sessionId), + }; + } + if ( + compaction.state === "running" && + (!sessionId || !next.internalWorkers.some((worker) => + worker.worker.session_id === sessionId + )) + ) { + line.streaming = false; + line.compaction = { + ...line.compaction!, + state: "interrupted", + endedAtMs: envelope.observedAtMs ?? Date.now(), + }; + } + } break; } case "internal_worker": { @@ -841,7 +981,7 @@ export function applyProtocolEvent( }; if (existingIndex >= 0) next.internalWorkers[existingIndex] = updated; else next.internalWorkers.push(updated); - break; + return refreshCompactionActivity(next, event.data.worker.session_id); } case "internal_worker_removed": { const existingIndex = next.internalWorkers.findIndex((worker) => @@ -905,21 +1045,9 @@ export function applyProtocolEvent( // them to the conversation surface; browser Console should not either. break; case "compact_start": - upsertStatusLine(next, "compact", envelope.eventId, "Compacting…", true); - break; case "compact_done": - upsertStatusLine(next, "compact", envelope.eventId, "Compacted.", false); - break; case "compact_failed": - upsertStatusLine( - next, - "compact", - envelope.eventId, - `Compact failed: ${event.data.error}`, - false, - true, - ); - break; + return applyCompactionLifecycle(next, event.data.lifecycle); case "shutdown": next.status = "shutdown"; break; @@ -1880,7 +2008,22 @@ function applyExtensionEntry( return; } const payload = entry["payload"]; - if (!isRecord(payload) || payload["kind"] !== "compaction_block") { + if (!isRecord(payload)) return; + if ( + typeof payload["compaction_id"] === "string" && + typeof payload["revision"] === "number" && + typeof payload["state"] === "string" && + typeof payload["started_at_ms"] === "number" + ) { + const updated = applyCompactionLifecycle( + projection, + payload as unknown as CompactionLifecycle, + ); + projection.lines = updated.lines; + return; + } + // Schema v1 remains readable historical evidence. + if (payload["kind"] !== "compaction_block") { return; } const blockId = stringField(payload, "block_id") || "compact"; diff --git a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts index 009cce36..02ada805 100644 --- a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts +++ b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts @@ -926,8 +926,9 @@ Deno.test("Web Console switches main and direct SubWorker views from the Tasks r ); assert( consoleModel.includes("consoleWorkerViews") && - consoleModel.includes("projection.internalWorkers.map") && + consoleModel.includes('worker.worker.kind === "sub_worker"') && + consoleModel.includes("children.map") && consoleModel.includes("resolveConsoleWorkerView"), - "Worker view selection should use direct Internal Worker session identities with main fallback", + "Worker view selection should expose only direct SubWorker session identities with main fallback", ); });