diff --git a/src/tasks/task-executor.test.ts b/src/tasks/task-executor.test.ts index b59e85aac581..e30ee65d8a86 100644 --- a/src/tasks/task-executor.test.ts +++ b/src/tasks/task-executor.test.ts @@ -19,7 +19,6 @@ import { createRunningTaskRun as createRunningTaskRunOrNull, failTaskRunByRunId, recordTaskRunProgressByRunId, - retryBlockedFlowAsQueuedTaskRun, runTaskInFlow, runTaskInFlowForOwner, setDetachedTaskDeliveryStatusByRunId, @@ -381,75 +380,6 @@ describe("task-executor", () => { }); }); - it("records blocked metadata on one-task flows and reuses the same flow for queued retries", async () => { - await withTaskExecutorStateDir(async () => { - const created = createRunningTaskRun({ - runtime: "acp", - ownerKey: "agent:main:main", - scopeKind: "session", - requesterOrigin: { - channel: "notifychat", - to: "notifychat:123", - }, - childSessionKey: "agent:codex:acp:child", - runId: "run-executor-blocked", - task: "Patch file", - startedAt: 10, - deliveryStatus: "pending", - }); - - completeTaskRunByRunId({ - runId: "run-executor-blocked", - endedAt: 40, - lastEventAt: 40, - terminalOutcome: "blocked", - terminalSummary: "Writable session required.", - }); - - const blockedTask = getTaskById(created.taskId); - expect(blockedTask?.taskId).toBe(created.taskId); - expect(blockedTask?.status).toBe("succeeded"); - expect(blockedTask?.terminalOutcome).toBe("blocked"); - expect(blockedTask?.terminalSummary).toBe("Writable session required."); - const parentFlowId = expectParentFlowId(created); - const blockedFlow = getTaskFlowById(parentFlowId); - expect(blockedFlow?.flowId).toBe(parentFlowId); - expect(blockedFlow?.status).toBe("blocked"); - expect(blockedFlow?.blockedTaskId).toBe(created.taskId); - expect(blockedFlow?.blockedSummary).toBe("Writable session required."); - expect(blockedFlow?.endedAt).toBe(40); - - const retried = retryBlockedFlowAsQueuedTaskRun({ - flowId: parentFlowId, - runId: "run-executor-retry", - childSessionKey: "agent:codex:acp:retry-child", - }); - - expect(retried.found).toBe(true); - expect(retried.retried).toBe(true); - if (!retried.retried) { - throw new Error("Expected blocked flow retry"); - } - if (!retried.previousTask || !retried.task) { - throw new Error("Expected retry result payload"); - } - expect(retried.previousTask.taskId).toBe(created.taskId); - expect(retried.task.parentFlowId).toBe(parentFlowId); - expect(retried.task.parentTaskId).toBe(created.taskId); - expect(retried.task.status).toBe("queued"); - expect(retried.task.runId).toBe("run-executor-retry"); - const queuedFlow = getTaskFlowById(parentFlowId); - expect(queuedFlow?.flowId).toBe(parentFlowId); - expect(queuedFlow?.status).toBe("queued"); - expect(findLatestTaskForFlowId(parentFlowId)?.runId).toBe("run-executor-retry"); - const original = findTaskByRunId("run-executor-blocked"); - expect(original?.taskId).toBe(created.taskId); - expect(original?.status).toBe("succeeded"); - expect(original?.terminalOutcome).toBe("blocked"); - expect(original?.terminalSummary).toBe("Writable session required."); - }); - }); - it("cancels active tasks linked to a managed TaskFlow", async () => { await withTaskExecutorStateDir(async () => { hoisted.cancelSessionMock.mockResolvedValue(undefined); diff --git a/src/tasks/task-executor.ts b/src/tasks/task-executor.ts index 549b3753fc5f..d081f265f54e 100644 --- a/src/tasks/task-executor.ts +++ b/src/tasks/task-executor.ts @@ -10,7 +10,6 @@ import { getRegisteredDetachedTaskLifecycleRuntime } from "./detached-task-runti import { cancelTaskById, createTaskRecord, - findLatestTaskForFlowId, getTaskById, isParentFlowLinkError, linkTaskToFlowById, @@ -216,141 +215,6 @@ export function setDetachedTaskDeliveryStatusByRunId(params: { return setTaskRunDeliveryStatusByRunId(params); } -type RetryBlockedFlowResult = { - found: boolean; - retried: boolean; - reason?: string; - previousTask?: TaskRecord; - task?: TaskRecord; -}; - -type RetryBlockedFlowParams = { - flowId: string; - sourceId?: string; - requesterOrigin?: TaskDeliveryState["requesterOrigin"]; - childSessionKey?: string; - agentId?: string; - runId?: string; - label?: string; - task?: string; - preferMetadata?: boolean; - notifyPolicy?: TaskNotifyPolicy; - deliveryStatus?: TaskDeliveryStatus; - status: "queued" | "running"; - startedAt?: number; - lastEventAt?: number; - progressSummary?: string | null; -}; - -function resolveRetryableBlockedFlowTask(flowId: string): { - flowFound: boolean; - retryable: boolean; - latestTask?: TaskRecord; - reason?: string; -} { - const flow = getTaskFlowById(flowId); - if (!flow) { - return { - flowFound: false, - retryable: false, - reason: "Flow not found.", - }; - } - const latestTask = findLatestTaskForFlowId(flowId); - if (!latestTask) { - return { - flowFound: true, - retryable: false, - reason: "Flow has no retryable task.", - }; - } - if (flow.status !== "blocked") { - return { - flowFound: true, - retryable: false, - latestTask, - reason: "Flow is not blocked.", - }; - } - if (latestTask.status !== "succeeded" || latestTask.terminalOutcome !== "blocked") { - return { - flowFound: true, - retryable: false, - latestTask, - reason: "Latest TaskFlow task is not blocked.", - }; - } - return { - flowFound: true, - retryable: true, - latestTask, - }; -} - -function retryBlockedFlowTask(params: RetryBlockedFlowParams): RetryBlockedFlowResult { - const resolved = resolveRetryableBlockedFlowTask(params.flowId); - if (!resolved.retryable || !resolved.latestTask) { - return { - found: resolved.flowFound, - retried: false, - reason: resolved.reason, - }; - } - const flow = getTaskFlowById(params.flowId); - if (!flow) { - return { - found: false, - retried: false, - reason: "Flow not found.", - previousTask: resolved.latestTask, - }; - } - const task = createTaskRecord({ - runtime: resolved.latestTask.runtime, - sourceId: params.sourceId ?? resolved.latestTask.sourceId, - ownerKey: flow.ownerKey, - scopeKind: "session", - requesterOrigin: params.requesterOrigin ?? flow.requesterOrigin, - parentFlowId: flow.flowId, - childSessionKey: params.childSessionKey, - parentTaskId: resolved.latestTask.taskId, - agentId: params.agentId ?? resolved.latestTask.agentId, - runId: params.runId, - label: params.label ?? resolved.latestTask.label, - task: params.task ?? resolved.latestTask.task, - preferMetadata: params.preferMetadata, - notifyPolicy: params.notifyPolicy ?? resolved.latestTask.notifyPolicy, - deliveryStatus: params.deliveryStatus ?? "pending", - status: params.status, - startedAt: params.startedAt, - lastEventAt: params.lastEventAt, - progressSummary: params.progressSummary, - }); - if (!task) { - return { - found: true, - retried: false, - reason: "Task persistence failed.", - previousTask: resolved.latestTask, - }; - } - return { - found: true, - retried: true, - previousTask: resolved.latestTask, - task, - }; -} - -export function retryBlockedFlowAsQueuedTaskRun( - params: Omit, -): RetryBlockedFlowResult { - return retryBlockedFlowTask({ - ...params, - status: "queued", - }); -} - type CancelFlowResult = { found: boolean; cancelled: boolean;