feat(workflow): let hosts check or redirect run-scoped writes

Author: TheGreatAxios

README.md

@@ -101,6 +101,17 @@ Run-scoped routes for a workflow run, which authenticates with a bearer token an
 | `uploadPolicy`    | `UploadPolicy`        | Optional. MIME types `POST /artifacts/binary` accepts. Defaults to `ARTIFACT_UPLOAD_POLICY`.                                                                     |
 | `maxBinaryBytes`  | `number`              | Optional. Byte ceiling for `POST /artifacts/binary`. Defaults to `MAX_UPLOAD_BYTES`.                                                                             |
 | `maxContentChars` | `number`              | Optional. Character ceiling on the resulting content — full text or edited text — for `POST /artifacts` and `PATCH /artifacts/:id/versions`. Defaults to 64,000. |
+| `beforeWrite`     | `WorkflowBeforeWrite` | Optional. Checks each text create (`POST /artifacts`) and revision (`PATCH /artifacts/:id`) before it is written. Omitted, every valid write lands.              |
+
+`beforeWrite({ scope, kind, write, content })` gets the run scope, the artifact kind, `write` as `{ op: "create" }` or `{ op: "revise", artifactId }`, and the content the version will hold after any edits. It returns one of:
+
+- `{ action: "allow" }` writes as usual.
+- `{ action: "refuse", code, message }` answers `422` with `{ code, message }`, and the agent's tool error is `message`. `code` must match `/^[a-z0-9_]+$/`; `message` must be 1 to 1,000 characters.
+- `{ action: "redirect", artifactId, expectedVersion? }` turns a create into a revision of that artifact, which must be of the same kind and in the run's tenant, answering `200` with its `{ id, version }`. The revision takes the create's title. A missing, archived or other-tenant artifact answers `404`, and a stale `expectedVersion` answers `409` telling the agent to read the existing artifact and revise it.
+
+Any other decision, a redirect returned for a revision, or a redirect to an artifact of another kind answers `500` and writes nothing. Binary uploads and linked files are not checked.
+
+A create's check is not serialized: two concurrent creates can both be allowed, so a policy like "one plan per tenant" can still end with two plans when runs race. A revision's check runs inside the transaction holding the artifact's row lock, so it should be fast and must not write that artifact.
 
 ### `runArtifactMigrations(config, { schema })`
 

e2e/reference-host.test.ts

@@ -722,3 +722,39 @@ describe("the migration runner is re-runnable", () => {
     expect(survived.artifacts.length).toBeGreaterThan(0);
   });
 });
+
+describe("the host's beforeWrite policy on agent writes", () => {
+  const runPost = (body: unknown): RequestInit => ({
+    method: "POST",
+    headers: { "content-type": "application/json", ...host.runHeaders },
+    body: JSON.stringify(body),
+  });
+
+  test("refuses a plan with no heading, then folds a second plan into the first", async () => {
+    const refused = await host.request(
+      "/api/workflow-artifacts/artifacts",
+      runPost({ title: "Plan", kind: "plan", content: "Ship." }),
+    );
+    expect(refused.status).toBe(422);
+    expect(await refused.json()).toEqual({
+      code: "plan_needs_heading",
+      message: "Start the plan with a '# ' heading.",
+    });
+
+    const first = await host.request(
+      "/api/workflow-artifacts/artifacts",
+      runPost({ title: "Plan", kind: "plan", content: "# Plan\nShip." }),
+    );
+    expect(first.status).toBe(201);
+    const { data: created } = await json<{ data: { id: string } }>(first);
+
+    const second = await host.request(
+      "/api/workflow-artifacts/artifacts",
+      runPost({ title: "Plan v2", kind: "plan", content: "# Plan\nShip now." }),
+    );
+    expect(second.status).toBe(200);
+    expect(await second.json()).toEqual({
+      data: { id: created.id, version: 2 },
+    });
+  });
+});

e2e/workflow-mount.test.ts

@@ -3,9 +3,16 @@ import { sql } from "drizzle-orm";
 import {
   createWorkflowArtifactRoutes,
   type ResolvedWorkflowRunScope,
+  type WorkflowBeforeWrite,
+  type WorkflowWriteCheck,
+  type WorkflowWriteDecision,
 } from "../src/workflow-mount.js";
 import { InlineContentStore } from "../src/content-store.js";
-import { getArtifact, setArtifactArchived } from "../src/artifacts.js";
+import {
+  getArtifact,
+  listArtifacts,
+  setArtifactArchived,
+} from "../src/artifacts.js";
 import { seedArtifact } from "./fixtures.js";
 import { testDb } from "./helpers.js";
 import type { ArtifactDb } from "../src/db.js";
@@ -24,6 +31,7 @@ function host(
   opts: {
     resolves?: ResolvedWorkflowRunScope | null;
     maxContentChars?: number;
+    beforeWrite?: WorkflowBeforeWrite;
   } = {},
 ) {
   return createWorkflowArtifactRoutes({
@@ -32,6 +40,9 @@ function host(
     ...(opts.maxContentChars !== undefined
       ? { maxContentChars: opts.maxContentChars }
       : {}),
+    ...(opts.beforeWrite !== undefined
+      ? { beforeWrite: opts.beforeWrite }
+      : {}),
     resolveRunScope: (token, address) => {
       if (opts.resolves === undefined) {
         return token === VALID_TOKEN && address === VALID_ADDRESS
@@ -806,3 +817,160 @@ describe("user-input errors", () => {
     expect((await read("?version=2147483648")).status).toBe(400);
   });
 });
+
+describe("beforeWrite", () => {
+  const plan = { title: "Plan", kind: "plan", content: "## Plan\nShip." };
+  const created = async (res: Response) =>
+    ((await res.json()) as { data: { id: string; version: number } }).data;
+
+  test("a refused create is 422 with the code and message, and writes nothing", async () => {
+    const db = await testDb();
+    const checks: WorkflowWriteCheck[] = [];
+    const app = host(db, {
+      beforeWrite: (check) => {
+        checks.push(check);
+        return {
+          action: "refuse",
+          code: "plan_exists",
+          message: "Revise the existing plan instead.",
+        };
+      },
+    });
+    const res = await app.request("/artifacts", json(plan));
+    expect(res.status).toBe(422);
+    expect(await res.json()).toEqual({
+      code: "plan_exists",
+      message: "Revise the existing plan instead.",
+    });
+    expect(checks).toEqual([
+      {
+        scope: RUN_SCOPE,
+        kind: "plan",
+        write: { op: "create" },
+        content: plan.content,
+      },
+    ]);
+    expect((await listArtifacts(db, "acme", {})).rows).toHaveLength(0);
+  });
+
+  test("a refused revision sees the edited content and leaves the version alone", async () => {
+    const db = await testDb();
+    const row = await seedArtifact(db, {
+      tenantId: "acme",
+      kind: "design",
+      content: "## Plan\nShip.",
+    });
+    const checks: WorkflowWriteCheck[] = [];
+    const app = host(db, {
+      beforeWrite: (check) => {
+        checks.push(check);
+        return { action: "refuse", code: "too_vague", message: "Add a date." };
+      },
+    });
+    const res = await app.request(
+      `/artifacts/${row.id}`,
+      patchJson({ edits: [{ oldText: "Ship.", newText: "Ship soon." }] }),
+    );
+    expect(res.status).toBe(422);
+    expect(checks).toEqual([
+      {
+        scope: RUN_SCOPE,
+        kind: "design",
+        write: { op: "revise", artifactId: row.id },
+        content: "## Plan\nShip soon.",
+      },
+    ]);
+    expect((await getArtifact(db, row.id))?.version).toBe(1);
+  });
+
+  test("a redirected create revises the named artifact under expectedVersion", async () => {
+    const db = await testDb();
+    const row = await seedArtifact(db, { tenantId: "acme", kind: "plan" });
+    const redirectTo = (expectedVersion: number) =>
+      host(db, {
+        beforeWrite: () => ({
+          action: "redirect",
+          artifactId: row.id,
+          expectedVersion,
+        }),
+      });
+
+    const res = await redirectTo(1).request("/artifacts", json(plan));
+    expect(res.status).toBe(200);
+    expect(await created(res)).toEqual({ id: row.id, version: 2 });
+    const revised = await getArtifact(db, row.id);
+    expect(revised?.title).toBe("Plan");
+    expect(revised?.content).toBe(plan.content);
+    expect((await listArtifacts(db, "acme", {})).rows).toHaveLength(1);
+
+    const stale = await redirectTo(1).request("/artifacts", json(plan));
+    expect(stale.status).toBe(409);
+    expect(await stale.json()).toEqual({
+      error: `A plan already exists as artifact ${row.id} at version 2. Read it, then revise it.`,
+    });
+    expect((await getArtifact(db, row.id))?.version).toBe(2);
+  });
+
+  test("a redirect to another tenant's artifact is 404 and writes nothing", async () => {
+    const db = await testDb();
+    const foreign = await seedArtifact(db, { tenantId: "other" });
+    const app = host(db, {
+      beforeWrite: () => ({ action: "redirect", artifactId: foreign.id }),
+    });
+    const res = await app.request("/artifacts", json(plan));
+    expect(res.status).toBe(404);
+    expect(await res.json()).toEqual({
+      error:
+        "This plan could not be created: the existing plan it should revise was not found.",
+    });
+    expect((await getArtifact(db, foreign.id))?.version).toBe(1);
+    expect((await listArtifacts(db, "acme", {})).rows).toHaveLength(0);
+  });
+
+  test("a redirect to an artifact of another kind fails closed", async () => {
+    const db = await testDb();
+    const design = await seedArtifact(db, { tenantId: "acme", kind: "design" });
+    const app = host(db, {
+      beforeWrite: () => ({ action: "redirect", artifactId: design.id }),
+    });
+    expect((await app.request("/artifacts", json(plan))).status).toBe(500);
+    const unchanged = await getArtifact(db, design.id);
+    expect(unchanged?.version).toBe(1);
+    expect(unchanged?.content).not.toBe(plan.content);
+    expect((await listArtifacts(db, "acme", {})).rows).toHaveLength(1);
+  });
+
+  test("a malformed decision fails closed", async () => {
+    const db = await testDb();
+    const row = await seedArtifact(db, { tenantId: "acme" });
+    const malformed: unknown[] = [
+      { action: "deny" },
+      { action: "refuse", code: "Bad Code", message: "x" },
+      { action: "refuse", code: "no_message" },
+      { action: "refuse", code: "empty_message", message: " " },
+      { action: "refuse", code: "long_message", message: "x".repeat(1_001) },
+    ];
+    for (const decision of malformed) {
+      // A host bug that the decision type alone cannot rule out at runtime.
+      const app = host(db, {
+        beforeWrite: () => decision as WorkflowWriteDecision,
+      });
+      expect((await app.request("/artifacts", json(plan))).status).toBe(500);
+      const revise = await app.request(
+        `/artifacts/${row.id}`,
+        patchJson({ content: "changed" }),
+      );
+      expect(revise.status).toBe(500);
+    }
+    const reviseRedirect = host(db, {
+      beforeWrite: () => ({ action: "redirect", artifactId: row.id }),
+    });
+    const res = await reviseRedirect.request(
+      `/artifacts/${row.id}`,
+      patchJson({ content: "changed" }),
+    );
+    expect(res.status).toBe(500);
+    expect((await getArtifact(db, row.id))?.version).toBe(1);
+    expect((await listArtifacts(db, "acme", {})).rows).toHaveLength(1);
+  });
+});

examples/reference-host/src/index.ts

@@ -41,12 +41,16 @@ import {
 import {
   InlineContentStore,
   createArtifactRoutes,
+  createWorkflowArtifactRoutes,
+  listArtifacts,
   runArtifactMigrations,
   type ArtifactDb,
   type ResolvedPrincipal,
   type ContentStore,
   type SerializedArtifactBase,
+  type WorkflowBeforeWrite,
 } from "../../../src/index.js";
+import { WORKFLOW_ARTIFACTS_BASE_PATH } from "../../../src/sidecar-bundle.js";
 import { databaseConfig, DATABASE_URL } from "../../../e2e/helpers.js";
 
 const EPOCH = new Date(0);
@@ -66,6 +70,39 @@ async function decorate(
 
 export type Session = { userId: string } | null;
 
+/** The one workflow run this host resolves, standing in for its run table. */
+const RUN_TOKEN = "reference-sidecar-token";
+const RUN_ADDRESS = "reference-run";
+
+/**
+ * The host's write policy for agents: a plan opens with a heading, and a
+ * tenant keeps one plan, so a second plan create revises the first.
+ */
+function planPolicy(db: ArtifactDb): WorkflowBeforeWrite {
+  return async ({ scope, kind, write, content }) => {
+    if (kind !== "plan") return { action: "allow" };
+    if (!content.startsWith("# ")) {
+      return {
+        action: "refuse",
+        code: "plan_needs_heading",
+        message: "Start the plan with a '# ' heading.",
+      };
+    }
+    if (write.op === "revise") return { action: "allow" };
+    const { rows } = await listArtifacts(db, scope.tenantId, {
+      kind: "plan",
+      limit: 1,
+    });
+    const [existing] = rows;
+    if (existing === undefined) return { action: "allow" };
+    return {
+      action: "redirect",
+      artifactId: existing.id,
+      expectedVersion: existing.version,
+    };
+  };
+}
+
 export type ReferenceHost = {
   db: ArtifactDb;
   /** The `DBConfig` the host migrates with. */
@@ -81,6 +118,8 @@ export type ReferenceHost = {
    * `createFileArtifact` with this.
    */
   scope: () => ResolvedPrincipal;
+  /** Headers a workflow run presents to `WORKFLOW_ARTIFACTS_BASE_PATH`. */
+  runHeaders: Record<string, string>;
   /** Who the hub's session middleware will report. `null` means signed out. */
   setSession: (session: Session) => void;
   /** Request the default host (InlineContentStore, real DB-backed grants). */
@@ -355,6 +394,22 @@ export async function createReferenceHost(): Promise<ReferenceHost> {
         decorate,
       }),
     );
+    app.route(
+      WORKFLOW_ARTIFACTS_BASE_PATH,
+      createWorkflowArtifactRoutes({
+        db,
+        contentStore,
+        resolveRunScope: (token, address) =>
+          token === RUN_TOKEN && address === RUN_ADDRESS
+            ? {
+                tenantId: tenant.id,
+                principalId: agentPrincipal,
+                runId: address,
+              }
+            : null,
+        beforeWrite: planPolicy(db),
+      }),
+    );
     const mounted = app.route("/api", api);
     // `Hono#request` may answer synchronously; normalize to a promise so every
     // caller can simply await it.
@@ -371,6 +426,10 @@ export async function createReferenceHost(): Promise<ReferenceHost> {
     config,
     tenantId: tenant.id,
     agentPrincipal,
+    runHeaders: {
+      authorization: `Bearer ${RUN_TOKEN}`,
+      "x-workflow-run-address": RUN_ADDRESS,
+    },
     scope: () => ({
       tenantId: tenant.id,
       principalId: principals.find(

src/artifacts.ts

@@ -419,6 +419,9 @@ export async function reviseArtifactVersion(
      * thrown and nothing is written. Omitted preserves today's behavior.
      */
     expectedVersion?: number;
+    /** Runs under the lock with the content this version will hold; a throw
+     * rolls the revision back. */
+    beforeWrite?: (locked: ArtifactRow, content: string) => Promise<void>;
   },
   now: Date,
 ): Promise<ArtifactRow> {
@@ -478,6 +481,7 @@ export async function reviseArtifactVersion(
       );
     }
   }
+  if (args.beforeWrite !== undefined) await args.beforeWrite(existing, content);
   const metadata =
     args.metadata === undefined
       ? (existing.metadata as Record<string, unknown> | null)
@@ -542,6 +546,8 @@ export async function writeArtifactVersion(
     parentVersionIds?: string[] | null;
     /** See `VersionConflictError` — omitted preserves today's behavior. */
     expectedVersion?: number;
+    /** See {@link reviseArtifactVersion}. */
+    beforeWrite?: (locked: ArtifactRow, content: string) => Promise<void>;
   },
 ): Promise<{
   artifactId: string;

src/index.ts

@@ -11,7 +11,10 @@ export type {
   CreateWorkflowArtifactRoutesDeps,
   ResolvedWorkflowRunScope,
   WorkflowArtifactEnv,
+  WorkflowBeforeWrite,
   WorkflowRunResolver,
+  WorkflowWriteCheck,
+  WorkflowWriteDecision,
 } from "./workflow-mount.js";
 
 export {