From f900a1a89e918c4132e519af8cd4c7d29d1750be Mon Sep 17 00:00:00 2001 From: Joe Date: Wed, 12 Aug 2026 06:20:02 +0200 Subject: [PATCH] feat(workflows): add ctx.workflows.dispatch() API for programmatic workflow triggering --- docs/writing-extensions.md | 36 ++++++ src/extensions/core/workflows/engine.ts | 10 +- src/extensions/core/workflows/index.test.ts | 3 + src/extensions/core/workflows/index.ts | 24 ++++ .../engine/extensionContext.test.ts | 121 ++++++++++++++++++ src/extensions/engine/extensionContext.ts | 29 +++++ src/extensions/types.ts | 42 ++++++ 7 files changed, 256 insertions(+), 9 deletions(-) create mode 100644 src/extensions/engine/extensionContext.test.ts diff --git a/docs/writing-extensions.md b/docs/writing-extensions.md index 82c2d4d..cb27a87 100644 --- a/docs/writing-extensions.md +++ b/docs/writing-extensions.md @@ -263,6 +263,42 @@ Avoid unnecessary prefixes like `/admin/` - all extension routes are already beh | `ctx.messaging.broadcast(message)` | Push a WebSocket message to all frontend clients | | `ctx.messaging.push(sessionId, content, opts?)` | Send a push message to a session | +### Workflows (`ctx.workflows`) + +| Method | Description | +| --- | --- | +| `ctx.workflows.dispatch(name, payload?)` | Trigger a named workflow run and return the run ID + step job IDs | + +Dispatch a workflow programmatically without HTTP self-calls. The method looks up the workflow definition by name, validates it is enabled, creates a run with the provided payload as trigger data, and broadcasts a `workflow_started` WebSocket event. + +```typescript +async initialize(ctx) { + ctx.routes.register("POST", "/process", async (reqCtx) => { + const body = reqCtx.body as { filePath: string }; + + const result = await ctx.workflows.dispatch("invoice-process", { + filePath: body.filePath, + project: "default", + }); + + return Response.json({ + workflowRunId: result.workflowRunId, + jobIds: result.jobIds, + }); + }); +} +``` + +**Error cases:** + +| Condition | Error message | +| --- | --- | +| Workflow name not found in loaded definitions | `Workflow not found: ` | +| Workflow exists but has `enabled: false` | `Workflow is disabled: ` | +| Called before the workflows core extension initializes | `Workflows extension not initialized` | + +Extensions that use `ctx.workflows.dispatch()` should declare `"workflows"` in their manifest `dependencies` to ensure correct initialization order. + ### Agent Execution (`ctx.agent`) | Method | Description | diff --git a/src/extensions/core/workflows/engine.ts b/src/extensions/core/workflows/engine.ts index d338a42..310bbfe 100644 --- a/src/extensions/core/workflows/engine.ts +++ b/src/extensions/core/workflows/engine.ts @@ -3,7 +3,7 @@ * via bunqueue's {@link FlowProducer.addChain}. */ -import type { Logger } from "@ext/types"; +import type { Logger, WorkflowDispatchResult } from "@ext/types"; import type { FlowProducer, FlowStep } from "bunqueue/client"; import type { WorkflowDefinition } from "./schemas"; import type { WorkflowStepJobData } from "./types"; @@ -16,14 +16,6 @@ export interface SessionFactory { create(opts: { source: string; sourceId?: string; metadata?: Record }): { id: string }; } -/** Result of dispatching a workflow. */ -export interface WorkflowDispatchResult { - /** Unique identifier for this workflow run. */ - workflowRunId: string; - /** Job IDs for each step in the chain. */ - jobIds: string[]; -} - /** Queue name used for all workflow step jobs. */ export const WORKFLOW_STEPS_QUEUE = "workflows:steps"; diff --git a/src/extensions/core/workflows/index.test.ts b/src/extensions/core/workflows/index.test.ts index 7af5cee..caee86d 100644 --- a/src/extensions/core/workflows/index.test.ts +++ b/src/extensions/core/workflows/index.test.ts @@ -189,6 +189,9 @@ function createMockContext(workDir: string) { dynamicItems: { register: () => {}, }, + workflows: { + dispatch: async () => ({ workflowRunId: "run-1", jobIds: ["job-1"] }), + }, }; return { ctx, routes }; diff --git a/src/extensions/core/workflows/index.ts b/src/extensions/core/workflows/index.ts index 5d9465a..4ca4498 100644 --- a/src/extensions/core/workflows/index.ts +++ b/src/extensions/core/workflows/index.ts @@ -26,6 +26,7 @@ import type { Extension, ExtensionContext, ExtensionManifest, Logger } from "@ex import type { AgentEvent } from "@mariozechner/pi-agent-core"; import { Value } from "@sinclair/typebox/value"; import { serverOrigin } from "@src/config"; +import { setWorkflowDispatchFn } from "@src/extensions/engine/extensionContext"; import { SANDBOX_TOOL_NAMES } from "@src/tools/file"; import type { SessionFactory } from "./engine"; import { dispatchWorkflow } from "./engine"; @@ -279,6 +280,29 @@ export function createExtension(): Extension { for (const [k, v] of loaded) store.set(k, v); logger.info(`Loaded ${store.size} workflow definition(s)`); + // Register the dispatch function so all extension contexts can use ctx.workflows.dispatch() + setWorkflowDispatchFn(async (name, payload) => { + const wf = store.get(name); + if (!wf) { + throw new Error(`Workflow not found: ${name}`); + } + if (wf.enabled === false) { + throw new Error(`Workflow is disabled: ${name}`); + } + const result = await dispatchWorkflow(flowProducer, wf, payload ?? null, logger, sessionFactory); + ctx.messaging.broadcast({ + type: "workflow_started", + workflowRunId: result.workflowRunId, + workflowName: wf.name, + steps: wf.steps.map((s, i) => ({ + slug: s.slug, + type: s.type, + jobId: result.jobIds[i], + })), + }); + return result; + }); + // Watch for file changes and hot-reload try { state.watcher = watch(state.workflowsDir, (_event, filename) => { diff --git a/src/extensions/engine/extensionContext.test.ts b/src/extensions/engine/extensionContext.test.ts new file mode 100644 index 0000000..494bc5a --- /dev/null +++ b/src/extensions/engine/extensionContext.test.ts @@ -0,0 +1,121 @@ +/** + * Integration tests for the ctx.workflows.dispatch late-bound dispatch mechanism. + */ + +import { describe, expect, test } from "bun:test"; +import type { WorkflowDispatchResult } from "@ext/types"; +import { EventBus } from "./eventBus"; +import { createExtensionContext, setWorkflowDispatchFn } from "./extensionContext"; + +/** Creates minimal deps for createExtensionContext (only what workflows.dispatch needs). */ +function createMinimalDeps() { + return { + extensionName: "test-extension", + workDir: "/tmp/test-work", + dataDir: "/tmp/test-data", + extensionsDir: "/tmp/test-extensions", + toolNameSet: new Set(), + routeKeySet: new Set(), + stepTypeNameSet: new Set(), + eventBus: new EventBus(), + flowProducer: { addChain: async () => ({ jobs: [] }) } as any, + runAgentFn: async () => ({ answer: "", state: null, timestamp: Date.now() }), + database: {} as any, + sessionStore: { + create: () => ({ id: "s-1", source: "test", messages: [], createdAt: Date.now(), updatedAt: Date.now() }), + } as any, + isExtensionEnabledFn: () => true, + }; +} + +describe("ctx.workflows.dispatch", () => { + describe("late-binding behavior", () => { + test("throws when dispatch function has not been set", async () => { + // Reset the module-level dispatch function by setting it to cause the "not initialized" error + // We simulate the pre-initialization state by creating a fresh context without setting the fn + // Note: since the module-level state is shared, we need to explicitly unset it + setWorkflowDispatchFn(null as any); + + const deps = createMinimalDeps(); + const { context } = createExtensionContext(deps); + + expect(context.workflows.dispatch("any-workflow")).rejects.toThrow("Workflows extension not initialized"); + }); + + test("succeeds after dispatch function is set", async () => { + const mockResult: WorkflowDispatchResult = { + workflowRunId: "run-123", + jobIds: ["job-1", "job-2"], + }; + + setWorkflowDispatchFn(async (_name, _payload) => mockResult); + + const deps = createMinimalDeps(); + const { context } = createExtensionContext(deps); + + const result = await context.workflows.dispatch("test-workflow", { key: "value" }); + expect(result).toEqual(mockResult); + }); + }); + + describe("dispatch error handling", () => { + test("throws when workflow is not found", async () => { + setWorkflowDispatchFn(async (name) => { + throw new Error(`Workflow not found: ${name}`); + }); + + const deps = createMinimalDeps(); + const { context } = createExtensionContext(deps); + + expect(context.workflows.dispatch("nonexistent")).rejects.toThrow("Workflow not found: nonexistent"); + }); + + test("throws when workflow is disabled", async () => { + setWorkflowDispatchFn(async (name) => { + throw new Error(`Workflow is disabled: ${name}`); + }); + + const deps = createMinimalDeps(); + const { context } = createExtensionContext(deps); + + expect(context.workflows.dispatch("my-workflow")).rejects.toThrow("Workflow is disabled: my-workflow"); + }); + }); + + describe("dispatch passes arguments correctly", () => { + test("passes name and payload to the dispatch function", async () => { + let capturedName: string | undefined; + let capturedPayload: unknown; + + setWorkflowDispatchFn(async (name, payload) => { + capturedName = name; + capturedPayload = payload; + return { workflowRunId: "run-1", jobIds: [] }; + }); + + const deps = createMinimalDeps(); + const { context } = createExtensionContext(deps); + + await context.workflows.dispatch("invoice-process", { filePath: "inbox/test.pdf" }); + + expect(capturedName).toBe("invoice-process"); + expect(capturedPayload).toEqual({ filePath: "inbox/test.pdf" }); + }); + + test("payload defaults to undefined when not provided", async () => { + let capturedPayload: unknown = "sentinel"; + + setWorkflowDispatchFn(async (_name, payload) => { + capturedPayload = payload; + return { workflowRunId: "run-1", jobIds: [] }; + }); + + const deps = createMinimalDeps(); + const { context } = createExtensionContext(deps); + + await context.workflows.dispatch("simple-workflow"); + + expect(capturedPayload).toBeUndefined(); + }); + }); +}); diff --git a/src/extensions/engine/extensionContext.ts b/src/extensions/engine/extensionContext.ts index 2918e74..a81777d 100644 --- a/src/extensions/engine/extensionContext.ts +++ b/src/extensions/engine/extensionContext.ts @@ -34,12 +34,33 @@ import type { RouteHandler, RunAgentOptions, StepTypeHandler, + WorkflowDispatchResult, } from "../types"; import { createConfigResolver } from "./configResolver"; import type { EventBus } from "./eventBus"; const logger = createLogger("ExtensionContext"); +// --------------------------------------------------------------------------- +// Late-bound workflow dispatch function +// --------------------------------------------------------------------------- + +/** + * Module-level dispatch function set by the workflows core extension during + * its initialization. Shared across all extension contexts via closure. + */ +let workflowDispatchFn: ((name: string, payload?: unknown) => Promise) | undefined; + +/** + * Registers the workflow dispatch implementation. Called once by the workflows + * core extension during its `initialize()` phase. + * + * @param fn - The dispatch function that looks up a workflow by name and triggers a run + */ +export function setWorkflowDispatchFn(fn: (name: string, payload?: unknown) => Promise): void { + workflowDispatchFn = fn; +} + /** * Dependencies injected from the registry so the context can wire into * the core system without owning those resources directly. @@ -580,6 +601,14 @@ export function createExtensionContext(deps: ExtensionContextDeps): { }, broadcast, }, + workflows: { + async dispatch(name: string, payload?: unknown): Promise { + if (!workflowDispatchFn) { + throw new Error("Workflows extension not initialized"); + } + return workflowDispatchFn(name, payload); + }, + }, db: database, fetch: authenticatedFetch, isEnabled, diff --git a/src/extensions/types.ts b/src/extensions/types.ts index c121d9c..9746e0d 100644 --- a/src/extensions/types.ts +++ b/src/extensions/types.ts @@ -371,6 +371,25 @@ export interface StepTypeInfo { // Agent execution // --------------------------------------------------------------------------- +// --------------------------------------------------------------------------- +// Workflow dispatch +// --------------------------------------------------------------------------- + +/** + * Result returned by {@link ExtensionContext.workflows.dispatch}. + * Contains the IDs needed to track the dispatched workflow run. + */ +export interface WorkflowDispatchResult { + /** Unique identifier for the created workflow run. */ + workflowRunId: string; + /** Job IDs for each step in the dispatched workflow (in execution order). */ + jobIds: string[]; +} + +// --------------------------------------------------------------------------- +// Agent execution +// --------------------------------------------------------------------------- + /** Result returned by {@link ExtensionContext.runAgent}. */ export interface AgentProcessorResult { /** The assistant's final text response. */ @@ -666,6 +685,29 @@ export interface ExtensionContext { broadcast(message: WebSocketMessage): void; }; + // ------------------------------------------------------------------------- + // Workflows + // ------------------------------------------------------------------------- + + /** Programmatic workflow dispatch (triggers named workflow runs without HTTP self-calls). */ + readonly workflows: { + /** + * Dispatch a named workflow run. + * + * Looks up the workflow definition by name, validates it is enabled, + * creates a new run with the provided payload as trigger data, and + * broadcasts a `workflow_started` WebSocket event. + * + * @param name - The workflow definition name (matches the `name` field in the workflow JSON5) + * @param payload - Optional trigger payload passed to the workflow as `{{trigger.*}}` + * @returns The run ID and step job IDs + * @throws {Error} If the workflow is not found ("Workflow not found: ") + * @throws {Error} If the workflow is disabled ("Workflow is disabled: ") + * @throws {Error} If the workflows extension has not initialized yet ("Workflows extension not initialized") + */ + dispatch(name: string, payload?: unknown): Promise; + }; + // ------------------------------------------------------------------------- // Database // -------------------------------------------------------------------------