diff --git a/src/plugins/runtime/runtime-taskflow.test.ts b/src/plugins/runtime/runtime-taskflow.test.ts index 0fd752d97757..6e3ce40bb321 100644 --- a/src/plugins/runtime/runtime-taskflow.test.ts +++ b/src/plugins/runtime/runtime-taskflow.test.ts @@ -1,6 +1,6 @@ // Runtime task-flow tests cover plugin task-flow registration and execution behavior. import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import { getTaskFlowById } from "../../tasks/task-flow-registry.js"; +import { createTaskFlowForTask, getTaskFlowById } from "../../tasks/task-flow-registry.js"; import { getTaskById } from "../../tasks/task-registry.js"; import { installRuntimeTaskDeliveryMock, @@ -8,6 +8,9 @@ import { } from "./runtime-task-test-harness.js"; import { createRuntimeTaskFlow } from "./runtime-taskflow.js"; +type BoundTaskFlow = ReturnType["bindSession"]>; +type MutationName = "setWaiting" | "resume" | "finish" | "fail" | "requestCancel"; + function requireCreatedFlow(flow: T | null): T { if (!flow) { throw new Error("expected managed TaskFlow creation to succeed"); @@ -137,4 +140,119 @@ describe("runtime TaskFlow", () => { expect(summary.total).toBe(1); expect(summary.active).toBe(1); }); + + it("applies each managed transition exactly once with its explicit payload", () => { + const taskFlow = createRuntimeTaskFlow().bindSession({ sessionKey: "agent:main:main" }); + const created = requireCreatedFlow( + taskFlow.createManaged({ + controllerId: "tests/runtime-taskflow/transitions", + goal: "Apply transitions", + }), + ); + const transitions: Array<[name: MutationName, input: Record, status: string]> = + [ + [ + "setWaiting", + { + currentStep: "await_review", + stateJson: { phase: "waiting" }, + waitJson: { kind: "approval" }, + blockedTaskId: "task-review", + blockedSummary: "Review required", + updatedAt: 20, + }, + "blocked", + ], + [ + "resume", + { + status: "running", + currentStep: "continue_work", + stateJson: { phase: "running" }, + updatedAt: 30, + }, + "running", + ], + ["finish", { stateJson: { phase: "done" }, updatedAt: 40, endedAt: 41 }, "succeeded"], + [ + "fail", + { + stateJson: { phase: "failed" }, + blockedTaskId: "task-failed", + blockedSummary: "Task failed", + updatedAt: 50, + endedAt: 51, + }, + "failed", + ], + ["requestCancel", { cancelRequestedAt: 60 }, "failed"], + ]; + + for (const [index, [name, input, status]] of transitions.entries()) { + const mutate = taskFlow[name] as BoundTaskFlow["setWaiting"]; + const result = mutate({ + flowId: created.flowId, + expectedRevision: index, + ...input, + }); + expect(result.applied, name).toBe(true); + if (!result.applied) { + throw new Error(`expected ${name} to apply`); + } + expect(result.flow, name).toMatchObject({ + ...input, + status, + flowId: created.flowId, + revision: index + 1, + }); + expect(getTaskFlowById(created.flowId)?.revision, name).toBe(index + 1); + } + }); + + it("rejects invalid mutation targets before writing and preserves conflict mapping", () => { + const runtime = createRuntimeTaskFlow(); + const ownerTaskFlow = runtime.bindSession({ sessionKey: "agent:main:main" }); + const otherTaskFlow = runtime.bindSession({ sessionKey: "agent:main:other" }); + const managed = requireCreatedFlow( + ownerTaskFlow.createManaged({ + controllerId: "tests/runtime-taskflow/auth", + goal: "Keep ownership", + }), + ); + + const denied = otherTaskFlow.setWaiting({ + flowId: managed.flowId, + expectedRevision: managed.revision, + }); + expect(denied).toEqual({ applied: false, code: "not_found" }); + expect(getTaskFlowById(managed.flowId)?.revision).toBe(0); + + const mirrored = requireCreatedFlow( + createTaskFlowForTask({ + task: { + ownerKey: "agent:main:main", + taskId: "task-mirrored", + notifyPolicy: "done_only", + status: "running", + task: "Mirror this task", + createdAt: 10, + lastEventAt: 10, + }, + }), + ); + const wrongMode = ownerTaskFlow.resume({ + flowId: mirrored.flowId, + expectedRevision: mirrored.revision, + }); + expect(wrongMode).toMatchObject({ applied: false, code: "not_managed" }); + + const conflict = ownerTaskFlow.finish({ flowId: managed.flowId, expectedRevision: 1 }); + expect(conflict).toMatchObject({ applied: false, code: "revision_conflict" }); + expect(getTaskFlowById(managed.flowId)).toMatchObject({ + revision: 0, + status: "queued", + }); + expect(getTaskFlowById(managed.flowId)?.endedAt).toBeUndefined(); + expect(getTaskFlowById(mirrored.flowId)?.revision).toBe(0); + }); }); diff --git a/src/plugins/runtime/runtime-taskflow.ts b/src/plugins/runtime/runtime-taskflow.ts index 464fb06a4131..1975598bb623 100644 --- a/src/plugins/runtime/runtime-taskflow.ts +++ b/src/plugins/runtime/runtime-taskflow.ts @@ -46,26 +46,6 @@ function asManagedTaskFlowRecord( return flow as ManagedTaskFlowRecord; } -function resolveManagedFlowForOwner(params: { - flowId: string; - ownerKey: string; -}): - | { ok: true; flow: ManagedTaskFlowRecord } - | { ok: false; code: "not_found" | "not_managed"; current?: TaskFlowRecord } { - const flow = getTaskFlowByIdForOwner({ - flowId: params.flowId, - callerOwnerKey: params.ownerKey, - }); - if (!flow) { - return { ok: false, code: "not_found" }; - } - const managed = asManagedTaskFlowRecord(flow); - if (!managed) { - return { ok: false, code: "not_managed", current: flow }; - } - return { ok: true, flow: managed }; -} - function mapFlowUpdateResult(result: TaskFlowUpdateResult): ManagedTaskFlowMutationResult { if (result.applied) { const managed = asManagedTaskFlowRecord(result.flow); @@ -88,6 +68,26 @@ function mapFlowUpdateResult(result: TaskFlowUpdateResult): ManagedTaskFlowMutat }; } +function applyManagedFlowMutationForOwner(params: { + flowId: string; + ownerKey: string; + mutate: (flowId: string) => TaskFlowUpdateResult; +}): ManagedTaskFlowMutationResult { + // Authorization and mode checks must complete before the mutation can touch persistence. + const flow = getTaskFlowByIdForOwner({ + flowId: params.flowId, + callerOwnerKey: params.ownerKey, + }); + if (!flow) { + return { applied: false, code: "not_found" }; + } + const managed = asManagedTaskFlowRecord(flow); + if (!managed) { + return { applied: false, code: "not_managed", current: flow }; + } + return mapFlowUpdateResult(params.mutate(managed.flowId)); +} + function createBoundTaskFlowRuntime(params: { sessionKey: string; requesterOrigin?: TaskDeliveryState["requesterOrigin"]; @@ -154,120 +154,75 @@ function createBoundTaskFlowRuntime(params: { }); return flow ? getFlowTaskSummary(flow.flowId) : undefined; }, - setWaiting: (input) => { - const flow = resolveManagedFlowForOwner({ + setWaiting: (input) => + applyManagedFlowMutationForOwner({ flowId: input.flowId, ownerKey, - }); - if (!flow.ok) { - return { - applied: false, - code: flow.code, - ...(flow.current ? { current: flow.current } : {}), - }; - } - return mapFlowUpdateResult( - setFlowWaiting({ - flowId: flow.flow.flowId, - expectedRevision: input.expectedRevision, - currentStep: input.currentStep, - stateJson: input.stateJson, - waitJson: input.waitJson, - blockedTaskId: input.blockedTaskId, - blockedSummary: input.blockedSummary, - updatedAt: input.updatedAt, - }), - ); - }, - resume: (input) => { - const flow = resolveManagedFlowForOwner({ + mutate: (flowId) => + setFlowWaiting({ + flowId, + expectedRevision: input.expectedRevision, + currentStep: input.currentStep, + stateJson: input.stateJson, + waitJson: input.waitJson, + blockedTaskId: input.blockedTaskId, + blockedSummary: input.blockedSummary, + updatedAt: input.updatedAt, + }), + }), + resume: (input) => + applyManagedFlowMutationForOwner({ flowId: input.flowId, ownerKey, - }); - if (!flow.ok) { - return { - applied: false, - code: flow.code, - ...(flow.current ? { current: flow.current } : {}), - }; - } - return mapFlowUpdateResult( - resumeFlow({ - flowId: flow.flow.flowId, - expectedRevision: input.expectedRevision, - status: input.status, - currentStep: input.currentStep, - stateJson: input.stateJson, - updatedAt: input.updatedAt, - }), - ); - }, - finish: (input) => { - const flow = resolveManagedFlowForOwner({ + mutate: (flowId) => + resumeFlow({ + flowId, + expectedRevision: input.expectedRevision, + status: input.status, + currentStep: input.currentStep, + stateJson: input.stateJson, + updatedAt: input.updatedAt, + }), + }), + finish: (input) => + applyManagedFlowMutationForOwner({ flowId: input.flowId, ownerKey, - }); - if (!flow.ok) { - return { - applied: false, - code: flow.code, - ...(flow.current ? { current: flow.current } : {}), - }; - } - return mapFlowUpdateResult( - finishFlow({ - flowId: flow.flow.flowId, - expectedRevision: input.expectedRevision, - stateJson: input.stateJson, - updatedAt: input.updatedAt, - endedAt: input.endedAt, - }), - ); - }, - fail: (input) => { - const flow = resolveManagedFlowForOwner({ + mutate: (flowId) => + finishFlow({ + flowId, + expectedRevision: input.expectedRevision, + stateJson: input.stateJson, + updatedAt: input.updatedAt, + endedAt: input.endedAt, + }), + }), + fail: (input) => + applyManagedFlowMutationForOwner({ flowId: input.flowId, ownerKey, - }); - if (!flow.ok) { - return { - applied: false, - code: flow.code, - ...(flow.current ? { current: flow.current } : {}), - }; - } - return mapFlowUpdateResult( - failFlow({ - flowId: flow.flow.flowId, - expectedRevision: input.expectedRevision, - stateJson: input.stateJson, - blockedTaskId: input.blockedTaskId, - blockedSummary: input.blockedSummary, - updatedAt: input.updatedAt, - endedAt: input.endedAt, - }), - ); - }, - requestCancel: (input) => { - const flow = resolveManagedFlowForOwner({ + mutate: (flowId) => + failFlow({ + flowId, + expectedRevision: input.expectedRevision, + stateJson: input.stateJson, + blockedTaskId: input.blockedTaskId, + blockedSummary: input.blockedSummary, + updatedAt: input.updatedAt, + endedAt: input.endedAt, + }), + }), + requestCancel: (input) => + applyManagedFlowMutationForOwner({ flowId: input.flowId, ownerKey, - }); - if (!flow.ok) { - return { - applied: false, - code: flow.code, - ...(flow.current ? { current: flow.current } : {}), - }; - } - return mapFlowUpdateResult( - requestFlowCancel({ - flowId: flow.flow.flowId, - expectedRevision: input.expectedRevision, - cancelRequestedAt: input.cancelRequestedAt, - }), - ); - }, + mutate: (flowId) => + requestFlowCancel({ + flowId, + expectedRevision: input.expectedRevision, + cancelRequestedAt: input.cancelRequestedAt, + }), + }), cancel: ({ flowId, cfg }) => cancelFlowByIdForOwner({ cfg,