diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 3ec8a9a0a..087e55e73 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -5298,7 +5298,12 @@ impl Worker { } pub(crate) async fn retry_pending_compaction_cleanup(&self) -> Result<(), WorkerError> { - self.release_pending_compaction_service().await + let cleanup_pending = self.has_pending_compaction_cleanup(); + self.release_pending_compaction_service().await?; + if cleanup_pending { + self.set_compaction_progress(None); + } + Ok(()) } pub(crate) async fn finish_pending_compaction_cleanup_for_shutdown( @@ -8979,6 +8984,15 @@ mod build_summary_prompt_tests { let (cleanup_session_id, _events) = registry.install_service_for_test(); registry.fail_service_stops_for_test(&cleanup_session_id, 4); worker.internal_worker_registry = Some(Arc::clone(®istry)); + let (event_tx, mut event_rx) = broadcast::channel(16); + worker.attach_working_event_tx(event_tx.clone()); + let in_flight = InFlightEvents::new(event_tx); + worker.attach_in_flight_events(in_flight.clone()); + worker.set_compaction_progress(Some(InFlightCompaction { + phase: CompactionPhase::Summarizing, + started_at_ms: 1, + trigger: CompactionTrigger::Manual, + })); *worker .pending_compaction_cleanup .lock() @@ -8995,12 +9009,44 @@ mod build_summary_prompt_tests { } assert!(worker.has_pending_compaction_cleanup()); assert!(registry.has_service_for_test(&cleanup_session_id)); + let in_flight_guard = in_flight.snapshot_guard(); + assert!( + crate::in_flight::snapshot_from_guard(&in_flight_guard) + .compaction + .is_some(), + "failed cleanup retries must keep reconnect compaction state active" + ); + drop(in_flight_guard); + while let Ok(event) = event_rx.try_recv() { + assert!( + !matches!(event, Event::CompactionProgress { compaction: None }), + "terminal progress must not publish before the service stops" + ); + } tokio::time::timeout(Duration::from_secs(1), cleanup) .await .expect("shutdown cleanup must converge after retry") .unwrap(); assert!(!worker.has_pending_compaction_cleanup()); assert!(!registry.has_service_for_test(&cleanup_session_id)); + let in_flight_guard = in_flight.snapshot_guard(); + assert!( + crate::in_flight::snapshot_from_guard(&in_flight_guard) + .compaction + .is_none(), + "successful cleanup retry must clear reconnect compaction state" + ); + drop(in_flight_guard); + let mut saw_clear = false; + while let Ok(event) = event_rx.try_recv() { + if matches!(event, Event::CompactionProgress { compaction: None }) { + saw_clear = true; + } + } + assert!( + saw_clear, + "successful cleanup retry must publish terminal progress" + ); } #[tokio::test] diff --git a/web/workspace/src/lib/workspace/console/ConsoleLineItem.svelte b/web/workspace/src/lib/workspace/console/ConsoleLineItem.svelte index ad631e467..7a4fdf526 100644 --- a/web/workspace/src/lib/workspace/console/ConsoleLineItem.svelte +++ b/web/workspace/src/lib/workspace/console/ConsoleLineItem.svelte @@ -12,7 +12,7 @@ let nowMs = $state(Date.now()); $effect(() => { - if (item.compaction?.state !== 'running') return; + if (!item.compaction) return; nowMs = Date.now(); const timer = window.setInterval(() => { nowMs = Date.now(); @@ -23,7 +23,7 @@ function compactionElapsedMs(line: ConsoleLine): number { const compaction = line.compaction; if (!compaction) return 0; - return Math.max(0, (compaction.endedAtMs ?? nowMs) - compaction.startedAtMs); + return Math.max(0, nowMs - compaction.startedAtMs); } function formatElapsed(ms: number): string { @@ -33,8 +33,12 @@ return `${minutes}m ${seconds % 60}s`; } - function compactionState(line: ConsoleLine): string { - return line.compaction?.state.replace('_', ' ') ?? ''; + function compactionHeading(line: ConsoleLine): string { + const compaction = line.compaction; + if (!compaction) return ''; + return compaction.phase + ? `Compacting · ${compaction.phase.replace('_', ' ')}` + : 'Compacting…'; } function lineClass(line: ConsoleLine): string { @@ -80,7 +84,7 @@ > {#if item.compaction}
- Compaction · {compactionState(item)} + {compactionHeading(item)} {formatElapsed(compactionElapsedMs(item))}
{#if item.compaction.activity.length > 0} @@ -90,16 +94,6 @@ {/each} {/if} - {#if item.compaction.summary} -
{item.compaction.summary}
- {:else if item.compaction.candidate} -
- candidate - {item.compaction.candidate} -
- {:else if item.compaction.error} -
{item.compaction.error}
- {/if} {:else if shouldRenderHeading(item)}
{item.title} @@ -119,7 +113,7 @@
{/if} {#if item.compaction} - + {:else if item.kind === 'tool'} {#if toolBodyText(item)}

@@ -216,24 +210,6 @@ font-size: var(--font-size-compact); } - .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: var(--font-size-compact); - } - - .compaction-error { - color: var(--danger); - } - .activity-summary { font-size: var(--font-size-body); } diff --git a/web/workspace/src/lib/workspace/console/model.test.ts b/web/workspace/src/lib/workspace/console/model.test.ts index 665427f8f..69914f007 100644 --- a/web/workspace/src/lib/workspace/console/model.test.ts +++ b/web/workspace/src/lib/workspace/console/model.test.ts @@ -1,4 +1,8 @@ -import type { Event, WorkerStateSnapshot, WorkerStatus } from "$lib/generated/protocol"; +import type { + Event, + WorkerStateSnapshot, + WorkerStatus, +} from "$lib/generated/protocol"; import { type ConsoleEventInput, type ConsoleLine, @@ -22,15 +26,13 @@ declare const Deno: { function workerState(status: WorkerStatus): WorkerStateSnapshot { return { last_command_id: 0, - state: status === "idle" - ? { kind: "idle" } - : { - kind: "busy", - state: { - kind: "run", - state: status === "paused" ? "paused" : "running", - }, + state: status === "idle" ? { kind: "idle" } : { + kind: "busy", + state: { + kind: "run", + state: status === "paused" ? "paused" : "running", }, + }, }; } @@ -61,7 +63,8 @@ function consoleLine(id: string, kind: ConsoleLine["kind"]): ConsoleLine { function canonicalSession(logEntries: unknown[]): Event extends { event: "snapshot"; data: infer D; -} ? D extends { session: infer S } ? S : never : never { +} ? D extends { session: infer S } ? S : never + : never { const entries: Record[] = []; let sequence = 0; const itemEntry = (item: Record) => { @@ -80,9 +83,15 @@ function canonicalSession(logEntries: unknown[]): Event extends { case "segment_start": case "annotated_segment_start": entries.length = 0; - for (const history of Array.isArray(entry["history"]) ? entry["history"] : []) { + for ( + const history of Array.isArray(entry["history"]) + ? entry["history"] + : [] + ) { const value = history as Record; - itemEntry((value["item"] as Record | undefined) ?? value); + itemEntry( + (value["item"] as Record | undefined) ?? value, + ); } break; case "user_input": @@ -99,15 +108,18 @@ function canonicalSession(logEntries: unknown[]): Event extends { case "annotated_assistant_item": case "annotated_tool_result": { const annotated = entry["entry"] as Record | undefined; - itemEntry((annotated?.["item"] as Record | undefined) ?? - (entry["item"] as Record)); + itemEntry( + (annotated?.["item"] as Record | undefined) ?? + (entry["item"] as Record), + ); break; } case "system_item": case "annotated_system_item": { const annotated = entry["entry"] as Record | undefined; - const item = (annotated?.["item"] as Record | undefined) ?? - (entry["item"] as Record); + const item = + (annotated?.["item"] as Record | undefined) ?? + (entry["item"] as Record); entries.push({ entry_id: `legacy-test-${sequence++}`, provenance: "legacy_unknown", @@ -168,7 +180,10 @@ Deno.test("large paste segments project compact artifact metadata", () => { source_entry_id: "entry-1", }, }]); - assert(text.includes("019ca7c8-57b6-7f05-8edf-524147aba7b2"), "artifact id is visible"); + assert( + text.includes("019ca7c8-57b6-7f05-8edf-524147aba7b2"), + "artifact id is visible", + ); assert(text.includes("65536 bytes"), "bounded size metadata is visible"); assert(text.includes("text_plain_utf8"), "media type is visible"); assert(text.includes("available"), "availability is visible"); @@ -267,7 +282,9 @@ Deno.test("Worker state events and acknowledgements replace the full state", () assertEquals(projection.workerState, freshIdle); assertEquals(projection.status, "idle"); assert( - !projection.lines.some((line) => line.eventId?.includes("worker-state-conflict")), + !projection.lines.some((line) => + line.eventId?.includes("worker-state-conflict") + ), "full snapshots must not be rejected by a client-side version comparison", ); }); @@ -284,7 +301,10 @@ Deno.test("snapshot replaces a live error with one durable run_errored row", () }, { eventId: "idle-after-error", - event: { event: "worker_state", data: { snapshot: workerState("idle") } } satisfies Event, + event: { + event: "worker_state", + data: { snapshot: workerState("idle") }, + } satisfies Event, }, ]); @@ -722,7 +742,9 @@ Deno.test("projectConsole streams distinct Bash stdout and stderr through termin Deno.test("snapshot restores bounded in-flight Bash command output", () => { const snapshot = snapshotEvent("/repo"); - if (snapshot.event !== "snapshot") throw new Error("snapshot fixture expected"); + if (snapshot.event !== "snapshot") { + throw new Error("snapshot fixture expected"); + } snapshot.data.state = workerState("running"); snapshot.data.in_flight = { blocks: [{ @@ -750,7 +772,10 @@ Deno.test("snapshot restores bounded in-flight Bash command output", () => { }], }; - const projection = projectConsole([{ eventId: "snapshot-command", event: snapshot }]); + const projection = projectConsole([{ + eventId: "snapshot-command", + event: snapshot, + }]); const [line] = projection.lines.filter((line) => line.kind === "tool"); assertEquals(line.toolCallLabel, "Bash($ slow)"); assertEquals(line.toolStatus, "running…"); @@ -798,14 +823,20 @@ Deno.test("projectConsole caps default preview but keeps complete detail body", const [line] = projection.lines.filter((line) => line.kind === "tool"); assertEquals(line.title, "Call · CustomTool"); - assertEquals(line.toolCallLabel, 'CustomTool("first":"one","second":"two","third":"three","fourth":"four")'); + assertEquals( + line.toolCallLabel, + 'CustomTool("first":"one","second":"two","third":"three","fourth":"four")', + ); assertEquals(line.toolStatus, "done"); assertEquals(line.body.split("\n").length, 3); assert( line.body.includes("out1") && line.body.includes("… +3 more lines"), "normal display should retain the capped response preview", ); - assert(!line.body.includes("first"), "request arguments should stay in the Call signature and detail"); + assert( + !line.body.includes("first"), + "request arguments should stay in the Call signature and detail", + ); assert( line.detail?.includes("arguments:\nfirst: one") === true && line.detail?.includes("fourth: four") === true, @@ -873,7 +904,9 @@ Deno.test("projectConsole renders JSON tool responses as YAML", () => { const toolLines = projection.lines.filter((line) => line.kind === "tool"); const jsonLine = toolLines.find((line) => line.id.includes("json-tool")); - const invalidLine = toolLines.find((line) => line.id.includes("invalid-json-tool")); + const invalidLine = toolLines.find((line) => + line.id.includes("invalid-json-tool") + ); assert(jsonLine, "JSON tool line should be projected"); assert(invalidLine, "invalid JSON tool line should be projected"); assert( @@ -924,7 +957,10 @@ Deno.test("projectConsole caps Grep preview but keeps complete detail body", () assertEquals(line.toolStatus, "done"); assert(line.body.includes("hit1"), "first result should be shown"); assert(line.body.includes("hit5"), "fifth result should be shown"); - assert(!line.body.includes("hit6"), "normal preview should retain its result cap"); + assert( + !line.body.includes("hit6"), + "normal preview should retain its result cap", + ); assert( line.body.includes("… +1 more results"), "preview should show the omitted result count", @@ -1018,7 +1054,7 @@ Deno.test("projectConsole renders alert events", () => { assertEquals(projection.lines[1].error, true); }); -Deno.test("projectConsole upserts compaction lifecycle by stable id", () => { +Deno.test("legacy compaction lifecycle events are inert", () => { const running = { schema_version: 2, compaction_id: "compaction-1", @@ -1033,23 +1069,14 @@ Deno.test("projectConsole upserts compaction lifecycle by stable id", () => { } as const; const projection = projectConsole([ { - eventId: "compact-1", - event: { event: "compact_start", data: { lifecycle: running } } satisfies Event, - }, - ]); - - assertEquals(projection.lines.length, 1); - 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", data: { lifecycle: running } } satisfies Event, + eventId: "compact-start", + event: { + event: "compact_start", + data: { lifecycle: running }, + } satisfies Event, }, { - eventId: "compact-2", + eventId: "compact-done", event: { event: "compact_done", data: { @@ -1064,17 +1091,32 @@ Deno.test("projectConsole upserts compaction lifecycle by stable id", () => { }, } satisfies Event, }, + { + eventId: "compact-failed", + event: { + event: "compact_failed", + data: { + lifecycle: { + ...running, + revision: 3, + state: "failed", + ended_at_ms: 5_000, + error: "legacy failure", + }, + }, + } satisfies Event, + }, ]); - assertEquals(completed.lines.length, 1); - 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); + assertEquals(projection.lines, []); + assertEquals(projection.compaction, null); }); Deno.test("snapshot restores running compaction without staged content", () => { - const snapshot = snapshotEvent("/repo") as Extract; + const snapshot = snapshotEvent("/repo") as Extract< + Event, + { event: "snapshot" } + >; snapshot.data.in_flight = { blocks: [], compaction: { @@ -1083,37 +1125,223 @@ Deno.test("snapshot restores running compaction without staged content", () => { trigger: "manual", }, }; + snapshot.data.internal_workers = [{ + worker: { + session_id: "snapshot-compactor", + name: "Compaction", + parent_session_id: "parent-session", + kind: { service: { kind: "compaction" } }, + }, + revision: 1, + session: canonicalSession([]), + status: "running", + in_flight: { + blocks: [{ + kind: "tool_call", + id: "snapshot-read", + name: "ReadEntry", + args: JSON.stringify({ entry_ref: "E1" }), + state: "pending", + }], + }, + internal_workers: [], + }]; const projection = projectConsole([{ eventId: "snapshot", event: snapshot }]); - assertEquals(projection.lines.length, 0); + assertEquals(projection.lines.length, 1); + assertEquals(projection.lines[0].id, "compaction-runtime"); + assertEquals(projection.lines[0].compaction?.phase, "summarizing"); + assertEquals( + projection.lines[0].compaction?.internalWorkerSessionId, + "snapshot-compactor", + ); + assertEquals(projection.lines[0].compaction?.activity, [ + "ReadEntry — pending", + ]); assertEquals(projection.compaction?.phase, "summarizing"); assertEquals(projection.compaction?.trigger, "manual"); }); -Deno.test("compaction service activity stays nested in one lifecycle item", () => { +Deno.test("snapshot authoritatively clears and restores compaction progress", () => { + const projector = createConsoleProjector(); + let projection = projector.append([{ + eventId: "live-progress", + event: { + event: "compaction_progress", + data: { + compaction: { + phase: "summarizing", + started_at_ms: 1_000, + trigger: "manual", + }, + }, + } satisfies Event, + }]); + assertEquals(projection.lines[0]?.id, "compaction-runtime"); + + projection = projector.append([{ + eventId: "clear-snapshot", + event: snapshotEvent("/repo"), + }]); + assertEquals(projection.compaction, null); + assertEquals(projection.lines, []); + + const restore = snapshotEvent("/repo") as Extract< + Event, + { event: "snapshot" } + >; + restore.data.in_flight = { + blocks: [], + compaction: { + phase: "committing", + started_at_ms: 2_000, + trigger: "request_threshold", + }, + }; + projection = projector.append([{ + eventId: "restore-snapshot", + event: restore, + }]); + assertEquals(projection.lines.length, 1); + assertEquals(projection.lines[0].id, "compaction-runtime"); + assertEquals(projection.lines[0].compaction?.phase, "committing"); + assertEquals(projection.compaction?.started_at_ms, 2_000); +}); + +Deno.test("snapshot does not guess between multiple compaction services", () => { + const snapshot = snapshotEvent("/repo") as Extract< + Event, + { event: "snapshot" } + >; + snapshot.data.in_flight = { + blocks: [], + compaction: { + phase: "summarizing", + started_at_ms: 1_000, + trigger: "manual", + }, + }; + snapshot.data.internal_workers = ["compactor-a", "compactor-b"].map( + (sessionId) => ({ + worker: { + session_id: sessionId, + name: "Compaction", + parent_session_id: "parent-session", + kind: { service: { kind: "compaction" } }, + }, + revision: 1, + session: canonicalSession([]), + status: "running" as const, + in_flight: { + blocks: [{ + kind: "tool_call" as const, + id: `read-${sessionId}`, + name: "ReadEntry", + args: "{}", + state: "pending" as const, + }], + }, + internal_workers: [], + }), + ); + + const projection = projectConsole([{ eventId: "snapshot", event: snapshot }]); + + assertEquals(projection.lines.length, 1); + assertEquals( + projection.lines[0].compaction?.internalWorkerSessionId, + undefined, + ); + assertEquals(projection.lines[0].compaction?.activity, []); + assertEquals(consoleWorkerViews(projection).length, 1); +}); + +Deno.test("live compaction keeps its first service binding until terminal progress", () => { + const workerA = { + session_id: "compactor-a", + name: "Compaction A", + parent_session_id: "parent-session", + kind: { service: { kind: "compaction" } }, + } as const; + const workerB = { + ...workerA, + session_id: "compactor-b", + name: "Compaction B", + } as const; + const projection = projectConsole([ + { + eventId: "compaction-progress", + event: { + event: "compaction_progress", + data: { + compaction: { + phase: "summarizing", + started_at_ms: 1_000, + trigger: "manual", + }, + }, + } satisfies Event, + }, + { + eventId: "compactor-a-read", + event: { + event: "internal_worker", + data: { + worker: workerA, + revision: 1, + event: { + event: "tool_call_start", + data: { id: "read-a", name: "ReadEntry" }, + }, + }, + } satisfies Event, + }, + { + eventId: "compactor-b-search", + event: { + event: "internal_worker", + data: { + worker: workerB, + revision: 1, + event: { + event: "tool_call_start", + data: { id: "search-b", name: "SearchSessionEntries" }, + }, + }, + } satisfies Event, + }, + ]); + + assertEquals( + projection.lines[0].compaction?.internalWorkerSessionId, + workerA.session_id, + ); + assertEquals(projection.lines[0].compaction?.activity, [ + "ReadEntry — pending", + ]); +}); + +Deno.test("compaction summary candidate stays private", () => { 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-progress", + event: { + event: "compaction_progress", + data: { + compaction: { + phase: "summarizing", + started_at_ms: 1_000, + trigger: "manual", + }, + }, + } satisfies Event, }, { eventId: "compaction-tool", @@ -1150,15 +1378,115 @@ Deno.test("compaction service activity stays nested in one lifecycle item", () = ]); 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"); + assertEquals(projection.lines[0].id, "compaction-runtime"); + assertEquals(projection.lines[0].compaction?.activity, [ + "write_summary — running", + ]); + assert( + !JSON.stringify(projection.lines[0]).includes("draft candidate"), + "uncommitted summary content is not projected into the parent Console", + ); assert( consoleWorkerViews(projection).length === 1, "service is not a selectable SubWorker pane", ); }); +Deno.test("runtime compaction progress nests service activity and clears it", () => { + const worker = { + session_id: "runtime-compactor-session", + name: "Compaction", + parent_session_id: "parent-session", + kind: { service: { kind: "compaction" } }, + } as const; + const projector = createConsoleProjector(); + let projection = projector.append([{ + eventId: "compaction-progress", + event: { + event: "compaction_progress", + data: { + compaction: { + phase: "summarizing", + started_at_ms: 1_000, + trigger: "request_threshold", + }, + }, + } satisfies Event, + }, { + eventId: "compaction-read", + event: { + event: "internal_worker", + data: { + worker, + revision: 1, + event: { + event: "tool_call_done", + data: { + id: "call-read", + name: "ReadEntry", + arguments: JSON.stringify({ entry_ref: "E1" }), + }, + }, + }, + } satisfies Event, + }, { + eventId: "compaction-mark", + event: { + event: "internal_worker", + data: { + worker, + revision: 2, + event: { + event: "tool_call_done", + data: { + id: "call-mark", + name: "mark_read_required", + arguments: JSON.stringify({ path: "src/lib.rs" }), + }, + }, + }, + } satisfies Event, + }]); + + assertEquals(projection.lines.length, 1); + assertEquals(projection.lines[0].id, "compaction-runtime"); + assertEquals( + projection.lines[0].compaction?.internalWorkerSessionId, + worker.session_id, + ); + assertEquals(projection.lines[0].compaction?.activity, [ + "ReadEntry — running", + "mark_read_required — running", + ]); + assert( + consoleWorkerViews(projection).length === 1, + "service activity remains nested instead of becoming a selectable Worker view", + ); + + projection = projector.append([{ + eventId: "compaction-service-removed", + event: { + event: "internal_worker_removed", + data: { worker, revision: 3 }, + } satisfies Event, + }]); + assertEquals(projection.internalWorkers, []); + assertEquals(projection.lines[0].compaction?.activity, [ + "ReadEntry — running", + "mark_read_required — running", + ]); + + projection = projector.append([{ + eventId: "compaction-clear", + event: { + event: "compaction_progress", + data: { compaction: null }, + } satisfies Event, + }]); + assertEquals(projection.compaction, null); + assertEquals(projection.lines, []); +}); + Deno.test("snapshot excludes storage-only compaction extension records", () => { const projection = projectConsole([{ eventId: "snapshot", @@ -1186,61 +1514,126 @@ Deno.test("snapshot excludes storage-only compaction extension records", () => { assertEquals(projection.lines.length, 0); }); -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, +Deno.test("segment rotation preserves compaction progress in either event order", () => { + const progress = { + event: "compaction_progress", + data: { + compaction: { + phase: "committing", + started_at_ms: 1_000, + trigger: "manual", + }, + }, + } satisfies Event; + const rotation = { + event: "segment_rotated", + data: { session: canonicalSession([]) }, + } satisfies Event; + + const progressThenRotation = projectConsole([ + { eventId: "progress-first", event: progress }, + { eventId: "rotation-second", event: rotation }, + ]); + assertEquals(progressThenRotation.lines.length, 1); + assertEquals(progressThenRotation.lines[0].id, "compaction-runtime"); + assertEquals(progressThenRotation.lines[0].compaction?.phase, "committing"); + + const rotationThenProgress = projectConsole([ + { eventId: "rotation-first", event: rotation }, + { eventId: "progress-second", event: progress }, + ]); + assertEquals(rotationThenProgress.lines.length, 1); + assertEquals(rotationThenProgress.lines[0].id, "compaction-runtime"); + assertEquals(rotationThenProgress.lines[0].compaction?.phase, "committing"); +}); + +Deno.test("back-to-back compactions do not retain prior activity", () => { + const worker = { + session_id: "first-compactor", + name: "Compaction", + parent_session_id: "parent-session", + kind: { service: { kind: "compaction" } }, } as const; + const projector = createConsoleProjector(); let projection = projector.append([ { - eventId: "compact-identity-1", + eventId: "first-progress", event: { - event: "user_message", - data: { segments: [{ kind: "text", content: "hello" }] }, - } satisfies Event, - }, - { - eventId: "compact-identity-2", - event: { event: "compact_start", data: { lifecycle: base } } satisfies Event, - }, - ]); - const userLine = projection.lines[0]; - - projection = projector.append([ - { - eventId: "compact-identity-3", - event: { - event: "compact_done", + event: "compaction_progress", data: { - lifecycle: { - ...base, - revision: 2, - state: "done", - ended_at_ms: 2_000, - summary: "accepted", - new_segment_id: "00000000-0000-0000-0000-000000000001", + compaction: { + phase: "summarizing", + started_at_ms: 1_000, + trigger: "manual", }, }, } satisfies Event, }, { - eventId: "compact-identity-stale", - event: { event: "compact_start", data: { lifecycle: base } } satisfies Event, + eventId: "first-read", + event: { + event: "internal_worker", + data: { + worker, + revision: 1, + event: { + event: "tool_call_done", + data: { id: "read-1", name: "ReadEntry", arguments: "{}" }, + }, + }, + } satisfies Event, + }, + { + eventId: "first-removed", + event: { + event: "internal_worker_removed", + data: { worker, revision: 2 }, + } satisfies Event, + }, + { + eventId: "first-clear", + event: { + event: "compaction_progress", + data: { compaction: null }, + } satisfies Event, + }, + { + eventId: "second-progress", + event: { + event: "compaction_progress", + data: { + compaction: { + phase: "preparing", + started_at_ms: 2_000, + trigger: "request_threshold", + }, + }, + } satisfies Event, }, ]); - assert(projection.lines[0] === userLine, "unrelated line retains identity"); - assertEquals(projection.lines[1].compaction?.state, "done"); - assertEquals(projection.lines[1].compaction?.summary, "accepted"); + assertEquals(projection.lines.length, 1); + assertEquals(projection.lines[0].compaction?.startedAtMs, 2_000); + assertEquals( + projection.lines[0].compaction?.internalWorkerSessionId, + undefined, + ); + assertEquals(projection.lines[0].compaction?.activity, []); + + projection = projector.append([{ + eventId: "second-commit", + event: { + event: "compaction_progress", + data: { + compaction: { + phase: "committing", + started_at_ms: 2_000, + trigger: "request_threshold", + }, + }, + } satisfies Event, + }]); + assertEquals(projection.lines[0].compaction?.activity, []); }); Deno.test("projectConsole keeps streaming tool call updates in the same Call block", () => { @@ -1491,7 +1884,10 @@ Deno.test("projectConsole hides lifecycle events and renders system items", () = const projection = projectConsole([ { eventId: "30", - event: { event: "worker_state", data: { snapshot: workerState("running") } } satisfies Event, + event: { + event: "worker_state", + data: { snapshot: workerState("running") }, + } satisfies Event, }, { eventId: "31", @@ -1629,7 +2025,9 @@ Deno.test("projectConsole renders snapshot entries and in-flight output", () => assertEquals(projection.status, "running"); assertEquals( projection.lines.map((line) => - `${line.kind}:${line.toolCallLabel ? `${line.toolCallLabel}\n${line.body}` : line.body}:${line.streaming}` + `${line.kind}:${ + line.toolCallLabel ? `${line.toolCallLabel}\n${line.body}` : line.body + }:${line.streaming}` ), [ "user:seed user:false", @@ -1872,7 +2270,8 @@ Deno.test("projectConsole relativizes known tool path displays from snapshot cwd ); assert( toolLines.some((line) => - line.toolCallLabel === "Write(out.txt)" && line.body.includes("Wrote out.txt") + line.toolCallLabel === "Write(out.txt)" && + line.body.includes("Wrote out.txt") ), "Write signature and known result path should be relative", ); @@ -2010,7 +2409,10 @@ Deno.test("console Worker views expose only direct Internal Workers", () => { kind: "sub_worker", }, revision: 1, - event: { event: "worker_state", data: { snapshot: workerState("running") } }, + event: { + event: "worker_state", + data: { snapshot: workerState("running") }, + }, }, }, }, @@ -2029,7 +2431,10 @@ Deno.test("console Worker views expose only direct Internal Workers", () => { kind: "sub_worker", }, revision: 1, - event: { event: "worker_state", data: { snapshot: workerState("idle") } }, + event: { + event: "worker_state", + data: { snapshot: workerState("idle") }, + }, }, }, }]); @@ -2121,7 +2526,10 @@ Deno.test("parent snapshot authoritatively replaces Internal Worker projections" kind: "sub_worker", }, revision: 1, - event: { event: "worker_state", data: { snapshot: workerState("running") } }, + event: { + event: "worker_state", + data: { snapshot: workerState("running") }, + }, }, }, }]); diff --git a/web/workspace/src/lib/workspace/console/model.ts b/web/workspace/src/lib/workspace/console/model.ts index e8e99008e..e267a7222 100644 --- a/web/workspace/src/lib/workspace/console/model.ts +++ b/web/workspace/src/lib/workspace/console/model.ts @@ -3,7 +3,6 @@ import type { CommandEvent, CommandSnapshot, CommandStreamSlice, - CompactionLifecycle, Event as ProtocolEvent, InFlightBlock, InFlightCompaction, @@ -73,14 +72,8 @@ export type ConsoleDiffLine = { export type ConsoleViewMode = "overview" | "normal"; export type ConsoleCompaction = { - id: string; - revision: number; - state: "running" | "done" | "failed" | "interrupted"; + phase?: InFlightCompaction["phase"]; startedAtMs: number; - endedAtMs?: number; - summary?: string; - candidate?: string; - error?: string; internalWorkerSessionId?: string; activity: string[]; }; @@ -514,7 +507,9 @@ function appendSnapshotCommands( commands: CommandSnapshot[], eventId: string, ): void { - commands.forEach((command) => upsertCommandSnapshot(projection, eventId, command)); + commands.forEach((command) => + upsertCommandSnapshot(projection, eventId, command) + ); } function upsertCommandSnapshot( @@ -631,7 +626,8 @@ function appendCommandStream( start_offset: endOffset - tail.length, end_offset: endOffset, content: tail, - truncated: existing.truncated || !contiguous || tail.length < combined.length || + truncated: existing.truncated || !contiguous || + tail.length < combined.length || startOffset > 0, }; } @@ -678,107 +674,115 @@ function projectInternalWorkerSnapshot( return { worker: snapshot.worker, revision: snapshot.revision, console }; } -function compactionCandidate( +const RUNTIME_COMPACTION_LINE_ID = "compaction-runtime"; + +function isCompactionService(worker: InternalWorkerRef): boolean { + return worker.kind !== "sub_worker" && + worker.kind.service.kind === "compaction"; +} + +function activeCompactionServiceSessionId( projection: ConsoleProjection, - sessionId: string | undefined, ): string | undefined { - if (!sessionId) return undefined; - const worker = projection.internalWorkers.find( - (candidate) => candidate.worker.session_id === sessionId, + const candidates = projection.internalWorkers.filter((worker) => + isCompactionService(worker.worker) ); - 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; - } + return candidates.length === 1 ? candidates[0]?.worker.session_id : undefined; } function compactionActivity( projection: ConsoleProjection, - sessionId: string | undefined + sessionId: string | undefined, ): string[] { if (!sessionId) return []; const worker = projection.internalWorkers.find( - (candidate) => candidate.worker.session_id === sessionId + (candidate) => candidate.worker.session_id === sessionId, ); if (!worker) return []; return worker.console.lines - .filter((line) => line.kind === "tool" || line.kind === "status" || line.kind === "error") + .filter((line) => line.kind === "tool") .slice(-12) .map((line) => - line.toolCall?.name === "write_summary" - ? `write_summary — ${line.toolCall.state}` + line.toolCall + ? `${line.toolCall.name} — ${line.toolCall.state}` : line.body || line.title ) - .filter((value, index, values) => value.length > 0 && values.indexOf(value) === index); + .filter((value, index, values) => + value.length > 0 && values.indexOf(value) === index + ); } function applyInFlightCompaction( projection: ConsoleProjection, progress: InFlightCompaction | null, ): ConsoleProjection { - return { ...projection, compaction: progress }; -} + const lineIndex = projection.lines.findIndex((line) => + line.id === RUNTIME_COMPACTION_LINE_ID + ); + const existing = lineIndex >= 0 + ? projection.lines[lineIndex]?.compaction + : undefined; + if (!progress) { + return { + ...projection, + compaction: null, + lines: projection.lines.filter((line) => + line.id !== RUNTIME_COMPACTION_LINE_ID + ), + }; + } -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, + const internalWorkerSessionId = existing?.internalWorkerSessionId ?? + activeCompactionServiceSessionId(projection); + const observedActivity = compactionActivity( + projection, internalWorkerSessionId, - activity: compactionActivity(projection, internalWorkerSessionId) + ); + const compaction: ConsoleCompaction = { + phase: progress.phase, + startedAtMs: progress.started_at_ms, + internalWorkerSessionId, + activity: observedActivity.length > 0 + ? observedActivity + : existing?.activity ?? [], }; const line: ConsoleLine = { - id: lineId, + id: RUNTIME_COMPACTION_LINE_ID, source: "event", - kind: compaction.state === "failed" ? "error" : "status", + kind: "status", title: "Compaction", - body: compaction.summary ?? compaction.error ?? "", - streaming: compaction.state === "running", - error: compaction.state === "failed", - compaction + body: "", + streaming: true, + compaction, }; - const index = projection.lines.findIndex((candidate) => candidate.id === lineId); const lines = [...projection.lines]; - if (index >= 0) lines[index] = line; + if (lineIndex >= 0) lines[lineIndex] = line; else lines.push(line); - return { ...projection, lines }; + return { ...projection, compaction: progress, lines }; } function refreshCompactionActivity( projection: ConsoleProjection, - sessionId: string + worker: InternalWorkerRef, ): ConsoleProjection { let changed = false; const lines = projection.lines.map((line) => { - if (line.compaction?.internalWorkerSessionId !== sessionId) return line; + const compaction = line.compaction; + if (!compaction) return line; + const internalWorkerSessionId = compaction.internalWorkerSessionId ?? + (line.id === RUNTIME_COMPACTION_LINE_ID && isCompactionService(worker) && + activeCompactionServiceSessionId(projection) === worker.session_id + ? worker.session_id + : undefined); + if (internalWorkerSessionId !== worker.session_id) return line; changed = true; return { ...line, compaction: { - ...line.compaction, - candidate: compactionCandidate(projection, sessionId), - activity: compactionActivity(projection, sessionId) - } + ...compaction, + internalWorkerSessionId, + activity: compactionActivity(projection, internalWorkerSessionId), + }, }; }); return changed ? { ...projection, lines } : projection; @@ -938,26 +942,12 @@ export function applyProtocolEvent( projectInternalWorkerSnapshot(worker, envelope.eventId, next.cwd) ); next.removedInternalWorkers = {}; - if (event.data.in_flight?.compaction) { - const withCompaction = applyInFlightCompaction( - next, - event.data.in_flight.compaction, - ); - next.lines = withCompaction.lines; - next.compaction = withCompaction.compaction; - } - 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), - }; - } - } + const withCompaction = applyInFlightCompaction( + next, + event.data.in_flight?.compaction ?? null, + ); + next.lines = withCompaction.lines; + next.compaction = withCompaction.compaction; applyWorkerStateSnapshot(next, event.data.state); break; } @@ -991,7 +981,7 @@ export function applyProtocolEvent( }; if (existingIndex >= 0) next.internalWorkers[existingIndex] = updated; else next.internalWorkers.push(updated); - return refreshCompactionActivity(next, event.data.worker.session_id); + return refreshCompactionActivity(next, event.data.worker); } case "internal_worker_removed": { const existingIndex = next.internalWorkers.findIndex((worker) => @@ -1017,12 +1007,15 @@ export function applyProtocolEvent( break; case "segment_rotated": { const retainedErrors = next.lines.filter((line) => line.kind === "error"); + const retainedCompaction = next.lines.filter((line) => + line.id === RUNTIME_COMPACTION_LINE_ID + ); const segment = snapshotProjectionFromSession( envelope.eventId, event.data.session, next.cwd, ); - next.lines = [...segment.lines, ...retainedErrors]; + next.lines = [...segment.lines, ...retainedErrors, ...retainedCompaction]; next.tasks = segment.tasks; next.taskNextId = segment.taskNextId; break; @@ -1062,7 +1055,9 @@ export function applyProtocolEvent( case "compact_start": case "compact_done": case "compact_failed": - return applyCompactionLifecycle(next, event.data.lifecycle); + // Durable lifecycle events are retained only as historical protocol + // compatibility. Runtime progress is the sole Console block authority. + break; case "shutdown": next.status = "shutdown"; break; @@ -1432,16 +1427,25 @@ function toolCallSignature(toolCall: ToolCallView): string { return `Read(${readPath(toolCall)})`; case "Write": case "Edit": { - const path = displayPath(stringField(args, "file_path") ?? "?", toolCall.cwd); + const path = displayPath( + stringField(args, "file_path") ?? "?", + toolCall.cwd, + ); return `${toolCall.name}(${path})`; } case "Glob": - return `Glob(${stringField(args, "pattern") ?? genericCallArguments(toolCall)})`; + return `Glob(${ + stringField(args, "pattern") ?? genericCallArguments(toolCall) + })`; case "Grep": - return `Grep(${stringField(args, "pattern") ?? genericCallArguments(toolCall)})`; + return `Grep(${ + stringField(args, "pattern") ?? genericCallArguments(toolCall) + })`; case "Bash": { const command = stringField(args, "command"); - return `Bash(${command ? `$ ${singleLine(command)}` : genericCallArguments(toolCall)})`; + return `Bash(${ + command ? `$ ${singleLine(command)}` : genericCallArguments(toolCall) + })`; } default: return `${toolCall.name}(${genericCallArguments(toolCall)})`; @@ -1462,7 +1466,9 @@ function singleLine(value: string): string { } function toolCallStatus(toolCall: ToolCallView): string { - return toolCall.name === "Bash" ? commandStateSuffix(toolCall) : stateSuffix(toolCall.state); + return toolCall.name === "Bash" + ? commandStateSuffix(toolCall) + : stateSuffix(toolCall.state); } function aggregateReadToolLines(lines: ConsoleLine[]): ConsoleLine[] { @@ -1661,7 +1667,9 @@ function commandStateSuffix(toolCall: ToolCallView): string { : `completed (exit ${command.exit_code})`; } if (command.status === "failed") { - return command.exit_code === null ? "failed" : `failed (exit ${command.exit_code})`; + return command.exit_code === null + ? "failed" + : `failed (exit ${command.exit_code})`; } if (command.status === "timed_out") return "timed out"; if (command.status === "cancelled") return "cancelled"; @@ -1679,7 +1687,9 @@ function commandTiming(command?: CommandSnapshot): string | undefined { 0, command.last_output_at_ms - command.started_at_ms, ); - return `elapsed ${durationLabel(elapsed)} · last output at +${durationLabel(lastOutputElapsed)}`; + return `elapsed ${durationLabel(elapsed)} · last output at +${ + durationLabel(lastOutputElapsed) + }`; } function durationLabel(milliseconds: number): string { @@ -1687,7 +1697,9 @@ function durationLabel(milliseconds: number): string { return `${(milliseconds / 1000).toFixed(milliseconds < 10_000 ? 1 : 0)}s`; } -function renderLiveCommandOutput(command?: CommandSnapshot): string | undefined { +function renderLiveCommandOutput( + command?: CommandSnapshot, +): string | undefined { if (!command) return undefined; const stdout = compactLines([ command.stdout.truncated ? "[… earlier stdout omitted]" : undefined, @@ -1712,7 +1724,9 @@ function toolCallDetail(toolCall: ToolCallView): string { return compactLines([ `id: ${toolCall.id}`, `state: ${stateSuffix(toolCall.state)}`, - toolCall.command ? `command: ${commandTiming(toolCall.command)}` : undefined, + toolCall.command + ? `command: ${commandTiming(toolCall.command)}` + : undefined, toolCall.summary ? `summary: ${ normalizeKnownToolResult(toolCall.name, toolCall.summary, toolCall.cwd) @@ -2032,63 +2046,6 @@ function applySessionEntry( } } -function applyExtensionEntry( - projection: ConsoleProjection, - eventId: string, - entry: Record, -) { - if (entry["domain"] !== "yoi.compaction") { - return; - } - const payload = entry["payload"]; - 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"; - const state = stringField(payload, "state") || "running"; - const message = stringField(payload, "message") || - compactMessageForState(state, payload); - upsertStatusLine( - projection, - blockId, - eventId, - message, - state === "running", - state === "failed", - ); -} - -function compactMessageForState( - state: string, - payload: Record, -): string { - switch (state) { - case "done": - return "Compacted."; - case "failed": { - const error = stringField(payload, "error"); - return error ? `Compact failed: ${error}` : "Compact failed."; - } - default: - return "Compacting…"; - } -} - function applyLoggedUserInput( projection: ConsoleProjection, eventId: string, diff --git a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts index aef46f99d..037adf4a2 100644 --- a/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts +++ b/web/workspace/src/lib/workspace/console/worker-console.ui.test.ts @@ -560,6 +560,20 @@ Deno.test("Worker Console owns its narrower centered shell width", async () => { ); }); +Deno.test("Worker Console renders runtime compaction as one activity block", async () => { + const consoleLine = await Deno.readTextFile( + new URL("./ConsoleLineItem.svelte", import.meta.url), + ); + + assert( + consoleLine.includes("function compactionHeading") && + consoleLine.includes("Compacting ·") && + consoleLine.includes("{#each item.compaction.activity as activity}") && + consoleLine.includes("

  • {activity}
    • "), + "Compaction should render as one live block with nested service activity", + ); +}); + Deno.test("Worker Console overview activity summaries use 14px text", async () => { const consoleLine = await Deno.readTextFile( new URL("./ConsoleLineItem.svelte", import.meta.url),