feat(pipeline): replace Mastra workflow with @calca/pipeline package

.agents/backlog-policy.md

@@ -0,0 +1,19 @@
+# Backlog Policy
+
+GitHub Project: https://github.com/users/espetro/projects/6
+
+## Refined task fields
+
+- **Milestone** — maps to iteration/quarter (e.g. `2026 Q3`). Create the milestone in the repo if it doesn't exist yet.
+- **Size** — effort estimate: `XS`/`S`/`M`/`L`/`XL`. Include testing + bug potential per contact surface.
+- **Start date** / **Target date** — scheduled window for the task.
+- **Label** — classification, drives client positioning: `feature`, `bug`, `cosmetic`, `infra`. Create the label in the repo if it doesn't exist yet.
+
+## Workflow
+
+1. Create a GitHub issue in `espetro/calca` with title, body, classification label, milestone.
+2. Add it to project 6: `gh project item-add 6 --owner espetro --url <issue-url>`.
+3. Set Size/Start date/Target date via `gh project item-edit --project-id <id> --id <item-id> --field-id <field-id> ...`.
+4. Reference the issue number in the PR description and in the corresponding `.agents/plans/` doc.
+
+No orphan work: every plan/implementation must link back to a refined task here.

apps/server/package.json

@@ -23,9 +23,10 @@
     "@app/core": "workspace:*",
     "@app/logger": "workspace:*",
     "@app/shared": "workspace:*",
+    "@calca/pipeline": "workspace:*",
     "@hono/node-server": "^1.19.14",
     "@openfeature/server-sdk": "^1.20.2",
-    "ai": "^6.0.156",
+    "ai": "^7.0.0",
     "hono": "^4.12.14",
     "zod": "^4.3.6"
   },

apps/server/src/routes/__tests__/workflow.test.ts

@@ -1,21 +1,18 @@
+import type { GenerateOptions } from "@app/core/ai/client";
 import type { Context } from "hono";
 import { beforeEach, describe, expect, it, vi } from "vitest";
 
-// Mock dependencies before importing the handler
-vi.mock("@mastra/ai-sdk", () => ({
-  handleWorkflowStream: vi.fn(),
+vi.mock("@app/core/ai/client", () => ({
+  generateWithFallback: vi.fn(),
+  streamAnthropic: vi.fn(),
 }));
 
-vi.mock("ai", () => ({
-  createUIMessageStreamResponse: vi.fn(),
+vi.mock("@app/core/pipeline/images", () => ({
+  generateImages: vi.fn(),
 }));
 
-vi.mock("../../workflows/mastra", () => ({
-  mastra: { _mock: true },
-}));
-
-import { handleWorkflowStream } from "@mastra/ai-sdk";
-import { createUIMessageStreamResponse } from "ai";
+import { generateWithFallback, streamAnthropic } from "@app/core/ai/client";
+import { generateImages } from "@app/core/pipeline/images";
 
 import { handleWorkflow } from "../workflow";
 
@@ -27,59 +24,189 @@ function createMockContext(body: unknown): Context {
   } as unknown as Context;
 }
 
+async function readStream(
+  response: Response,
+): Promise<Array<{ type: string; [key: string]: unknown }>> {
+  const reader = response.body!.getReader();
+  const decoder = new TextDecoder();
+  let buffer = "";
+  const parts: Array<{ type: string; [key: string]: unknown }> = [];
+
+  while (true) {
+    const { done, value } = await reader.read();
+    if (done) break;
+    buffer += decoder.decode(value, { stream: true });
+    const lines = buffer.split("\n");
+    buffer = lines.pop() ?? "";
+    for (const line of lines) {
+      const colonIdx = line.indexOf(":");
+      if (colonIdx === -1) continue;
+      try {
+        parts.push(
+          JSON.parse(line.slice(colonIdx + 1)) as { type: string; [key: string]: unknown },
+        );
+      } catch {
+        // ignore malformed lines
+      }
+    }
+  }
+  if (buffer.trim()) {
+    const colonIdx = buffer.indexOf(":");
+    if (colonIdx !== -1) {
+      try {
+        parts.push(
+          JSON.parse(buffer.slice(colonIdx + 1)) as { type: string; [key: string]: unknown },
+        );
+      } catch {
+        // ignore
+      }
+    }
+  }
+  return parts;
+}
+
+function buildInput(overrides?: Record<string, unknown>): Record<string, unknown> {
+  return {
+    prompt: "a pricing card",
+    mode: "sequential",
+    model: "claude-model",
+    ...overrides,
+  };
+}
+
+function mockGenerateWithFallback() {
+  (generateWithFallback as ReturnType<typeof vi.fn>).mockImplementation(
+    async (options: GenerateOptions) => {
+      const functionId = options.functionId ?? "";
+
+      if (functionId === "plan") {
+        return {
+          result: {
+            text: JSON.stringify([{ name: "Minimal", direction: "Clean" }]),
+          } as Awaited<ReturnType<typeof generateWithFallback>>["result"],
+          usedModel: options.model ?? "model",
+        };
+      }
+
+      if (functionId.startsWith("review")) {
+        return {
+          result: { text: "<div>reviewed</div>" } as Awaited<
+            ReturnType<typeof generateWithFallback>
+          >["result"],
+          usedModel: options.model ?? "model",
+        };
+      }
+
+      if (functionId.startsWith("critique")) {
+        return {
+          result: { text: "Looks good" } as Awaited<
+            ReturnType<typeof generateWithFallback>
+          >["result"],
+          usedModel: options.model ?? "model",
+        };
+      }
+
+      if (functionId === "summary") {
+        return {
+          result: { text: JSON.stringify({ rationale: "nice" }) } as Awaited<
+            ReturnType<typeof generateWithFallback>
+          >["result"],
+          usedModel: options.model ?? "model",
+        };
+      }
+
+      return {
+        result: { text: "" } as Awaited<ReturnType<typeof generateWithFallback>>["result"],
+        usedModel: options.model ?? "model",
+      };
+    },
+  );
+}
+
+function mockStreamAnthropic() {
+  (streamAnthropic as ReturnType<typeof vi.fn>).mockResolvedValue({
+    text: Promise.resolve(`<!--size:400x300-->\n<div>hello</div>`),
+  });
+}
+
+function mockGenerateImages() {
+  (generateImages as ReturnType<typeof vi.fn>).mockImplementation(async () => ({
+    html: `<div>hello</div>`,
+    imageCount: 0,
+    skipped: true,
+    reason: "no keys",
+  }));
+}
+
 describe("handleWorkflow", () => {
   beforeEach(() => {
     vi.clearAllMocks();
+    mockGenerateWithFallback();
+    mockStreamAnthropic();
+    mockGenerateImages();
   });
 
-  it("returns a UI message stream response on success", async () => {
-    const mockStream = { [Symbol.asyncIterator]: vi.fn() };
-    const mockResponse = new Response(null, { status: 200 });
+  it("returns an SSE response with correct headers", async () => {
+    const ctx = createMockContext(buildInput());
+    const response = await handleWorkflow(ctx);
 
-    (handleWorkflowStream as ReturnType<typeof vi.fn>).mockResolvedValue(mockStream);
-    (createUIMessageStreamResponse as ReturnType<typeof vi.fn>).mockReturnValue(mockResponse);
-
-    const ctx = createMockContext({ prompt: "a pricing card" });
-    const result = await handleWorkflow(ctx);
+    expect(response.headers.get("Content-Type")).toBe("text/event-stream; charset=utf-8");
+    expect(response.headers.get("Cache-Control")).toBe("no-cache, no-transform");
+    expect(response.headers.get("X-Accel-Buffering")).toBe("no");
+  });
 
-    expect(ctx.req.json).toHaveBeenCalledOnce();
-    expect(handleWorkflowStream).toHaveBeenCalledOnce();
-    expect(handleWorkflowStream).toHaveBeenCalledWith({
-      mastra: expect.anything(),
-      params: { inputData: { prompt: "a pricing card" } },
-      version: "v6",
-      workflowId: "designPipeline",
+  it("streams data-workflow parts through the pipeline", async () => {
+    const ctx = createMockContext(buildInput());
+    const response = await handleWorkflow(ctx);
+    const parts = await readStream(response);
+
+    const workflowParts = parts.filter((p) => p.type === "data-workflow");
+    expect(workflowParts.length).toBeGreaterThan(0);
+
+    const first = workflowParts[0] as unknown as {
+      data: { name: string; status: string; steps: Record<string, unknown> };
+    };
+    expect(first.data.name).toBe("designPipeline");
+    expect(first.data.status).toBe("running");
+    expect(first.data.steps.plan).toMatchObject({ name: "plan", status: "running" });
+
+    const last = workflowParts[workflowParts.length - 1] as unknown as {
+      data: {
+        status: string;
+        steps: Record<string, { output?: unknown }>;
+      };
+    };
+    expect(last.data.status).toBe("success");
+    expect(last.data.steps.collectResults?.output).toMatchObject({
+      frames: [
+        expect.objectContaining({
+          html: "<div>reviewed</div>",
+          label: "Variation 1",
+        }),
+      ],
+      summary: expect.stringContaining("rationale"),
     });
-    expect(createUIMessageStreamResponse).toHaveBeenCalledWith({ stream: mockStream });
-    expect(result).toBe(mockResponse);
   });
 
-  it("propagates errors from handleWorkflowStream", async () => {
-    (handleWorkflowStream as ReturnType<typeof vi.fn>).mockRejectedValue(
-      new Error("workflow exploded"),
-    );
-
-    const ctx = createMockContext({ prompt: "fail" });
+  it("passes the full request body to the pipeline", async () => {
+    const body = { conceptCount: 4, mode: "quick", prompt: "hero section", model: "claude-model" };
+    const ctx = createMockContext(body);
+    const response = await handleWorkflow(ctx);
+    await readStream(response);
 
-    await expect(handleWorkflow(ctx)).rejects.toThrow("workflow exploded");
-    expect(createUIMessageStreamResponse).not.toHaveBeenCalled();
+    expect(ctx.req.json).toHaveBeenCalledOnce();
+    expect(streamAnthropic).toHaveBeenCalledWith(expect.objectContaining({ model: body.model }));
   });
 
-  it("passes the full request body as inputData", async () => {
-    const body = { conceptCount: 4, preset: "marketing", prompt: "hero section" };
-    const mockStream = {};
-    const mockResponse = new Response(null, { status: 200 });
+  it("emits an error part when the pipeline throws", async () => {
+    (streamAnthropic as ReturnType<typeof vi.fn>).mockRejectedValue(new Error("layout exploded"));
 
-    (handleWorkflowStream as ReturnType<typeof vi.fn>).mockResolvedValue(mockStream);
-    (createUIMessageStreamResponse as ReturnType<typeof vi.fn>).mockReturnValue(mockResponse);
-
-    const ctx = createMockContext(body);
-    await handleWorkflow(ctx);
+    const ctx = createMockContext(buildInput());
+    const response = await handleWorkflow(ctx);
+    const parts = await readStream(response);
 
-    expect(handleWorkflowStream).toHaveBeenCalledWith(
-      expect.objectContaining({
-        params: { inputData: body },
-      }),
-    );
+    const errorParts = parts.filter((p) => p.type === "error");
+    expect(errorParts.length).toBeGreaterThan(0);
+    expect(errorParts[0]).toMatchObject({ type: "error", errorText: "layout exploded" });
   });
 });

apps/server/src/routes/workflow.ts

@@ -1,24 +1,18 @@
-import { handleWorkflowStream } from "@mastra/ai-sdk";
-import { createUIMessageStreamResponse } from "ai";
+import { designPipelineStream } from "@calca/pipeline";
 import { type Context, Hono } from "hono";
 
-import { mastra } from "../workflows/mastra";
-
 export async function handleWorkflow(c: Context) {
   const body = await c.req.json();
-
-  const stream = await handleWorkflowStream({
-    mastra,
-    params: { inputData: body },
-    version: "v6",
-    workflowId: "designPipeline",
+  const stream = designPipelineStream(body);
+  return new Response(stream, {
+    headers: {
+      "Content-Type": "text/event-stream; charset=utf-8",
+      "Cache-Control": "no-cache, no-transform",
+      "X-Accel-Buffering": "no",
+    },
   });
-
-  return createUIMessageStreamResponse({ stream });
 }
 
-const route = new Hono()
-  //
-  .post("/", handleWorkflow);
+const route = new Hono().post("/", handleWorkflow);
 
 export default route;