From 14e2691a58d279fa4dff0606d9ba8f84da83c88b Mon Sep 17 00:00:00 2001 From: Michael Suchacz <203725896+ibetitsmike@users.noreply.github.com> Date: Fri, 28 Aug 2026 08:40:16 +0000 Subject: [PATCH 1/3] =?UTF-8?q?fix(workflows):=20=F0=9F=A4=96=20render=20k?= =?UTF-8?q?ernel-nested=20workflow=20runs=20as=20live=20cards?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Tools/Shared/NestedToolRenderer.tsx | 5 + .../Tools/Shared/NestedToolsContainer.tsx | 1 + .../Tools/Shared/codeExecutionTypes.ts | 4 + .../Tools/Shared/getToolComponent.test.ts | 12 +++ .../features/Tools/Shared/getToolComponent.ts | 8 +- .../Tools/WorkflowRunToolCall.test.tsx | 78 +++++++++++++++ .../features/Tools/WorkflowRunToolCall.tsx | 72 +++++++++++--- src/browser/stores/WorkspaceStore.ts | 25 +++-- src/common/orpc/schemas/message.ts | 5 + src/common/types/message.ts | 2 + src/common/utils/tools/kernelBoundedMarker.ts | 21 ++++ src/node/services/ptc/quickjsRuntime.ts | 49 +++++++++- src/node/services/ptc/runtime.ts | 25 +++++ src/node/services/ptc/toolBridge.test.ts | 1 + src/node/services/ptc/toolBridge.ts | 19 +++- src/node/services/ptc/types.test.ts | 51 ++++++++++ src/node/services/ptc/types.ts | 38 +++++++ src/node/services/streamManager.test.ts | 56 +++++++++++ src/node/services/streamManager.ts | 42 ++++++++ .../services/tools/code_execution.test.ts | 98 +++++++++++++++++++ .../services/workflows/WorkflowRunner.test.ts | 3 + 21 files changed, 586 insertions(+), 29 deletions(-) create mode 100644 src/common/utils/tools/kernelBoundedMarker.ts diff --git a/src/browser/features/Tools/Shared/NestedToolRenderer.tsx b/src/browser/features/Tools/Shared/NestedToolRenderer.tsx index 044915dbd05..f63b8c1b184 100644 --- a/src/browser/features/Tools/Shared/NestedToolRenderer.tsx +++ b/src/browser/features/Tools/Shared/NestedToolRenderer.tsx @@ -3,6 +3,7 @@ import type { ToolStatus } from "./toolUtils"; import { getToolComponent } from "./getToolComponent"; import { HookOutputDisplay, extractHookDuration, extractHookOutput } from "./HookOutputDisplay"; import { ToolNameProvider } from "../../Messages/ToolNameContext"; +import type { WorkflowRunToolAttachment } from "@/common/orpc/schemas/message"; interface NestedToolRendererProps { toolName: string; @@ -12,6 +13,8 @@ interface NestedToolRendererProps { workspaceId?: string; toolCallId?: string; toolCallTimestamp?: number; + /** Persisted run identity for nested workflow tool calls; only the workflow card consumes it. */ + workflowRunHint?: WorkflowRunToolAttachment; } /** @@ -26,6 +29,7 @@ export const NestedToolRenderer: React.FC = ({ workspaceId, toolCallId, toolCallTimestamp, + workflowRunHint, }) => { const ToolComponent = getToolComponent(toolName, input); const hookOutput = extractHookOutput(output); @@ -43,6 +47,7 @@ export const NestedToolRenderer: React.FC = ({ workspaceId={workspaceId} toolCallId={toolCallId} toolCallTimestamp={toolCallTimestamp} + workflowRunHint={workflowRunHint} /> {hookOutput && } diff --git a/src/browser/features/Tools/Shared/NestedToolsContainer.tsx b/src/browser/features/Tools/Shared/NestedToolsContainer.tsx index 3844b850aa0..df5f6b73fcb 100644 --- a/src/browser/features/Tools/Shared/NestedToolsContainer.tsx +++ b/src/browser/features/Tools/Shared/NestedToolsContainer.tsx @@ -42,6 +42,7 @@ export const NestedToolsContainer: React.FC = ({ workspaceId={workspaceId} toolCallId={call.toolCallId} toolCallTimestamp={call.timestamp ?? toolCallTimestamp} + workflowRunHint={call.workflowRun} /> ); })} diff --git a/src/browser/features/Tools/Shared/codeExecutionTypes.ts b/src/browser/features/Tools/Shared/codeExecutionTypes.ts index 96584be5a62..7b935b35c91 100644 --- a/src/browser/features/Tools/Shared/codeExecutionTypes.ts +++ b/src/browser/features/Tools/Shared/codeExecutionTypes.ts @@ -1,3 +1,5 @@ +import type { WorkflowRunToolAttachment } from "@/common/orpc/schemas/message"; + export type { CodeExecutionConsoleRecord as ConsoleRecord, CodeExecutionResult, @@ -13,4 +15,6 @@ export interface NestedToolCall { state: "input-available" | "output-available" | "output-redacted"; failed?: boolean; timestamp?: number; + /** Durable run identity persisted for nested workflow tool calls. */ + workflowRun?: WorkflowRunToolAttachment; } diff --git a/src/browser/features/Tools/Shared/getToolComponent.test.ts b/src/browser/features/Tools/Shared/getToolComponent.test.ts index 5233206e50f..baf2d44e7b4 100644 --- a/src/browser/features/Tools/Shared/getToolComponent.test.ts +++ b/src/browser/features/Tools/Shared/getToolComponent.test.ts @@ -38,6 +38,18 @@ describe("getToolComponent", () => { expect(component).toBe(WorkflowRunToolCall); }); + test("routes kernel-bounded workflow_run args to the workflow card", () => { + // Kernel-nested calls with oversized launch args arrive as a marker; the + // card renders from the attached durable run instead of raw JSON. + const marker = { __kernelBounded: true, bytes: 18_457, preview: '{"script_path":"skill…' }; + expect(getToolComponent("workflow_run", marker)).toBe(WorkflowRunToolCall); + expect( + getToolComponent("workflow_run", { ...marker, script_path: "skill://demo/workflow.js" }) + ).toBe(WorkflowRunToolCall); + // Other tools keep the generic fallback for bounded args. + expect(getToolComponent("bash", marker)).toBe(GenericToolCall); + }); + test("returns WorkflowResumeToolCall for workflow_resume", () => { const component = getToolComponent("workflow_resume", { run_id: "wfr_123" }); expect(component).toBe(WorkflowResumeToolCall); diff --git a/src/browser/features/Tools/Shared/getToolComponent.ts b/src/browser/features/Tools/Shared/getToolComponent.ts index ac38aef0381..6623c45010e 100644 --- a/src/browser/features/Tools/Shared/getToolComponent.ts +++ b/src/browser/features/Tools/Shared/getToolComponent.ts @@ -229,8 +229,14 @@ const TOOL_REGISTRY: Record = { schema: TaskWorkspaceLifecycleToolArgsSchema, }, workflow_run: { + // Kernel-nested calls can arrive with a __kernelBounded args marker + // (oversized launch args); the card still renders from the attached + // durable run, so don't bounce those to the generic JSON renderer. component: WorkflowRunToolCall, - schema: TOOL_DEFINITIONS.workflow_run.schema, + schema: z.union([ + TOOL_DEFINITIONS.workflow_run.schema, + z.object({ __kernelBounded: z.literal(true), script_path: z.string().nullish() }), + ]), }, workflow_resume: { component: WorkflowResumeToolCall, diff --git a/src/browser/features/Tools/WorkflowRunToolCall.test.tsx b/src/browser/features/Tools/WorkflowRunToolCall.test.tsx index bd63dc6b7da..f90ad0f0441 100644 --- a/src/browser/features/Tools/WorkflowRunToolCall.test.tsx +++ b/src/browser/features/Tools/WorkflowRunToolCall.test.tsx @@ -2136,6 +2136,84 @@ describe("WorkflowRunToolCall", () => { ); }); + test("renders a workflow card from kernel-bounded args via the attached run hint", async () => { + // A kernel-nested workflow_run whose launch args were replaced by a + // __kernelBounded marker must still render the live card: identity comes + // from the attached run hint, and the marker offers nothing for the + // heuristic name+args discovery to match (listRuns must stay untouched). + const attachedRun = createWorkflowRunForExpansionTest({ id: "wfr_kernel", status: "running" }); + const listRuns = mock(async () => []); + const getRun = mock(async () => attachedRun); + + const view = render( + + + + + + + + ); + + // The workflow name renders in both the header and the script card. + expect(view.getAllByText("deep-research").length).toBeGreaterThan(0); + expect(view.getByText("wfr_kernel")).toBeTruthy(); + expect(listRuns).not.toHaveBeenCalled(); + await waitFor(() => + expect(getRun).toHaveBeenCalledWith({ workspaceId: TEST_WORKSPACE_ID, runId: "wfr_kernel" }) + ); + }); + + test("recovers run identity from a kernel-bounded result marker", async () => { + // With both args and result bounded, the retained runId on the result + // marker is the only identity; expanding the card must refetch the + // durable run by that id. + const completedRun = createWorkflowRunForExpansionTest({ + id: "wfr_bounded_result", + status: "completed", + }); + const getRun = mock(async () => completedRun); + + const view = render( + + + + + + + + ); + + fireEvent.click(getWorkflowHeader(view)); + await waitFor(() => + expect(getRun).toHaveBeenCalledWith({ + workspaceId: TEST_WORKSPACE_ID, + runId: "wfr_bounded_result", + }) + ); + await waitFor(() => expect(view.getAllByText("deep-research").length).toBeGreaterThan(0)); + }); + test("uses resume result status when the attachment run snapshot is stale", async () => { const staleRun = { ...createWorkflowRunForExpansionTest({ id: "wfr_resume_stale", status: "running" }), diff --git a/src/browser/features/Tools/WorkflowRunToolCall.tsx b/src/browser/features/Tools/WorkflowRunToolCall.tsx index 55104fd7beb..b2c54650d4f 100644 --- a/src/browser/features/Tools/WorkflowRunToolCall.tsx +++ b/src/browser/features/Tools/WorkflowRunToolCall.tsx @@ -72,12 +72,32 @@ import { type WorkflowToolLiveRunState, } from "@/browser/stores/WorkspaceStore"; import { workflowScriptMatchesPath } from "@/browser/utils/workflowRunScriptPaths"; +import { + isKernelBoundedMarker, + type KernelBoundedMarker, +} from "@/common/utils/tools/kernelBoundedMarker"; import { MarkdownRenderer } from "../Messages/MarkdownRenderer"; -type WorkflowRunToolDisplayArgs = +type WorkflowRunToolLaunchArgs = | WorkflowRunToolArgs | (Omit & { script_path?: string; name: string }); +/** + * Kernel-nested calls can arrive with their launch args replaced by a + * __kernelBounded marker; capture retains script_path when it fits (see + * retainPersistenceCriticalArgsFields). Run identity then comes from the + * workflow-run-attached hint or the result's retained runId. + */ +type KernelBoundedWorkflowRunArgs = KernelBoundedMarker & { script_path?: string | null }; + +type WorkflowRunToolDisplayArgs = WorkflowRunToolLaunchArgs | KernelBoundedWorkflowRunArgs; + +function getWorkflowRunLaunchArgs( + args: WorkflowRunToolDisplayArgs +): WorkflowRunToolLaunchArgs | null { + return isKernelBoundedMarker(args) ? null : args; +} + interface WorkflowRunToolCallProps { args: WorkflowRunToolDisplayArgs; result?: WorkflowRunToolResult; @@ -204,7 +224,21 @@ async function updateWorkflowRunFromAction(input: { function isWorkflowRunSuccessResult( value: WorkflowRunToolResult | undefined ): value is WorkflowRunToolSuccessResult { - return value != null && !isToolErrorResult(value); + return value != null && !isToolErrorResult(value) && !isKernelBoundedMarker(value); +} + +/** RunId/status retained on a kernel-bounded result marker (see retainWorkflowResultIdentityFields). */ +function getKernelBoundedResultIdentity( + result: unknown +): { runId: string; status?: string } | null { + if (!isKernelBoundedMarker(result)) { + return null; + } + const { runId, status } = result as { runId?: unknown; status?: unknown }; + if (typeof runId !== "string" || runId.length === 0) { + return null; + } + return { runId, ...(typeof status === "string" ? { status } : {}) }; } // Schema-shaped workflow agent reports carry structuredOutput only; their placeholder @@ -1133,12 +1167,16 @@ function getWorkflowRunDisplayName(args: WorkflowRunToolDisplayArgs): string { if (scriptPath.length > 0) { return scriptPath; } + if (isKernelBoundedMarker(args)) { + // Placeholder until the attached durable run supplies the real name. + return "workflow"; + } return args.script_source != null ? "inline workflow" : ""; } function workflowRunMatchesLaunchArgs( run: WorkflowRunRecord, - args: WorkflowRunToolDisplayArgs + args: WorkflowRunToolLaunchArgs ): boolean { const invocationArgs = args.args ?? {}; if (!workflowArgsEqual(run.args ?? {}, invocationArgs)) { @@ -1154,7 +1192,7 @@ function workflowRunMatchesLaunchArgs( function findForegroundWorkflowRun(input: { runs: readonly WorkflowRunRecord[]; - args: WorkflowRunToolDisplayArgs; + args: WorkflowRunToolLaunchArgs; startedAt?: number; }): WorkflowRunRecord | null { const candidates = input.runs.filter( @@ -1241,6 +1279,10 @@ export const WorkflowRunToolCall: React.FC = ({ const registerCommandSource = commandRegistry?.registerSource; const errorResult = isToolErrorResult(result) ? result : null; const successResult = isWorkflowRunSuccessResult(result) ? result : null; + // Kernel-nested calls: launch args may be a bounded marker (no launch args + // to match against) and the result marker may retain only runId/status. + const launchArgs = getWorkflowRunLaunchArgs(args); + const boundedResultIdentity = getKernelBoundedResultIdentity(result); const liveWorkflowRunHint = useWorkflowToolLiveRun(workspaceId, toolCallId); const workflowRunHint = explicitWorkflowRunHint ?? liveWorkflowRunHint; const [refreshedRun, setRefreshedRun] = useState(null); @@ -1260,7 +1302,8 @@ export const WorkflowRunToolCall: React.FC = ({ const selectedRun = selectWorkflowRunSnapshot({ // knownRunId (workflow_resume) and workflowRunHint provide exact identities before any // result arrives, which also disables the heuristic name+args foreground discovery below. - runId: successResult?.runId ?? knownRunId ?? workflowRunHint?.runId, + runId: + successResult?.runId ?? boundedResultIdentity?.runId ?? knownRunId ?? workflowRunHint?.runId, baseRun, refreshedRun, }); @@ -1274,13 +1317,13 @@ export const WorkflowRunToolCall: React.FC = ({ const displayStatus = successResult?.run == null && successResult?.status != null && !hasRefreshedRunSnapshot ? successResult.status - : (run?.status ?? successResult?.status ?? status); + : (run?.status ?? successResult?.status ?? boundedResultIdentity?.status ?? status); const parentRunActive = isWorkflowDisplayStatusActive(displayStatus); const displayEventSequence = getLatestWorkflowEventSequence(run); const resultValue = successResult?.result ?? getLatestResultEvent(run); const reportMarkdown = getReportMarkdown(resultValue); const structuredOutput = getStructuredOutput(resultValue); - const invocationArgs = run?.args ?? args.args ?? {}; + const invocationArgs = run?.args ?? launchArgs?.args ?? {}; const events = run?.events ?? []; const displayRows = getWorkflowDisplayRows(events); const headerStatus = toToolStatus(displayStatus); @@ -1335,7 +1378,8 @@ export const WorkflowRunToolCall: React.FC = ({ // A uniquely discovered foreground run is actionable before the blocking tool call returns. const discoveredForegroundRunConfirmed = status === "executing" && - args.run_in_background !== true && + launchArgs != null && + launchArgs.run_in_background !== true && workspaceId != null && refreshedRun != null && runId === refreshedRun.id && @@ -1353,6 +1397,7 @@ export const WorkflowRunToolCall: React.FC = ({ (workflowRunHint.run?.workspaceId == null || workflowRunHint.run.workspaceId === workspaceId); const runIdentityConfirmed = successResult?.runId != null || + boundedResultIdentity?.runId != null || baseRun?.id != null || discoveredForegroundRunConfirmed || discoveredKnownRunConfirmed || @@ -1496,7 +1541,10 @@ export const WorkflowRunToolCall: React.FC = ({ workspaceId == null || runId != null || status !== "executing" || - args.run_in_background === true + // Bounded marker args carry nothing to match against; identity arrives + // via the workflow-run-attached hint instead. + launchArgs == null || + launchArgs.run_in_background === true ) { return; } @@ -1507,7 +1555,7 @@ export const WorkflowRunToolCall: React.FC = ({ const runs = await apiState.api.workflows.listRuns({ workspaceId }); const foregroundRun = findForegroundWorkflowRun({ runs, - args, + args: launchArgs, startedAt: discoveryFreshnessBound, }); if (!ignore && foregroundRun != null) { @@ -1526,9 +1574,9 @@ export const WorkflowRunToolCall: React.FC = ({ ignore = true; window.clearInterval(interval); }; - }, [apiState?.api, args, runId, discoveryFreshnessBound, status, workspaceId]); + }, [apiState?.api, launchArgs, runId, discoveryFreshnessBound, status, workspaceId]); - const exactDiscoveryRunId = knownRunId ?? workflowRunHint?.runId; + const exactDiscoveryRunId = knownRunId ?? workflowRunHint?.runId ?? boundedResultIdentity?.runId; useEffect(() => { // workflow_resume args and workflowRunHint carry exact run identity, so fetch by ID while diff --git a/src/browser/stores/WorkspaceStore.ts b/src/browser/stores/WorkspaceStore.ts index 324f8a400f4..10357f7f5f4 100644 --- a/src/browser/stores/WorkspaceStore.ts +++ b/src/browser/stores/WorkspaceStore.ts @@ -1939,17 +1939,26 @@ export class WorkspaceStore { const activeBashToolCallIds = new Set(); const activeAdvisorToolCallIds = new Set(); const activeWorkflowToolCallIds = new Set(); + const collectToolCallId = (toolName: string, toolCallId: string) => { + if (toolName === "bash") { + activeBashToolCallIds.add(toolCallId); + } + if (toolName === "advisor") { + activeAdvisorToolCallIds.add(toolCallId); + } + if (isWorkflowRunEmittingToolName(toolName)) { + activeWorkflowToolCallIds.add(toolCallId); + } + }; for (const msg of aggregator.getDisplayedMessages()) { if (msg.type !== "tool") continue; - if (msg.toolName === "bash") { - activeBashToolCallIds.add(msg.toolCallId); - } - if (msg.toolName === "advisor") { - activeAdvisorToolCallIds.add(msg.toolCallId); - } - if (isWorkflowRunEmittingToolName(msg.toolName)) { - activeWorkflowToolCallIds.add(msg.toolCallId); + collectToolCallId(msg.toolName, msg.toolCallId); + // Kernel-nested calls (code_execution) render inside the parent card but + // key their live state by their own nested ids; without collecting them + // the sweep would prune a nested workflow/bash card's live state mid-run. + for (const nested of msg.nestedCalls ?? []) { + collectToolCallId(nested.toolName, nested.toolCallId); } } diff --git a/src/common/orpc/schemas/message.ts b/src/common/orpc/schemas/message.ts index e71091f6c48..56b9da49c4e 100644 --- a/src/common/orpc/schemas/message.ts +++ b/src/common/orpc/schemas/message.ts @@ -70,6 +70,11 @@ export const NestedToolCallSchema = z.object({ state: z.enum(["input-available", "output-available", "output-redacted"]), failed: z.boolean().optional(), timestamp: z.number().optional(), + // Durable run identity for nested workflow_run/workflow_resume calls, set + // by streamManager when the workflow-run-attached event targets a nested + // call. Kernel bounding can replace the nested args/result with a marker, + // so without this the transcript card loses the run after reload. + workflowRun: WorkflowRunToolAttachmentSchema.optional(), }); export type NestedToolCall = z.infer; diff --git a/src/common/types/message.ts b/src/common/types/message.ts index 9b16234a106..66beb1dfef2 100644 --- a/src/common/types/message.ts +++ b/src/common/types/message.ts @@ -1217,6 +1217,8 @@ export type DisplayedMessage = state: "input-available" | "output-available" | "output-redacted"; failed?: boolean; timestamp?: number; + /** Durable run identity for nested workflow tool calls (see NestedToolCallSchema). */ + workflowRun?: MuxToolPart["workflowRun"]; }>; } | { diff --git a/src/common/utils/tools/kernelBoundedMarker.ts b/src/common/utils/tools/kernelBoundedMarker.ts new file mode 100644 index 00000000000..7274ed847cd --- /dev/null +++ b/src/common/utils/tools/kernelBoundedMarker.ts @@ -0,0 +1,21 @@ +/** + * Marker shape produced by the code_execution kernel when a nested tool + * call's args or result exceed the kernel record caps (see + * QuickJSRuntime.boundCapture). Retained attribution fields (a workflow_run's + * script_path, a workflow result's runId/status, a file edit's path) may ride + * alongside the marker fields; the marker fields themselves always win on key + * collisions at capture time. + */ +export interface KernelBoundedMarker { + __kernelBounded: true; + bytes?: number; + preview?: string; +} + +export function isKernelBoundedMarker(value: unknown): value is KernelBoundedMarker { + return ( + typeof value === "object" && + value !== null && + (value as { __kernelBounded?: unknown }).__kernelBounded === true + ); +} diff --git a/src/node/services/ptc/quickjsRuntime.ts b/src/node/services/ptc/quickjsRuntime.ts index c2e4cd54589..b21aa5ccc00 100644 --- a/src/node/services/ptc/quickjsRuntime.ts +++ b/src/node/services/ptc/quickjsRuntime.ts @@ -180,6 +180,10 @@ export class QuickJSRuntime implements IJSRuntime { private disposed = false; private eventHandler?: (event: PTCEvent) => void; private abortController?: AbortController; + /** See IJSRuntime.takeActiveHostCallId: set synchronously right before a + * registered host function is invoked, consumed by the tool bridge inside + * that same synchronous window. */ + private activeHostCallId?: string; private abortRequested = false; // Track abort requests before eval() starts private limits: RuntimeLimits = {}; private consoleSetup = false; @@ -341,6 +345,9 @@ export class QuickJSRuntime implements IJSRuntime { }); try { + // Hand the record callId to the dispatched function (tool bridge) + // through the synchronous window contract; see takeActiveHostCallId. + this.activeHostCallId = callId; const result = await fn(...args); const endTime = Date.now(); const duration_ms = endTime - startTime; @@ -723,17 +730,41 @@ export class QuickJSRuntime implements IJSRuntime { } // Budget exhausted: fall back to normal bounding — oversized results // become honest-size markers, small results still pass inline. - return QuickJSRuntime.preserveSuccessBit( - this.boundCapture(sanitized, this.kernelRecordBounds.resultCapBytes), - retained + return this.applyRetainedResultFields( + QuickJSRuntime.preserveSuccessBit( + this.boundCapture(sanitized, this.kernelRecordBounds.resultCapBytes), + retained + ), + toolName, + sanitized ); } - return QuickJSRuntime.preserveSuccessBit( - this.boundCapture(sanitized, this.kernelRecordBounds.resultCapBytes), + return this.applyRetainedResultFields( + QuickJSRuntime.preserveSuccessBit( + this.boundCapture(sanitized, this.kernelRecordBounds.resultCapBytes), + sanitized + ), + toolName, sanitized ); } + /** Merge captureResultRetained identity fields under a __kernelBounded + * result marker (marker fields win on collisions, mirroring the + * captureArgsRetained merge in boundCaptureArgs). No-op for results that + * survived bounding inline. */ + private applyRetainedResultFields(bounded: unknown, toolName: string, source: unknown): unknown { + if ( + typeof bounded !== "object" || + bounded === null || + (bounded as { __kernelBounded?: boolean }).__kernelBounded !== true + ) { + return bounded; + } + const retained = this.kernelRecordBounds?.captureResultRetained?.(toolName, source); + return retained !== undefined ? { ...retained, ...bounded } : bounded; + } + /** A boolean success bit is preserved onto EVERY __kernelBounded result * marker (r29 — not just the retained-budget-exhausted branch): compaction * folds result.success===false into the compact ok bit, and a FAILED call @@ -952,6 +983,8 @@ export class QuickJSRuntime implements IJSRuntime { }); try { + // Same synchronous-window handoff as registerFunction. + this.activeHostCallId = callId; const result = await fn(...args); const endTime = Date.now(); const duration_ms = endTime - startTime; @@ -1202,6 +1235,12 @@ export class QuickJSRuntime implements IJSRuntime { this.abortController?.abort(); } + takeActiveHostCallId(): string | undefined { + const callId = this.activeHostCallId; + this.activeHostCallId = undefined; + return callId; + } + getAbortSignal(): AbortSignal | undefined { return this.abortController?.signal; } diff --git a/src/node/services/ptc/runtime.ts b/src/node/services/ptc/runtime.ts index 6aa0ddffa45..14e9597fa21 100644 --- a/src/node/services/ptc/runtime.ts +++ b/src/node/services/ptc/runtime.ts @@ -157,6 +157,19 @@ export interface IJSRuntime extends Disposable { */ getAbortSignal(): AbortSignal | undefined; + /** + * Consume the record callId of the host call currently being dispatched. + * The runtime sets it synchronously immediately before invoking a + * registered host function, and the tool bridge must read it as its FIRST + * synchronous operation (before any await): the value is only coherent + * inside that same synchronous window. Consuming (clear-on-read) prevents a + * stale id from leaking into a host function invoked outside the runtime + * dispatch path. Bridges use it as the nested tool call's toolCallId so + * UI events emitted by the tool (workflow-run-attached, task-created, live + * bash output) land on the SAME id the transcript's nested record carries. + */ + takeActiveHostCallId(): string | undefined; + /** * Clean up resources. Called automatically with `using` declarations. */ @@ -186,6 +199,18 @@ export interface KernelRecordBounds { * retained diff. Marker fields win on key collisions. */ captureArgsRetained?: (toolName: string, args: unknown) => Record | undefined; + /** + * Identity fields merged onto a __kernelBounded RESULT marker when bounding + * replaces the result of a record (see retainWorkflowResultIdentityFields): + * an oversized workflow_run/workflow_resume result would otherwise lose the + * runId and status the transcript card needs to re-render the durable run + * after reload. Marker fields win on key collisions, so retained fields can + * never spoof __kernelBounded/bytes/preview. + */ + captureResultRetained?: ( + toolName: string, + result: unknown + ) => Record | undefined; } /** diff --git a/src/node/services/ptc/toolBridge.test.ts b/src/node/services/ptc/toolBridge.test.ts index 3dbd7f2e324..748fed60b4f 100644 --- a/src/node/services/ptc/toolBridge.test.ts +++ b/src/node/services/ptc/toolBridge.test.ts @@ -42,6 +42,7 @@ function createMockRuntime(overrides: Partial = {}): IJSRuntime { onEvent: mock((_handler: (event: PTCEvent) => void) => undefined), abort: mock(() => undefined), getAbortSignal: mock(() => undefined), + takeActiveHostCallId: mock(() => undefined), dispose: mock(() => undefined), [Symbol.dispose]: mock(() => undefined), ...overrides, diff --git a/src/node/services/ptc/toolBridge.ts b/src/node/services/ptc/toolBridge.ts index 2ab16c14d66..878f785b5f5 100644 --- a/src/node/services/ptc/toolBridge.ts +++ b/src/node/services/ptc/toolBridge.ts @@ -35,6 +35,7 @@ import { import { retainExemptKernelRecordResult, retainPersistenceCriticalArgsFields, + retainWorkflowResultIdentityFields, sanitizeMediaRecordCapture, } from "./types"; @@ -279,6 +280,7 @@ export class ToolBridge { resultCapBytes: RESULT_HANDLE_OFFLOAD_THRESHOLD_BYTES, captureRetained: retainExemptKernelRecordResult, captureArgsRetained: retainPersistenceCriticalArgsFields, + captureResultRetained: retainWorkflowResultIdentityFields, } : undefined ); @@ -304,6 +306,13 @@ export class ToolBridge { const toolName = name; xumObj[name] = async (args: unknown) => { + // MUST be the first operation: the runtime hands over the nested + // record's callId through a synchronous window (see + // IJSRuntime.takeActiveHostCallId). Using it as the execute + // toolCallId lets tool-emitted UI events (workflow-run-attached, + // task-created, live bash output) target the transcript's nested + // tool call instead of an id no rendered card carries. + const bridgedToolCallId = runtime.takeActiveHostCallId() ?? syntheticToolCallId(toolName); // Defense in depth: re-check the grant at call time so a stale or // mutated bridge can never invoke a non-granted tool. if (!isBridgeToolGranted(this.grants, toolName)) { @@ -329,10 +338,10 @@ export class ToolBridge { } // Execute tool with full options (toolCallId and messages are required by type - // but not used by most tools - generate synthetic values for sandbox context) + // but not used by most tools; messages stay synthetic for sandbox context) const result: unknown = await boundTool.execute!(validatedArgs, { abortSignal, - toolCallId: syntheticToolCallId(toolName), + toolCallId: bridgedToolCallId, messages: [], context: undefined, }); @@ -371,6 +380,10 @@ export class ToolBridge { const taskTool = this.bridgeableTools.get("task"); if (taskTool !== undefined) { xumObj.task_spawn = async (args: unknown) => { + // First operation, same synchronous-window contract as the regular + // bridged tools above. + const bridgedToolCallId = + runtime.takeActiveHostCallId() ?? syntheticToolCallId("task_spawn"); // task_spawn is subject to the same grant as task (defense in depth, // mirroring the per-call re-check on regular bridged tools). if (!isBridgeToolGranted(this.grants, "task")) { @@ -394,7 +407,7 @@ export class ToolBridge { } const result: unknown = await taskTool.execute!(validatedArgs, { abortSignal, - toolCallId: syntheticToolCallId("task_spawn"), + toolCallId: bridgedToolCallId, messages: [], context: undefined, }); diff --git a/src/node/services/ptc/types.test.ts b/src/node/services/ptc/types.test.ts index aa02d5d63a4..28b226978a6 100644 --- a/src/node/services/ptc/types.test.ts +++ b/src/node/services/ptc/types.test.ts @@ -10,6 +10,7 @@ import { createCaptureSanitizerBudget, retainExemptKernelRecordResult, retainPersistenceCriticalArgsFields, + retainWorkflowResultIdentityFields, sanitizeCapturedMediaValue, sanitizeMediaRecordCapture, SANITIZER_BUDGET_EXHAUSTED_STUB, @@ -559,4 +560,54 @@ describe("retainPersistenceCriticalArgsFields", () => { retainPersistenceCriticalArgsFields("file_edit_insert", { path: "\uD800".repeat(1_000) }) ).toBeUndefined(); }); + + it("retains script_path for workflow_run launch args", () => { + expect( + retainPersistenceCriticalArgsFields("workflow_run", { + script_path: "skill://demo/workflow.js", + args: { problem: "x".repeat(5_000) }, + }) + ).toEqual({ script_path: "skill://demo/workflow.js" }); + // Inline-source launches have no path to retain. + expect( + retainPersistenceCriticalArgsFields("workflow_run", { + script_source: "export default function workflow() {}", + }) + ).toBeUndefined(); + expect( + retainPersistenceCriticalArgsFields("workflow_run", { script_path: "p".repeat(5_000) }) + ).toBeUndefined(); + }); +}); + +describe("retainWorkflowResultIdentityFields", () => { + it("retains runId and validated status for workflow tools", () => { + expect( + retainWorkflowResultIdentityFields("workflow_run", { + status: "completed", + runId: "wfr_1", + result: { reportMarkdown: "big" }, + }) + ).toEqual({ runId: "wfr_1", status: "completed" }); + expect( + retainWorkflowResultIdentityFields("workflow_resume", { status: "running", runId: "wfr_2" }) + ).toEqual({ runId: "wfr_2", status: "running" }); + }); + + it("drops non-workflow tools, missing runIds, and invalid statuses", () => { + expect( + retainWorkflowResultIdentityFields("bash", { runId: "wfr_1", status: "completed" }) + ).toBeUndefined(); + expect( + retainWorkflowResultIdentityFields("workflow_run", { status: "completed" }) + ).toBeUndefined(); + expect(retainWorkflowResultIdentityFields("workflow_run", "not-an-object")).toBeUndefined(); + // A guest-spoofable status is validated; the runId still survives alone. + expect( + retainWorkflowResultIdentityFields("workflow_run", { runId: "wfr_1", status: "exploded" }) + ).toEqual({ runId: "wfr_1" }); + expect( + retainWorkflowResultIdentityFields("workflow_run", { runId: "r".repeat(5_000) }) + ).toBeUndefined(); + }); }); diff --git a/src/node/services/ptc/types.ts b/src/node/services/ptc/types.ts index bc0907e5941..70394c839e4 100644 --- a/src/node/services/ptc/types.ts +++ b/src/node/services/ptc/types.ts @@ -11,6 +11,7 @@ import type { CodeExecutionToolCallRecord, } from "@/common/types/codeExecution"; import { FILE_EDIT_TOOL_NAMES } from "@/common/types/tools"; +import { WorkflowRunStatusSchema } from "@/common/orpc/schemas/workflow"; import { isSupportedAttachmentMediaType } from "@/common/utils/attachments/supportedAttachmentMediaTypes"; import { getToolOutputUiOnly } from "@/common/utils/tools/toolOutputUiOnly"; import { extractToolFilePath } from "@/common/utils/tools/toolInputFilePath"; @@ -646,6 +647,19 @@ export function retainPersistenceCriticalArgsFields( toolName: string, args: unknown ): Record | undefined { + if (toolName === "workflow_run") { + // A workflow_run whose args exceed the kernel args cap (large inline + // script_source or invocation args) would otherwise lose script_path, + // leaving the transcript workflow card without a display name until the + // durable run loads. Same serialized-byte bound as file paths below. + const scriptPath = (args as { script_path?: unknown } | null | undefined)?.script_path; + if (typeof scriptPath !== "string" || scriptPath.length === 0) return undefined; + const scriptPathBytes = serializedJsonByteLength(scriptPath); + if (scriptPathBytes === undefined || scriptPathBytes > KERNEL_RETAINED_PATH_MAX_BYTES) { + return undefined; + } + return { script_path: scriptPath }; + } if (!isPersistenceCriticalRecordToolName(toolName)) return undefined; const path = extractToolFilePath(args); if (path === undefined) return undefined; @@ -660,6 +674,30 @@ export function retainPersistenceCriticalArgsFields( return { path }; } +/** + * Run-identity fields merged onto a __kernelBounded RESULT marker (see + * KernelRecordBounds.captureResultRetained): a workflow_run/workflow_resume + * result embedding a large run record or workflow output exceeds the kernel + * result cap, and a bare marker would strip the runId/status the transcript + * card needs to re-fetch the durable run after reload. Only the validated + * identity fields are preserved; the run record and workflow output stay + * bounded (never exempt workflow results from bounding: they can carry + * megabytes of delegated-agent output). + */ +export function retainWorkflowResultIdentityFields( + toolName: string, + result: unknown +): Record | undefined { + if (toolName !== "workflow_run" && toolName !== "workflow_resume") return undefined; + if (typeof result !== "object" || result === null) return undefined; + const { runId, status } = result as { runId?: unknown; status?: unknown }; + if (typeof runId !== "string" || runId.length === 0) return undefined; + const runIdBytes = serializedJsonByteLength(runId); + if (runIdBytes === undefined || runIdBytes > KERNEL_RETAINED_PATH_MAX_BYTES) return undefined; + const parsedStatus = WorkflowRunStatusSchema.safeParse(status); + return { runId, ...(parsedStatus.success ? { status: parsedStatus.data } : {}) }; +} + /** See isKernelRecordResultExempt (persistence-critical branch). */ export function isPersistenceCriticalRecordToolName(toolName: string): boolean { return ( diff --git a/src/node/services/streamManager.test.ts b/src/node/services/streamManager.test.ts index 52c7f4abc2e..abe496c4054 100644 --- a/src/node/services/streamManager.test.ts +++ b/src/node/services/streamManager.test.ts @@ -282,6 +282,62 @@ describe("StreamManager - workflow run attachments", () => { }); }); + test("persists workflow attachments onto nested kernel tool calls", async () => { + const streamManager = new StreamManager(historyService); + const workspaceId = "workflow-nested-attachment-workspace"; + const messageId = "workflow-nested-attachment-message"; + const timestamp = Date.now(); + const streamInfo = createStreamInfoForTests({ + messageId, + lastPartialWriteTime: timestamp, + parts: [ + { + type: "dynamic-tool", + toolCallId: "code-exec-1", + toolName: "code_execution", + input: { code: "mux.workflow_run({...})" }, + state: "input-available", + timestamp, + nestedCalls: [ + { + toolCallId: "nested-workflow-1", + toolName: "workflow_run", + // Kernel bounding replaced the launch args with a marker; the + // attachment is the only durable run identity for this call. + input: { __kernelBounded: true, bytes: 18_457, preview: "{…}" }, + state: "input-available", + timestamp, + }, + ], + }, + ], + }); + + getWorkspaceStreamsForTests(streamManager).set(workspaceId, streamInfo); + + const attached = await streamManager.attachWorkflowRunToToolCall({ + type: "workflow-run-attached", + workspaceId, + messageId, + toolCallId: "nested-workflow-1", + runId: "wfr_nested", + timestamp: timestamp + 1, + }); + + expect(attached).toBe(true); + const partial = await historyService.readPartial(workspaceId); + const part = partial?.parts[0]; + if (part?.type !== "dynamic-tool") { + throw new Error("Expected code_execution tool part in persisted partial"); + } + expect(part.nestedCalls?.[0]?.workflowRun).toEqual({ + runId: "wfr_nested", + timestamp: timestamp + 1, + }); + // The attachment landed on the nested record, not the pending map. + expect((streamInfo.pendingWorkflowRunAttachments as Map).size).toBe(0); + }); + test("persists workflow attachments that arrive before the tool part", async () => { const streamManager = new StreamManager(historyService); const workspaceId = "workflow-attachment-race-workspace"; diff --git a/src/node/services/streamManager.ts b/src/node/services/streamManager.ts index a25265be36b..5cafcfc1208 100644 --- a/src/node/services/streamManager.ts +++ b/src/node/services/streamManager.ts @@ -795,6 +795,18 @@ export class StreamManager extends EventEmitter { (part) => part.type === "dynamic-tool" && part.toolCallId === event.toolCallId ); if (partIndex === -1) { + // Kernel-launched workflows target a NESTED call persisted on a + // code_execution part (see emitNestedToolEvent), not a top-level part. + if ( + await this.attachWorkflowRunToNestedCall( + workspaceId, + streamInfo, + event.toolCallId, + attachment + ) + ) { + return true; + } (streamInfo.pendingWorkflowRunAttachments ??= new Map()).set(event.toolCallId, attachment); return true; } @@ -814,6 +826,36 @@ export class StreamManager extends EventEmitter { return true; } + /** + * Persist a workflow run attachment onto the nested call record it targets, + * so a kernel-launched run's identity survives reload even when kernel + * bounding replaced the nested args/result with a marker. The live event + * still reaches the frontend through the normal emit path. + */ + private async attachWorkflowRunToNestedCall( + workspaceId: WorkspaceId, + streamInfo: WorkspaceStreamInfo, + toolCallId: string, + attachment: WorkflowRunToolAttachment + ): Promise { + for (const part of streamInfo.parts) { + if (part.type !== "dynamic-tool") { + continue; + } + const parentPart = part as { nestedCalls?: NestedToolCall[] }; + const nestedCalls = parentPart.nestedCalls; + const nestedIndex = + nestedCalls?.findIndex((nested) => nested.toolCallId === toolCallId) ?? -1; + if (nestedCalls == null || nestedIndex === -1) { + continue; + } + nestedCalls[nestedIndex] = { ...nestedCalls[nestedIndex], workflowRun: attachment }; + await this.flushPartialWrite(workspaceId, streamInfo); + return true; + } + return false; + } + /** * Record on the dynamic-tool part when its execute() actually began running and notify * the UI. Returns false when the part has not landed in streamInfo.parts yet. diff --git a/src/node/services/tools/code_execution.test.ts b/src/node/services/tools/code_execution.test.ts index d2c44430f50..bef9955d6f9 100644 --- a/src/node/services/tools/code_execution.test.ts +++ b/src/node/services/tools/code_execution.test.ts @@ -1633,6 +1633,104 @@ describe("createCodeExecutionTool", () => { await host.disposeScope("ws-event-bound"); }); + it("threads the nested record callId into bridged execute options", async () => { + // The UI keys nested cards and live events (workflow-run-attached, + // task-created, live bash output) by the PTC record callId; execute() + // must observe the SAME id or those events target an id no rendered + // card carries. + using tmp = new DisposableTempDir("code-exec-callid"); + const host = new SandboxHostService(); + const executed: Array<{ toolCallId: string; tag: string }> = []; + const probeTool: Tool = { + description: "Probe tool", + inputSchema: z.object({ tag: z.string() }), + execute: (args, options) => { + executed.push({ toolCallId: options.toolCallId, tag: (args as { tag: string }).tag }); + return Promise.resolve({ success: true }); + }, + }; + const emitted: Array<{ + type?: string; + callId?: string; + toolName?: string; + args?: { tag?: string }; + }> = []; + const tool = await createCodeExecutionTool( + runtimeFactory, + new ToolBridge({ probe: probeTool }), + (event) => emitted.push(event as (typeof emitted)[number]), + persistentRunner(host, "ws-callid", tmp.path) + ); + + const result = (await tool.execute!( + { code: 'mux.probe({tag: "a"}); mux.probe({tag: "b"}); return true;' }, + mockToolCallOptions + )) as PTCExecutionResult; + expect(result.success).toBe(true); + + const starts = emitted.filter((e) => e.type === "tool-call-start" && e.toolName === "probe"); + expect(starts).toHaveLength(2); + expect(executed).toHaveLength(2); + for (const call of executed) { + const start = starts.find((e) => e.args?.tag === call.tag); + expect(start?.callId).toBe(call.toolCallId); + } + expect(executed[0].toolCallId).not.toBe(executed[1].toolCallId); + await host.disposeScope("ws-callid"); + }); + + it("retains workflow identity on kernel-bounded workflow_run args and results", async () => { + // Oversized workflow launches and results collapse to markers, but the + // transcript card needs script_path (display) and runId/status (durable + // run refetch) to render the live workflow instead of raw JSON. + using tmp = new DisposableTempDir("code-exec-wf-bound"); + const host = new SandboxHostService(); + const workflowTools: Record = { + workflow_run: createMockTool( + "workflow_run", + z.object({ script_path: z.string(), args: z.unknown() }), + () => ({ + status: "completed", + runId: "wfr_bounded", + result: { reportMarkdown: "r".repeat(64 * 1024) }, + }) + ), + }; + const emitted: Array<{ type?: string; toolName?: string; args?: unknown; result?: unknown }> = + []; + const tool = await createCodeExecutionTool( + runtimeFactory, + new ToolBridge(workflowTools), + (event) => emitted.push(event as (typeof emitted)[number]), + persistentRunner(host, "ws-wf-bound", tmp.path) + ); + + const result = (await tool.execute!( + { + code: 'const r = mux.workflow_run({script_path: "skill://demo/workflow.js", args: {problem: "p".repeat(4096)}}); return r.runId;', + }, + mockToolCallOptions + )) as PTCExecutionResult; + expect(result.success).toBe(true); + // The guest still sees the full result. + expect(result.result).toBe("wfr_bounded"); + + const end = emitted.find((e) => e.type === "tool-call-end" && e.toolName === "workflow_run"); + expect(end).toBeDefined(); + const argsMarker = end!.args as { __kernelBounded?: boolean; script_path?: string }; + expect(argsMarker.__kernelBounded).toBe(true); + expect(argsMarker.script_path).toBe("skill://demo/workflow.js"); + const resultMarker = end!.result as { + __kernelBounded?: boolean; + runId?: string; + status?: string; + }; + expect(resultMarker.__kernelBounded).toBe(true); + expect(resultMarker.runId).toBe("wfr_bounded"); + expect(resultMarker.status).toBe("completed"); + await host.disposeScope("ws-wf-bound"); + }); + it("bounds oversized nested-call args in compact records (no echo of kernel data)", async () => { using tmp = new DisposableTempDir("code-exec-offload"); const host = new SandboxHostService(); diff --git a/src/node/services/workflows/WorkflowRunner.test.ts b/src/node/services/workflows/WorkflowRunner.test.ts index ceb2035252c..dca50a8042b 100644 --- a/src/node/services/workflows/WorkflowRunner.test.ts +++ b/src/node/services/workflows/WorkflowRunner.test.ts @@ -3237,6 +3237,9 @@ describe("WorkflowRunner", () => { getAbortSignal() { return undefined; }, + takeActiveHostCallId() { + return undefined; + }, async eval() { evalSawLimits = limitsApplied; return { From 916ee813df1cb48365dcfde9c8ea543945632e71 Mon Sep 17 00:00:00 2001 From: Michael Suchacz <203725896+ibetitsmike@users.noreply.github.com> Date: Fri, 28 Aug 2026 11:45:53 +0000 Subject: [PATCH 2/3] =?UTF-8?q?fix(workflows):=20=F0=9F=A4=96=20bound=20ne?= =?UTF-8?q?sted=20attachments,=20buffer=20racing=20nested=20events,=20repl?= =?UTF-8?q?ay=20nested=20calls?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../StreamingMessageAggregator.test.ts | 20 ++ .../messages/StreamingMessageAggregator.ts | 7 + src/node/services/streamManager.test.ts | 200 ++++++++++++++++++ src/node/services/streamManager.ts | 142 ++++++++++++- 4 files changed, 367 insertions(+), 2 deletions(-) diff --git a/src/browser/utils/messages/StreamingMessageAggregator.test.ts b/src/browser/utils/messages/StreamingMessageAggregator.test.ts index 03a5a3d3a5c..2222108ff03 100644 --- a/src/browser/utils/messages/StreamingMessageAggregator.test.ts +++ b/src/browser/utils/messages/StreamingMessageAggregator.test.ts @@ -3795,6 +3795,26 @@ describe("StreamingMessageAggregator", () => { } }); + test("skips duplicate nested starts (reconnect replays re-emit them with the parent part)", () => { + const aggregator = createTestAggregator(); + startParentTool(aggregator); + for (let i = 0; i < 2; i++) { + startToolCall(aggregator, { + toolCallId: "nested-tool-1", + toolName: "workflow_run", + args: { script_path: "wf.js" }, + timestamp: 1100, + parentToolCallId: "parent-tool-1", + }); + } + + const toolMsg = parentToolMessage(aggregator); + if (toolMsg?.type !== "tool") { + throw new Error("Expected parent tool message"); + } + expect(toolMsg.nestedCalls).toHaveLength(1); + }); + test("updates nested call with output on tool-call-end with parentToolCallId", () => { const aggregator = createTestAggregator(); startParentTool(aggregator); diff --git a/src/browser/utils/messages/StreamingMessageAggregator.ts b/src/browser/utils/messages/StreamingMessageAggregator.ts index d1932332705..8a37344721c 100644 --- a/src/browser/utils/messages/StreamingMessageAggregator.ts +++ b/src/browser/utils/messages/StreamingMessageAggregator.ts @@ -2503,6 +2503,13 @@ export class StreamingMessageAggregator { if (parentPart) { // Initialize nestedCalls array if needed parentPart.nestedCalls ??= []; + // Reconnect replays re-emit nested starts for rows the renderer may + // already have (the parent part replays whenever any of its nested + // activity is newer than the cursor); skip duplicates like the + // top-level path below does. + if (parentPart.nestedCalls.some((nc) => nc.toolCallId === data.toolCallId)) { + return; + } parentPart.nestedCalls.push({ toolCallId: data.toolCallId, toolName: data.toolName, diff --git a/src/node/services/streamManager.test.ts b/src/node/services/streamManager.test.ts index abe496c4054..6ad591df6b9 100644 --- a/src/node/services/streamManager.test.ts +++ b/src/node/services/streamManager.test.ts @@ -12,6 +12,7 @@ import type { WorkflowRunAttachedEvent, } from "@/common/types/stream"; import type { MuxMessage } from "@/common/types/message"; +import type { WorkflowRunRecord } from "@/common/types/workflow"; import { Ok, Err } from "@/common/types/result"; import type { ToolPolicy } from "@/common/utils/tools/toolPolicy"; import type { ToolSearchStreamState } from "@/common/utils/tools/toolCatalog"; @@ -213,6 +214,7 @@ function createStreamInfoForTests( lastPartTimestamp: now, toolCompletionTimestamps: new Map(), pendingWorkflowRunAttachments: new Map(), + pendingNestedCalls: new Map(), pendingToolExecutionStarts: new Map(), model, metadataModel: overrides.metadataModel ?? model, @@ -237,6 +239,27 @@ function createStreamInfoForTests( }; } +function createWorkflowRunRecordForTests(runId: string, workspaceId: string): WorkflowRunRecord { + return { + id: runId, + workspaceId, + workflow: { + name: "deep-research", + description: "test workflow", + scope: "project", + executable: true, + }, + source: "export default function workflow() { return { reportMarkdown: 'x'.repeat(64) }; }", + sourceHash: "sha256:test", + args: {}, + status: "running", + createdAt: "2026-05-29T00:00:00.000Z", + updatedAt: "2026-05-29T00:00:01.000Z", + events: [], + steps: [], + }; +} + describe("StreamManager - workflow run attachments", () => { test("persists attached workflow run metadata to partial immediately", async () => { const streamManager = new StreamManager(historyService); @@ -321,6 +344,9 @@ describe("StreamManager - workflow run attachments", () => { messageId, toolCallId: "nested-workflow-1", runId: "wfr_nested", + // The live event carries the full run record (large source/args), + // exactly what kernel bounding keeps out of the nested record. + run: createWorkflowRunRecordForTests("wfr_nested", workspaceId), timestamp: timestamp + 1, }); @@ -330,6 +356,8 @@ describe("StreamManager - workflow run attachments", () => { if (part?.type !== "dynamic-tool") { throw new Error("Expected code_execution tool part in persisted partial"); } + // Identity only: persisting the run record (source, args) would bypass the + // kernel record caps via partial.json. expect(part.nestedCalls?.[0]?.workflowRun).toEqual({ runId: "wfr_nested", timestamp: timestamp + 1, @@ -415,6 +443,178 @@ describe("StreamManager - workflow run attachments", () => { }); }); +describe("StreamManager - nested kernel call race and replay", () => { + test("buffers nested events that beat the parent part and persists them (with run identity) on merge", async () => { + const streamManager = new StreamManager(historyService); + const workspaceId = "nested-race-workspace"; + const messageId = "nested-race-message"; + const timestamp = Date.now(); + const streamInfo = createStreamInfoForTests({ + messageId, + lastPartialWriteTime: timestamp, + parts: [], + }); + getWorkspaceStreamsForTests(streamManager).set(workspaceId, streamInfo); + + // execute() wins the race: nested start arrives before the parent part exists. + streamManager.emitNestedToolEvent(workspaceId, messageId, { + type: "tool-call-start", + callId: "nested-race-workflow", + toolName: "workflow_run", + args: { __kernelBounded: true, bytes: 18_457, preview: "{…}" }, + parentToolCallId: "code-exec-race", + startTime: timestamp, + }); + + // The workflow attachment lands while the nested record is still buffered. + const attached = await streamManager.attachWorkflowRunToToolCall({ + type: "workflow-run-attached", + workspaceId, + messageId, + toolCallId: "nested-race-workflow", + runId: "wfr_race_nested", + run: createWorkflowRunRecordForTests("wfr_race_nested", workspaceId), + timestamp: timestamp + 1, + }); + expect(attached).toBe(true); + // Nothing persisted yet: the parent part has not landed. + expect(await historyService.readPartial(workspaceId)).toBeNull(); + + const appendPartAndEmit = getPrivateMethodForTests< + ( + workspaceId: string, + streamInfo: Record, + part: CompletedMessagePart, + schedulePartialWrite?: boolean + ) => Promise + >(streamManager, "appendPartAndEmit"); + await appendPartAndEmit.call( + streamManager, + workspaceId, + streamInfo, + { + type: "dynamic-tool", + toolCallId: "code-exec-race", + toolName: "code_execution", + input: { code: "mux.workflow_run({...})" }, + state: "input-available", + timestamp: timestamp + 2, + }, + false + ); + + const partial = await historyService.readPartial(workspaceId); + const part = partial?.parts[0]; + if (part?.type !== "dynamic-tool") { + throw new Error("Expected code_execution tool part in persisted partial"); + } + expect(part.nestedCalls).toHaveLength(1); + expect(part.nestedCalls?.[0]?.toolCallId).toBe("nested-race-workflow"); + // Run identity only (no run record), same bound as the direct attach path. + expect(part.nestedCalls?.[0]?.workflowRun).toEqual({ + runId: "wfr_race_nested", + timestamp: timestamp + 1, + }); + // Both holding areas were consumed. + expect((streamInfo.pendingNestedCalls as Map).size).toBe(0); + expect((streamInfo.pendingWorkflowRunAttachments as Map).size).toBe(0); + }); + + test("replays persisted nested calls (start, attachment, end) with the parent part", async () => { + const streamManager = new StreamManager(historyService); + const workspaceId = "nested-replay-workspace"; + const messageId = "nested-replay-message"; + const timestamp = Date.now(); + const streamInfo = createStreamInfoForTests({ + messageId, + lastPartialWriteTime: timestamp, + parts: [ + { + type: "dynamic-tool", + toolCallId: "code-exec-replay", + toolName: "code_execution", + input: { code: "mux.workflow_run({...})" }, + state: "input-available", + timestamp, + nestedCalls: [ + { + toolCallId: "nested-replay-workflow", + toolName: "workflow_run", + input: { __kernelBounded: true, bytes: 18_457, preview: "{…}" }, + state: "output-available", + output: { __kernelBounded: true, runId: "wfr_replay", status: "running" }, + timestamp: timestamp + 1, + workflowRun: { runId: "wfr_replay", timestamp: timestamp + 2 }, + }, + ], + }, + ], + }); + getWorkspaceStreamsForTests(streamManager).set(workspaceId, streamInfo); + + const events: Array> = []; + for (const type of ["tool-call-start", "workflow-run-attached", "tool-call-end"] as const) { + streamManager.on(type, (event: Record) => events.push(event)); + } + + await streamManager.replayStream(workspaceId); + + const nestedStart = events.find( + (e) => e.type === "tool-call-start" && e.toolCallId === "nested-replay-workflow" + ); + expect(nestedStart?.parentToolCallId).toBe("code-exec-replay"); + expect(nestedStart?.replay).toBe(true); + const nestedAttach = events.find( + (e) => e.type === "workflow-run-attached" && e.toolCallId === "nested-replay-workflow" + ); + expect(nestedAttach?.runId).toBe("wfr_replay"); + const nestedEnd = events.find( + (e) => e.type === "tool-call-end" && e.toolCallId === "nested-replay-workflow" + ); + expect(nestedEnd?.parentToolCallId).toBe("code-exec-replay"); + }); + + test("incremental replay keeps a parent whose only fresh activity is nested", async () => { + const streamManager = new StreamManager(historyService); + const workspaceId = "nested-replay-cursor-workspace"; + const messageId = "nested-replay-cursor-message"; + const timestamp = Date.now(); + const streamInfo = createStreamInfoForTests({ + messageId, + lastPartialWriteTime: timestamp, + parts: [ + { + type: "dynamic-tool", + toolCallId: "code-exec-cursor", + toolName: "code_execution", + input: { code: "mux.workflow_run({...})" }, + state: "input-available", + // Parent part predates the reconnect cursor... + timestamp, + nestedCalls: [ + { + toolCallId: "nested-cursor-workflow", + toolName: "workflow_run", + input: {}, + state: "input-available", + // ...but the nested workflow started after it. + timestamp: timestamp + 100, + }, + ], + }, + ], + }); + getWorkspaceStreamsForTests(streamManager).set(workspaceId, streamInfo); + + const starts: Array> = []; + streamManager.on("tool-call-start", (event: Record) => starts.push(event)); + + await streamManager.replayStream(workspaceId, { afterTimestamp: timestamp + 50 }); + + expect(starts.some((e) => e.toolCallId === "nested-cursor-workflow")).toBe(true); + }); +}); + describe("StreamManager - nested tool call normalization", () => { test("normalizes zero-arg nested calls to {} for the persisted record and the wire event", () => { const streamManager = new StreamManager(historyService); diff --git a/src/node/services/streamManager.ts b/src/node/services/streamManager.ts index 5cafcfc1208..0437dc4060f 100644 --- a/src/node/services/streamManager.ts +++ b/src/node/services/streamManager.ts @@ -570,6 +570,12 @@ interface WorkspaceStreamInfo { // attachment and apply it as soon as the matching dynamic-tool part lands. pendingWorkflowRunAttachments: Map; + // Kernel-nested tool events can arrive before the fullStream consumer stores the parent + // code_execution part (same race as pendingToolExecutionStarts). Buffer the nested records + // by parent toolCallId and merge them when the parent part lands, so an interrupt in that + // window doesn't lose the nested calls (or their workflow run identity) from partial.json. + pendingNestedCalls: Map; + // execute() can begin (lock acquired in withSequentialExecution) before the fullStream // consumer has stored the matching dynamic-tool part. Keep the execution-start timestamp // and apply it as soon as the part lands. @@ -826,6 +832,20 @@ export class StreamManager extends EventEmitter { return true; } + /** + * A nested call's persisted attachment keeps only the run identity: the full + * WorkflowRunRecord carries the workflow source and invocation args, which + * would ride into partial.json unbounded and defeat the kernel record caps + * that already replaced the nested args/result with markers. The frontend + * re-fetches the durable run by runId. (The live workflow-run-attached event + * still carries the full run snapshot.) + */ + private static toNestedWorkflowRunAttachment( + attachment: WorkflowRunToolAttachment + ): WorkflowRunToolAttachment { + return { runId: attachment.runId, timestamp: attachment.timestamp }; + } + /** * Persist a workflow run attachment onto the nested call record it targets, * so a kernel-launched run's identity survives reload even when kernel @@ -838,6 +858,7 @@ export class StreamManager extends EventEmitter { toolCallId: string, attachment: WorkflowRunToolAttachment ): Promise { + const nestedAttachment = StreamManager.toNestedWorkflowRunAttachment(attachment); for (const part of streamInfo.parts) { if (part.type !== "dynamic-tool") { continue; @@ -849,10 +870,19 @@ export class StreamManager extends EventEmitter { if (nestedCalls == null || nestedIndex === -1) { continue; } - nestedCalls[nestedIndex] = { ...nestedCalls[nestedIndex], workflowRun: attachment }; + nestedCalls[nestedIndex] = { ...nestedCalls[nestedIndex], workflowRun: nestedAttachment }; await this.flushPartialWrite(workspaceId, streamInfo); return true; } + // The nested record may still be buffered because its parent part has not + // landed yet; attach there so the merge in appendPartAndEmit persists it. + for (const buffered of streamInfo.pendingNestedCalls?.values() ?? []) { + const nestedIndex = buffered.findIndex((nested) => nested.toolCallId === toolCallId); + if (nestedIndex !== -1) { + buffered[nestedIndex] = { ...buffered[nestedIndex], workflowRun: nestedAttachment }; + return true; + } + } return false; } @@ -1368,6 +1398,53 @@ export class StreamManager extends EventEmitter { } satisfies WorkflowRunAttachedEvent); } + // Replays rebuild renderer state from persisted parts, but nested kernel + // calls (and their workflow run identities) live on the parent part and + // are only ever live-emitted by emitNestedToolEvent; without re-emitting + // them here a reconnect rebuilds the code_execution card without its + // nested rows. Live appends never carry nestedCalls (buffered nested + // records merge after this emit), so this only fires on replay in practice. + const nestedCalls = (part as { nestedCalls?: NestedToolCall[] }).nestedCalls ?? []; + for (const nested of nestedCalls) { + this.emit("tool-call-start", { + type: "tool-call-start", + workspaceId: workspaceId as string, + messageId, + ...(isReplay ? { replay: true } : {}), + toolCallId: nested.toolCallId, + toolName: nested.toolName, + args: nested.input ?? {}, + tokens: 0, + timestamp: nested.timestamp ?? timestamp, + parentToolCallId: part.toolCallId, + }); + if (nested.workflowRun != null) { + this.emit("workflow-run-attached", { + type: "workflow-run-attached", + workspaceId: workspaceId as string, + messageId, + ...(isReplay ? { replay: true } : {}), + toolCallId: nested.toolCallId, + runId: nested.workflowRun.runId, + ...(nested.workflowRun.run != null ? { run: nested.workflowRun.run } : {}), + timestamp: nested.workflowRun.timestamp, + } satisfies WorkflowRunAttachedEvent); + } + if (nested.state === "output-available") { + this.emit("tool-call-end", { + type: "tool-call-end", + workspaceId: workspaceId as string, + messageId, + ...(isReplay ? { replay: true } : {}), + toolCallId: nested.toolCallId, + toolName: nested.toolName, + result: nested.output, + timestamp: Date.now(), + parentToolCallId: part.toolCallId, + }); + } + } + // If tool has output, emit completion if (part.state === "output-available") { this.emit("tool-call-end", { @@ -1402,6 +1479,7 @@ export class StreamManager extends EventEmitter { let partToPersist = part; let pendingAttachment: WorkflowRunToolAttachment | undefined; let pendingExecutionStart: number | undefined; + let bufferedNestedCalls: NestedToolCall[] | undefined; if (part.type === "dynamic-tool") { pendingAttachment = this.takePendingWorkflowRunAttachment(streamInfo, part.toolCallId); // execute() may have started (lock acquired) before the fullStream consumer @@ -1410,13 +1488,25 @@ export class StreamManager extends EventEmitter { if (pendingExecutionStart !== undefined) { streamInfo.pendingToolExecutionStarts.delete(part.toolCallId); } - if (pendingAttachment != null || pendingExecutionStart !== undefined) { + // Nested kernel events may have arrived (and been buffered) in the same + // race window; merge them so they persist with the parent part. Their + // live events already reached the frontend at emission time. + bufferedNestedCalls = streamInfo.pendingNestedCalls?.get(part.toolCallId); + if (bufferedNestedCalls !== undefined) { + streamInfo.pendingNestedCalls.delete(part.toolCallId); + } + if ( + pendingAttachment != null || + pendingExecutionStart !== undefined || + bufferedNestedCalls !== undefined + ) { partToPersist = { ...part, ...(pendingAttachment != null ? { workflowRun: pendingAttachment } : {}), ...(pendingExecutionStart !== undefined ? { executionStartedAt: pendingExecutionStart } : {}), + ...(bufferedNestedCalls !== undefined ? { nestedCalls: bufferedNestedCalls } : {}), }; } } @@ -1430,6 +1520,11 @@ export class StreamManager extends EventEmitter { timestamp: pendingExecutionStart, } satisfies ToolCallExecutionStartEvent); } + if (bufferedNestedCalls !== undefined && pendingAttachment == null) { + // Buffered nested calls can carry a workflow run identity that must + // survive an interrupt; don't wait for the debounced write. + await this.flushPartialWrite(workspaceId, streamInfo); + } if (pendingAttachment != null && part.type === "dynamic-tool") { await this.flushPartialWrite(workspaceId, streamInfo); this.emitWorkflowRunAttachedFromAttachment({ @@ -2172,6 +2267,7 @@ export class StreamManager extends EventEmitter { lastPartTimestamp: startTime, toolCompletionTimestamps: new Map(), pendingWorkflowRunAttachments: new Map(), + pendingNestedCalls: new Map(), pendingToolExecutionStarts: new Map(), model: modelString, metadataModel, @@ -2427,6 +2523,31 @@ export class StreamManager extends EventEmitter { // Schedule partial write so nested calls survive crashes void this.schedulePartialWrite(workspaceId as WorkspaceId, streamInfo); + } else { + // execute() can win the race against the fullStream consumer (same window as + // pendingToolExecutionStarts): buffer the nested record and merge it when the + // parent part lands, instead of silently dropping it from persistence. + const pendingNestedCalls = (streamInfo.pendingNestedCalls ??= new Map()); + const buffered = pendingNestedCalls.get(event.parentToolCallId) ?? []; + if (event.type === "tool-call-start") { + buffered.push({ + toolCallId: event.callId, + toolName: event.toolName, + input: args, + state: "input-available", + timestamp: event.startTime, + }); + } else if (event.type === "tool-call-end") { + const idx = buffered.findIndex((n) => n.toolCallId === event.callId); + if (idx !== -1) { + buffered[idx] = { + ...buffered[idx], + output: event.result ?? (event.error ? { error: event.error } : undefined), + state: "output-available", + }; + } + } + pendingNestedCalls.set(event.parentToolCallId, buffered); } } @@ -4940,6 +5061,23 @@ export class StreamManager extends EventEmitter { return true; } + // Nested kernel calls carry their own activity (start or workflow + // attach) after the cursor even while the parent part's own + // timestamps are older; replay the parent so emitPartAsEvent can + // rebuild the nested rows. + const nestedCalls = (part as { nestedCalls?: NestedToolCall[] }).nestedCalls ?? []; + if ( + nestedCalls.some( + (nested) => + // Legacy rows can miss the timestamp; replay defensively. + nested.timestamp === undefined || + nested.timestamp > afterTimestamp || + (nested.workflowRun != null && nested.workflowRun.timestamp > afterTimestamp) + ) + ) { + return true; + } + if (part.state === "output-available") { const completionTimestamp = streamInfo.toolCompletionTimestamps.get( part.toolCallId From f4283b2c143bcd3e53749e279ade7fd8d03a1f3e Mon Sep 17 00:00:00 2001 From: Michael Suchacz <203725896+ibetitsmike@users.noreply.github.com> Date: Fri, 28 Aug 2026 12:02:25 +0000 Subject: [PATCH 3/3] =?UTF-8?q?fix(workflows):=20=F0=9F=A4=96=20re-deliver?= =?UTF-8?q?=20raced=20nested=20events=20and=20track=20nested=20completion?= =?UTF-8?q?=20for=20replay?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../StreamingMessageAggregator.test.ts | 37 ++--- .../messages/StreamingMessageAggregator.ts | 40 +++--- src/node/services/streamManager.test.ts | 111 ++++++++++++++ src/node/services/streamManager.ts | 135 ++++++++++++------ 4 files changed, 244 insertions(+), 79 deletions(-) diff --git a/src/browser/utils/messages/StreamingMessageAggregator.test.ts b/src/browser/utils/messages/StreamingMessageAggregator.test.ts index 2222108ff03..dcaa4070817 100644 --- a/src/browser/utils/messages/StreamingMessageAggregator.test.ts +++ b/src/browser/utils/messages/StreamingMessageAggregator.test.ts @@ -3795,6 +3795,26 @@ describe("StreamingMessageAggregator", () => { } }); + test("drops nested starts whose parent part has not streamed in (no ghost top-level row)", () => { + const aggregator = createTestAggregator(); + startTestStream(aggregator, { messageId: "msg-1" }); + // No parent tool part exists yet; streamManager re-emits the nested + // events after the parent lands, so the early event must be dropped. + startToolCall(aggregator, { + toolCallId: "nested-early-1", + toolName: "workflow_run", + args: {}, + timestamp: 1100, + parentToolCallId: "parent-tool-1", + }); + + expect( + aggregator + .getDisplayedMessages() + .some((m) => m.type === "tool" && m.toolCallId === "nested-early-1") + ).toBe(false); + }); + test("skips duplicate nested starts (reconnect replays re-emit them with the parent part)", () => { const aggregator = createTestAggregator(); startParentTool(aggregator); @@ -3892,23 +3912,6 @@ describe("StreamingMessageAggregator", () => { } }); - test("falls through to create regular tool if parent not found", () => { - // Defensive behavior: out-of-order nested calls should become regular tool parts. - const aggregator = createTestAggregator(); - startTestStream(aggregator, { messageId: "msg-1" }); - startToolCall(aggregator, { - toolCallId: "nested-orphan", - toolName: "file_read", - args: { filePath: "test.txt" }, - timestamp: 1000, - parentToolCallId: "non-existent-parent", - }); - - const toolParts = aggregator.getDisplayedMessages().filter((m) => m.type === "tool"); - expect(toolParts).toHaveLength(1); - expect(toolParts[0].toolCallId).toBe("nested-orphan"); - }); - test("nested call end is ignored if nested call not found in parent", () => { const aggregator = createTestAggregator(); startParentTool(aggregator); diff --git a/src/browser/utils/messages/StreamingMessageAggregator.ts b/src/browser/utils/messages/StreamingMessageAggregator.ts index 8a37344721c..94b785d581a 100644 --- a/src/browser/utils/messages/StreamingMessageAggregator.ts +++ b/src/browser/utils/messages/StreamingMessageAggregator.ts @@ -2500,26 +2500,30 @@ export class StreamingMessageAggregator { (part): part is DynamicToolPart => part.type === "dynamic-tool" && part.toolCallId === data.parentToolCallId ); - if (parentPart) { - // Initialize nestedCalls array if needed - parentPart.nestedCalls ??= []; - // Reconnect replays re-emit nested starts for rows the renderer may - // already have (the parent part replays whenever any of its nested - // activity is newer than the cursor); skip duplicates like the - // top-level path below does. - if (parentPart.nestedCalls.some((nc) => nc.toolCallId === data.toolCallId)) { - return; - } - parentPart.nestedCalls.push({ - toolCallId: data.toolCallId, - toolName: data.toolName, - state: "input-available", - input: data.args, - timestamp: data.timestamp, - }); - this.markMessageDirty(data.messageId); + if (!parentPart) { + // execute() can emit nested events before the parent part streams in. + // Never fall through to creating a ghost top-level row: streamManager + // buffers the nested record and re-emits its events (start, workflow + // attachment, end) right after the parent part lands. + return; + } + // Initialize nestedCalls array if needed + parentPart.nestedCalls ??= []; + // Buffered-merge and reconnect replays re-deliver nested starts the + // renderer may already have; skip duplicates like the top-level path + // below does. + if (parentPart.nestedCalls.some((nc) => nc.toolCallId === data.toolCallId)) { return; } + parentPart.nestedCalls.push({ + toolCallId: data.toolCallId, + toolName: data.toolName, + state: "input-available", + input: data.args, + timestamp: data.timestamp, + }); + this.markMessageDirty(data.messageId); + return; } // Check if this tool call already exists to prevent duplicates diff --git a/src/node/services/streamManager.test.ts b/src/node/services/streamManager.test.ts index 6ad591df6b9..88a87b5fb8b 100644 --- a/src/node/services/streamManager.test.ts +++ b/src/node/services/streamManager.test.ts @@ -488,6 +488,12 @@ describe("StreamManager - nested kernel call race and replay", () => { schedulePartialWrite?: boolean ) => Promise >(streamManager, "appendPartAndEmit"); + // The renderer dropped the original raced events, so the merge must + // re-deliver them once the parent part exists. + const reEmitted: Array> = []; + for (const type of ["tool-call-start", "workflow-run-attached"] as const) { + streamManager.on(type, (event: Record) => reEmitted.push(event)); + } await appendPartAndEmit.call( streamManager, workspaceId, @@ -502,6 +508,15 @@ describe("StreamManager - nested kernel call race and replay", () => { }, false ); + const reEmittedStart = reEmitted.find( + (e) => e.type === "tool-call-start" && e.toolCallId === "nested-race-workflow" + ); + expect(reEmittedStart?.parentToolCallId).toBe("code-exec-race"); + expect( + reEmitted.some( + (e) => e.type === "workflow-run-attached" && e.toolCallId === "nested-race-workflow" + ) + ).toBe(true); const partial = await historyService.readPartial(workspaceId); const part = partial?.parts[0]; @@ -574,6 +589,102 @@ describe("StreamManager - nested kernel call race and replay", () => { expect(nestedEnd?.parentToolCallId).toBe("code-exec-replay"); }); + test("incremental replay notices a nested call that completed while disconnected", async () => { + const streamManager = new StreamManager(historyService); + const workspaceId = "nested-replay-completion-workspace"; + const messageId = "nested-replay-completion-message"; + const timestamp = Date.now(); + const makeStreamInfo = (completedAt: number) => + createStreamInfoForTests({ + messageId, + lastPartialWriteTime: timestamp, + toolCompletionTimestamps: new Map([["nested-completed-workflow", completedAt]]), + parts: [ + { + type: "dynamic-tool", + toolCallId: "code-exec-completion", + toolName: "code_execution", + input: { code: "mux.workflow_run({...})" }, + state: "input-available", + timestamp, + nestedCalls: [ + { + toolCallId: "nested-completed-workflow", + toolName: "workflow_run", + input: {}, + state: "output-available", + output: { runId: "wfr_done" }, + // Start predates the cursor; only the completion is fresh. + timestamp, + }, + ], + }, + ], + }); + const cursor = timestamp + 50; + + // Completed after the cursor: the parent must replay. + getWorkspaceStreamsForTests(streamManager).set(workspaceId, makeStreamInfo(timestamp + 100)); + const fresh: Array> = []; + const onStart = (event: Record) => fresh.push(event); + streamManager.on("tool-call-start", onStart); + await streamManager.replayStream(workspaceId, { afterTimestamp: cursor }); + expect(fresh.some((e) => e.toolCallId === "nested-completed-workflow")).toBe(true); + streamManager.off("tool-call-start", onStart); + + // Completed before the cursor: nothing fresh, no replay. + getWorkspaceStreamsForTests(streamManager).set(workspaceId, makeStreamInfo(timestamp + 10)); + const stale: Array> = []; + streamManager.on("tool-call-start", (event: Record) => stale.push(event)); + await streamManager.replayStream(workspaceId, { afterTimestamp: cursor }); + expect(stale.some((e) => e.toolCallId === "nested-completed-workflow")).toBe(false); + }); + + test("emitNestedToolEvent records nested completion timestamps", () => { + const streamManager = new StreamManager(historyService); + const workspaceId = "nested-completion-record-workspace"; + const messageId = "nested-completion-record-message"; + const timestamp = Date.now(); + const streamInfo = createStreamInfoForTests({ + messageId, + parts: [ + { + type: "dynamic-tool", + toolCallId: "code-exec-ts", + toolName: "code_execution", + input: {}, + state: "input-available", + timestamp, + nestedCalls: [ + { + toolCallId: "nested-ts", + toolName: "bash", + input: {}, + state: "input-available", + timestamp, + }, + ], + }, + ], + }); + getWorkspaceStreamsForTests(streamManager).set(workspaceId, streamInfo); + + streamManager.emitNestedToolEvent(workspaceId, messageId, { + type: "tool-call-end", + callId: "nested-ts", + toolName: "bash", + args: {}, + parentToolCallId: "code-exec-ts", + startTime: timestamp, + endTime: timestamp + 5, + result: { ok: true }, + }); + + expect((streamInfo.toolCompletionTimestamps as Map).get("nested-ts")).toBe( + timestamp + 5 + ); + }); + test("incremental replay keeps a parent whose only fresh activity is nested", async () => { const streamManager = new StreamManager(historyService); const workspaceId = "nested-replay-cursor-workspace"; diff --git a/src/node/services/streamManager.ts b/src/node/services/streamManager.ts index 0437dc4060f..57c905eb8a7 100644 --- a/src/node/services/streamManager.ts +++ b/src/node/services/streamManager.ts @@ -1405,45 +1405,10 @@ export class StreamManager extends EventEmitter { // nested rows. Live appends never carry nestedCalls (buffered nested // records merge after this emit), so this only fires on replay in practice. const nestedCalls = (part as { nestedCalls?: NestedToolCall[] }).nestedCalls ?? []; - for (const nested of nestedCalls) { - this.emit("tool-call-start", { - type: "tool-call-start", - workspaceId: workspaceId as string, - messageId, - ...(isReplay ? { replay: true } : {}), - toolCallId: nested.toolCallId, - toolName: nested.toolName, - args: nested.input ?? {}, - tokens: 0, - timestamp: nested.timestamp ?? timestamp, - parentToolCallId: part.toolCallId, - }); - if (nested.workflowRun != null) { - this.emit("workflow-run-attached", { - type: "workflow-run-attached", - workspaceId: workspaceId as string, - messageId, - ...(isReplay ? { replay: true } : {}), - toolCallId: nested.toolCallId, - runId: nested.workflowRun.runId, - ...(nested.workflowRun.run != null ? { run: nested.workflowRun.run } : {}), - timestamp: nested.workflowRun.timestamp, - } satisfies WorkflowRunAttachedEvent); - } - if (nested.state === "output-available") { - this.emit("tool-call-end", { - type: "tool-call-end", - workspaceId: workspaceId as string, - messageId, - ...(isReplay ? { replay: true } : {}), - toolCallId: nested.toolCallId, - toolName: nested.toolName, - result: nested.output, - timestamp: Date.now(), - parentToolCallId: part.toolCallId, - }); - } - } + this.emitNestedCallEvents(workspaceId, messageId, part.toolCallId, nestedCalls, { + replay: isReplay, + fallbackTimestamp: timestamp, + }); // If tool has output, emit completion if (part.state === "output-available") { @@ -1461,6 +1426,62 @@ export class StreamManager extends EventEmitter { } } + /** + * Emit a nested call's full event sequence (start, workflow attachment, end) + * on behalf of its parent part. Used when the renderer could not have applied + * the original live events: reconnect replays, and the buffered-merge path in + * appendPartAndEmit where the original events raced ahead of the parent part. + * The aggregator dedupes re-delivered nested starts by toolCallId. + */ + private emitNestedCallEvents( + workspaceId: WorkspaceId, + messageId: string, + parentToolCallId: string, + nestedCalls: NestedToolCall[], + options: { replay?: boolean; fallbackTimestamp: number } + ): void { + const isReplay = options.replay === true; + for (const nested of nestedCalls) { + this.emit("tool-call-start", { + type: "tool-call-start", + workspaceId: workspaceId as string, + messageId, + ...(isReplay ? { replay: true } : {}), + toolCallId: nested.toolCallId, + toolName: nested.toolName, + args: nested.input ?? {}, + tokens: 0, + timestamp: nested.timestamp ?? options.fallbackTimestamp, + parentToolCallId, + }); + if (nested.workflowRun != null) { + this.emit("workflow-run-attached", { + type: "workflow-run-attached", + workspaceId: workspaceId as string, + messageId, + ...(isReplay ? { replay: true } : {}), + toolCallId: nested.toolCallId, + runId: nested.workflowRun.runId, + ...(nested.workflowRun.run != null ? { run: nested.workflowRun.run } : {}), + timestamp: nested.workflowRun.timestamp, + } satisfies WorkflowRunAttachedEvent); + } + if (nested.state === "output-available") { + this.emit("tool-call-end", { + type: "tool-call-end", + workspaceId: workspaceId as string, + messageId, + ...(isReplay ? { replay: true } : {}), + toolCallId: nested.toolCallId, + toolName: nested.toolName, + result: nested.output, + timestamp: Date.now(), + parentToolCallId, + }); + } + } + } + private async appendPartAndEmit( workspaceId: WorkspaceId, streamInfo: WorkspaceStreamInfo, @@ -1520,10 +1541,22 @@ export class StreamManager extends EventEmitter { timestamp: pendingExecutionStart, } satisfies ToolCallExecutionStartEvent); } - if (bufferedNestedCalls !== undefined && pendingAttachment == null) { - // Buffered nested calls can carry a workflow run identity that must - // survive an interrupt; don't wait for the debounced write. - await this.flushPartialWrite(workspaceId, streamInfo); + if (bufferedNestedCalls !== undefined && part.type === "dynamic-tool") { + if (pendingAttachment == null) { + // Buffered nested calls can carry a workflow run identity that must + // survive an interrupt; don't wait for the debounced write. + await this.flushPartialWrite(workspaceId, streamInfo); + } + // The original nested events raced ahead of the parent part, so the + // renderer dropped them (it refuses parented events with no parent + // row); re-deliver them now that the parent start has been emitted. + this.emitNestedCallEvents( + workspaceId, + streamInfo.messageId, + part.toolCallId, + bufferedNestedCalls, + { fallbackTimestamp: part.timestamp ?? Date.now() } + ); } if (pendingAttachment != null && part.type === "dynamic-tool") { await this.flushPartialWrite(workspaceId, streamInfo); @@ -2491,6 +2524,12 @@ export class StreamManager extends EventEmitter { // Persist nested calls to streamInfo.parts for crash/interrupt resilience const streamInfo = this.workspaceStreams.get(workspaceId as WorkspaceId); if (streamInfo) { + if (event.type === "tool-call-end") { + // Nested records never store an end time, so incremental replay needs + // this to notice a nested call that completed while the renderer was + // disconnected (same mechanism as top-level tool completions). + streamInfo.toolCompletionTimestamps.set(event.callId, event.endTime ?? Date.now()); + } const parentPartIndex = streamInfo.parts.findIndex( (p): p is CompletedMessagePart & { type: "dynamic-tool"; toolCallId: string } => p.type === "dynamic-tool" && "toolCallId" in p && p.toolCallId === event.parentToolCallId @@ -5072,7 +5111,15 @@ export class StreamManager extends EventEmitter { // Legacy rows can miss the timestamp; replay defensively. nested.timestamp === undefined || nested.timestamp > afterTimestamp || - (nested.workflowRun != null && nested.workflowRun.timestamp > afterTimestamp) + (nested.workflowRun != null && nested.workflowRun.timestamp > afterTimestamp) || + // The nested record stores no end time, so a call that + // completed while the renderer was disconnected is only + // visible through the recorded completion timestamp. + // Missing entry => replay defensively; re-delivered nested + // events are deduped by the aggregator. + (nested.state === "output-available" && + (streamInfo.toolCompletionTimestamps.get(nested.toolCallId) ?? + Number.POSITIVE_INFINITY) > afterTimestamp) ) ) { return true;