Two runtime-hardening fixes: interaction inline limit, worker readiness

CHANGELOG.md

@@ -88,6 +88,8 @@ All notable changes to Agrippa are documented here. The format follows
 - **The compose stack's volumes were named `infra_*`** — with no `name:` key the project name defaulted to the compose file's directory, so `pgdata`/`artifacts`/`workspaces` were created under an `infra` prefix: surprising when restoring a backup, and liable to collide with any other stack living in an `infra/` directory. Now pinned to `agrippa`. **Upgrading from `v0.2.0` or earlier requires a one-time volume migration** — otherwise `docker compose up -d` builds a second, empty stack beside the old one and collides on the published port. The documented procedure guards both ends. It refuses to copy into a non-empty target, because `docker volume create` succeeds silently on an existing volume and `cp -a` leaves destination-only files in place, so copying onto a started-then-abandoned stack would merge two Postgres clusters. And it preflights the sources, because Docker creates a missing named volume on the fly — a mistyped name would otherwise mount an empty one, copy nothing, and leave a fresh empty database that passes `/healthz` while the real data sits elsewhere. Step-by-step procedure: [Operations → Upgrading from a deployment created before the compose project was named](docs/manual/en/06-operations.md#upgrades--scaling).
 - **`AGRIPPA_PORT` exposed the plain-HTTP API on `0.0.0.0`** — it is the left side of the compose port mapping, so it can carry an interface. Documented as `127.0.0.1:3000` for the reverse-proxy topology, which is the only supported way to run it publicly.
 - **The worker could be SIGKILLed mid-drain** — it handles SIGTERM by awaiting pg-boss's graceful stop, but compose's default `stop_grace_period` is 10s. Raised to 60s.
+- **A thorough review report could kill its run at the review gate** — checkpoint-driving artifacts (review reports, questions) had to fit the store's general 64 KB inline threshold, but the interaction schemas admit a valid report of ~248K characters (up to ~1.5 MB as escaped JSON), so a schema-valid report of a large change failed the run with `contract_violation` instead of opening the gate. Interaction artifacts now get their own inline allowance (`INTERACTION_ARTIFACT_MAX_BYTES`, 2 MiB), sized to dominate any schema-valid payload — so only genuinely invalid (or padded) content can still trip the distinct too-large error. Both previously-untested failure paths (store-time and the resume-time legacy-row backstop) are now covered by the engine compliance suite.
+- **Deploy verification now proves each worker replica actually consumes, not just that one registered** (closes #15) — `worker_ok()` accepted a fresh `executor_registrations` row, which the worker writes *before* starting its pg-boss consumers, so a worker that registered and then wedged inside consumer setup read as a successful deploy while nothing consumed the queue. Workers now upsert a per-container row into the new `worker_heartbeats` table (migration `0012`) — hostname inside compose is the container id — with `consumers_ready_at` written only after `boss.work()` has returned for every consumer, plus a 60 s liveness bump from the sweeper (rows silent for a week are pruned). `deploy.sh` requires one fresh consumers-ready container per expected `WORKER_REPLICAS`; the replica-count and RestartCount checks stay, each still catching what the row count cannot (out-of-band scaling; crash loops that get past consumer setup).
 
 - **Dead clicks after confirming a retry, and sidebar collapse quirks** — Radix modal layers lock `<body>` with `pointer-events: none` while open; the retry confirm dialog navigating to the new run could unmount it mid-exit-animation and leave the lock behind, after which every click (the sidebar toggle included) silently did nothing:
   - *Stale body locks are cleared when a navigation settles* — a router subscription resets the lock, covering every dialog/dropdown that a route change can unmount mid-close.

apps/worker/src/deps/artifacts.test.ts

@@ -120,6 +120,54 @@ describe("DiskArtifactStore path containment", () => {
     }
   });
 
+  it("inlines past 64 KB when the caller raises the inline limit", async () => {
+    const ws = freshWorkspace();
+    await mkdir(path.join(ws, ".agrippa/artifacts"), { recursive: true });
+    const content = "r".repeat(100 * 1024); // 100 KB — over the default threshold
+    writeFileSync(path.join(ws, ".agrippa/artifacts/report.json"), content);
+
+    // without the override, the same file goes to disk
+    const onDisk = await store.store(
+      "run-1",
+      "report",
+      "json",
+      { path: ".agrippa/artifacts/report.json" },
+      ws,
+    );
+    expect(onDisk.inline).toBeNull();
+    expect(onDisk.storageRef).not.toBeNull();
+
+    // with it (the engine's interaction-artifact path), it inlines whole
+    const inlined = await store.store(
+      "run-2",
+      "report",
+      "json",
+      { path: ".agrippa/artifacts/report.json" },
+      ws,
+      { inlineLimitBytes: 2 * 1024 * 1024 },
+    );
+    expect(inlined.inline).toBe(content);
+    expect(inlined.storageRef).toBeNull();
+
+    // engine-provided inline sources honor the same override
+    const inlineSource = await store.store("run-3", "report", "json", { inline: content }, ws, {
+      inlineLimitBytes: 2 * 1024 * 1024,
+    });
+    expect(inlineSource.inline).toBe(content);
+    expect(inlineSource.storageRef).toBeNull();
+  });
+
+  it("spills to disk when content exceeds even a raised inline limit", async () => {
+    const ws = freshWorkspace();
+    const content = "x".repeat(200 * 1024);
+    const stored = await store.store("run-1", "report", "json", { inline: content }, ws, {
+      inlineLimitBytes: 128 * 1024,
+    });
+    expect(stored.inline).toBeNull();
+    expect(stored.storageRef).not.toBeNull();
+    expect(stored.size).toBe(200 * 1024);
+  });
+
   it("falls back to the default cap when the size env is not a valid number", async () => {
     const ws = freshWorkspace();
     await mkdir(path.join(ws, ".agrippa/artifacts"), { recursive: true });

apps/worker/src/deps/artifacts.ts

@@ -47,21 +47,27 @@ async function resolveContainedPath(workspaceDir: string, rel: string): Promise<
   return real;
 }
 
-/** ≤64 KB inline in Postgres; larger content on the artifacts volume. */
+/**
+ * ≤64 KB inline in Postgres; larger content on the artifacts volume. The
+ * engine raises the inline threshold per call (opts.inlineLimitBytes) for
+ * checkpoint-driving artifacts, which must inline whole to survive resume.
+ */
 export class DiskArtifactStore implements ArtifactStore {
   async store(
     runId: string,
     key: string,
     kind: ArtifactKind,
     source: { inline?: unknown; path?: string },
     workspaceDir: string,
+    opts?: { inlineLimitBytes?: number },
   ): Promise<StoredArtifact> {
+    const inlineLimit = opts?.inlineLimitBytes ?? INLINE_LIMIT;
     // engine-provided inline content (patch diffs, links) is always text
     if (source.inline !== undefined) {
       const content =
         typeof source.inline === "string" ? source.inline : JSON.stringify(source.inline);
       const mime = kind === "json" ? "application/json" : "text/markdown";
-      return this.storeText(runId, key, content, mime);
+      return this.storeText(runId, key, content, mime, inlineLimit);
     }
     if (!source.path) return EMPTY;
 
@@ -78,8 +84,8 @@ export class DiskArtifactStore implements ArtifactStore {
 
     // small text can inline in Postgres; `file`-kind (possibly binary) and any
     // large artifact stream straight to disk byte-exact, never fully buffered
-    if (kind !== "file" && size <= INLINE_LIMIT) {
-      return this.storeText(runId, key, await file.text(), mime);
+    if (kind !== "file" && size <= inlineLimit) {
+      return this.storeText(runId, key, await file.text(), mime, inlineLimit);
     }
     const storageRef = await this.writeToDisk(runId, key, file);
     return { inline: null, storageRef, size, mime };
@@ -90,11 +96,12 @@ export class DiskArtifactStore implements ArtifactStore {
     key: string,
     content: string,
     mime: string | null,
+    inlineLimit: number,
   ): Promise<StoredArtifact> {
     const size = Buffer.byteLength(content);
     if (size === 0) return EMPTY;
     if (size > maxArtifactSize()) throw new ArtifactTooLargeError(key, size);
-    if (size <= INLINE_LIMIT) return { inline: content, storageRef: null, size, mime };
+    if (size <= inlineLimit) return { inline: content, storageRef: null, size, mime };
     const storageRef = await this.writeToDisk(runId, key, content);
     return { inline: null, storageRef, size, mime };
   }

apps/worker/src/deps/readiness.test.ts

@@ -0,0 +1,85 @@
+import { beforeAll, describe, expect, it } from "bun:test";
+import { createDb, migrateDb, workerHeartbeats } from "@agrippa/db";
+import { eq, sql } from "drizzle-orm";
+import { markConsumersReady, touchWorkerHeartbeat } from "./readiness";
+
+const TEST_DATABASE_URL = process.env.TEST_DATABASE_URL ?? "postgres://localhost:5432/agrippa_test";
+
+// one pool for the whole suite — a pool per fixture exhausts max_connections
+const db = createDb(TEST_DATABASE_URL);
+let dbUp = true;
+try {
+  await db.execute(sql`select 1`);
+} catch {
+  dbUp = false;
+  console.warn("[test] postgres unreachable — skipping worker readiness suite");
+}
+
+const rowFor = async (containerId: string) => {
+  const [row] = await db
+    .select()
+    .from(workerHeartbeats)
+    .where(eq(workerHeartbeats.containerId, containerId));
+  return row;
+};
+
+describe.skipIf(!dbUp)("worker readiness heartbeats", () => {
+  beforeAll(async () => {
+    // drop the drizzle schema too — the migrator journals there, and dropping
+    // only public makes the migrations silently no-op on the next run
+    await db.execute(sql`drop schema public cascade`);
+    await db.execute(sql`create schema public`);
+    await db.execute(sql`drop schema if exists drizzle cascade`);
+    await migrateDb(db);
+  });
+
+  it("writes a consumers-ready row for this container", async () => {
+    await markConsumersReady(db, "ctr-ready");
+    const row = await rowFor("ctr-ready");
+    expect(row?.consumersReadyAt).not.toBeNull();
+    expect(row?.heartbeatAt).not.toBeNull();
+  });
+
+  it("re-freshens readiness on restart of the same container", async () => {
+    await markConsumersReady(db, "ctr-restart");
+    const stale = new Date(Date.now() - 60 * 60 * 1000);
+    await db
+      .update(workerHeartbeats)
+      .set({ consumersReadyAt: stale, heartbeatAt: stale })
+      .where(eq(workerHeartbeats.containerId, "ctr-restart"));
+
+    // same container id boots again (restart: unless-stopped) — the deploy
+    // freshness window must see the NEW boot, not the old one
+    await markConsumersReady(db, "ctr-restart");
+    const row = await rowFor("ctr-restart");
+    expect(row?.consumersReadyAt?.getTime()).toBeGreaterThan(stale.getTime());
+    expect(row?.heartbeatAt?.getTime()).toBeGreaterThan(stale.getTime());
+  });
+
+  it("touch bumps liveness without ever granting readiness", async () => {
+    // a fresh row via touch alone (e.g. after an out-of-band delete) must not
+    // count as ready — readiness is written only after boss.work() returns
+    await touchWorkerHeartbeat(db, "ctr-touch-only");
+    expect((await rowFor("ctr-touch-only"))?.consumersReadyAt).toBeNull();
+
+    // and on a ready row, touch must preserve the boot-time readiness stamp
+    await markConsumersReady(db, "ctr-touch-ready");
+    const before = (await rowFor("ctr-touch-ready"))?.consumersReadyAt;
+    await touchWorkerHeartbeat(db, "ctr-touch-ready");
+    expect((await rowFor("ctr-touch-ready"))?.consumersReadyAt?.getTime()).toBe(before?.getTime());
+  });
+
+  it("prunes containers silent for over a week", async () => {
+    const ancient = new Date(Date.now() - 8 * 24 * 60 * 60 * 1000);
+    await db.insert(workerHeartbeats).values({
+      containerId: "ctr-gone",
+      startedAt: ancient,
+      consumersReadyAt: ancient,
+      heartbeatAt: ancient,
+    });
+
+    await markConsumersReady(db, "ctr-alive");
+    expect(await rowFor("ctr-gone")).toBeUndefined();
+    expect(await rowFor("ctr-alive")).toBeDefined();
+  });
+});

apps/worker/src/deps/readiness.ts

@@ -0,0 +1,42 @@
+import { type Db, workerHeartbeats } from "@agrippa/db";
+import { lt, sql } from "drizzle-orm";
+
+/**
+ * Per-container liveness rows (issue #15). Registrations are global per
+ * executor, so they cannot prove each replica came up; deploy verification
+ * (infra/deploy.sh worker_ok) counts fresh consumers_ready_at rows instead —
+ * one per expected replica.
+ */
+
+/**
+ * Call only after boss.work() has returned for every consumer. Writing it any
+ * earlier reopens the register-then-wedge gap this row exists to close: a
+ * worker that hangs in consumer setup must never look ready.
+ */
+export async function markConsumersReady(db: Db, containerId: string): Promise<void> {
+  const now = new Date();
+  await db
+    .insert(workerHeartbeats)
+    .values({ containerId, startedAt: now, consumersReadyAt: now, heartbeatAt: now })
+    .onConflictDoUpdate({
+      target: workerHeartbeats.containerId,
+      set: { consumersReadyAt: now, heartbeatAt: now },
+    });
+  // containers are recreated on every deploy, so rows accumulate one per
+  // container forever; a week of silence is far past any freshness window
+  await db
+    .delete(workerHeartbeats)
+    .where(lt(workerHeartbeats.heartbeatAt, sql`now() - interval '7 days'`));
+}
+
+/** Sweeper-interval liveness bump; never touches consumersReadyAt. */
+export async function touchWorkerHeartbeat(db: Db, containerId: string): Promise<void> {
+  const now = new Date();
+  await db
+    .insert(workerHeartbeats)
+    .values({ containerId, startedAt: now, heartbeatAt: now })
+    .onConflictDoUpdate({
+      target: workerHeartbeats.containerId,
+      set: { heartbeatAt: now },
+    });
+}

apps/worker/src/index.ts

@@ -1,3 +1,4 @@
+import { hostname } from "node:os";
 import {
   type ApprovalExpirePayload,
   EXECUTOR_CATALOG,
@@ -29,11 +30,15 @@ import { and, eq, lt, sql } from "drizzle-orm";
 import type { Job, JobWithMetadata } from "pg-boss";
 import { DiskArtifactStore } from "./deps/artifacts";
 import { DemoExecutor } from "./deps/demo-executor";
+import { markConsumersReady, touchWorkerHeartbeat } from "./deps/readiness";
 import { DbResourceMaterializer } from "./deps/resources";
 import { GitScmService } from "./deps/scm";
 import { GitWorkspaceManager } from "./deps/workspace";
 
 const db = createDb();
+// inside a compose container the hostname IS the container id — the identity
+// deploy verification counts readiness rows by
+const containerId = hostname();
 const bus = process.env.REDIS_URL
   ? new RedisEventBus(process.env.REDIS_URL)
   : new InProcessEventBus();
@@ -178,6 +183,10 @@ await queue.boss.work(QUEUE_APPROVAL_EXPIRE, async (jobs: Job<ApprovalExpirePayl
   }
 });
 
+// every boss.work() above has returned — only now is this replica actually
+// consuming, which is the signal deploy verification counts (issue #15)
+await markConsumersReady(db, containerId);
+
 async function scheduleApprovalExpiry(runId: string): Promise<void> {
   const rows = await db
     .select()
@@ -230,12 +239,16 @@ setInterval(async () => {
 
     // executor-availability heartbeat (the API's live window is minutes-wide)
     await registerExecutors();
+    // per-container liveness (readiness stays untouched — boot-time only)
+    await touchWorkerHeartbeat(db, containerId);
   } catch (err) {
     deps.logger.warn("sweeper failed", { err: String(err) });
   }
 }, 60_000);
 
-console.log(`[worker] up — slots=${SLOTS} redis=${Boolean(process.env.REDIS_URL)}`);
+console.log(
+  `[worker] up — container=${containerId} slots=${SLOTS} redis=${Boolean(process.env.REDIS_URL)}`,
+);
 
 process.on("SIGTERM", async () => {
   deps.logger.info("draining…");

docs/design/01-domain-model.md

@@ -222,7 +222,7 @@ audit_logs (id pk, org_id fk, project_id fk null,
 - **Why `model_resolution` frozen at run start**: role→tier→model resolution depends on project grants, which can change mid-run; freezing makes runs reproducible and usage attribution unambiguous. The freeze is per **run**, never per task — a retry re-resolves against current configuration and freezes its own resolution (ADR-0014).
 - **Why `token_usage.attempt`**: a retried step consumes tokens again; rows keyed by `(run_id, step_id, attempt)` let the usage meter sum persisted totals on resume without double-counting a partially-executed attempt (the attempt's rows are written incrementally and summed as-is — the tokens were really spent even when the attempt failed).
 - **Why `mcp_servers.config_revision`**: MCP config is mutable head state (no full versioning in M1 — configs are small and secrets rotate); runs record the revision they resolved so audits can detect drift.
-- **Storage**: artifacts ≤ 64 KB are stored `inline` (jsonb/text); larger ones go to a disk-backed store at `storage_ref` (a Docker volume path in M1; the indirection allows S3 later).
+- **Storage**: artifacts ≤ 64 KB are stored `inline` (jsonb/text); larger ones go to a disk-backed store at `storage_ref` (a Docker volume path in M1; the indirection allows S3 later). Checkpoint-driving (interaction) artifacts get a raised inline allowance — `INTERACTION_ARTIFACT_MAX_BYTES`, 2 MiB, sized to dominate any schema-valid payload — because resume re-reads them from the row; an artifact that only exists on disk cannot drive its checkpoint.
 
 ## Drizzle Package Layout
 

docs/design/04-execution-runtime.md

@@ -109,7 +109,7 @@ Checkpoints **do not hold a worker slot**. When a checkpoint step pauses: a `che
 - approval `rejected` → run → `failed` with `error.code = "approval_rejected"`.
 - expiry → per-template `onTimeout` (`cancel | reject | approve` for approvals; `cancel` otherwise).
 
-**Gate-without-evidence rule.** Artifacts that drive an input/review-gate checkpoint are validated against the shared interaction schemas **at store time** — a malformed questions/review-report artifact fails the *producing step* with `contract_violation` while its attempt is still open, so template `retry`/`onFailure` apply. The checkpoint-time read is a strict backstop (it protects resumed runs whose artifact rows predate the validation): an **absent** review report fails the run — a gate must never pass on missing evidence — while an absent/empty questions list is the designed "nothing to ask" auto-pass; an artifact too large to inline gets a distinct error rather than being read as empty.
+**Gate-without-evidence rule.** Artifacts that drive an input/review-gate checkpoint are validated against the shared interaction schemas **at store time** — a malformed questions/review-report artifact fails the *producing step* with `contract_violation` while its attempt is still open, so template `retry`/`onFailure` apply. The checkpoint-time read is a strict backstop (it protects resumed runs whose artifact rows predate the validation): an **absent** review report fails the run — a gate must never pass on missing evidence — while an absent/empty questions list is the designed "nothing to ask" auto-pass. Interaction artifacts get a raised inline allowance (`INTERACTION_ARTIFACT_MAX_BYTES`, 2 MiB — sized so every schema-valid report/questions payload inlines whole), so only schema-invalid or padded content can be too large to inline — and that gets a distinct too-large error rather than being read as empty.
 
 **Work branch naming.** `git.branch` defaults to `agrippa/run-${run.number}-${run.shortId}`: run numbers are unique per *task*, so the run id's random tail (`run.shortId`, the last 12 hex chars of the UUIDv7 — 48 random bits — the head is timestamp bits) disambiguates across tasks. Unique branches are also what makes `pr.open`'s duplicate-recovery safe: a provider duplicate status on retry (GitHub 422, GitLab 409, GitCode any 4xx — its status is undocumented) looks up the existing open PR by head/base and returns its URL.