mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
refactor(tasks): centralize managed flow mutations (#104994)
This commit is contained in:
Vincent Koc
@@ -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<ReturnType<typeof createRuntimeTaskFlow>["bindSession"]>;
|
||||
type MutationName = "setWaiting" | "resume" | "finish" | "fail" | "requestCancel";
|
||||
|
||||
function requireCreatedFlow<T>(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<string, unknown>, 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);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user