Security & correctness hardening (P0 + P1)

ARCHITECTURE.md

@@ -1,6 +1,6 @@
 # Architecture
 
-This document is the orientation map for contributors. The authoritative design lives in `docs/design/` (10 documents) and `docs/adr/` (8 decision records); this file tells you what exists, where it lives, and which invariants hold it together.
+This document is the orientation map for contributors. The authoritative design lives in `docs/design/` (10 documents) and `docs/adr/` (9 decision records); this file tells you what exists, where it lives, and which invariants hold it together.
 
 ## Bird's-eye view
 
@@ -35,7 +35,7 @@ Dependency direction is enforced by `scripts/check-deps.ts` (runtime deps only).
 
 ## Key flows
 
-**Submit → run.** `POST /projects/:id/tasks` validates params against the compiled input schema (the same schema the SPA rendered the form from), verifies skill/MCP grants, resolves model roles → concrete granted models (frozen into `runs.model_resolution`), checks hard-stop quota headroom, then inserts task+run in one transaction and sends a pg-boss job keyed by run id. A worker sweeper re-enqueues stragglers, so a run can never be stranded nor double-queued.
+**Submit → run.** `POST /projects/:id/tasks` validates params against the compiled input schema (the same schema the SPA rendered the form from), verifies each `repoRef` is owned by the project, resolves model roles → concrete granted models (frozen into `runs.model_resolution`), pins the authorized skills/MCP into `runs.resource_manifest` (required grants enforced, optional included only when granted), checks hard-stop quota headroom, then inserts task+run in one transaction and sends a pg-boss job keyed by run id. A worker sweeper re-enqueues stragglers (and runs left paused by a lost approval-resume), so a run can never be stranded nor double-queued.
 
 **Engine loop.** The engine (worker-side) walks phases/steps: `when:` conditions, optional-resource skips, per-step retries, budget checks at every boundary. Agent steps stream normalized executor events which the engine persists to `run_events` (per-run monotonic `seq`), mirrors to the bus, and projects into `run_steps`/`token_usage`/`artifacts`. At the end it enforces the artifact contract — *succeeded* always means the contracted outputs exist.
 
@@ -52,7 +52,10 @@ Dependency direction is enforced by `scripts/check-deps.ts` (runtime deps only).
 3. **Steps are the idempotency unit** — restart-safe by template rule, resumable by session id where the executor supports it.
 4. **Usage rows are keyed `(run, step, attempt)`** so retries re-incur cost without ever double-counting.
 5. **Executors are stateless I/O**: all inputs in the request, all outputs as events; they never touch the database.
-6. **Secrets never leave as plaintext**: encrypted at rest (AES-256-GCM), write-only in the API, scrubbed from git remotes before agent code runs.
+6. **Secrets never leave as plaintext**: encrypted at rest (AES-256-GCM), write-only in the API, scrubbed from git remotes before agent code runs, and stripped from the agent subprocess environment (the master key and datastore URLs never reach a tool call).
+7. **Containment goes through one seam**: every tool call and the subprocess env are decided by `packages/executor-core/isolation.ts`; the adapter never reimplements it (ADR-0009).
+8. **The worker trusts only the pinned manifest**: repos are project-scoped and skills/MCP resolve solely from `runs.resource_manifest`, never the mutable global registry.
+9. **Lifecycle mutations are atomic**: run status transitions are compare-and-swap on the expected status and event `seq` is allocated by the database, so concurrent writers can't clobber a status or collide on a seq (`run-lifecycle.ts`).
 
 ## Where to look
 

CHANGELOG.md

@@ -6,6 +6,20 @@ All notable changes to Agrippa are documented here. The format follows
 
 ## [Unreleased]
 
+### Security
+
+- **Executor isolation seam** — one enforceable place (`packages/executor-core/isolation.ts`) decides every tool call and scrubs the subprocess environment. Read-only workspaces now actually deny shell and confine writes to the artifact directory; read-write workspaces confine writes to the workspace with a boundary-safe check (the previous `startsWith` let a sibling `<workspace>-evil` path through, and `Bash` bypassed the check entirely). The SDK subprocess runs with the platform secrets (`AGRIPPA_SECRET_KEY`, datastore URLs) stripped from its environment, the OS `sandbox` enabled where available, `strictMcpConfig`, and repo-supplied `.claude`/`.mcp.json` removed after checkout so a checked-out repository can't inject hooks or permission overrides. The worker image runs as a non-root user.
+- **Artifact path containment** — artifact ingestion resolves sources through `realpath` and rejects any that escape the workspace, closing a symlink disclosure (e.g. `ln -s /proc/self/environ`) that could exfiltrate secrets or other runs' files through the download endpoint.
+- **Cross-tenant resource authorization** — submission rejects a `repoConnectionId` that isn't owned by the project, and the worker loads repo connections scoped to the run's project. Optional skills/MCP servers are now grant-checked: an authorized-resource manifest is pinned onto the run at submit (required grants enforced, optional resources included only when granted) and the worker resolves resources only from it, so a project without a grant can no longer receive the platform's global credential (e.g. the shared GitHub token).
+
+### Fixed
+
+- **Crash recovery for no-retry steps** — a worker that died mid-step no longer silently skips (or spuriously fails) a step without template retries; the crashed attempt no longer consumes the retry budget, and the executor session is carried onto the recovery attempt so resume works.
+- **Atomic run lifecycle** — run status transitions are compare-and-swap on the expected status (a late finalize can't overwrite a cancellation), event sequence numbers are allocated by the database (no `max+1` collisions), and approval decisions are CAS on `pending` with the sweeper re-enqueuing any run left paused by a lost resume enqueue.
+- **Quota accounting** — the engine now counts the same monthly window as the submit gate, excludes the run's own spend from the headroom it checks (no double-count on resume), and re-reads project usage at each step boundary so concurrent runs can't jointly overspend.
+- **Artifact output contract** — patch steps no longer hand-write the diff (the engine generates it from `git diff` as intended), collected artifacts are validated against the step's declared keys/kinds, files from earlier steps aren't re-emitted, and missing/empty sources don't create zero-byte artifact rows.
+- **SSE live gap** — the events stream subscribes before replaying history, so an event committed in the replay/subscribe window is delivered live instead of only at the terminal replay (ADR-0007).
+
 ## [0.1.0] — 2026-07-17
 
 The M1 milestone: all three layers of the platform, working end to end.

apps/api/src/lib/usage.ts

@@ -45,8 +45,10 @@ export async function projectUsage(db: Db, projectId: string): Promise<ProjectUs
 
 /**
  * Submit-time quota gate (docs/design/04): hard-stop quotas reject new work
- * once the period's spend has reached either limit. The engine re-checks at
- * every step boundary for mid-run enforcement.
+ * once the current month's spend has reached either limit. The engine re-reads
+ * the same month-scoped project usage at every step boundary (excluding the
+ * run's own spend, which its budget meter already carries) for mid-run
+ * enforcement, so this gate and the engine agree on the accounting window.
  */
 export async function assertQuotaHeadroom(db: Db, projectId: string): Promise<void> {
   const [quota] = await db

apps/api/src/routes/execution.ts

@@ -18,13 +18,16 @@ import {
   templateVersions,
 } from "@agrippa/db";
 import {
+  appendRunEvent as allocateRunEvent,
+  authorizeResources,
   buildParamsValidator,
+  decideApproval,
   resolveModelRoles,
   SubmitError,
   type TemplateDoc,
-  verifyResourceGrants,
+  verifyRepoRefs,
 } from "@agrippa/orchestration";
-import { and, asc, desc, eq, gt, max } from "drizzle-orm";
+import { and, asc, desc, eq, gt } from "drizzle-orm";
 import { Hono } from "hono";
 import { streamSSE } from "hono/streaming";
 import type { AppEnv } from "../context";
@@ -46,26 +49,15 @@ async function loadRunScoped(
   return run;
 }
 
-/** Appends an API-originated event to the run log (seq = max + 1) and the bus. */
+/** Appends an API-originated event to the run log (DB-allocated seq) and the bus. */
 async function appendRunEvent(
   c: { var: AppEnv["Variables"] },
   runId: string,
   type: string,
   payload: Record<string, unknown>,
 ): Promise<void> {
-  const [maxSeq] = await c.var.db
-    .select({ v: max(runEvents.seq) })
-    .from(runEvents)
-    .where(eq(runEvents.runId, runId));
-  const seq = (maxSeq?.v ?? 0) + 1;
-  await c.var.db.insert(runEvents).values({ runId, seq, type, payload });
-  await c.var.bus?.publish({
-    runId,
-    seq,
-    type,
-    payload,
-    createdAt: new Date().toISOString(),
-  });
+  const { seq, createdAt } = await allocateRunEvent(c.var.db, { runId, type, payload });
+  await c.var.bus?.publish({ runId, seq, type, payload, createdAt: createdAt.toISOString() });
 }
 
 export const executionRoutes = new Hono<AppEnv>()
@@ -108,11 +100,15 @@ export const executionRoutes = new Hono<AppEnv>()
       await assertQuotaHeadroom(db, projectId);
 
       try {
+        // every repoRef must reference a connection owned by this project
+        await verifyRepoRefs(db, projectId, compiled.spec.inputs, parsed.data);
+
         const skillRows = await db.select({ id: skills.id, slug: skills.slug }).from(skills);
         const mcpRows = await db
           .select({ id: mcpServers.id, slug: mcpServers.slug })
           .from(mcpServers);
-        await verifyResourceGrants(db, projectId, compiled, {
+        // required grants enforced; optional resources pinned only when granted
+        const resourceManifest = await authorizeResources(db, projectId, compiled, {
           skillIdBySlug: new Map(skillRows.map((s) => [s.slug, s.id])),
           mcpIdBySlug: new Map(mcpRows.map((m) => [m.slug, m.id])),
         });
@@ -142,6 +138,7 @@ export const executionRoutes = new Hono<AppEnv>()
               executorId: DEFAULT_EXECUTOR,
               paramsSnapshot: parsed.data,
               modelResolution,
+              resourceManifest,
               budget: compiled.spec.budgets as unknown as Record<string, unknown>,
               createdBy: user.id,
             })
@@ -239,6 +236,7 @@ export const executionRoutes = new Hono<AppEnv>()
         executorId: latest.executorId,
         paramsSnapshot: latest.paramsSnapshot,
         modelResolution: latest.modelResolution,
+        resourceManifest: latest.resourceManifest,
         budget: latest.budget,
         createdBy: c.var.user.id,
       })
@@ -303,19 +301,20 @@ export const executionRoutes = new Hono<AppEnv>()
       .from(approvals)
       .where(and(eq(approvals.id, c.req.param("approvalId")), eq(approvals.runId, run.id)));
     if (!approval) throw AppError.notFound("Approval");
-    if (approval.status !== "pending") {
-      throw AppError.conflict("already_decided", `Approval is ${approval.status}`);
+    // compare-and-swap on status='pending' so a user decision and the expiry
+    // worker cannot overwrite each other
+    const updated = await decideApproval(c.var.db, approval.id, {
+      status: input.decision,
+      decidedBy: c.var.user.id,
+      comment: input.comment,
+    });
+    if (!updated) {
+      // already decided, OR a prior attempt decided then failed to enqueue: in
+      // both cases the durable state is correct, so re-enqueue to unstick the
+      // run (the sweeper also backstops this) and report the conflict
+      await c.var.queue?.enqueueRun(run.id);
+      throw AppError.conflict("already_decided", "Approval is already decided");
     }
-    const [updated] = await c.var.db
-      .update(approvals)
-      .set({
-        status: input.decision,
-        decidedBy: c.var.user.id,
-        decidedAt: new Date(),
-        comment: input.comment,
-      })
-      .where(eq(approvals.id, approval.id))
-      .returning();
     await appendRunEvent(c, run.id, "approval.decided", {
       approvalId: approval.id,
       checkpointId: approval.checkpointId,
@@ -419,9 +418,6 @@ export const executionRoutes = new Hono<AppEnv>()
         return rows;
       };
 
-      // 1) replay history (gap-free by construction)
-      await replay();
-
       const isTerminal = async (): Promise<boolean> => {
         const [row] = await db
           .select({ status: runs.status })
@@ -430,9 +426,11 @@ export const executionRoutes = new Hono<AppEnv>()
         return row ? isTerminalRunStatus(row.status) : true;
       };
 
-      if (await isTerminal()) return;
-
-      // 2) live: bridge the bus when present, else poll the DB
+      // Live: bridge the bus when present, else poll the DB. With a bus we must
+      // subscribe BEFORE replaying Postgres — an event committed and published
+      // in the window between replay and subscribe would otherwise be delivered
+      // only at the terminal replay, contradicting ADR-0007's gap-free
+      // guarantee. Buffered live events dedupe against the replay by seq.
       if (bus) {
         const queue: Array<() => Promise<void>> = [];
         let notify: (() => void) | null = null;
@@ -445,6 +443,7 @@ export const executionRoutes = new Hono<AppEnv>()
           notify?.();
         });
         try {
+          await replay(); // history first; the subscription is already buffering
           while (!closed) {
             while (queue.length > 0) {
               const job = queue.shift();
@@ -464,6 +463,8 @@ export const executionRoutes = new Hono<AppEnv>()
           unsubscribe();
         }
       } else {
+        await replay();
+        if (await isTerminal()) return;
         while (!closed) {
           await new Promise((resolve) => setTimeout(resolve, 1000));
           await replay();

apps/api/src/test/execution.integration.test.ts

@@ -1,6 +1,6 @@
 import { beforeAll, describe, expect, it } from "bun:test";
 import type { RunQueue } from "@agrippa/core";
-import { runs } from "@agrippa/db";
+import { repoConnections, runs } from "@agrippa/db";
 import { FakeExecutor, type FakeStepBehavior } from "@agrippa/executor-core";
 import {
   type EngineDeps,
@@ -43,6 +43,7 @@ describe.skipIf(!dbUp)("execution api (submit → engine → approve → artifac
   let admin: TestClient;
   let viewer: TestClient;
   let projectId: string;
+  let repoConnectionId: string;
   let taskTypeId: string;
   let taskId: string;
   let runId: string;
@@ -85,6 +86,13 @@ describe.skipIf(!dbUp)("execution api (submit → engine → approve → artifac
       json: { email: "vera@example.com", role: "viewer" },
     });
 
+    // a repo connection owned by this project — submissions must reference one
+    const [conn] = await db
+      .insert(repoConnections)
+      .values({ projectId, provider: "github", url: "https://github.com/acme/widget.git" })
+      .returning();
+    repoConnectionId = conn?.id as string;
+
     const types = await jsonOf<Array<{ id: string; slug: string }>>(
       await admin.request("/api/v1/scenarios/software-development/task-types"),
     );
@@ -96,7 +104,7 @@ describe.skipIf(!dbUp)("execution api (submit → engine → approve → artifac
     title: "Fix the widget",
     params: {
       bugReport: "It crashes",
-      repo: { repoConnectionId: Bun.randomUUIDv7() },
+      repo: { repoConnectionId },
     },
   });
 
@@ -141,6 +149,33 @@ describe.skipIf(!dbUp)("execution api (submit → engine → approve → artifac
     expect(res.status).toBe(403);
   });
 
+  it("refuses a repoConnectionId belonging to another project", async () => {
+    // a connection owned by a different project — cross-tenant IDOR attempt
+    const otherId = (
+      await jsonOf<{ id: string }>(
+        await admin.request("/api/v1/projects", {
+          method: "POST",
+          json: { slug: "other", name: "Other" },
+        }),
+      )
+    ).id;
+    const [foreign] = await db
+      .insert(repoConnections)
+      .values({ projectId: otherId, provider: "github", url: "https://github.com/acme/secret.git" })
+      .returning();
+
+    const res = await admin.request(`/api/v1/projects/${projectId}/tasks`, {
+      method: "POST",
+      json: {
+        taskTypeId,
+        title: "IDOR",
+        params: { bugReport: "It crashes", repo: { repoConnectionId: foreign?.id } },
+      },
+    });
+    expect(res.status).toBe(400);
+    expect((await jsonOf<{ code: string }>(res)).code).toBe("repo_not_in_project");
+  });
+
   it("accepts a valid submission and enqueues the run", async () => {
     const res = await admin.request(`/api/v1/projects/${projectId}/tasks`, {
       method: "POST",
@@ -215,6 +250,10 @@ describe.skipIf(!dbUp)("execution api (submit → engine → approve → artifac
 
     // resume from the middle: only later events are replayed
     const ids = [...text.matchAll(/^id: (\d+)$/gm)].map((m) => Number(m[1]));
+    // subscribe-before-replay must not deliver any event twice: seq strictly
+    // increasing, deduped by cursor (ADR-0007)
+    expect(new Set(ids).size).toBe(ids.length);
+    expect([...ids].sort((a, b) => a - b)).toEqual(ids);
     const middle = ids[Math.floor(ids.length / 2)] as number;
     const partial = await viewer.request(`/api/v1/runs/${runId}/events`, {
       headers: { "last-event-id": String(middle) },

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

@@ -0,0 +1,70 @@
+import { afterEach, describe, expect, it } from "bun:test";
+import { mkdtempSync, rmSync, symlinkSync, writeFileSync } from "node:fs";
+import { mkdir } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { DiskArtifactStore } from "./artifacts";
+
+const store = new DiskArtifactStore();
+const dirs: string[] = [];
+
+function freshWorkspace(): string {
+  const dir = mkdtempSync(path.join(tmpdir(), "agrippa-ws-"));
+  dirs.push(dir);
+  return dir;
+}
+
+afterEach(() => {
+  for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true });
+});
+
+describe("DiskArtifactStore path containment", () => {
+  it("stores a normal in-workspace file inline", async () => {
+    const ws = freshWorkspace();
+    await mkdir(path.join(ws, ".agrippa/artifacts"), { recursive: true });
+    writeFileSync(path.join(ws, ".agrippa/artifacts/report.md"), "# ok");
+
+    const stored = await store.store(
+      "run-1",
+      "report",
+      "markdown",
+      {
+        path: ".agrippa/artifacts/report.md",
+      },
+      ws,
+    );
+    expect(stored.inline).toBe("# ok");
+    expect(stored.size).toBeGreaterThan(0);
+  });
+
+  it("rejects a symlink escaping the workspace", async () => {
+    const ws = freshWorkspace();
+    await mkdir(path.join(ws, ".agrippa/artifacts"), { recursive: true });
+    // secret file outside the workspace, reachable only via the symlink
+    const outside = mkdtempSync(path.join(tmpdir(), "agrippa-secret-"));
+    dirs.push(outside);
+    const secret = path.join(outside, "environ");
+    writeFileSync(secret, "AGRIPPA_SECRET_KEY=master");
+    symlinkSync(secret, path.join(ws, ".agrippa/artifacts/leak.md"));
+
+    await expect(
+      store.store("run-1", "leak", "markdown", { path: ".agrippa/artifacts/leak.md" }, ws),
+    ).rejects.toThrow(/escapes the run workspace/);
+  });
+
+  it("treats a missing file as no content, not a zero-byte artifact", async () => {
+    const ws = freshWorkspace();
+    const stored = await store.store(
+      "run-1",
+      "nope",
+      "markdown",
+      {
+        path: ".agrippa/artifacts/nope.md",
+      },
+      ws,
+    );
+    expect(stored.inline).toBeNull();
+    expect(stored.storageRef).toBeNull();
+    expect(stored.size).toBe(0);
+  });
+});