feat(workflows): add ctx.workflows.dispatch() API for programmatic wo…

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: <name>` |
+| Workflow exists but has `enabled: false` | `Workflow is disabled: <name>` |
+| 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 |

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<string, unknown> }): { 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";
 

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 };

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) => {

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<string>(),
+    routeKeySet: new Set<string>(),
+    stepTypeNameSet: new Set<string>(),
+    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();
+    });
+  });
+});

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<WorkflowDispatchResult>) | 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<WorkflowDispatchResult>): 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<WorkflowDispatchResult> {
+        if (!workflowDispatchFn) {
+          throw new Error("Workflows extension not initialized");
+        }
+        return workflowDispatchFn(name, payload);
+      },
+    },
     db: database,
     fetch: authenticatedFetch,
     isEnabled,

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: <name>")
+     * @throws {Error} If the workflow is disabled ("Workflow is disabled: <name>")
+     * @throws {Error} If the workflows extension has not initialized yet ("Workflows extension not initialized")
+     */
+    dispatch(name: string, payload?: unknown): Promise<WorkflowDispatchResult>;
+  };
+
   // -------------------------------------------------------------------------
   // Database
   // -------------------------------------------------------------------------