fix: isolate worker command telemetry
This commit is contained in:
+19
-10
@@ -103,10 +103,20 @@ impl WorkdirToolBroker {
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
command_start_gate: Mutex::new(None),
|
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 {
|
Self {
|
||||||
session: authority.clone(),
|
session: authority.clone(),
|
||||||
authority,
|
authority,
|
||||||
event_forwarder: None,
|
event_forwarder,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1179,11 +1189,7 @@ impl WorkdirSession for ScopedWorkdirSession {
|
|||||||
{
|
{
|
||||||
return None;
|
return None;
|
||||||
}
|
}
|
||||||
if self.scope.is_none() {
|
Some(self.command_events.subscribe())
|
||||||
self.source.subscribe_command_events()
|
|
||||||
} else {
|
|
||||||
Some(self.command_events.subscribe())
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
fn command_snapshot(&self) -> Vec<CommandSnapshot> {
|
fn command_snapshot(&self) -> Vec<CommandSnapshot> {
|
||||||
@@ -1193,9 +1199,6 @@ impl WorkdirSession for ScopedWorkdirSession {
|
|||||||
{
|
{
|
||||||
return Vec::new();
|
return Vec::new();
|
||||||
}
|
}
|
||||||
if self.scope.is_none() {
|
|
||||||
return self.source.command_snapshot();
|
|
||||||
}
|
|
||||||
let owned = self
|
let owned = self
|
||||||
.owned_commands
|
.owned_commands
|
||||||
.lock()
|
.lock()
|
||||||
@@ -1768,11 +1771,12 @@ mod tests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[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();
|
let root = TempDir::new().unwrap();
|
||||||
fs::create_dir_all(root.path().join("one")).unwrap();
|
fs::create_dir_all(root.path().join("one")).unwrap();
|
||||||
fs::create_dir_all(root.path().join("two")).unwrap();
|
fs::create_dir_all(root.path().join("two")).unwrap();
|
||||||
let parent = session(root.path());
|
let parent = session(root.path());
|
||||||
|
let mut parent_events = parent.subscribe_command_events().unwrap();
|
||||||
let first = parent
|
let first = parent
|
||||||
.scope(request("one", WorkdirToolScopePermission::Write))
|
.scope(request("one", WorkdirToolScopePermission::Write))
|
||||||
.await
|
.await
|
||||||
@@ -1799,6 +1803,11 @@ mod tests {
|
|||||||
first_events.recv().await.unwrap(),
|
first_events.recv().await.unwrap(),
|
||||||
CommandEvent::Started { .. }
|
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!(
|
assert!(matches!(
|
||||||
tokio::time::timeout(std::time::Duration::from_millis(50), second_events.recv()).await,
|
tokio::time::timeout(std::time::Duration::from_millis(50), second_events.recv()).await,
|
||||||
Err(_)
|
Err(_)
|
||||||
|
|||||||
@@ -740,6 +740,121 @@ Deno.test("projectConsole streams distinct Bash stdout and stderr through termin
|
|||||||
assertEquals(line.error, true);
|
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", () => {
|
Deno.test("snapshot restores bounded in-flight Bash command output", () => {
|
||||||
const snapshot = snapshotEvent("/repo");
|
const snapshot = snapshotEvent("/repo");
|
||||||
if (snapshot.event !== "snapshot") {
|
if (snapshot.event !== "snapshot") {
|
||||||
|
|||||||
@@ -516,15 +516,34 @@ function upsertCommandSnapshot(
|
|||||||
projection: ConsoleProjection,
|
projection: ConsoleProjection,
|
||||||
eventId: string,
|
eventId: string,
|
||||||
command: CommandSnapshot,
|
command: CommandSnapshot,
|
||||||
|
source: "snapshot" | CommandEvent["kind"] = "snapshot",
|
||||||
): void {
|
): 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 existingIndex = findToolCallLineIndex(projection, toolCallId);
|
||||||
const existing = existingIndex >= 0
|
if (existingIndex < 0) {
|
||||||
? projection.lines[existingIndex].toolCall
|
reportCommandCorrelationError(
|
||||||
: undefined;
|
projection,
|
||||||
|
eventId,
|
||||||
|
command.command_id,
|
||||||
|
source,
|
||||||
|
toolCallId,
|
||||||
|
);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
const existing = projection.lines[existingIndex].toolCall!;
|
||||||
upsertToolCall(projection, eventId, toolCallId, {
|
upsertToolCall(projection, eventId, toolCallId, {
|
||||||
name: existing?.name ?? "Bash",
|
name: existing.name,
|
||||||
state: existing?.state ?? "running",
|
state: existing.state,
|
||||||
command,
|
command,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
@@ -545,7 +564,7 @@ function applyCommandEvent(
|
|||||||
stdout: emptyCommandStream(),
|
stdout: emptyCommandStream(),
|
||||||
stderr: emptyCommandStream(),
|
stderr: emptyCommandStream(),
|
||||||
exit_code: null,
|
exit_code: null,
|
||||||
});
|
}, "started");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -553,20 +572,12 @@ function applyCommandEvent(
|
|||||||
line.toolCall?.command?.command_id === event.command_id
|
line.toolCall?.command?.command_id === event.command_id
|
||||||
);
|
);
|
||||||
if (index < 0) {
|
if (index < 0) {
|
||||||
if (event.kind === "output") {
|
reportCommandCorrelationError(
|
||||||
const stream = commandStreamFromEvent(event);
|
projection,
|
||||||
upsertCommandSnapshot(projection, eventId, {
|
eventId,
|
||||||
command_id: event.command_id,
|
event.command_id,
|
||||||
tool_call_id: null,
|
event.kind,
|
||||||
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,
|
|
||||||
});
|
|
||||||
}
|
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -577,7 +588,7 @@ function applyCommandEvent(
|
|||||||
status: event.status,
|
status: event.status,
|
||||||
exit_code: event.exit_code,
|
exit_code: event.exit_code,
|
||||||
observed_at_ms: event.observed_at_ms,
|
observed_at_ms: event.observed_at_ms,
|
||||||
});
|
}, "terminal");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
const updatedStream = appendCommandStream(
|
const updatedStream = appendCommandStream(
|
||||||
@@ -592,6 +603,30 @@ function applyCommandEvent(
|
|||||||
last_output_at_ms: event.observed_at_ms,
|
last_output_at_ms: event.observed_at_ms,
|
||||||
stdout: event.stream === "stdout" ? updatedStream : existing.stdout,
|
stdout: event.stream === "stdout" ? updatedStream : existing.stdout,
|
||||||
stderr: event.stream === "stderr" ? updatedStream : existing.stderr,
|
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 };
|
return { start_offset: 0, end_offset: 0, content: "", truncated: false };
|
||||||
}
|
}
|
||||||
|
|
||||||
function commandStreamFromEvent(
|
|
||||||
event: Extract<CommandEvent, { kind: "output" }>,
|
|
||||||
): CommandStreamSlice {
|
|
||||||
return appendCommandStream(
|
|
||||||
emptyCommandStream(),
|
|
||||||
event.start_offset,
|
|
||||||
event.end_offset,
|
|
||||||
event.content,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
|
|
||||||
function appendCommandStream(
|
function appendCommandStream(
|
||||||
existing: CommandStreamSlice,
|
existing: CommandStreamSlice,
|
||||||
startOffset: number,
|
startOffset: number,
|
||||||
|
|||||||
Reference in New Issue
Block a user