diff --git a/crates/workdir/src/scope.rs b/crates/workdir/src/scope.rs index 7eadad4e9..abd7d1024 100644 --- a/crates/workdir/src/scope.rs +++ b/crates/workdir/src/scope.rs @@ -103,10 +103,20 @@ impl WorkdirToolBroker { #[cfg(test)] command_start_gate: Mutex::new(None), }); + let event_forwarder = forward_owned_command_events( + authority.source.subscribe_command_events(), + authority.owned_commands.clone(), + authority.pending_command_events.clone(), + authority.starting_tool_calls.clone(), + authority.forwarded_starts.clone(), + authority.forwarded_terminals.clone(), + authority.command_events.clone(), + ) + .map(|handle| Arc::new(Mutex::new(Some(handle)))); Self { session: authority.clone(), authority, - event_forwarder: None, + event_forwarder, } } @@ -1179,11 +1189,7 @@ impl WorkdirSession for ScopedWorkdirSession { { return None; } - if self.scope.is_none() { - self.source.subscribe_command_events() - } else { - Some(self.command_events.subscribe()) - } + Some(self.command_events.subscribe()) } fn command_snapshot(&self) -> Vec { @@ -1193,9 +1199,6 @@ impl WorkdirSession for ScopedWorkdirSession { { return Vec::new(); } - if self.scope.is_none() { - return self.source.command_snapshot(); - } let owned = self .owned_commands .lock() @@ -1768,11 +1771,12 @@ mod tests { } #[tokio::test] - async fn scoped_commands_use_child_cwd_and_do_not_leak_between_siblings() { + async fn scoped_commands_use_child_cwd_and_do_not_leak_to_parent_or_siblings() { let root = TempDir::new().unwrap(); fs::create_dir_all(root.path().join("one")).unwrap(); fs::create_dir_all(root.path().join("two")).unwrap(); let parent = session(root.path()); + let mut parent_events = parent.subscribe_command_events().unwrap(); let first = parent .scope(request("one", WorkdirToolScopePermission::Write)) .await @@ -1799,6 +1803,11 @@ mod tests { first_events.recv().await.unwrap(), CommandEvent::Started { .. } )); + assert!(matches!( + tokio::time::timeout(std::time::Duration::from_millis(50), parent_events.recv()).await, + Err(_) + )); + assert!(parent.command_snapshot().is_empty()); assert!(matches!( tokio::time::timeout(std::time::Duration::from_millis(50), second_events.recv()).await, Err(_) diff --git a/web/workspace/src/lib/workspace/console/model.test.ts b/web/workspace/src/lib/workspace/console/model.test.ts index 69914f007..139372b9d 100644 --- a/web/workspace/src/lib/workspace/console/model.test.ts +++ b/web/workspace/src/lib/workspace/console/model.test.ts @@ -740,6 +740,121 @@ Deno.test("projectConsole streams distinct Bash stdout and stderr through termin assertEquals(line.error, true); }); +Deno.test("projectConsole reports command telemetry without its tool call", () => { + const projection = projectConsole([ + { + eventId: "command-only-started", + event: { + event: "command", + data: { + event: { + kind: "started", + command_id: "command-only-1", + tool_call_id: "command-only-call", + observed_at_ms: 1000, + }, + }, + } satisfies Event, + }, + { + eventId: "command-only-output", + event: { + event: "command", + data: { + event: { + kind: "output", + command_id: "command-only-1", + stream: "stdout", + start_offset: 0, + end_offset: 6, + content: "ready\n", + observed_at_ms: 1100, + }, + }, + } satisfies Event, + }, + { + eventId: "command-only-terminal", + event: { + event: "command", + data: { + event: { + kind: "terminal", + command_id: "command-only-1", + status: "completed", + exit_code: 0, + stdout_end_offset: 6, + stderr_end_offset: 0, + observed_at_ms: 1200, + }, + }, + } satisfies Event, + }, + ]); + + const toolLines = projection.lines.filter((line) => line.kind === "tool"); + assertEquals(toolLines.length, 0); + const errorLines = projection.lines.filter((line) => line.kind === "error"); + assertEquals(errorLines.length, 1); + assertEquals( + errorLines[0].id, + "command-correlation-error-command-only-1", + ); + assertEquals(errorLines[0].title, "Command telemetry error"); + assertEquals( + errorLines[0].body, + "Command started telemetry for `command-only-1` could not be correlated with tool call `command-only-call`.", + ); + assertEquals(errorLines[0].error, true); +}); + +Deno.test("snapshot reports command telemetry without a tool call id", () => { + const snapshot = snapshotEvent("/repo"); + if (snapshot.event !== "snapshot") { + throw new Error("snapshot fixture expected"); + } + snapshot.data.in_flight = { + blocks: [], + commands: [{ + command_id: "command-without-call", + tool_call_id: null, + status: "running", + started_at_ms: 1000, + observed_at_ms: 1000, + last_output_at_ms: null, + stdout: { + start_offset: 0, + end_offset: 0, + content: "", + truncated: false, + }, + stderr: { + start_offset: 0, + end_offset: 0, + content: "", + truncated: false, + }, + exit_code: null, + }], + }; + + const projection = projectConsole([{ + eventId: "snapshot-command-error", + event: snapshot, + }]); + + assertEquals( + projection.lines.filter((line) => line.kind === "tool").length, + 0, + ); + const errorLines = projection.lines.filter((line) => line.kind === "error"); + assertEquals(errorLines.length, 1); + assertEquals( + errorLines[0].body, + "Command snapshot telemetry for `command-without-call` could not be correlated with a tool call.", + ); +}); + Deno.test("snapshot restores bounded in-flight Bash command output", () => { const snapshot = snapshotEvent("/repo"); if (snapshot.event !== "snapshot") { diff --git a/web/workspace/src/lib/workspace/console/model.ts b/web/workspace/src/lib/workspace/console/model.ts index e267a7222..08757ef7b 100644 --- a/web/workspace/src/lib/workspace/console/model.ts +++ b/web/workspace/src/lib/workspace/console/model.ts @@ -516,15 +516,34 @@ function upsertCommandSnapshot( projection: ConsoleProjection, eventId: string, command: CommandSnapshot, + source: "snapshot" | CommandEvent["kind"] = "snapshot", ): void { - const toolCallId = command.tool_call_id ?? `command:${command.command_id}`; + const toolCallId = command.tool_call_id; + if (toolCallId === null) { + reportCommandCorrelationError( + projection, + eventId, + command.command_id, + source, + toolCallId, + ); + return; + } const existingIndex = findToolCallLineIndex(projection, toolCallId); - const existing = existingIndex >= 0 - ? projection.lines[existingIndex].toolCall - : undefined; + if (existingIndex < 0) { + reportCommandCorrelationError( + projection, + eventId, + command.command_id, + source, + toolCallId, + ); + return; + } + const existing = projection.lines[existingIndex].toolCall!; upsertToolCall(projection, eventId, toolCallId, { - name: existing?.name ?? "Bash", - state: existing?.state ?? "running", + name: existing.name, + state: existing.state, command, }); } @@ -545,7 +564,7 @@ function applyCommandEvent( stdout: emptyCommandStream(), stderr: emptyCommandStream(), exit_code: null, - }); + }, "started"); return; } @@ -553,20 +572,12 @@ function applyCommandEvent( line.toolCall?.command?.command_id === event.command_id ); if (index < 0) { - if (event.kind === "output") { - const stream = commandStreamFromEvent(event); - upsertCommandSnapshot(projection, eventId, { - command_id: event.command_id, - tool_call_id: null, - status: "running", - started_at_ms: event.observed_at_ms, - observed_at_ms: event.observed_at_ms, - last_output_at_ms: event.observed_at_ms, - stdout: event.stream === "stdout" ? stream : emptyCommandStream(), - stderr: event.stream === "stderr" ? stream : emptyCommandStream(), - exit_code: null, - }); - } + reportCommandCorrelationError( + projection, + eventId, + event.command_id, + event.kind, + ); return; } @@ -577,7 +588,7 @@ function applyCommandEvent( status: event.status, exit_code: event.exit_code, observed_at_ms: event.observed_at_ms, - }); + }, "terminal"); return; } const updatedStream = appendCommandStream( @@ -592,6 +603,30 @@ function applyCommandEvent( last_output_at_ms: event.observed_at_ms, stdout: event.stream === "stdout" ? updatedStream : existing.stdout, stderr: event.stream === "stderr" ? updatedStream : existing.stderr, + }, "output"); +} + +function reportCommandCorrelationError( + projection: ConsoleProjection, + eventId: string, + commandId: string, + source: "snapshot" | CommandEvent["kind"], + toolCallId?: string | null, +): void { + const id = `command-correlation-error-${commandId}`; + if (projection.lines.some((item) => item.id === id)) return; + const target = toolCallId ? `tool call \`${toolCallId}\`` : "a tool call"; + projection.lines.push({ + ...line( + eventId, + "error", + "Command telemetry error", + `Command ${source} telemetry for \`${commandId}\` could not be correlated with ${target}.`, + undefined, + false, + true, + ), + id, }); } @@ -599,17 +634,6 @@ function emptyCommandStream(): CommandStreamSlice { return { start_offset: 0, end_offset: 0, content: "", truncated: false }; } -function commandStreamFromEvent( - event: Extract, -): CommandStreamSlice { - return appendCommandStream( - emptyCommandStream(), - event.start_offset, - event.end_offset, - event.content, - ); -} - function appendCommandStream( existing: CommandStreamSlice, startOffset: number,