diff --git a/docs/content/api/conductor.md b/docs/content/api/conductor.md index 782bce8..1bfbdef 100644 --- a/docs/content/api/conductor.md +++ b/docs/content/api/conductor.md @@ -78,7 +78,7 @@ const task = conductor.createTask( timeoutMs?: number; // Fail an attempt that runs longer (default: none) concurrency?: number; // Max concurrent executions (default: unlimited) window?: [string, string]; // Time window (e.g., ["09:00", "17:00"]) - removeOnComplete?: boolean | { days: number }; // Retention policy (true: at the next maintenance run) + removeOnComplete?: boolean | { days: number }; // Retention policy (true: when it settles) removeOnFail?: boolean | { days: number }; // Retention policy batch?: { size: number; timeoutMs: number }; // Batch processing config deadLetter?: { queue: string; task?: Task }; // Final-failure destination diff --git a/docs/content/crafting-tasks/child-invocation.md b/docs/content/crafting-tasks/child-invocation.md index 70c7f0a..fb32b18 100644 --- a/docs/content/crafting-tasks/child-invocation.md +++ b/docs/content/crafting-tasks/child-invocation.md @@ -175,6 +175,8 @@ const workflow = conductor.createTask( Unlike `ctx.invokeAndWait()`, the outcome is a value: a failed or cancelled execution resolves the wait instead of failing the caller, and a timeout leaves the execution running. To race the execution against events, pass it as an `{ execution: id }` branch to [`ctx.waitForAny()`](../api/task-context.md#ctxwaitforany). +To report an outcome to another execution, see [Reporting Outcomes](report-outcomes.md). + Outside a task, await an execution with [`conductor.waitForResult()`](../api/conductor.md) or read it with `conductor.getExecution()`. ## What's Next? diff --git a/docs/content/crafting-tasks/report-outcomes.md b/docs/content/crafting-tasks/report-outcomes.md new file mode 100644 index 0000000..474a651 --- /dev/null +++ b/docs/content/crafting-tasks/report-outcomes.md @@ -0,0 +1,131 @@ +# Report an Execution's Outcome to Another Execution + +Postgres Conductor has no hook that runs when an execution settles. To tell another execution how one ended, start a small task that waits for the outcome with [`ctx.waitForResult()`](../api/task-context.md#ctxwaitforresult) and reports it with a signal. The examples on this page are compiled and run by `tests/integration/outcome-recipe.test.ts`. + +## Supervisor: Report to the Target + +A supervisor starts the work, waits for its outcome and sends it to the target with [`ctx.signalWithStart()`](../task-execution/signals.md#signal-or-start): + +```typescript +const watchToolCall = conductor.createTask( + { name: "watch-tool-call" }, + { invocable: true }, + async (event, ctx) => { + const id = await ctx.invoke("start", { name: "tool-call" }, event.payload); + const outcome = await ctx.waitForResult("result", { name: "tool-call" }, id); + await ctx.signalWithStart("report", { + task: "session", + key: event.payload.sessionKey, + signal: { + name: "tool-settled", + id: `tool-settled:${id}`, + payload: { executionId: id, outcome }, + }, + }); + } +); + +const session = conductor.createTask( + { name: "session" }, + { invocable: true }, + async (event, ctx) => { + ctx.onSignal("tool-settled", async (payload, { id }) => { + // payload is { executionId, outcome }, and outcome is { status: "completed", result }, + // { status: "failed", error }, { status: "cancelled", reason } or { status: "timeout" } + }); + // ... + } +); +``` + +Invoke the supervisor instead of the work: `conductor.invoke({ name: "watch-tool-call" }, { sessionKey, input })`. + +- The outcome is a value: a failed or cancelled execution resolves the wait instead of failing the supervisor. Pass `timeout` to `ctx.waitForResult()` to report `{ status: "timeout" }` and leave the work running +- `ctx.signalWithStart()` signals the session running for the key, or starts one that receives the signal first. A short-lived target, such as a session that finished before the tool did, is started again for the report. Register the listener before the handler awaits anything, as for [any signal](../task-execution/signals.md#listening-for-signals) + +## Deterministic Signal IDs + +Derive the signal id from the watched execution, as `tool-settled:${id}` does, never from a random value or the time. + +`ctx.signalWithStart()` is memoized by its step key, so a supervisor that is retried or resumed after the report does not send it again. `ctx.sendSignal()` is not memoized and runs again on every replay, and a signal id is stored once per execution, so a deterministic id keeps a replayed send from storing a second signal. Delivery to `ctx.onSignal()` is at least once per id either way, so the target dedupes by `id` as well. + +## Watcher: React to the Caller + +The other direction is an execution that watches its caller and shuts down the work it owns when the caller fails or is cancelled. The work starts the watcher, linked with `cancelWithParent` so the watcher ends with it: + +```typescript +const sandbox = conductor.createTask( + { name: "sandbox" }, + { invocable: true }, + async (event, ctx) => { + ctx.onSignal("shutdown", async () => { + await stopSandbox(); + }); + await ctx.invoke( + "watch-caller", + { name: "watch-caller" }, + { callerId: event.payload.callerId, workId: ctx.executionId }, + { cancelWithParent: true } + ); + // ... + } +); + +const watchCaller = conductor.createTask( + { name: "watch-caller" }, + { invocable: true }, + async (event, ctx) => { + const { callerId, workId } = event.payload; + const outcome = await ctx.waitForResult("caller", callerId); + if (outcome.status === "completed") return; + await ctx.sendSignal(workId, { + name: "shutdown", + id: `shutdown:${callerId}`, + payload: outcome, + }); + } +); +``` + +The caller starts the sandbox with `ctx.invoke("sandbox", { name: "sandbox" }, { callerId: ctx.executionId })`. To stop the work without letting it clean up, call `ctx.cancel(workId, { reason })` instead of sending a signal. Starting the work with [`cancelWithParent`](child-invocation.md#cancelling-with-the-parent) does that without a watcher, but only for work the caller starts itself. + +## Race an Execution Against Events + +A `{ execution: id }` branch of [`ctx.waitForAny()`](../api/task-context.md#ctxwaitforany) settles with the execution, so the first of an event and an outcome wins. For example, remove an approval card when the deploy ends before anyone decides: + +```typescript +const approvalCard = conductor.createTask( + { name: "approval-card" }, + { invocable: true }, + async (event, ctx) => { + const { deployId } = event.payload; + const decided = await ctx.subscribe("decided", { + event: approvalDecided, + filter: { executionId: [deployId] }, + }); + const winner = await ctx.waitForAny("decision", { + decided, + deploy: { execution: deployId }, + }); + if (winner.key === "deploy") { + await ctx.step("remove-card", () => removeCard(deployId)); + } + } +); +``` + +## Retention + +The waiting task needs the watched execution's row, which retention removes. `ctx.waitForResult()` and execution branches of `ctx.waitForAny()` register on the row: + +- A wait registered before the execution is removed always receives the outcome, also when the execution is removed at the moment it settles (`removeOnComplete: true`) +- A wait registered after the execution was removed throws `Execution not found`. Uncaught, this fails the attempt like any error, and its retries throw again + +[`conductor.waitForResult()`](../api/conductor.md) polls the row instead, so it rejects with `Execution not found` once the execution is removed, even if it started before. + +Give the watched task a retention that outlives the wait, or start the wait before the work can finish. The supervisor above calls `ctx.invoke()` and `ctx.waitForResult()` in one attempt, so with `removeOnComplete: true` or `removeOnFail: true` on the work, a tool that finishes in between is removed before the wait registers. With `{ days: n }`, maintenance removes an execution no earlier than `n` days after it settled, which leaves the supervisor ample time. + +## Compared to a Settlement Hook + +- Every watched call costs one extra execution: the supervisor or watcher +- The report is only as reliable as the supervisor: if it is cancelled or fails permanently and lands in a [dead-letter queue](../task-execution/dead-letter-queue.md), nothing is sent diff --git a/docs/content/crafting-tasks/retention.md b/docs/content/crafting-tasks/retention.md index 5f59002..1f1e604 100644 --- a/docs/content/crafting-tasks/retention.md +++ b/docs/content/crafting-tasks/retention.md @@ -40,7 +40,7 @@ Old executions are automatically removed by the maintenance task. } ``` -**Remove completed at the next maintenance run:** +**Remove completed as soon as it settles:** ```typescript { @@ -108,7 +108,7 @@ Postgres Conductor runs daily maintenance automatically on each queue: No manual intervention required. Maintenance does not release executions locked by a crashed worker: running orchestrators recover those through their heartbeats. -An execution's result is stored on the execution, so [`conductor.getExecution()` and `conductor.waitForResult()`](../api/conductor.md#conductorgetexecution) can only read it until the execution is removed. [`ctx.waitForResult()`](../api/task-context.md#ctxwaitforresult) receives the outcome if it started waiting before the removal, and throws if it starts after. +An execution's result is stored on the execution, so [`conductor.getExecution()` and `conductor.waitForResult()`](../api/conductor.md#conductorgetexecution) can only read it until the execution is removed. [`ctx.waitForResult()`](../api/task-context.md#ctxwaitforresult) receives the outcome if it started waiting before the removal, and throws if it starts after; see [Reporting Outcomes](report-outcomes.md#retention). ## What's Next? diff --git a/docs/zensical.toml b/docs/zensical.toml index 0ed7d61..0c89baf 100644 --- a/docs/zensical.toml +++ b/docs/zensical.toml @@ -58,6 +58,7 @@ nav = [ { "Task Triggers" = "crafting-tasks/triggers.md" }, { "Retries and Steps" = "crafting-tasks/retries-and-steps.md" }, { "Child Invocation" = "crafting-tasks/child-invocation.md" }, + { "Reporting Outcomes" = "crafting-tasks/report-outcomes.md" }, { "Long-Running Tasks" = "crafting-tasks/long-running.md" }, { "Custom Context" = "crafting-tasks/custom-context.md" }, { "Execution Metadata" = "crafting-tasks/execution-metadata.md" }, diff --git a/packages/pgconductor-js/tests/integration/outcome-recipe.test.ts b/packages/pgconductor-js/tests/integration/outcome-recipe.test.ts new file mode 100644 index 0000000..4c950dc --- /dev/null +++ b/packages/pgconductor-js/tests/integration/outcome-recipe.test.ts @@ -0,0 +1,241 @@ +import { afterAll, afterEach, beforeAll, describe, expect, test } from "vitest"; +import { z } from "zod"; +import { Conductor } from "../../src/conductor"; +import { Orchestrator } from "../../src/orchestrator"; +import { defineEvent } from "../../src/event-definition"; +import { defineTask } from "../../src/task-definition"; +import { TaskSchemas } from "../../src/schemas"; +import type { Task } from "../../src/task"; +import type { JsonValue } from "../../src/database-client"; +import { Deferred } from "../../src/lib/deferred"; +import { TestDatabasePool, type TestDatabase } from "../fixtures/test-database"; +import { waitForCondition } from "../test-utils"; + +// The examples of docs/content/crafting-tasks/report-outcomes.md + +const approvalDecided = defineEvent({ + name: "approval.decided", + payload: z.object({ executionId: z.string(), approved: z.boolean() }), + filterable: ["executionId"], +}); +const toolCall = defineTask({ + name: "tool-call", + payload: z.object({ sessionKey: z.string(), input: z.string() }), + returns: z.object({ output: z.string() }), +}); +const watchToolCall = defineTask({ + name: "watch-tool-call", + payload: z.object({ sessionKey: z.string(), input: z.string() }), +}); +const session = defineTask({ name: "session" }); +const agentRun = defineTask({ name: "agent-run" }); +const sandbox = defineTask({ name: "sandbox", payload: z.object({ callerId: z.string() }) }); +const watchCaller = defineTask({ + name: "watch-caller", + payload: z.object({ callerId: z.string(), workId: z.string() }), +}); +const deploy = defineTask({ name: "deploy" }); +const approvalCard = defineTask({ + name: "approval-card", + payload: z.object({ deployId: z.string() }), +}); + +describe("report an execution's outcome to another execution", () => { + let pool: TestDatabasePool; + const databases: TestDatabase[] = []; + const orchestrators: Orchestrator[] = []; + + beforeAll(async () => { + pool = await TestDatabasePool.create(); + }, 60000); + + afterEach(async () => { + await Promise.allSettled(orchestrators.map((orchestrator) => orchestrator.stop())); + orchestrators.length = 0; + await Promise.all(databases.map((database) => database.destroy())); + databases.length = 0; + }); + + afterAll(async () => { + await pool?.destroy(); + }); + + async function setup() { + const database = await pool.child(); + databases.push(database); + const conductor = Conductor.create({ + sql: database.sql, + tasks: TaskSchemas.fromSchema([ + toolCall, + watchToolCall, + session, + agentRun, + sandbox, + watchCaller, + deploy, + approvalCard, + ]), + events: [approvalDecided], + context: {}, + }); + await conductor.ensureInstalled(); + return { database, conductor }; + } + + async function start( + conductor: Awaited>["conductor"], + tasks: readonly Task[], + ) { + const orchestrator = Orchestrator.create({ + conductor, + tasks, + defaultWorker: { pollIntervalMs: 10, flushIntervalMs: 10, concurrency: 8 }, + }); + orchestrators.push(orchestrator); + await orchestrator.start(); + } + + test("a supervisor reports the outcome to a session, restarting it when it finished", async () => { + const { database, conductor } = await setup(); + const received: { sessionId: string; id: string; payload: JsonValue }[] = []; + await start(conductor, [ + conductor.createTask( + { name: "tool-call", maxAttempts: 1 }, + { invocable: true }, + async (event) => { + if (event.payload.input === "fail") throw new Error("tool failed"); + return { output: event.payload.input.toUpperCase() }; + }, + ), + conductor.createTask({ name: "watch-tool-call" }, { invocable: true }, async (event, ctx) => { + const id = await ctx.invoke("start", { name: "tool-call" }, event.payload); + const outcome = await ctx.waitForResult("result", { name: "tool-call" }, id); + await ctx.signalWithStart("report", { + task: "session", + key: event.payload.sessionKey, + signal: { + name: "tool-settled", + id: `tool-settled:${id}`, + payload: { executionId: id, outcome }, + }, + }); + }), + conductor.createTask( + { name: "session", maxAttempts: 1 }, + { invocable: true }, + async (_event, ctx) => { + ctx.onSignal("tool-settled", async (payload, { id }) => { + received.push({ sessionId: ctx.executionId, id, payload }); + }); + }, + ), + ]); + + const first = await conductor.invoke( + { name: "watch-tool-call" }, + { sessionKey: "s1", input: "hi" }, + ); + await conductor.waitForResult(first, { pollIntervalMs: 10 }); + await waitForCondition(() => received.length === 1); + await conductor.waitForResult(received[0]?.sessionId || "", { pollIntervalMs: 10 }); + const second = await conductor.invoke( + { name: "watch-tool-call" }, + { sessionKey: "s1", input: "fail" }, + ); + await conductor.waitForResult(second, { pollIntervalMs: 10 }); + await waitForCondition(() => received.length === 2); + + const toolIds = await database.sql<{ id: string }[]>` + select id from pgconductor._private_executions where task_key = 'tool-call' order by created_at + `; + const [okId, failedId] = toolIds.map((row) => row.id); + expect(received.map(({ id, payload }) => ({ id, payload }))).toEqual([ + { + id: `tool-settled:${okId}`, + payload: { executionId: okId, outcome: { status: "completed", result: { output: "HI" } } }, + }, + { + id: `tool-settled:${failedId}`, + payload: { executionId: failedId, outcome: { status: "failed", error: "tool failed" } }, + }, + ]); + expect(received[0]?.sessionId).not.toBe(received[1]?.sessionId); + }); + + test("a watcher shuts down the work it owns when its caller is cancelled", async () => { + const { conductor } = await setup(); + const sandboxIds = new Deferred(); + const stopped = new Deferred(); + await start(conductor, [ + conductor.createTask({ name: "agent-run" }, { invocable: true }, async (_event, ctx) => { + const id = await ctx.invoke("sandbox", { name: "sandbox" }, { callerId: ctx.executionId }); + sandboxIds.resolve(id); + await ctx.sleep("idle", 3_600_000); + }), + conductor.createTask({ name: "sandbox" }, { invocable: true }, async (event, ctx) => { + const shutdown = new Deferred(); + ctx.onSignal("shutdown", async (payload) => { + stopped.resolve(payload); + shutdown.resolve(); + }); + await ctx.invoke( + "watch-caller", + { name: "watch-caller" }, + { callerId: event.payload.callerId, workId: ctx.executionId }, + { cancelWithParent: true }, + ); + await shutdown.promise; + }), + conductor.createTask({ name: "watch-caller" }, { invocable: true }, async (event, ctx) => { + const { callerId, workId } = event.payload; + const outcome = await ctx.waitForResult("caller", callerId); + if (outcome.status === "completed") return; + await ctx.sendSignal(workId, { + name: "shutdown", + id: `shutdown:${callerId}`, + payload: outcome, + }); + }), + ]); + + const runId = await conductor.invoke({ name: "agent-run" }, {}); + const sandboxId = await sandboxIds.promise; + await waitForCondition(async () => (await conductor.getExecution(runId))?.status === "pending"); + expect(await conductor.cancel(runId, { reason: "user stopped the run" })).toBe(true); + + expect(await stopped.promise).toEqual({ status: "cancelled", reason: "user stopped the run" }); + await conductor.waitForResult(sandboxId, { pollIntervalMs: 10 }); + }); + + test("an approval card is removed when the deploy ends before a decision", async () => { + const { conductor } = await setup(); + const release = new Deferred(); + const removed = new Deferred(); + await start(conductor, [ + conductor.createTask({ name: "deploy" }, { invocable: true }, async () => release.promise), + conductor.createTask({ name: "approval-card" }, { invocable: true }, async (event, ctx) => { + const { deployId } = event.payload; + const decided = await ctx.subscribe("decided", { + event: approvalDecided, + filter: { executionId: [deployId] }, + }); + const winner = await ctx.waitForAny("decision", { + decided, + deploy: { execution: deployId }, + }); + if (winner.key === "deploy") { + await ctx.step("remove-card", async () => removed.resolve(winner.outcome.status)); + } + }), + ]); + + const deployId = await conductor.invoke({ name: "deploy" }, {}); + const cardId = await conductor.invoke({ name: "approval-card" }, { deployId }); + await waitForCondition( + async () => (await conductor.getExecution(cardId))?.status === "pending", + ); + release.resolve(); + + expect(await removed.promise).toBe("completed"); + }); +}); diff --git a/packages/pgconductor-js/tests/integration/wait-for-result.test.ts b/packages/pgconductor-js/tests/integration/wait-for-result.test.ts index ab2d596..ac51c2b 100644 --- a/packages/pgconductor-js/tests/integration/wait-for-result.test.ts +++ b/packages/pgconductor-js/tests/integration/wait-for-result.test.ts @@ -62,7 +62,11 @@ describe("ctx.waitForResult and execution branches (Postgres integration)", () = }: { waiter: Handler; work?: (value: number, ctx: any) => Promise; - workOptions?: { maxAttempts?: number; removeOnComplete?: { days: number } }; + workOptions?: { + maxAttempts?: number; + removeOnComplete?: boolean | { days: number }; + removeOnFail?: boolean | { days: number }; + }; orchestratorCount?: number; }, ) { @@ -482,6 +486,49 @@ describe("ctx.waitForResult and execution branches (Postgres integration)", () = expect(await conductor.getExecution(ids[0] || "")).toBeNull(); }); + test.each([ + { status: "completed", work: async () => {}, workOptions: { removeOnComplete: true } }, + { + status: "failed", + work: async () => { + throw new Error("boom"); + }, + workOptions: { maxAttempts: 1, removeOnFail: true }, + }, + ])( + "a $status execution removed before the wait fails it as not found", + async ({ work, workOptions }) => { + const database = await db(); + const errors: string[] = []; + const ids: string[] = []; + const fail = (error: Error) => { + errors.push(error.message); + }; + const conductor = await setup(database, { + waiter: async (_id, ctx) => { + const id = await ctx.invoke("start", { name: "work" }, { value: 1 }); + ids.push(id); + await ctx.step("await-removal", () => + waitForCondition(async () => (await conductor.getExecution(id)) === null), + ); + await ctx.waitForResult("result", id).catch(fail); + await ctx.waitForAny("any", { work: { execution: id } }).catch(fail); + }, + work, + workOptions, + }); + + const waiterId = await conductor.invoke({ name: "waiter" }, { id: "removed-before" }); + await conductor.waitForResult(waiterId, { pollIntervalMs: 10 }); + const workId = ids[0] || ""; + await conductor.waitForResult(workId, { pollIntervalMs: 10 }).catch(fail); + + const notFound = `Execution ${workId} not found`; + expect(errors).toEqual([notFound, notFound, notFound]); + expect(await waits(database)).toBe(0); + }, + ); + test("an unknown execution fails the wait", async () => { const database = await db(); const errors: string[] = [];