Custom workflows

Pi harness workflows are normal @fragno-dev/workflows workflows. Use them when a Pi session needs more than the stock interactive chat loop: multiple agent steps, tool-result branching, parallel reviewers, races against user input, or explicit approvals.

Register workflow entries directly with createPiWorkflows(...) and with the Workflows fragment. There is intentionally no createPi() or definePiWorkflow(...) builder in @fragno-dev/pi-harness.

import { AgentHarness } from "@earendil-works/pi-agent-core";
import { createPiHarness, createPiWorkflows } from "@fragno-dev/pi-harness/factory";
import { piSessionCommandPayloadSchema } from "@fragno-dev/pi-harness/route-schemas";
import {
  applyWorkflowAgentHarnessStepResult,
  createPiHarnessSessionState,
  restoreWorkflowBackedSession,
  withWorkflowAgentHarness,
} from "@fragno-dev/pi-harness/workflows/workflow-agent-harness";
import { createWorkflowsFragment } from "@fragno-dev/workflows";
import { defineWorkflow } from "@fragno-dev/workflows/workflow";
import { z } from "zod";

const supportWorkflow = defineWorkflow(
  { name: "support", schema: z.object({ topic: z.string() }) },
  async (event, step) => {
    let state = createPiHarnessSessionState({
      metadata: { id: event.instanceId, createdAt: event.timestamp.toISOString() },
    });

    const initialResult = await step.do("initial", async (tx) => {
      const restored = restoreWorkflowBackedSession({
        operationId: `${supportWorkflow.name}:${event.instanceId}:initial`,
        state,
        previousEmissions: await tx.previousEmissions(),
        models,
      });
      const harness = new AgentHarness({
        models,
        model,
        systemPrompt: "You are a helpful support agent.",
        tools: [searchTool],
        ...restored.options,
      });

      return await withWorkflowAgentHarness({
        restored,
        harness,
        tx,
        runDurableStep: () => harness.prompt(`Help with ${event.payload.topic}`),
      });
    });
    state = applyWorkflowAgentHarnessStepResult(state, initialResult);

    while (true) {
      const commandEvent = await step.waitForEvent("wait-command", {
        type: "command",
        timeout: "7 days",
      });
      const command = piSessionCommandPayloadSchema.parse(commandEvent.payload);
      if (command.kind !== "prompt") continue;

      const result = await step.do(`command:${command.commandId}`, async (tx) => {
        const restored = restoreWorkflowBackedSession({
          operationId: `${supportWorkflow.name}:${event.instanceId}:command:${command.commandId}`,
          state,
          previousEmissions: await tx.previousEmissions(),
          models,
        });
        const harness = new AgentHarness({
          models,
          model,
          systemPrompt: "You are a helpful support agent.",
          tools: [searchTool],
          ...restored.options,
        });

        return await withWorkflowAgentHarness({
          restored,
          harness,
          tx,
          runDurableStep: () => harness.prompt(command.input.text),
        });
      });
      state = applyWorkflowAgentHarnessStepResult(state, result);
    }
  },
);

const piConfig = { workflows: [supportWorkflow] };
const workflows = createPiWorkflows(piConfig);

Create sessions by selecting the workflow and passing schema-validated input.

{
  "name": "Customer issue",
  "input": { "topic": "durable LLM workflows" }
}

Direct harness operations

Inside each step.do(...), restore the workflow-backed Pi session, construct a real AgentHarness, and call its APIs directly. This keeps normal Pi harness configuration and extension points under the workflow author's control.

To stop a prompt after a particular tool result, register the normal Pi harness hook before invoking the prompt:

const restored = restoreWorkflowBackedSession({
  operationId,
  state,
  previousEmissions: await tx.previousEmissions(),
  models,
});
const harness = new AgentHarness({ models, model, tools, ...restored.options });

harness.on("tool_result", (result) =>
  result.toolName === "classify_request" ? { terminate: true } : undefined,
);

return await withWorkflowAgentHarness({
  restored,
  harness,
  tx,
  runDurableStep: () => harness.prompt(event.payload.request),
});

Replay rules

Keep workflow structure deterministic

Workflow replay depends on stable step structure. Do not build step names from random IDs, current time, or partial streamed LLM output.

Step names should be string literals or derived from already completed durable data.
Keep runtime-only capabilities (env, streamFn, tool execute functions) in workflow closures.
Persist only serializable params and step results.
Use activeToolNames as a per-step policy when a turn should expose only a subset of registered tools.

Custom workflows

Direct harness operations

Replay rules

On this page