feat: expose compaction as an internal worker

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<InternalWorkerRef>,
+    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<u64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub summary: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub error: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub new_segment_id: Option<String>,
+}
+
+#[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:?}"),
         }
     }

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::<CommandStreamSlice>(&cfg, &mut output);
     push_decl::<CommandSnapshot>(&cfg, &mut output);
     push_decl::<CommandEvent>(&cfg, &mut output);
+    push_decl::<CompactionLifecycleState>(&cfg, &mut output);
+    push_decl::<CompactionLifecycle>(&cfg, &mut output);
     push_decl::<ScopeRule>(&cfg, &mut output);
     push_decl::<CompletionEntry>(&cfg, &mut output);
     push_decl::<RewindTargetId>(&cfg, &mut output);

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);
     }

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<usize>,
-    /// Maximum number of hits to return.
-    #[serde(default)]
-    pub limit: Option<usize>,
-}
-
-/// 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<Vec<Item>>,
-    view: SessionCapture,
-}
-
-struct SearchSessionLogTool {
-    state: Arc<SessionLogToolState>,
-}
-
-#[async_trait]
-impl Tool for SearchSessionLogTool {
-    async fn execute(
-        &self,
-        input_json: &str,
-        _ctx: agen::tool::ToolExecutionContext,
-    ) -> Result<ToolOutput, ToolError> {
-        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::<Vec<_>>();
-        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<SessionLogToolState>,
-}
-
-#[async_trait]
-impl Tool for ReadSessionItemsTool {
-    async fn execute(
-        &self,
-        input_json: &str,
-        _ctx: agen::tool::ToolExecutionContext,
-    ) -> Result<ToolOutput, ToolError> {
-        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(&params.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::<Vec<_>>()
-            .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<Self, ToolError> {
-        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<Mutex<CompactWorkerContext>>,
@@ -508,36 +278,63 @@ pub(crate) fn write_summary_tool(ctx: Arc<Mutex<CompactWorkerContext>>) -> ToolD
     })
 }
 
-pub(crate) fn search_session_log_tool(items: Arc<Vec<Item>>) -> 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<dyn Tool> = Arc::new(SearchSessionLogTool {
-            state: state.clone(),
-        });
-        (meta, tool)
-    })
+#[derive(Clone)]
+pub(crate) struct CompactionOutputFeature {
+    session: Option<WorkdirSessionHandle>,
+    tracker: tools::Tracker,
+    context: Arc<Mutex<CompactWorkerContext>>,
 }
 
-pub(crate) fn read_session_items_tool(items: Arc<Vec<Item>>) -> 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<dyn Tool> = Arc::new(ReadSessionItemsTool {
-            state: state.clone(),
-        });
-        (meta, tool)
-    })
+impl CompactionOutputFeature {
+    pub(crate) fn new(
+        session: Option<WorkdirSessionHandle>,
+        tracker: tools::Tracker,
+        context: Arc<Mutex<CompactWorkerContext>>,
+    ) -> 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<dyn Tool> = 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<dyn Tool> = 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), "");
-    }
 }

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,

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 {

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<workdir::WorkdirSessionHandle>,
 }
 
 pub(crate) struct InternalWorkerSpec {
@@ -53,6 +55,7 @@ pub(crate) struct InternalWorkerSpec {
     pub input: String,
     pub cache_key: Option<String>,
     pub max_turns: Option<u32>,
+    pub engine_configurator: Option<Box<dyn FnOnce(&mut Engine<Box<dyn LlmClient>>) + 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<InternalWorkerSessionHandle, InternalWorkerSessionError>,
+            > + 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::<Vec<_>>();
+        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::<Vec<_>>()
+                .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<Event>) {
     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,
             },
         }
     }

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()

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<AtomicU64>,
+    protocol_emit_lock: Arc<Mutex<()>>,
+    protocol_terminal: Arc<AtomicBool>,
+    forwarding_started: Arc<AtomicBool>,
+}
+
+impl InternalServiceWorkerRecord {
+    pub(crate) fn new(
+        service_kind: impl Into<String>,
+        display_name: impl Into<String>,
+        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<String>) -> 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<SpawnedWorkerRegistry>,
     worker_name: String,
@@ -214,6 +261,7 @@ impl Drop for InternalSpawnReservation {
 
 pub struct SpawnedWorkerRegistry {
     internal_records: std::sync::Mutex<Vec<InternalSpawnedWorkerRecord>>,
+    service_records: std::sync::Mutex<Vec<InternalServiceWorkerRecord>>,
     internal_names: std::sync::Mutex<HashSet<String>>,
     parent_scope: Option<SharedScope>,
     parent_protocol: Mutex<Option<(broadcast::Sender<Event>, String)>>,
@@ -226,10 +274,21 @@ pub struct SpawnedWorkerRegistryLoad {
 }
 
 impl SpawnedWorkerRegistry {
+    pub(crate) fn new_for_internal_services() -> Arc<Self> {
+        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<RuntimeDir>) -> Arc<Self> {
         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<Self> {
         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<InternalWorkerRef> {
+        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<bool> {
+        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<bool> {
+        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::<Vec<_>>();
+        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<InternalSpawnedWorkerRecord> {
@@ -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();

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<C: LlmClient, St: Store> {
     /// notifications, events sent here are NOT replayed to clients that
     /// connect after the fact — they are fire-and-forget broadcasts.
     event_tx: Option<broadcast::Sender<Event>>,
+    /// 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<Arc<crate::spawn::registry::SpawnedWorkerRegistry>>,
     in_flight: Option<InFlightEvents>,
     /// Monotonic counter incremented by worker event bridges when an
     /// assistant-side execution artifact becomes visible to clients before
@@ -1115,6 +1118,7 @@ impl<C: LlmClient + Clone + 'static, St: Store + Clone + 'static> Worker<C, St>
             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<C: LlmClient, St: Store> Worker<C, St> {
             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<C: LlmClient, St: Store> Worker<C, St> {
     /// Worker-internal operations (currently: compaction) can surface
     /// progress to connected clients.
     pub fn attach_event_tx(&mut self, event_tx: broadcast::Sender<Event>) {
+        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<crate::spawn::registry::SpawnedWorkerRegistry>,
+    ) {
+        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<AtomicUsize> {
@@ -2895,52 +2912,46 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
             .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<SegmentId>,
+        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<C: LlmClient, St: Store> Worker<C, St> {
                 .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<C: LlmClient, St: Store> Worker<C, St> {
                 }
                 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<C: LlmClient, St: Store> Worker<C, St> {
         }
 
         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<C: LlmClient, St: Store> Worker<C, St> {
         }
 
         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<C: LlmClient, St: Store> Worker<C, St> {
             }
             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<C: LlmClient, St: Store> Worker<C, St> {
         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<SegmentId, WorkerError> {
+        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<C: LlmClient, St: Store> Worker<C, St> {
             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<dyn LlmClient> = self.build_compactor_client()?;
         let summary_system_prompt = self
@@ -3396,51 +3433,129 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
             .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<dyn Fn(String) + Send + Sync>
         });
-        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<C: LlmClient, St: Store> Worker<C, St> {
             }
         };
         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<C: LlmClient, St: Store> Worker<C, St> {
                  {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<C: LlmClient, St: Store> Worker<C, St> {
             .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<C: LlmClient, St: Store> Worker<C, St> {
             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<C: LlmClient, St: Store> Worker<C, St> {
                 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 },
 

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::<Vec<_>>();
+    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")
+            ),
             _ => {}
         }
     }

resources/prompts/common/tickets.md

@@ -4,6 +4,6 @@ Use the available typed Ticket tools as the authority for Ticket reads and mutat
 
 Read the relevant Ticket before making implementation, routing, review, state, or closure decisions. Do not infer the current contract from an id, title, notification, or remembered summary alone. Check related or potentially duplicate Tickets when creating or materially rescoping work.
 
-Keep durable Ticket records centered on user intent, confirmed background, requirements, acceptance criteria, binding decisions, and implementation/review evidence. Use `QueryObjective` for bounded Objective discovery and `ShowObjective` for authoritative revision and linked-Ticket context when coordinating broader work. Separate confirmed facts from user claims, hypotheses, and open questions. Avoid prematurely turning implementation tactics into requirements.
+Keep durable Ticket records centered on user intent, confirmed background, requirements, acceptance criteria, binding decisions, and implementation/review evidence. For implementation and review workflows, keep routine revision, verdict, fix, and rereview evidence on the Merge Request; use Ticket comments only for blockers or decisions requiring orchestration attention and the final approved handoff. Use `QueryObjective` for bounded Objective discovery and `ShowObjective` for authoritative revision and linked-Ticket context when coordinating broader work. Separate confirmed facts from user claims, hypotheses, and open questions. Avoid prematurely turning implementation tactics into requirements.
 
 Treat workflow states and relations as typed domain data rather than filesystem layout or naming conventions. Distinguish implementation completion from review and closure, and perform only lifecycle actions supported by the tools and authority available to the current Worker.

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.

resources/prompts/role/coder.md

@@ -1,5 +1,7 @@
 You are the assigned Coder. Implement the requested scope in the provided Workdir and keep durable evidence on the Ticket and its Merge Request.
 
+Use the Merge Request as the routine authority for review requests, verdicts, fixes, and rereview cycles. Do not add a Ticket comment for each review or fix iteration. Add a Ticket comment only when a blocker or decision requires Orchestrator attention, or once after approval to hand off the final implementation and validation evidence.
+
 Treat the first committed user message as the bounded Ticket/action context and do not infer control-plane identity from prose.
 
 Before opening a Merge Request, publish only the committed Ticket work branch with a normal non-force push and verify that the Ticket repository remote resolves it to the exact local `HEAD`; a local branch name or dirty Workdir is not immutable review evidence. Do not push the target branch, tags, or unrelated refs, and never force-push.

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<InFlightBlock>, commands?: Array<CommandSnapshot>, };
 
-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<InternalWorkerSnapshot>, } } | { "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<CompletionEntry>, } } | { "event": "rewind_targets", "data": { head_entries: number, targets: Array<RewindTarget>, } } | { "event": "rewind_applied", "data": { entries: Array<unknown>, input: Array<Segment>, 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<InternalWorkerSnapshot>, } } | { "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<CompletionEntry>, } } | { "event": "rewind_targets", "data": { head_entries: number, targets: Array<RewindTarget>, } } | { "event": "rewind_applied", "data": { entries: Array<unknown>, input: Array<Segment>, 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" };

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}
+    <div class="compaction-heading">
+      <span>Compaction · {compactionState(item)}</span>
+      <span>{formatElapsed(compactionElapsedMs(item))}</span>
+    </div>
+    {#if item.compaction.activity.length > 0}
+      <ul class="compaction-activity">
+        {#each item.compaction.activity as activity}
+          <li>{activity}</li>
+        {/each}
+      </ul>
+    {/if}
+    {#if item.compaction.summary}
+      <div class="compaction-summary">{item.compaction.summary}</div>
+    {:else if item.compaction.candidate}
+      <div class="compaction-candidate">
+        <span class="compaction-candidate-label">candidate</span>
+        {item.compaction.candidate}
+      </div>
+    {:else if item.compaction.error}
+      <div class="compaction-error">{item.compaction.error}</div>
+    {/if}
+  {:else if shouldRenderHeading(item)}
     <div class="message-heading">
       <span>{item.title}</span>
     </div>
@@ -63,7 +112,9 @@
       <span class={`tool-suffix ${item.toolCall?.state ?? ''}`}>{toolSummary(item).suffix}</span>
     </div>
   {/if}
-  {#if item.kind === 'tool'}
+  {#if item.compaction}
+    <!-- rendered as one lifecycle item above -->
+  {:else if item.kind === 'tool'}
     {#if bodyTextAfterToolSummary(item)}
       <p class="console-plain-text">
         {#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;
   }

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", () => {

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<string, number>();
   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";

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",
   );
 });