Document pipeline (ai-pipeline)

A domain-neutral document pipeline that composes all three ai-pipeline step kinds (#258) into one run: ai.extract pulls a title and key points out of an uploaded document, ai.classify routes it against a review-action catalogue using that extraction, ai.generate writes a two-sentence reviewer summary from both. The result is persisted only once every step succeeds. This is the reference sample for the AI-step vocabulary — it shows the building blocks themselves, not a specific use case (Joel’s DD-intake pipeline is one possible consumer of this vocabulary and is deliberately not built here).

What it shows

• Steps chain through ctx.steps — ai.classify’s input is built from ai.extract’s own output (title + key points), and ai.generate’s input reads both prior results. The test asserts the actual extracted text reaches the classify and generate provider requests, not just that all three ran.
• No workflow-runner needed today — process-document calls runPipeline directly against a synthetic workflow context, the same pattern ai-pipeline’s own integration test harness uses. The dispatcher/ctx handed to runPipeline is real; only the “a workflow engine invoked this” wrapper is stood in for, pending the real workflow-runner (M.4, kumiko-framework) — not a simplification specific to this sample.
• Persist only on full success, but the pipeline itself has no early-exit — extract, classify and generate are handled as an explicit three-way union after all three steps have run; the first non-success branch returns a structured 422 naming which step failed, and nothing is written. A failed or disabled step does not stop later steps from still calling the provider (the tests script all three calls even on the failing/disabled-step paths). A real pipeline that wants to stop early and skip the wasted provider calls would wrap the remaining steps in r.step.branch — that’s left out here to keep the vocabulary demo linear; r.step.return still has to come after the branch, not inside it.
• The policy catalogue and the step usage share one const — the same extractSpec/classifySpec/generateSpec objects are passed both to aiExtractStep()/aiClassifyStep()/aiGenerateStep() (the step usage) and, as documentPipelineStepSpecs, to createAiPipelineFeature() (the policy catalogue) — AiStepRuntimeSpec is a structural superset of AiStepSpec, so one declaration serves both.
• “Upload” is just a payload field here — the document’s base64 PDF bytes ride directly on process-document’s write payload. Real file storage, OCR routing and the fileRef.created trigger are the document-ingest recipe’s job; composing the two is a separate concern from demoing the AI-step vocabulary.

When to reach for it

Any pipeline that needs to pull structured data out of a document, route it, and hand a human a summary before they act — invoice intake, contract triage, compliance document review. Swap the extraction schema, the action catalogue and the summary prompt for your domain; the three-step shape and the policy wiring stay the same.

Wiring

document-pipeline requires ai-pipeline, mounted with this recipe’s own step catalogue via createAiPipelineFeature(documentPipelineStepSpecs). ai-pipeline in turn requires ai-foundation, prompt-store and tenant. Mount an LLM provider plugin (ai-provider-anthropic, ai-provider-openai-compat, or a mock in tests) alongside, and set the ai-foundation:config:provider config-key (or a per-step providerId via ai-pipeline:write:set-policy) — none of this sample’s steps pin a provider by default.

import { createAiPipelineFeature } from "@cosmicdriftgamestudio/kumiko-ai-pipeline";
import { documentPipelineFeature, documentPipelineStepSpecs } from "./feature";

const features = [
  createAiPipelineFeature(documentPipelineStepSpecs),
  aiFoundationFeature,
  promptStoreFeature,
  yourLlmProviderPlugin,
  documentPipelineFeature,
  ...appFeatures,
];

The integration test under src/__tests__/ boots that exact stack against real Postgres + Redis and scripts the provider — including the failing-step and disabled-step paths — so every assertion above is proven, not described.

bun --env-file=../.env test --config=bunfig.integration.toml samples/recipes/ai-pipeline

---

Source code

The feature entry point — embedded straight from the source file, so the code here is exactly what runs. Multi-file samples keep their remaining files next to it on GitHub (link below):

// ai-pipeline sample — the vocabulary IS the recipe.
//
// A domain-neutral document pipeline wiring all three ai-pipeline step
// kinds (#258) into one run: ai.extract pulls a title + key points out of
// the uploaded document, ai.classify routes it against a review-action
// catalogue using that extraction, ai.generate writes a two-sentence
// reviewer summary from both. Every step's policy (enabled/provider/model/
// params) stays tenant-editable via ai-pipeline's own write handler — this
// feature only declares the step *structure*.
//
// No workflow-runner exists yet (M.4, kumiko-framework): `process-document`
// below calls `runPipeline` directly against a synthetic workflow context,
// exactly like the ai-pipeline package's own integration test harness
// (packages/ai-pipeline/src/__tests__/ai-step-vocabulary.integration.test.ts).
// The dispatcher/ctx handed to `runPipeline` is real; only the "a workflow
// engine invoked this" wrapper is stood in for, pending M.4.
//
// The pipeline has no early-exit: a failed or skipped step does not stop
// later steps from still calling the provider (see the write handler below,
// which persists only after checking all three results).
//
// "Upload" here means the caller hands over the document bytes directly on
// the write payload — no document-ingest-foundation/file-storage wiring.
// That's the `document-ingest` recipe's job; composing the two is a separate
// concern from demoing the AI-step vocabulary itself.

import { randomUUID } from "node:crypto";
import {
  createEntityExecutor,
  defineEntityDetailHandler,
  defineEntityListHandler,
  defineFeature,
  defineWorkflow,
  type PipelineCtx,
  runPipeline,
  stepsPipeline,
  withResponseData,
} from "@cosmicdrift/kumiko-framework/engine";
import { failUnprocessable } from "@cosmicdrift/kumiko-framework/errors";
import {
  type AiClassifyAction,
  type AiClassifyResult,
  type AiClassifyStepArgs,
  type AiExtractDocument,
  type AiExtractResult,
  type AiExtractStepArgs,
  type AiGenerateResult,
  type AiGenerateStepArgs,
  type AiStepSpec,
  aiClassifyStep,
  aiExtractStep,
  aiGenerateStep,
} from "@cosmicdriftgamestudio/kumiko-ai-pipeline";
import { z } from "zod";
import { pipelineRunEntity } from "./entity";

const EXTRACT_STEP_KEY = "document-pipeline:extract";
const CLASSIFY_STEP_KEY = "document-pipeline:classify";
const GENERATE_STEP_KEY = "document-pipeline:generate";

const paramsSchema = z.object({ maxTokens: z.number().int().positive().optional() });

// The catalogue is the classify step's integration contract (same rule as
// the ai-triage recipe): ai.classify never invents an action type, only
// picks from what the caller hands it.
const REVIEW_ACTIONS: readonly AiClassifyAction[] = [
  { type: "route-to-finance", description: "Send to finance for invoice or payment processing." },
  { type: "route-to-legal", description: "Send to legal for contract review." },
  { type: "archive", description: "No action needed; file for reference." },
];

function payloadOf(ctx: PipelineCtx): ProcessDocumentPayload {
  // @cast-boundary pipeline resolver — defineStep's `run()` is declared once
  // for every step instance across the app, so PipelineCtx carries no
  // per-instance payload type. Narrowed here to this feature's own schema.
  return ctx.event.payload as ProcessDocumentPayload;
}

function extractedTitleAndPoints(ctx: PipelineCtx): {
  title: string;
  keyPoints: readonly string[];
} {
  const extract = ctx.steps[EXTRACT_STEP_KEY] as AiExtractResult;
  if (extract.type !== "extracted") {
    return { title: "the document", keyPoints: [] };
  }
  return {
    title: extract.data["title"] as string,
    keyPoints: extract.data["keyPoints"] as readonly string[],
  };
}

const extractSpec: AiExtractStepArgs = {
  stepKey: EXTRACT_STEP_KEY,
  promptKey: "document-pipeline:extract:prompt",
  promptFallback: "Extract the document's title and up to five key points as a short bullet list.",
  paramsSchema,
  defaults: { enabled: true, params: {} },
  outputSchema: z.object({
    title: z.string(),
    keyPoints: z.array(z.string()).max(5),
  }),
  instructions: () => "Extract the title and up to five key points from the attached document.",
  document: (ctx) => payloadOf(ctx).document,
};

const classifySpec: AiClassifyStepArgs = {
  stepKey: CLASSIFY_STEP_KEY,
  promptKey: "document-pipeline:classify:prompt",
  promptFallback: "Classify the document and propose the right review action.",
  paramsSchema,
  defaults: { enabled: true, params: {} },
  actions: REVIEW_ACTIONS,
  input: (ctx) => {
    const { title, keyPoints } = extractedTitleAndPoints(ctx);
    return `Title: ${title}\nKey points: ${keyPoints.join("; ") || "(none extracted)"}`;
  },
};

const generateSpec: AiGenerateStepArgs = {
  stepKey: GENERATE_STEP_KEY,
  promptKey: "document-pipeline:generate:prompt",
  promptFallback: "Write a two-sentence reviewer summary of the document.",
  paramsSchema,
  defaults: { enabled: true, params: {} },
  input: (ctx) => {
    const { title } = extractedTitleAndPoints(ctx);
    const classify = ctx.steps[CLASSIFY_STEP_KEY] as AiClassifyResult;
    const category = classify.type === "classified" ? classify.category : "uncategorized";
    return `Write a two-sentence reviewer summary of "${title}" (category: ${category}).`;
  },
};

// Handed to `createAiPipelineFeature` by the consumer (see README Wiring) so
// TenantAdmins can retune enabled/provider/model/params per step without a
// deploy. Reusing the same const objects the workflow below calls
// aiExtractStep/aiClassifyStep/aiGenerateStep with is deliberate, not
// duplication avoided: `AiStepRuntimeSpec` (used by the step call) is a
// structural superset of `AiStepSpec` (all the policy catalogue needs).
export const documentPipelineStepSpecs: readonly AiStepSpec[] = [
  extractSpec,
  classifySpec,
  generateSpec,
];

export type ProcessDocumentPayload = {
  readonly label: string;
  readonly document: AiExtractDocument;
};

type DocumentPipelineStepResults = {
  readonly extract: AiExtractResult;
  readonly classify: AiClassifyResult;
  readonly generate: AiGenerateResult;
};

export const documentPipelineWorkflow = defineWorkflow<
  ProcessDocumentPayload,
  DocumentPipelineStepResults
>({
  name: "document-pipeline-demo",
  trigger: { kind: "event", eventType: "document-pipeline-sample:triggered" },
  steps: stepsPipeline<ProcessDocumentPayload, DocumentPipelineStepResults>(({ r }) => [
    aiExtractStep(extractSpec),
    aiClassifyStep(classifySpec),
    aiGenerateStep(generateSpec),
    r.step.return((ctx) => ({
      isSuccess: true,
      data: {
        extract: ctx.steps[EXTRACT_STEP_KEY] as AiExtractResult,
        classify: ctx.steps[CLASSIFY_STEP_KEY] as AiClassifyResult,
        generate: ctx.steps[GENERATE_STEP_KEY] as AiGenerateResult,
      },
    })),
  ]),
});

const processDocumentSchema = z.object({
  label: z.string().min(1),
  document: z.object({
    // ~6 MB PDF as base64 (~8e6 chars). Unbounded data would spike memory
    // and LLM cost on every TenantAdmin call through this sample.
    data: z.string().min(1).max(8_000_000),
    mimeType: z.literal("application/pdf"),
  }),
});

const adminOnly = { access: { roles: ["TenantAdmin", "SystemAdmin"] } } as const;

export const documentPipelineFeature = defineFeature("document-pipeline", (r) => {
  r.requires("ai-pipeline");
  r.entity("pipeline-run", pipelineRunEntity);

  const { executor } = createEntityExecutor("pipeline-run", pipelineRunEntity);

  r.queryHandler(defineEntityListHandler("pipeline-run", pipelineRunEntity, adminOnly));
  r.queryHandler(defineEntityDetailHandler("pipeline-run", pipelineRunEntity, adminOnly));

  // The point of this sample: run the three-step pipeline, then persist the
  // result only once every step succeeded — any non-success branch becomes a
  // structured 422 naming which step failed, nothing is stored half-done.
  r.writeHandler(
    "process-document",
    processDocumentSchema,
    async (event, ctx) => {
      const runResult = await runPipeline(documentPipelineWorkflow.pipelineDef, event, ctx, {
        runId: randomUUID(),
        workflowName: documentPipelineWorkflow.name,
        // Overwritten per step by runStepList before each step.run() call.
        stepIndex: 0,
      });
      if (!runResult.isSuccess) {
        // Unreachable in practice: every path through the workflow ends at
        // r.step.return with isSuccess: true — a step's own failure surfaces
        // as `type: "error"` on the union below, not as a WriteFailure.
        // Kept so the union stays exhaustively handled.
        return runResult;
      }
      const { extract, classify, generate } = runResult.data;

      if (extract.type !== "extracted") {
        return failUnprocessable("document extraction failed", {
          step: "extract",
          result: extract,
        });
      }
      if (classify.type !== "classified") {
        return failUnprocessable("document classification failed", {
          step: "classify",
          result: classify,
        });
      }
      if (generate.type !== "generated") {
        return failUnprocessable("summary generation failed", {
          step: "generate",
          result: generate,
        });
      }

      const created = await executor.create(
        {
          label: event.payload.label,
          title: extract.data["title"],
          keyPoints: extract.data["keyPoints"],
          category: classify.category,
          confidence: classify.confidence,
          proposedActions: classify.proposedActions,
          summary: generate.text,
        },
        event.user,
        ctx.db,
      );
      if (!created.isSuccess) return created;

      return withResponseData(created, { id: created.data.id, extract, classify, generate });
    },
    adminOnly,
  );
});

Enterprise recipe — samples/recipes/ai-pipeline/src/feature.ts in the private kumiko-enterprise workspace.