diff --git a/src/node/services/taskMessagePredicates.test.ts b/src/node/services/taskMessagePredicates.test.ts new file mode 100644 index 00000000000..88dba2d5555 --- /dev/null +++ b/src/node/services/taskMessagePredicates.test.ts @@ -0,0 +1,207 @@ +import { describe, expect, test } from "bun:test"; + +import { CONTEXT_BOUNDARY_KINDS } from "@/common/constants/contextBoundary"; +import type { MuxMessage } from "@/common/types/message"; +import { + hasTerminalWorkflowTaskAwaitInParts, + hasTerminalWorkflowToolOutputInParts, + isCompactionRequestMessage, + isWorkflowSupersessionMessage, +} from "@/node/services/taskMessagePredicates"; + +function message(fields: Record): MuxMessage { + return { id: "m", parts: [], ...fields } as unknown as MuxMessage; +} + +function autoCompaction(dispatchSource: string | undefined): MuxMessage { + return message({ + role: "user", + metadata: { + synthetic: true, + muxMetadata: { + type: "compaction-request", + source: "auto-compaction", + parsed: { followUpContent: { dispatchOptions: { source: dispatchSource } } }, + }, + }, + }); +} + +describe("isWorkflowSupersessionMessage", () => { + test("a real user message, a reset boundary or a manual auto-compaction supersedes", () => { + expect(isWorkflowSupersessionMessage(message({ role: "user" }))).toBe(true); + expect( + isWorkflowSupersessionMessage( + message({ + role: "assistant", + metadata: { contextBoundaryKind: CONTEXT_BOUNDARY_KINDS.RESET }, + }) + ) + ).toBe(true); + expect(isWorkflowSupersessionMessage(autoCompaction(undefined))).toBe(true); + }); + + test("a synthetic message or an internal-resume compaction does not supersede", () => { + expect( + isWorkflowSupersessionMessage(message({ role: "user", metadata: { synthetic: true } })) + ).toBe(false); + expect(isWorkflowSupersessionMessage(autoCompaction("internal-resume"))).toBe(false); + expect(isWorkflowSupersessionMessage(message({ role: "assistant" }))).toBe(false); + }); +}); + +describe("hasTerminalWorkflowTaskAwaitInParts", () => { + const awaitPart = (result: unknown, state = "output-available") => ({ + type: "dynamic-tool", + toolCallId: "c1", + toolName: "task_await", + state, + input: {}, + output: { results: [result] }, + }); + + test("completed, interrupted and failed-run errors are terminal for that run", () => { + expect( + hasTerminalWorkflowTaskAwaitInParts( + [awaitPart({ taskId: "wfr_1", status: "completed" })], + "wfr_1" + ) + ).toBe(true); + expect( + hasTerminalWorkflowTaskAwaitInParts( + [awaitPart({ taskId: "wfr_1", status: "interrupted" })], + "wfr_1" + ) + ).toBe(true); + expect( + hasTerminalWorkflowTaskAwaitInParts( + [awaitPart({ taskId: "wfr_1", status: "error", run: { id: "wfr_1", status: "failed" } })], + "wfr_1" + ) + ).toBe(true); + }); + + test("a running run, another run, a plain error or an unfinished call is not terminal", () => { + expect( + hasTerminalWorkflowTaskAwaitInParts( + [awaitPart({ taskId: "wfr_1", status: "running" })], + "wfr_1" + ) + ).toBe(false); + expect( + hasTerminalWorkflowTaskAwaitInParts( + [awaitPart({ taskId: "wfr_2", status: "completed" })], + "wfr_1" + ) + ).toBe(false); + expect( + hasTerminalWorkflowTaskAwaitInParts( + [awaitPart({ taskId: "wfr_1", status: "error" })], + "wfr_1" + ) + ).toBe(false); + expect( + hasTerminalWorkflowTaskAwaitInParts( + [awaitPart({ taskId: "wfr_1", status: "completed" }, "input-available")], + "wfr_1" + ) + ).toBe(false); + }); +}); + +describe("isCompactionRequestMessage", () => { + test("only a compaction-request row matches, and a row without metadata does not throw", () => { + expect( + isCompactionRequestMessage( + message({ role: "user", metadata: { muxMetadata: { type: "compaction-request" } } }) + ) + ).toBe(true); + expect( + isCompactionRequestMessage( + message({ role: "user", metadata: { muxMetadata: { type: "other" } } }) + ) + ).toBe(false); + expect(isCompactionRequestMessage(message({ role: "user" }))).toBe(false); + }); +}); + +describe("hasTerminalWorkflowTaskAwaitInParts: malformed parts", () => { + const part = (fields: Record) => ({ + type: "dynamic-tool", + toolCallId: "c1", + toolName: "task_await", + state: "output-available", + input: {}, + ...fields, + }); + const completed = { results: [{ taskId: "wfr_1", status: "completed" }] }; + + test("skips other tools, other part types and malformed outputs", () => { + expect( + hasTerminalWorkflowTaskAwaitInParts([part({ toolName: "task", output: completed })], "wfr_1") + ).toBe(false); + expect( + hasTerminalWorkflowTaskAwaitInParts([part({ type: "text", output: completed })], "wfr_1") + ).toBe(false); + expect(hasTerminalWorkflowTaskAwaitInParts([part({ output: null })], "wfr_1")).toBe(false); + expect(hasTerminalWorkflowTaskAwaitInParts([part({ output: "done" })], "wfr_1")).toBe(false); + expect( + hasTerminalWorkflowTaskAwaitInParts([part({ output: { results: "done" } })], "wfr_1") + ).toBe(false); + expect( + hasTerminalWorkflowTaskAwaitInParts([part({ output: { results: [null, 5] } })], "wfr_1") + ).toBe(false); + }); + + test("an error result is terminal only with a failed snapshot of the same run", () => { + const error = (run: unknown) => [ + part({ output: { results: [{ taskId: "wfr_1", status: "error", run }] } }), + ]; + expect(hasTerminalWorkflowTaskAwaitInParts(error(null), "wfr_1")).toBe(false); + expect( + hasTerminalWorkflowTaskAwaitInParts(error({ id: "wfr_1", status: "running" }), "wfr_1") + ).toBe(false); + expect( + hasTerminalWorkflowTaskAwaitInParts(error({ id: "wfr_2", status: "failed" }), "wfr_1") + ).toBe(false); + }); +}); + +describe("hasTerminalWorkflowToolOutputInParts", () => { + const runPart = (fields: Record) => ({ + type: "dynamic-tool", + toolCallId: "c1", + toolName: "workflow_run", + state: "output-available", + input: {}, + output: { runId: "wfr_1", status: "completed" }, + ...fields, + }); + + test("a finished workflow_run output for the run is terminal", () => { + expect(hasTerminalWorkflowToolOutputInParts([runPart({})], "wfr_1")).toBe(true); + }); + + test("an unfinished call or another part type is not", () => { + expect( + hasTerminalWorkflowToolOutputInParts([runPart({ state: "input-available" })], "wfr_1") + ).toBe(false); + expect(hasTerminalWorkflowToolOutputInParts([runPart({ type: "text" })], "wfr_1")).toBe(false); + }); +}); + +describe("auto-compaction rows with partial metadata", () => { + test("a row without follow-up options counts as a manual supersession", () => { + const partial = (parsed: unknown) => + message({ + role: "user", + metadata: { + synthetic: true, + muxMetadata: { type: "compaction-request", source: "auto-compaction", parsed }, + }, + }); + expect(isWorkflowSupersessionMessage(partial({}))).toBe(true); + expect(isWorkflowSupersessionMessage(partial({ followUpContent: {} }))).toBe(true); + expect(isCompactionRequestMessage(message({ role: "user", metadata: {} }))).toBe(false); + }); +}); diff --git a/src/node/services/taskMessagePredicates.ts b/src/node/services/taskMessagePredicates.ts new file mode 100644 index 00000000000..a1134da2fd2 --- /dev/null +++ b/src/node/services/taskMessagePredicates.ts @@ -0,0 +1,107 @@ +// Pure predicates over chat messages that TaskService uses to find supersession points and +// terminal workflow results. Moved out of taskService.ts unchanged; they hold no state. +import { CONTEXT_BOUNDARY_KINDS } from "@/common/constants/contextBoundary"; +import type { MuxMessage } from "@/common/types/message"; +import { isDynamicToolPart } from "@/common/types/toolParts"; +import { isTerminalWorkflowRunToolOutput } from "@/common/utils/workflowRunMessages"; + +export function isInternalResumeAutoCompactionMessage(message: MuxMessage): boolean { + const muxMetadata = message.metadata?.muxMetadata; + if (muxMetadata?.type !== "compaction-request" || muxMetadata.source !== "auto-compaction") { + return false; + } + return muxMetadata.parsed.followUpContent?.dispatchOptions?.source === "internal-resume"; +} + +export function isSyntheticManualSupersessionMessage(message: MuxMessage): boolean { + const muxMetadata = message.metadata?.muxMetadata; + return ( + message.metadata?.synthetic === true && + muxMetadata?.type === "compaction-request" && + muxMetadata.source === "auto-compaction" && + !isInternalResumeAutoCompactionMessage(message) + ); +} + +export function isManualUserSupersessionMessage(message: MuxMessage): boolean { + return ( + message.role === "user" && + (message.metadata?.synthetic !== true || isSyntheticManualSupersessionMessage(message)) + ); +} + +export function isResetBoundaryMessage(message: MuxMessage): boolean { + return message.metadata?.contextBoundaryKind === CONTEXT_BOUNDARY_KINDS.RESET; +} + +export function isCompactionRequestMessage(message: MuxMessage): boolean { + return message.metadata?.muxMetadata?.type === "compaction-request"; +} + +export function isWorkflowSupersessionMessage(message: MuxMessage): boolean { + return isManualUserSupersessionMessage(message) || isResetBoundaryMessage(message); +} + +export function isFailedWorkflowRunSnapshot(value: unknown, runId: string): boolean { + if (value == null || typeof value !== "object") { + return false; + } + const record = value as Record; + return record.id === runId && record.status === "failed"; +} + +export function isTerminalWorkflowTaskAwaitRecord( + record: Record, + runId: string +): boolean { + if (record.taskId !== runId) { + return false; + } + if (record.status === "completed" || record.status === "interrupted") { + return true; + } + if (record.status === "error") { + return isFailedWorkflowRunSnapshot(record.run, runId); + } + return false; +} + +export function hasTerminalWorkflowTaskAwaitInParts( + parts: readonly unknown[], + runId: string +): boolean { + return parts.some((part) => { + if (!isDynamicToolPart(part) || part.toolName !== "task_await") { + return false; + } + if (part.state !== "output-available") { + return false; + } + const output = part.output; + if (output == null || typeof output !== "object") { + return false; + } + const results = (output as Record).results; + if (!Array.isArray(results)) { + return false; + } + return results.some((result) => { + if (result == null || typeof result !== "object") { + return false; + } + return isTerminalWorkflowTaskAwaitRecord(result as Record, runId); + }); + }); +} + +export function hasTerminalWorkflowToolOutputInParts( + parts: readonly unknown[], + runId: string +): boolean { + return parts.some( + (part) => + isDynamicToolPart(part) && + part.state === "output-available" && + isTerminalWorkflowRunToolOutput(part.toolName, part.output, runId) + ); +} diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 97479fc0383..e5a01e9b78d 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -6,7 +6,6 @@ import { getValidMessagingTag, getValidUnrelatedWorkspaceConsent, } from "@/common/orpc/schemas/workspace"; -import { isPlainObject } from "@/common/utils/isPlainObject"; import assert from "node:assert/strict"; import * as path from "node:path"; import * as fsPromises from "fs/promises"; @@ -152,7 +151,6 @@ import { markChildGoalPauseOwed, } from "@/node/services/taskUtils"; import { validateWorkspaceName } from "@/common/utils/validation/workspaceValidation"; -import { getTaskGroupCount } from "@/common/utils/tools/taskGroups"; import { stripTrailingSlashes } from "@/node/utils/pathUtils"; import { isErrnoException } from "@/node/utils/fs"; import { Ok, Err, type Result } from "@/common/types/result"; @@ -271,15 +269,12 @@ import { import { isDynamicToolPart, type DynamicToolPart } from "@/common/types/toolParts"; import { buildWorkflowResultContextMessage, - isTerminalWorkflowRunToolOutput, isWorkflowDisplayOnlyMessage, - isWorkflowRunEmittingToolName, } from "@/common/utils/workflowRunMessages"; import { AgentReportInlineToolArgsSchema, AgentReportSubmittedReportSchema, TaskToolResultSchema, - TaskToolArgsSchema, type SubagentGitPatchArtifact, } from "@/common/utils/tools/toolDefinitions"; import { isPlanLikeInResolvedChain } from "@/common/utils/agentTools"; @@ -322,8 +317,22 @@ import { isWorkspaceArchived } from "@/common/utils/archive"; import type { StartupRecoveryState } from "@/node/services/startupRecovery"; import { judgeLifecycleMarkerOwner } from "@/node/services/lifecycleMarkerOwners"; import { LegacyTodoNudgePolicyTrace, type ToolPolicy } from "@/common/utils/tools/toolPolicy"; -import { CONTEXT_BOUNDARY_KINDS } from "@/common/constants/contextBoundary"; import { WorkflowRunStore } from "@/node/services/workflows/WorkflowRunStore"; +import { + collectReferencedTaskIdsFromTaskToolOutput, + type CompletedTaskReportReceipt, + receiptMatchesReport, + collectCompletedTaskReportReceipts, + parseTaskToolInputForRecovery, + collectAgentReferencedWorkflowRunIdsFromParts, +} from "@/node/services/taskToolOutputs"; +import { + isResetBoundaryMessage, + isCompactionRequestMessage, + isWorkflowSupersessionMessage, + hasTerminalWorkflowTaskAwaitInParts, + hasTerminalWorkflowToolOutputInParts, +} from "@/node/services/taskMessagePredicates"; import { isActiveWorkspaceTurnTaskStatus, isWorkspaceTurnTaskId, @@ -337,7 +346,6 @@ import { } from "@/node/services/terminalAttentionStore"; import { readAgentWorkflowRunReferences } from "@/node/services/agentWorkflowRunReferences"; import type { AgentWorkflowRunStrictPin } from "@/node/services/agentWorkflowRunReferences"; -import { isWorkflowRunTaskId } from "@/node/services/tools/taskId"; import { normalizeWorkflowAgentReportPayloadForHostSchema } from "@/common/utils/tools/workflowReportPayload"; import { formatJsonSchemaValidationErrors, @@ -1759,436 +1767,10 @@ function buildBackgroundAwaitPrompt(params: { ); } -const isWorkflowRunId = isWorkflowRunTaskId; - -function collectWorkflowRunIdsFromToolOutput(output: unknown): string[] { - if (output == null || typeof output !== "object") { - return []; - } - - const record = output as Record; - if (isWorkflowRunId(record.runId)) { - return [record.runId]; - } - - const results = record.results; - if (!Array.isArray(results)) { - return []; - } - - const runIds: string[] = []; - for (const result of results) { - if (result == null || typeof result !== "object") { - continue; - } - const taskId = (result as Record).taskId; - if (isWorkflowRunId(taskId)) { - runIds.push(taskId); - } - } - return runIds; -} - -/** - * Tolerant extraction of the task IDs a persisted task tool output references. Recovery - * bookkeeping must survive outputs written by newer releases (extra fields would fail the - * strict result schema) so that a mid-stream downgrade cannot duplicate fallback reports. - */ -function collectReferencedTaskIdsFromTaskToolOutput(output: unknown, into: Set): void { - if (output == null || typeof output !== "object") { - return; - } - const addTaskId = (value: unknown): void => { - if (typeof value === "string" && value.length > 0) { - into.add(value); - } - }; - - const record = output as Record; - addTaskId(record.taskId); - if (Array.isArray(record.taskIds)) { - for (const taskId of record.taskIds) { - addTaskId(taskId); - } - } - for (const key of ["tasks", "reports"] as const) { - const rows = record[key]; - if (!Array.isArray(rows)) { - continue; - } - for (const row of rows) { - if (row != null && typeof row === "object") { - addTaskId((row as Record).taskId); - } - } - } -} - -type CompletedTaskReportReceipt = ( - | { kind: "initial"; taskId: string } - | { kind: "continuation"; taskId: string; handleId?: string; messageId: string } -) & { - reportMarkdown: string; - /** Where reportMarkdown starts in the full report, when the row is one paged page. */ - pageOffset?: number; -}; - -/** - * A paged row (task_await, or a foreground task since #6427) holds one page of a long report: - * its pageOffset is its startCharOffset, 0 for a first page that only carries nextCharOffset. - * Undefined for a whole report. - */ -function readReportPageOffset(row: Record): number | undefined { - if (typeof row.startCharOffset === "number") return row.startCharOffset; - return typeof row.nextCharOffset === "number" ? 0 : undefined; -} - -/** - * Whether a continuation receipt shows the owner this exact report: the whole text, or any - * non-empty page of it at the page's own offset. Reading one page (with its continuation note) - * delivers the result, so its terminal notification must not wake the owner again. - */ -function receiptMatchesReport( - receipt: Extract, - reportMarkdown: string | undefined -): boolean { - if (receipt.reportMarkdown === reportMarkdown) return true; - if (reportMarkdown == null || receipt.pageOffset == null) return false; - return ( - receipt.reportMarkdown.length > 0 && - reportMarkdown.startsWith(receipt.reportMarkdown, receipt.pageOffset) - ); -} - -/** Read only canonical task-result containers, never arbitrary JSON or prose. */ -function completedTaskReports(toolName: unknown, output: unknown): CompletedTaskReportReceipt[] { - if (!isPlainObject(output)) return []; - const rows: unknown[] = []; - if (toolName === "task") { - if (output.status === "completed") rows.push(output); - // A group can return completed members while other members are still running. - if (Array.isArray(output.reports)) output.reports.forEach((row: unknown) => rows.push(row)); - } else if (toolName === "task_await" && Array.isArray(output.results)) { - output.results.forEach((row: unknown) => { - if (isPlainObject(row) && row.status === "completed") rows.push(row); - }); - } - return rows.flatMap((row): CompletedTaskReportReceipt[] => { - if ( - !isPlainObject(row) || - typeof row.taskId !== "string" || - typeof row.reportMarkdown !== "string" - ) - return []; - const messageId = - row.messageId ?? - (isPlainObject(row.finalMessageRef) ? row.finalMessageRef.messageId : undefined); - // Workspace-target tools expose a handle; stable-child awaits expose the child itself. - const handleId = isWorkspaceTurnTaskId(row.taskId) ? row.taskId : undefined; - const taskId = handleId != null ? coerceNonEmptyString(row.workspaceId) : row.taskId; - if (taskId == null) return []; - if (typeof messageId === "string") { - return [ - { - kind: "continuation", - taskId, - handleId, - messageId, - reportMarkdown: row.reportMarkdown, - pageOffset: readReportPageOffset(row), - }, - ]; - } - return handleId == null - ? [ - { - kind: "initial", - taskId, - reportMarkdown: row.reportMarkdown, - pageOffset: readReportPageOffset(row), - }, - ] - : []; - }); -} - -/** - * Same text and same page position. A visible projection that drops a page's offset fields - * (nextCharOffset/startCharOffset) hides that more text exists and where to read it, so it does - * not prove delivery of the genuine paged row. - */ -function sameReportPage(a: CompletedTaskReportReceipt, b: CompletedTaskReportReceipt): boolean { - return a.reportMarkdown === b.reportMarkdown && a.pageOffset === b.pageOffset; -} - -function sameTaskReportReceipt( - a: CompletedTaskReportReceipt, - b: CompletedTaskReportReceipt -): boolean { - return ( - a.kind === b.kind && - a.taskId === b.taskId && - sameReportPage(a, b) && - (a.kind === "initial" || - (b.kind === "continuation" && a.messageId === b.messageId && a.handleId === b.handleId)) - ); -} - -/** Only model-visible successful tool results prove consumption. UI telemetry does not. */ -function collectCompletedTaskReportReceipts(message: MuxMessage): CompletedTaskReportReceipt[] { - const reports: CompletedTaskReportReceipt[] = []; - const succeeded = (call: unknown): call is Record => - isPlainObject(call) && - call.failed !== true && - call.error == null && - call.ok !== false && - (call.state == null || call.state === "output-available"); - const visit = (call: unknown, depth: number): void => { - if (depth > 30 || !succeeded(call)) return; - const output = call.output ?? call.result; - if (!isPlainObject(output)) return; - reports.push(...completedTaskReports(call.toolName, output)); - if (call.toolName !== "code_execution") return; - - // Classic executions expose these result records to the model, even when a later - // call fails. Persistent kernels replace them with summaries that contain no report. - if (Array.isArray(output.toolCalls)) { - for (const nested of output.toolCalls) visit(nested, depth + 1); - } - // Persistent nestedCalls hold UI-only full values. Use them solely as provenance for - // a complete canonical report deliberately returned through output.result. Offload - // handles, previews, hidden values, and unrelated returns must retain their wake. - if (Array.isArray(call.nestedCalls)) { - for (const nested of call.nestedCalls) { - if (!succeeded(nested)) continue; - const genuine = completedTaskReports(nested.toolName, nested.output); - const visible = completedTaskReports(nested.toolName, output.result); - reports.push( - ...visible.filter((receipt) => - genuine.some((source) => sameTaskReportReceipt(source, receipt)) - ) - ); - } - } - }; - for (const part of message.parts) { - if (isDynamicToolPart(part) && part.state === "output-available") visit(part, 0); - } - return reports; -} - -interface RecoveredTaskToolInput { - data: z.infer; - groupCount: number; - legacyVariants: boolean; -} - -/** - * Persisted partials may contain the removed `variants` input. Keep that field out of the - * model-facing schema while accepting it narrowly during crash recovery so completed legacy - * siblings can still finalize their parent tool call. - */ -function parseTaskToolInputForRecovery(input: unknown): RecoveredTaskToolInput | null { - const current = TaskToolArgsSchema.safeParse(input); - if (current.success) { - return { - data: current.data, - groupCount: getTaskGroupCount(current.data), - legacyVariants: false, - }; - } - - if (!input || typeof input !== "object" || Array.isArray(input)) { - return null; - } - - const sanitized = { ...(input as Record) }; - const variants = sanitized.variants; - delete sanitized.variants; - const parsed = TaskToolArgsSchema.safeParse(sanitized); - if (!parsed.success) { - return null; - } - - if (variants == null) { - return { - data: parsed.data, - groupCount: getTaskGroupCount(parsed.data), - legacyVariants: false, - }; - } - if (!Array.isArray(variants) || variants.length < 1 || variants.length > 20) { - return null; - } - - const labels = variants.map((variant) => (typeof variant === "string" ? variant.trim() : "")); - if (labels.some((label) => label.length === 0) || new Set(labels).size !== labels.length) { - return null; - } - - return { - data: parsed.data, - groupCount: labels.length, - legacyVariants: true, - }; -} - -function collectWorkflowRunIdsFromTaskAwaitInput(input: unknown): string[] { - if (input == null || typeof input !== "object") { - return []; - } - const taskIds = (input as Record).task_ids; - if (!Array.isArray(taskIds)) { - return []; - } - return taskIds.filter(isWorkflowRunId); -} - -function collectAgentReferencedWorkflowRunIdsFromParts( - parts: readonly unknown[], - knownAgentRunIds: ReadonlySet -): string[] { - const runIds = new Set(); - - for (const part of parts) { - if (!isDynamicToolPart(part) || part.state !== "output-available") { - continue; - } - // workflow_resume re-establishes agent provenance the same way workflow_run does: both - // outputs carry the runId of a run the agent explicitly owns. - if (!isWorkflowRunEmittingToolName(part.toolName)) { - continue; - } - for (const runId of collectWorkflowRunIdsFromToolOutput(part.output)) { - runIds.add(runId); - } - } - - const allowedTaskAwaitRunIds = new Set([...knownAgentRunIds, ...runIds]); - for (const part of parts) { - if (!isDynamicToolPart(part) || part.state !== "output-available") { - continue; - } - if (part.toolName !== "task_await") { - continue; - } - - // Omitted task_ids makes task_await discover every active run in the workspace, including - // slash-command runs. Only treat task_await output as agent provenance when the model either - // explicitly awaited that workflow ID in this turn or we already know the run was agent-started. - for (const runId of collectWorkflowRunIdsFromTaskAwaitInput(part.input)) { - runIds.add(runId); - allowedTaskAwaitRunIds.add(runId); - } - for (const runId of collectWorkflowRunIdsFromToolOutput(part.output)) { - if (allowedTaskAwaitRunIds.has(runId)) { - runIds.add(runId); - } - } - } - - return Array.from(runIds); -} - -function isInternalResumeAutoCompactionMessage(message: MuxMessage): boolean { - const muxMetadata = message.metadata?.muxMetadata; - if (muxMetadata?.type !== "compaction-request" || muxMetadata.source !== "auto-compaction") { - return false; - } - return muxMetadata.parsed.followUpContent?.dispatchOptions?.source === "internal-resume"; -} - -function isSyntheticManualSupersessionMessage(message: MuxMessage): boolean { - const muxMetadata = message.metadata?.muxMetadata; - return ( - message.metadata?.synthetic === true && - muxMetadata?.type === "compaction-request" && - muxMetadata.source === "auto-compaction" && - !isInternalResumeAutoCompactionMessage(message) - ); -} - -function isManualUserSupersessionMessage(message: MuxMessage): boolean { - return ( - message.role === "user" && - (message.metadata?.synthetic !== true || isSyntheticManualSupersessionMessage(message)) - ); -} - function withToolPolicy(toolPolicy: ToolPolicy | undefined): { toolPolicy?: ToolPolicy } { return toolPolicy != null ? { toolPolicy } : {}; } -function isResetBoundaryMessage(message: MuxMessage): boolean { - return message.metadata?.contextBoundaryKind === CONTEXT_BOUNDARY_KINDS.RESET; -} - -function isCompactionRequestMessage(message: MuxMessage): boolean { - return message.metadata?.muxMetadata?.type === "compaction-request"; -} - -function isWorkflowSupersessionMessage(message: MuxMessage): boolean { - return isManualUserSupersessionMessage(message) || isResetBoundaryMessage(message); -} - -function isFailedWorkflowRunSnapshot(value: unknown, runId: string): boolean { - if (value == null || typeof value !== "object") { - return false; - } - const record = value as Record; - return record.id === runId && record.status === "failed"; -} - -function isTerminalWorkflowTaskAwaitRecord( - record: Record, - runId: string -): boolean { - if (record.taskId !== runId) { - return false; - } - if (record.status === "completed" || record.status === "interrupted") { - return true; - } - if (record.status === "error") { - return isFailedWorkflowRunSnapshot(record.run, runId); - } - return false; -} - -function hasTerminalWorkflowTaskAwaitInParts(parts: readonly unknown[], runId: string): boolean { - return parts.some((part) => { - if (!isDynamicToolPart(part) || part.toolName !== "task_await") { - return false; - } - if (part.state !== "output-available") { - return false; - } - const output = part.output; - if (output == null || typeof output !== "object") { - return false; - } - const results = (output as Record).results; - if (!Array.isArray(results)) { - return false; - } - return results.some((result) => { - if (result == null || typeof result !== "object") { - return false; - } - return isTerminalWorkflowTaskAwaitRecord(result as Record, runId); - }); - }); -} - -function hasTerminalWorkflowToolOutputInParts(parts: readonly unknown[], runId: string): boolean { - return parts.some( - (part) => - isDynamicToolPart(part) && - part.state === "output-available" && - isTerminalWorkflowRunToolOutput(part.toolName, part.output, runId) - ); -} - function sanitizeAgentTypeForName(agentType: string): string { const normalized = agentType .trim() diff --git a/src/node/services/taskToolOutputs.test.ts b/src/node/services/taskToolOutputs.test.ts new file mode 100644 index 00000000000..e494119fe14 --- /dev/null +++ b/src/node/services/taskToolOutputs.test.ts @@ -0,0 +1,332 @@ +import { describe, expect, test } from "bun:test"; + +import type { MuxMessage } from "@/common/types/message"; +import { + collectAgentReferencedWorkflowRunIdsFromParts, + collectCompletedTaskReportReceipts, + collectReferencedTaskIdsFromTaskToolOutput, + parseTaskToolInputForRecovery, + receiptMatchesReport, +} from "@/node/services/taskToolOutputs"; + +function toolPart(toolName: string, output: unknown, input: unknown = {}) { + return { + type: "dynamic-tool", + toolCallId: `call-${toolName}`, + toolName, + state: "output-available", + input, + output, + }; +} + +function assistant(parts: unknown[]): MuxMessage { + return { id: "msg-1", role: "assistant", parts } as unknown as MuxMessage; +} + +describe("collectReferencedTaskIdsFromTaskToolOutput", () => { + test("collects task IDs from every container and skips empty values", () => { + const into = new Set(); + collectReferencedTaskIdsFromTaskToolOutput( + { + taskId: "t1", + taskIds: ["t2", "", 7], + tasks: [{ taskId: "t3" }, null], + reports: [{ taskId: "t4" }, { taskId: "" }], + extraFieldFromNewerRelease: true, + }, + into + ); + expect([...into].sort()).toEqual(["t1", "t2", "t3", "t4"]); + }); + + test("ignores an output that is not an object", () => { + const into = new Set(); + collectReferencedTaskIdsFromTaskToolOutput("t1", into); + expect(into.size).toBe(0); + }); +}); + +describe("receiptMatchesReport", () => { + const receipt = (reportMarkdown: string, pageOffset?: number) => ({ + kind: "continuation" as const, + taskId: "t1", + messageId: "m1", + reportMarkdown, + pageOffset, + }); + + test("matches the whole text, or a non-empty page at its own offset", () => { + expect(receiptMatchesReport(receipt("hello world"), "hello world")).toBe(true); + expect(receiptMatchesReport(receipt("world", 6), "hello world")).toBe(true); + expect(receiptMatchesReport(receipt("world", 0), "hello world")).toBe(false); + expect(receiptMatchesReport(receipt("", 0), "hello world")).toBe(false); + expect(receiptMatchesReport(receipt("world"), "hello world")).toBe(false); + expect(receiptMatchesReport(receipt("world", 6), undefined)).toBe(false); + }); +}); + +describe("collectCompletedTaskReportReceipts", () => { + test("reads completed task and task_await rows, with page offsets and continuation IDs", () => { + const receipts = collectCompletedTaskReportReceipts( + assistant([ + toolPart("task", { status: "completed", taskId: "t1", reportMarkdown: "r1" }), + toolPart("task_await", { + results: [ + { + status: "completed", + taskId: "t2", + reportMarkdown: "page", + startCharOffset: 40, + messageId: "m2", + }, + { status: "running", taskId: "t3", reportMarkdown: "not yet" }, + { + status: "completed", + taskId: "wst_abc", + workspaceId: "ws1", + reportMarkdown: "r4", + finalMessageRef: { messageId: "m4" }, + }, + { + status: "completed", + taskId: "wst_def", + workspaceId: "ws2", + reportMarkdown: "no message id", + }, + ], + }), + ]) + ); + expect(receipts).toEqual([ + { kind: "initial", taskId: "t1", reportMarkdown: "r1", pageOffset: undefined }, + { + kind: "continuation", + taskId: "t2", + handleId: undefined, + messageId: "m2", + reportMarkdown: "page", + pageOffset: 40, + }, + { + kind: "continuation", + taskId: "ws1", + handleId: "wst_abc", + messageId: "m4", + reportMarkdown: "r4", + pageOffset: undefined, + }, + ]); + }); + + test("a failed call proves nothing", () => { + const failed = { + ...toolPart("task", { status: "completed", taskId: "t1", reportMarkdown: "r1" }), + failed: true, + }; + expect(collectCompletedTaskReportReceipts(assistant([failed]))).toEqual([]); + }); + + test("a code_execution result counts only reports that a genuine nested call produced", () => { + const genuine = { status: "completed", taskId: "t1", reportMarkdown: "r1" }; + const receipts = collectCompletedTaskReportReceipts( + assistant([ + { + ...toolPart("code_execution", { + result: { status: "completed", taskId: "t1", reportMarkdown: "r1" }, + }), + nestedCalls: [{ toolName: "task", output: genuine }], + }, + { + ...toolPart("code_execution", { + result: { status: "completed", taskId: "t9", reportMarkdown: "forged" }, + }), + nestedCalls: [{ toolName: "task", output: genuine }], + }, + ]) + ); + expect(receipts).toEqual([ + { kind: "initial", taskId: "t1", reportMarkdown: "r1", pageOffset: undefined }, + ]); + }); +}); + +describe("parseTaskToolInputForRecovery", () => { + const base = { agentId: "explore", prompt: "Look around", title: "Explorer" }; + + test("parses current input with its group count", () => { + expect(parseTaskToolInputForRecovery({ ...base, n: 3 })).toMatchObject({ + groupCount: 3, + legacyVariants: false, + }); + }); + + test("accepts the removed variants field during recovery only with unique labels", () => { + expect(parseTaskToolInputForRecovery({ ...base, variants: ["a", "b"] })).toMatchObject({ + groupCount: 2, + legacyVariants: true, + }); + expect(parseTaskToolInputForRecovery({ ...base, variants: ["a", "a"] })).toBeNull(); + expect(parseTaskToolInputForRecovery({ ...base, variants: [" "] })).toBeNull(); + expect(parseTaskToolInputForRecovery({ ...base, variants: [] })).toBeNull(); + }); + + test("refuses input that fails the schema", () => { + expect(parseTaskToolInputForRecovery({ ...base, prompt: "" })).toBeNull(); + expect(parseTaskToolInputForRecovery(["not", "an", "object"])).toBeNull(); + }); +}); + +describe("collectAgentReferencedWorkflowRunIdsFromParts", () => { + test("keeps runs the agent started or awaited by ID, not runs that task_await discovered", () => { + const parts = [ + toolPart("workflow_run", { runId: "wfr_started" }), + toolPart( + "task_await", + { + results: [ + { taskId: "wfr_started" }, + { taskId: "wfr_named" }, + { taskId: "wfr_discovered" }, + { taskId: "wfr_known" }, + ], + }, + { task_ids: ["wfr_named"] } + ), + ]; + expect( + collectAgentReferencedWorkflowRunIdsFromParts(parts, new Set(["wfr_known"])).sort() + ).toEqual(["wfr_known", "wfr_named", "wfr_started"]); + }); +}); + +describe("edge cases of the tolerant parsers", () => { + test("ignores null outputs and rows, and keeps one-character task IDs", () => { + const into = new Set(); + collectReferencedTaskIdsFromTaskToolOutput(null, into); + collectReferencedTaskIdsFromTaskToolOutput({ taskId: "a" }, into); + expect([...into]).toEqual(["a"]); + }); + + test("a one-character page matches at its offset", () => { + expect( + receiptMatchesReport( + { kind: "continuation", taskId: "t1", messageId: "m1", reportMarkdown: "d", pageOffset: 3 }, + "abcd" + ) + ).toBe(true); + }); + + test("a first page that carries only nextCharOffset starts at 0", () => { + const [receipt] = collectCompletedTaskReportReceipts( + assistant([ + toolPart("task", { + status: "completed", + taskId: "t1", + reportMarkdown: "page", + nextCharOffset: 4, + }), + ]) + ); + expect(receipt?.pageOffset).toBe(0); + }); + + test("skips rows of other tools, malformed rows, unfinished calls and refused calls", () => { + const row = { status: "completed", taskId: "t1", reportMarkdown: "r1" }; + expect( + collectCompletedTaskReportReceipts( + assistant([ + toolPart("bash", { results: [row] }), + toolPart("task", { status: "completed", taskId: 7, reportMarkdown: "r" }), + toolPart("task", { status: "completed", taskId: "t2", reportMarkdown: 7 }), + toolPart("task", { reports: [null, "row"] }), + { ...toolPart("task", row), state: "input-available" }, + { ...toolPart("task", row), ok: false }, + { ...toolPart("task", row), type: "text" }, + ]) + ) + ).toEqual([]); + }); + + test("reads a group's completed members and a report in a classic code_execution's toolCalls", () => { + const receipts = collectCompletedTaskReportReceipts( + assistant([ + toolPart("task", { status: "running", reports: [{ taskId: "t1", reportMarkdown: "r1" }] }), + toolPart("code_execution", { + toolCalls: [ + { + toolName: "task", + output: { status: "completed", taskId: "t2", reportMarkdown: "r2" }, + }, + ], + }), + ]) + ); + expect(receipts.map((receipt) => receipt.taskId)).toEqual(["t1", "t2"]); + }); + + test("a code_execution projection must match the genuine page, kind, message and handle", () => { + const genuine = { + status: "completed", + taskId: "wst_a", + workspaceId: "ws1", + reportMarkdown: "page", + startCharOffset: 10, + messageId: "m1", + }; + const project = (visible: Record) => + collectCompletedTaskReportReceipts( + assistant([ + { + ...toolPart("code_execution", { result: { ...genuine, ...visible } }), + nestedCalls: [{ toolName: "task", output: genuine }], + }, + ]) + ); + expect(project({})).toHaveLength(1); + expect(project({ startCharOffset: undefined })).toEqual([]); + expect(project({ messageId: "m2" })).toEqual([]); + expect(project({ taskId: "wst_b" })).toEqual([]); + expect(project({ messageId: undefined })).toEqual([]); + }); + + test("a failed nested call proves nothing", () => { + const row = { status: "completed", taskId: "t1", reportMarkdown: "r1" }; + expect( + collectCompletedTaskReportReceipts( + assistant([ + { + ...toolPart("code_execution", { result: row }), + nestedCalls: [{ toolName: "task", output: row, error: "boom" }], + }, + ]) + ) + ).toEqual([]); + }); + + test("recovery accepts 1 to 20 variant labels and a null variants field", () => { + const base = { agentId: "explore", prompt: "Look around", title: "Explorer" }; + const labels = (count: number) => Array.from({ length: count }, (_, i) => `v${i}`); + expect(parseTaskToolInputForRecovery({ ...base, variants: labels(1) })?.groupCount).toBe(1); + expect(parseTaskToolInputForRecovery({ ...base, variants: labels(20) })?.groupCount).toBe(20); + expect(parseTaskToolInputForRecovery({ ...base, variants: labels(21) })).toBeNull(); + expect(parseTaskToolInputForRecovery({ ...base, variants: null })).toMatchObject({ + legacyVariants: false, + }); + }); + + test("workflow provenance ignores malformed parts and outputs", () => { + const parts = [ + { ...toolPart("workflow_run", { runId: "wfr_text" }), type: "text" }, + { ...toolPart("workflow_run", { runId: "wfr_pending" }), state: "input-available" }, + toolPart("workflow_run", null), + toolPart("workflow_run", { results: [null, { taskId: "wfr_row" }] }), + toolPart("task_await", { results: [{ taskId: "wfr_unasked" }] }, null), + { + ...toolPart("task_await", { results: [{ taskId: "wfr_x" }] }, { task_ids: ["wfr_x"] }), + state: "input-available", + }, + ]; + expect(collectAgentReferencedWorkflowRunIdsFromParts(parts, new Set())).toEqual(["wfr_row"]); + }); +}); diff --git a/src/node/services/taskToolOutputs.ts b/src/node/services/taskToolOutputs.ts new file mode 100644 index 00000000000..6476bde2ea3 --- /dev/null +++ b/src/node/services/taskToolOutputs.ts @@ -0,0 +1,348 @@ +// Pure parsers for task and workflow tool inputs, tool outputs and completed-report receipts. +// Moved out of taskService.ts unchanged so they read and test on their own; they hold no state. +import type { z } from "zod"; +import type { MuxMessage } from "@/common/types/message"; +import { isDynamicToolPart } from "@/common/types/toolParts"; +import { isPlainObject } from "@/common/utils/isPlainObject"; +import { getTaskGroupCount } from "@/common/utils/tools/taskGroups"; +import { TaskToolArgsSchema } from "@/common/utils/tools/toolDefinitions"; +import { isWorkflowRunEmittingToolName } from "@/common/utils/workflowRunMessages"; +import { isWorkspaceTurnTaskId } from "@/node/services/taskHandleStore"; +import { coerceNonEmptyString } from "@/node/services/taskUtils"; +import { isWorkflowRunTaskId } from "@/node/services/tools/taskId"; + +const isWorkflowRunId = isWorkflowRunTaskId; + +function collectWorkflowRunIdsFromToolOutput(output: unknown): string[] { + if (output == null || typeof output !== "object") { + return []; + } + + const record = output as Record; + if (isWorkflowRunId(record.runId)) { + return [record.runId]; + } + + const results = record.results; + if (!Array.isArray(results)) { + return []; + } + + const runIds: string[] = []; + for (const result of results) { + if (result == null || typeof result !== "object") { + continue; + } + const taskId = (result as Record).taskId; + if (isWorkflowRunId(taskId)) { + runIds.push(taskId); + } + } + return runIds; +} + +/** + * Tolerant extraction of the task IDs a persisted task tool output references. Recovery + * bookkeeping must survive outputs written by newer releases (extra fields would fail the + * strict result schema) so that a mid-stream downgrade cannot duplicate fallback reports. + */ +export function collectReferencedTaskIdsFromTaskToolOutput( + output: unknown, + into: Set +): void { + if (output == null || typeof output !== "object") { + return; + } + const addTaskId = (value: unknown): void => { + if (typeof value === "string" && value.length > 0) { + into.add(value); + } + }; + + const record = output as Record; + addTaskId(record.taskId); + if (Array.isArray(record.taskIds)) { + for (const taskId of record.taskIds) { + addTaskId(taskId); + } + } + for (const key of ["tasks", "reports"] as const) { + const rows = record[key]; + if (!Array.isArray(rows)) { + continue; + } + for (const row of rows) { + if (row != null && typeof row === "object") { + addTaskId((row as Record).taskId); + } + } + } +} + +export type CompletedTaskReportReceipt = ( + | { kind: "initial"; taskId: string } + | { kind: "continuation"; taskId: string; handleId?: string; messageId: string } +) & { + reportMarkdown: string; + /** Where reportMarkdown starts in the full report, when the row is one paged page. */ + pageOffset?: number; +}; + +/** + * A paged row (task_await, or a foreground task since #6427) holds one page of a long report: + * its pageOffset is its startCharOffset, 0 for a first page that only carries nextCharOffset. + * Undefined for a whole report. + */ +function readReportPageOffset(row: Record): number | undefined { + if (typeof row.startCharOffset === "number") return row.startCharOffset; + return typeof row.nextCharOffset === "number" ? 0 : undefined; +} + +/** + * Whether a continuation receipt shows the owner this exact report: the whole text, or any + * non-empty page of it at the page's own offset. Reading one page (with its continuation note) + * delivers the result, so its terminal notification must not wake the owner again. + */ +export function receiptMatchesReport( + receipt: Extract, + reportMarkdown: string | undefined +): boolean { + if (receipt.reportMarkdown === reportMarkdown) return true; + if (reportMarkdown == null || receipt.pageOffset == null) return false; + return ( + receipt.reportMarkdown.length > 0 && + reportMarkdown.startsWith(receipt.reportMarkdown, receipt.pageOffset) + ); +} + +/** Read only canonical task-result containers, never arbitrary JSON or prose. */ +function completedTaskReports(toolName: unknown, output: unknown): CompletedTaskReportReceipt[] { + if (!isPlainObject(output)) return []; + const rows: unknown[] = []; + if (toolName === "task") { + if (output.status === "completed") rows.push(output); + // A group can return completed members while other members are still running. + if (Array.isArray(output.reports)) output.reports.forEach((row: unknown) => rows.push(row)); + } else if (toolName === "task_await" && Array.isArray(output.results)) { + output.results.forEach((row: unknown) => { + if (isPlainObject(row) && row.status === "completed") rows.push(row); + }); + } + return rows.flatMap((row): CompletedTaskReportReceipt[] => { + if ( + !isPlainObject(row) || + typeof row.taskId !== "string" || + typeof row.reportMarkdown !== "string" + ) + return []; + const messageId = + row.messageId ?? + (isPlainObject(row.finalMessageRef) ? row.finalMessageRef.messageId : undefined); + // Workspace-target tools expose a handle; stable-child awaits expose the child itself. + const handleId = isWorkspaceTurnTaskId(row.taskId) ? row.taskId : undefined; + const taskId = handleId != null ? coerceNonEmptyString(row.workspaceId) : row.taskId; + if (taskId == null) return []; + if (typeof messageId === "string") { + return [ + { + kind: "continuation", + taskId, + handleId, + messageId, + reportMarkdown: row.reportMarkdown, + pageOffset: readReportPageOffset(row), + }, + ]; + } + return handleId == null + ? [ + { + kind: "initial", + taskId, + reportMarkdown: row.reportMarkdown, + pageOffset: readReportPageOffset(row), + }, + ] + : []; + }); +} + +/** + * Same text and same page position. A visible projection that drops a page's offset fields + * (nextCharOffset/startCharOffset) hides that more text exists and where to read it, so it does + * not prove delivery of the genuine paged row. + */ +function sameReportPage(a: CompletedTaskReportReceipt, b: CompletedTaskReportReceipt): boolean { + return a.reportMarkdown === b.reportMarkdown && a.pageOffset === b.pageOffset; +} + +function sameTaskReportReceipt( + a: CompletedTaskReportReceipt, + b: CompletedTaskReportReceipt +): boolean { + return ( + a.kind === b.kind && + a.taskId === b.taskId && + sameReportPage(a, b) && + (a.kind === "initial" || + (b.kind === "continuation" && a.messageId === b.messageId && a.handleId === b.handleId)) + ); +} + +/** Only model-visible successful tool results prove consumption. UI telemetry does not. */ +export function collectCompletedTaskReportReceipts( + message: MuxMessage +): CompletedTaskReportReceipt[] { + const reports: CompletedTaskReportReceipt[] = []; + const succeeded = (call: unknown): call is Record => + isPlainObject(call) && + call.failed !== true && + call.error == null && + call.ok !== false && + (call.state == null || call.state === "output-available"); + const visit = (call: unknown, depth: number): void => { + if (depth > 30 || !succeeded(call)) return; + const output = call.output ?? call.result; + if (!isPlainObject(output)) return; + reports.push(...completedTaskReports(call.toolName, output)); + if (call.toolName !== "code_execution") return; + + // Classic executions expose these result records to the model, even when a later + // call fails. Persistent kernels replace them with summaries that contain no report. + if (Array.isArray(output.toolCalls)) { + for (const nested of output.toolCalls) visit(nested, depth + 1); + } + // Persistent nestedCalls hold UI-only full values. Use them solely as provenance for + // a complete canonical report deliberately returned through output.result. Offload + // handles, previews, hidden values, and unrelated returns must retain their wake. + if (Array.isArray(call.nestedCalls)) { + for (const nested of call.nestedCalls) { + if (!succeeded(nested)) continue; + const genuine = completedTaskReports(nested.toolName, nested.output); + const visible = completedTaskReports(nested.toolName, output.result); + reports.push( + ...visible.filter((receipt) => + genuine.some((source) => sameTaskReportReceipt(source, receipt)) + ) + ); + } + } + }; + for (const part of message.parts) { + if (isDynamicToolPart(part) && part.state === "output-available") visit(part, 0); + } + return reports; +} + +interface RecoveredTaskToolInput { + data: z.infer; + groupCount: number; + legacyVariants: boolean; +} + +/** + * Persisted partials may contain the removed `variants` input. Keep that field out of the + * model-facing schema while accepting it narrowly during crash recovery so completed legacy + * siblings can still finalize their parent tool call. + */ +export function parseTaskToolInputForRecovery(input: unknown): RecoveredTaskToolInput | null { + const current = TaskToolArgsSchema.safeParse(input); + if (current.success) { + return { + data: current.data, + groupCount: getTaskGroupCount(current.data), + legacyVariants: false, + }; + } + + if (!input || typeof input !== "object" || Array.isArray(input)) { + return null; + } + + const sanitized = { ...(input as Record) }; + const variants = sanitized.variants; + delete sanitized.variants; + const parsed = TaskToolArgsSchema.safeParse(sanitized); + if (!parsed.success) { + return null; + } + + if (variants == null) { + return { + data: parsed.data, + groupCount: getTaskGroupCount(parsed.data), + legacyVariants: false, + }; + } + if (!Array.isArray(variants) || variants.length < 1 || variants.length > 20) { + return null; + } + + const labels = variants.map((variant) => (typeof variant === "string" ? variant.trim() : "")); + if (labels.some((label) => label.length === 0) || new Set(labels).size !== labels.length) { + return null; + } + + return { + data: parsed.data, + groupCount: labels.length, + legacyVariants: true, + }; +} + +function collectWorkflowRunIdsFromTaskAwaitInput(input: unknown): string[] { + if (input == null || typeof input !== "object") { + return []; + } + const taskIds = (input as Record).task_ids; + if (!Array.isArray(taskIds)) { + return []; + } + return taskIds.filter(isWorkflowRunId); +} + +export function collectAgentReferencedWorkflowRunIdsFromParts( + parts: readonly unknown[], + knownAgentRunIds: ReadonlySet +): string[] { + const runIds = new Set(); + + for (const part of parts) { + if (!isDynamicToolPart(part) || part.state !== "output-available") { + continue; + } + // workflow_resume re-establishes agent provenance the same way workflow_run does: both + // outputs carry the runId of a run the agent explicitly owns. + if (!isWorkflowRunEmittingToolName(part.toolName)) { + continue; + } + for (const runId of collectWorkflowRunIdsFromToolOutput(part.output)) { + runIds.add(runId); + } + } + + const allowedTaskAwaitRunIds = new Set([...knownAgentRunIds, ...runIds]); + for (const part of parts) { + if (!isDynamicToolPart(part) || part.state !== "output-available") { + continue; + } + if (part.toolName !== "task_await") { + continue; + } + + // Omitted task_ids makes task_await discover every active run in the workspace, including + // slash-command runs. Only treat task_await output as agent provenance when the model either + // explicitly awaited that workflow ID in this turn or we already know the run was agent-started. + for (const runId of collectWorkflowRunIdsFromTaskAwaitInput(part.input)) { + runIds.add(runId); + allowedTaskAwaitRunIds.add(runId); + } + for (const runId of collectWorkflowRunIdsFromToolOutput(part.output)) { + if (allowedTaskAwaitRunIds.has(runId)) { + runIds.add(runId); + } + } + } + + return Array.from(runIds); +} diff --git a/src/node/services/workspaceService.ts b/src/node/services/workspaceService.ts index 4d1b082ba35..2b72473945d 100644 --- a/src/node/services/workspaceService.ts +++ b/src/node/services/workspaceService.ts @@ -491,6 +491,11 @@ import { type WorkspaceLiveActivity, } from "@/node/services/taskWorkspaceSeam"; import { findWorkspaceEntry } from "@/node/services/taskUtils"; +import { + isManualUserSupersessionMessage, + isResetBoundaryMessage, + isTerminalWorkflowTaskAwaitRecord, +} from "@/node/services/taskMessagePredicates"; import { MINTED_SEND_ID_PREFIX } from "@/common/orpc/schemas/stream"; import { computeSendDigest, @@ -850,31 +855,6 @@ function isTerminalWorkflowToolResultMessage(message: MuxMessage, runId: string) ); } -function isInternalResumeAutoCompactionMessage(message: MuxMessage): boolean { - const muxMetadata = message.metadata?.muxMetadata; - if (muxMetadata?.type !== "compaction-request" || muxMetadata.source !== "auto-compaction") { - return false; - } - return muxMetadata.parsed.followUpContent?.dispatchOptions?.source === "internal-resume"; -} - -function isSyntheticManualSupersessionMessage(message: MuxMessage): boolean { - const muxMetadata = message.metadata?.muxMetadata; - return ( - message.metadata?.synthetic === true && - muxMetadata?.type === "compaction-request" && - muxMetadata.source === "auto-compaction" && - !isInternalResumeAutoCompactionMessage(message) - ); -} - -function isManualUserSupersessionMessage(message: MuxMessage): boolean { - return ( - message.role === "user" && - (message.metadata?.synthetic !== true || isSyntheticManualSupersessionMessage(message)) - ); -} - function isWorkflowResultContinuationMessage(message: MuxMessage, runId: string): boolean { return ( message.metadata?.muxMetadata?.type === WORKFLOW_RESULT_METADATA_TYPE && @@ -899,34 +879,6 @@ function isCoalescedWorkflowResultMessage(message: MuxMessage, runId: string): b ); } -function isResetBoundaryMessage(message: MuxMessage): boolean { - return message.metadata?.contextBoundaryKind === CONTEXT_BOUNDARY_KINDS.RESET; -} - -function isFailedWorkflowRunSnapshot(value: unknown, runId: string): boolean { - if (value == null || typeof value !== "object") { - return false; - } - const record = value as Record; - return record.id === runId && record.status === "failed"; -} - -function isTerminalWorkflowTaskAwaitRecord( - record: Record, - runId: string -): boolean { - if (record.taskId !== runId) { - return false; - } - if (record.status === "completed" || record.status === "interrupted") { - return true; - } - if (record.status === "error") { - return isFailedWorkflowRunSnapshot(record.run, runId); - } - return false; -} - function isTerminalWorkflowTaskAwaitResultMessage(message: MuxMessage, runId: string): boolean { if (message.role !== "assistant") { return false;