From 083fec357c713eea36841eb2a2c3d10b7515fb13 Mon Sep 17 00:00:00 2001 From: btipling Date: Sun, 23 Aug 2026 00:41:32 +0000 Subject: [PATCH 1/3] =?UTF-8?q?feat(backend-agents):=20turns=20runtime=20s?= =?UTF-8?q?pike=20=E2=80=94=20resumable=20AgentStreamEvent=20stream=20(pla?= =?UTF-8?q?n=20#787)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 2 + app/api/turns/[runId]/stream/route.test.ts | 235 +++++++++++++++++++++ app/api/turns/[runId]/stream/route.ts | 63 ++++++ app/api/turns/route.test.ts | 146 +++++++++++++ app/api/turns/route.ts | 58 +++++ docs/agent-stream.md | 23 ++ lib/workflows/turnsFixtureEvents.test.ts | 68 ++++++ lib/workflows/turnsFixtureEvents.ts | 28 +++ lib/workflows/turnsFixtureWorkflow.ts | 54 +++++ 9 files changed, 677 insertions(+) create mode 100644 app/api/turns/[runId]/stream/route.test.ts create mode 100644 app/api/turns/[runId]/stream/route.ts create mode 100644 app/api/turns/route.test.ts create mode 100644 app/api/turns/route.ts create mode 100644 lib/workflows/turnsFixtureEvents.test.ts create mode 100644 lib/workflows/turnsFixtureEvents.ts create mode 100644 lib/workflows/turnsFixtureWorkflow.ts diff --git a/AGENTS.md b/AGENTS.md index 58bbdb3..7634b1e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -222,6 +222,7 @@ ops inventory). | Optional SCIM (`SCIM_BEARER_TOKEN`) | **Not Done** | SCIM 2.0 Users at `/api/scim/v2` — set only when directory provisioning is intended; fail-closed 404 when unset | | Blob transcript store (`BLOB_READ_WRITE_TOKEN`, phase 0 #515) | **Not Done** (documented seam only) | Vercel Blob (or BYO S3/R2 behind the same seam) for the session **transcript**; Redis keeps the small envelope. Server-minted short-lived scoped upload URLs; client→Blob uploads; legacy full-record GET stays for roll-forward while old blobs stay small. Configure in Vercel/env-manager, never a laptop ritual; no migrate/backfill/seed. Docs: [docs/bring-your-own.md](docs/bring-your-own.md), [docs/session-model.md](docs/session-model.md). When unset the app runs an in-memory transcript store (dev/tests only) | | Vercel Workflows (backend-agents) | **Done (code)** | Workflow SDK wired (`workflow` dep + `withWorkflow` in `next.config.js`), authed smoke route (`app/api/workflows/smoke`), dispatch GHA **`workflows-smoke`** (always targets Production; no preview env is built). Workflows **activates automatically when you deploy with the SDK** — there is no dashboard enable button; the operator step is **deploy to Production** then dispatch **`workflows-smoke`** (`confirm=smoke`). Never the tab-owned `/api/agent` as the smoke path. Ops: [docs/feature-divide.md](docs/feature-divide.md) | +| Vercel Workflows turn probe (spike, plan #787) | `app/api/turns/*`, `lib/workflows/turnsFixtureWorkflow.ts` + `turnsFixtureEvents.ts` | **Spike-only, NOT the turn owner.** `POST /api/turns` → `x-workflow-run-id` + `{ runId }`; `GET /api/turns/:runId/stream?startIndex=` resumes the fixture's `AgentStreamEvents` emitted to `getWritable()` (reconnect primitive). Present on `main`, **preview/spike** — `/api/agent` stays the production turn owner until slice E swaps in; durable turn-owner/attach/detach is slice I (#772). Fail-closed 503 on a Workflows-disabled start; never falls back to `/api/agent` | **Agent behavior (origin):** @@ -332,6 +333,7 @@ invincible/ | DOM site chrome nav (hamburger Account menu) | `app/components/AppNav.tsx` (brand wordmark; optional `busy?: boolean` from `HarnessHost` only — TEAL outline + neon bloom + sine pulse + motes while the active harness turn is Busy; never poll the bridge; settings/admin omit the prop), `app/components/AuthNavLinks.tsx` (server: `soleMembership`+`canAccessAdmin` → `showAdmin`), `app/components/NavMenu.tsx` (client dropdown: ARIA `menu`, Arrow/Tab/Home/End, Escape + click-outside close + focus return, ≥44px touch targets, palette-only TEAL), `lib/navMenu.ts` + `lib/navMenu.test.ts` (`buildSignedInNavItems` — pure ordering/gating rule, unit-tested), footer slot `app/logout/LogoutButton.tsx`; unauth keeps inline `Sign in` header control. Client holds **zero** role-gate logic — it renders pre-gated inert `items` only | | API / AI Gateway / agent | `app/api/*`, `lib/agent/*`, `lib/sandbox/*` | | Agent SSE stream (tools + text + reasoning) | `lib/agent/agentStream.ts`, `lib/agent/runAgent.ts`, `lib/agent/reasoningConfig.ts`, `app/api/agent/route.ts`, `lib/agentApi.ts`, `docs/agent-stream.md` | +| Workflow turn probe (spike-only `/api/turns` prefix, plan #787) | `app/api/turns/route.ts`, `app/api/turns/[runId]/stream/route.ts`, `lib/workflows/turnsFixtureWorkflow.ts` (+ `turnsFixtureEvents.ts`) | **Vercel backend owns this; it is a throwaway probe, NOT the turn owner.** `/api/agent` stays the production turn owner until slice E (#768) swaps in; durable turn-owner/attach/detach is slice I (#772). Never add a client/Wasm path to these spike routes | | Agent read-before-edit / file freshness | `lib/agent/fileFreshness.ts`, `lib/agent/pathLock.ts` (per-path apply serialization), `lib/agent/tools.ts`, `lib/agent/runAgent.ts`, [docs/sandbox.md](docs/sandbox.md) | | Logical agent cwd + workspace-root↔abs canonicalization (`change_dir` / session / default env; **`sandbox_info`** is the bind/cwd/caps/env introspector — do not `exec env`; `canonicalizePath(R, p)` / `workspaceAbsToRel(R, abs)` / `resolvePathForTool(R, cwd, p)` / `rewriteExecRootToRel(R, text)` in `lib/agent/workPath.ts`) + **`search`** (read-grant-only code-grep via `rg`; `lib/agent/tools.ts`) | `lib/agent/workPath.ts`, `lib/agent/tools.ts`, `lib/agent/runAgent.ts`, `lib/agent/agentBody.ts`, `lib/sandbox/config.ts`, `lib/sessionStore.ts`, `lib/harnessChat.ts`, `lib/agentApi.ts`, `lib/sessionCloudCaps.ts` (shared client-safe `sanitizeSessionCwd` + Redis-safe opaque id predicate), [docs/sandbox.md](docs/sandbox.md), [docs/session-model.md](docs/session-model.md), [docs/agent-stream.md](docs/agent-stream.md). Tool paths accept **in-jail absolute paths** on all FS tools + `change_dir` + `exec` cwd: an absolute under the per-binding jail root R (`resolved.value.workspaceRoot` → `RunAgentParams.workspaceRoot` → `createAgentTools`) is canonicalized to the same workspace-relative freshness key as its relative form (BYO + Vercel parity); out-of-jail absolutes and `..`/symlink escapes fail closed. Absolute paths under `R` that **appear in `exec` stdout/stderr** are likewise rewritten to workspace-relative (`rewriteExecRootToRel` in `lib/agent/workPath.ts`, applied to `result.stdout`/`result.stderr` separately) so `exec pwd` ≡ `pwd`/annotations; when `R` is unresolvable the exec output passes through byte-for-byte (fail-open), and rewrites are capped and never throw. When R is unresolvable (BYO daemon down/pre-v2 — `workspaceRoot === null`) absolute is rejected (“root unavailable — use workspace-relative”) while relative + cwd still work. Initial request/session `cwd` stays relative-only; `.` is the workspace-root default session start (there is no `SANDBOX_DEFAULT_CWD` env knob), `..` walks up toward the workspace root and errors only past it, and an **exact ancestor** of cwd re-roots cleanly (`change_dir invincible` from `cwd=invincible/docs` → `invincible`, not the phantom `invincible/docs/invincible`) while a name-prefix sibling is never re-rooted. P1/GAP-1 (#452/#330): `cwd` + `activeSandboxId` are **session-owned** and ride the Redis record (`meta.{logicalCwd,activeSandboxId}`). `activeSandboxId` is now **server-resolved** (routing override via `lib/tenancy/resolveSandbox.ts` `requestedSandboxId`), not carry-only. A **confirmed successful `change_dir`** is persisted as the session cwd even when the turn later cancels / times out / hard-errors (`lib/harnessChat.ts` host-side `liveCwd`); the success path still prefers the authoritative `agentResult.cwd`, and only a confirmed `change_dir` (never an errored one) is stored on a failed/aborted turn. The **`exec` tool** returns a **compact summary**, not a raw dump: first `EXEC_LOG_HEAD_LINES` (10) + last `EXEC_LOG_TAIL_LINES` (10) lines per stdout/stderr with line/byte counts and `... (N lines truncated)`, each shown line byte-clipped (`EXEC_SUMMARY_LINE_MAX_BYTES`=4096) so a single fat stdio line can't inline the stream or truncate the `log:` pointer off; and when either stream is non-empty writes the full redacted output to `/.invincible/logs/exec--.log` via `client.write_file(..., mkdir: true)` (a brand-new hidden workspace dir; backends never auto-create parents; the `-` monotonic counter keeps same-ms parallel execs from overwriting), reporting two `read_file` pointers — `log: ` (cwd-relative, from nested cwd `../.invincible/logs/…`) and `log (root): ` (workspace-root-relative, read from the workspace root `cwd .`, so a depth-changing `change_dir` can never strand the full output); the write stays workspace-root, and `.invincible/` is gitignored; both pointers ride immediately after `exit=`/`TIMED_OUT` — empty output (`exec true`) writes no file, and a log-write failure fails soft with a `⚠ log write failed` note whose reason is **sanitized** (a backend/jail path never surfaces) (caps `EXEC_LOG_HEAD_LINES`/`EXEC_LOG_TAIL_LINES`=10 and `EXEC_LOG_MAX_BYTES`=8 MiB in `lib/sandbox/config.ts`) | | Cloud multi-device harness session (Redis multi-session, `/api/sessions*`, hybrid local+cloud; **phase 0 #515 envelope + Blob transcript carrier**) | `app/api/sessions/*` (+ `app/api/sessions/[id]/envelope/*`, `[id]/transcript/*`), `lib/sessionRepository.ts`, `lib/sessionCloudCaps.ts`, `lib/sessions/*` (+ `lib/sessions/blobStore.ts`, `blobStores.ts`), `lib/tenancy/harnessSessionsRedis.ts`, `lib/tenancy/harnessSessions.ts` (archive read + shared validator), `lib/di/index.ts` (root), `app/harness/HarnessHost.tsx`, `middleware.ts`, [docs/session-model.md](docs/session-model.md), [docs/bring-your-own.md](docs/bring-your-own.md), [SECURITY.md](SECURITY.md) — one-shot Postgres→Redis backfill: GHA **`sessions-redis-backfill`** (idempotent per-user marker); Postgres `harness_sessions` is a read-only archive. P1/GAP-1 (#452): session-carrier `meta.{logicalCwd,activeSandboxId}` folds into the PUT body and restores on pull/adopt; **plan #616 (source #610)** adds the reserved `meta.selectedModel` session carrier for the selected model pick (restore by id after the model catalog push; server **drops a poisoned value to unset**, never a 400). **Phase 0 (#515):** the transcript lives in **Vercel Blob** (`BLOB_READ_WRITE_TOKEN` / BYO S3-R2 seam) pointed to by `meta.transcriptPointer` on the small Redis envelope (`harness:envelope:*`); server mints short-lived scoped upload URLs for **client→Blob** uploads; legacy full-record GET stays for roll-forward while old blobs stay small. Envelope upsert/read: `PUT`/`GET /api/sessions/:id/envelope`; mint/read: `POST`/`GET /api/sessions/:id/transcript` | diff --git a/app/api/turns/[runId]/stream/route.test.ts b/app/api/turns/[runId]/stream/route.test.ts new file mode 100644 index 0000000..7884538 --- /dev/null +++ b/app/api/turns/[runId]/stream/route.test.ts @@ -0,0 +1,235 @@ +import { readFileSync } from 'node:fs'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { AGENT_STREAM_CONTENT_TYPE } from '../../../../../lib/agent/agentStream'; +import { TURNS_FIXTURE_SSE } from '../../../../../lib/workflows/turnsFixtureEvents'; + +/** + * backend-agents B spike (plan #787): unit tests for the resumable GET + * /api/turns/:runId/stream. Mocks `workflow/api` (`getRun`) — no real run here. + * + * Reconnect proof (DoD rows 6+7): POST → capture `runId`; abort the HTTP client + * (simulated); the run independently reaches `completed` (client abort ≠ + * cancel — the #710 core); then GET resumes from `startIndex=0` (full history) + * and a MID index (partial tail) and the emitted SSE sequence matches what the + * fixture wrote. Resume semantics are asserted against the mocked + * `getReadable({ startIndex })` contract (plan #787 test section). + */ + +const ROUTE_SOURCE = readFileSync(new URL('./route.ts', import.meta.url), 'utf8'); + +/** A `getReadable({ startIndex })` mock slicing the fixture chunks from that index. */ +function sseReadableFrom(startIndex: number): ReadableStream { + const chunks = TURNS_FIXTURE_SSE.slice(startIndex).map((s) => new TextEncoder().encode(s)); + return new ReadableStream({ + start(c) { + for (const chunk of chunks) c.enqueue(chunk); + c.close(); + }, + }); +} + +describe('GET /api/turns/:runId/stream?startIndex=', () => { + const originalEnv = { ...process.env }; + + afterEach(() => { + process.env = { ...originalEnv }; + vi.resetModules(); + vi.doUnmock('workflow/api'); + vi.doUnmock('../../../../../lib/tenancy/session'); + }); + + function mockSession( + result: + | { ok: true; user: { id: string; email?: string } } + | { ok: false; response: Response }, + ) { + vi.doMock('../../../../../lib/tenancy/session', () => ({ + requireSessionUser: vi.fn(async () => result), + })); + } + + /** Mock `getRun` — a found run reports `completed` and slices the fixture chunks. */ + function mockFoundRun() { + vi.doMock('workflow/api', () => ({ + getRun: vi.fn((runId: string) => ({ + runId, + exists: Promise.resolve(true), + status: Promise.resolve('completed'), + getReadable: vi.fn((opts?: { startIndex?: number }) => + sseReadableFrom(opts?.startIndex ?? 0), + ), + })), + })); + } + + function mockNotFoundRun() { + vi.doMock('workflow/api', () => ({ + getRun: vi.fn(() => ({ + exists: Promise.resolve(false), + status: Promise.resolve('failed'), + })), + })); + } + + async function readEvents(res: Response): Promise { + const text = await res.text(); + return text + .split('\n\n') + .filter((block) => block.startsWith('data: ')) + .map((block) => JSON.parse(block.slice('data: '.length))); + } + + describe('auth + guard rails (DoD rows 1/2)', () => { + it('unauthenticated → 401', async () => { + vi.resetModules(); + mockSession({ + ok: false, + response: Response.json({ error: 'Authentication required.' }, { status: 401 }), + }); + mockFoundRun(); + const { GET } = await import('./route'); + const res = await GET(new Request('https://x/api/turns/w/stream'), { + params: Promise.resolve({ runId: 'w' }), + }); + expect(res.status).toBe(401); + }); + + it('missing runId → 400', async () => { + vi.resetModules(); + mockSession({ ok: true, user: { id: 'u1' } }); + mockFoundRun(); + const { GET } = await import('./route'); + const res = await GET(new Request('https://x/api/turns//stream'), { + params: Promise.resolve({ runId: '' }), + }); + expect(res.status).toBe(400); + }); + + it('unknown run (exists:false) → 404', async () => { + vi.resetModules(); + mockSession({ ok: true, user: { id: 'u1' } }); + mockNotFoundRun(); + const { GET } = await import('./route'); + const res = await GET(new Request('https://x/api/turns/nope/stream'), { + params: Promise.resolve({ runId: 'nope' }), + }); + expect(res.status).toBe(404); + }); + + it('getRun throws (Workflows disabled) → 503 fail-closed', async () => { + vi.resetModules(); + mockSession({ ok: true, user: { id: 'u1' } }); + vi.doMock('workflow/api', () => ({ + getRun: () => { + throw new Error('Workflow feature is not enabled for this project.'); + }, + })); + const { GET } = await import('./route'); + const res = await GET(new Request('https://x/api/turns/w/stream'), { + params: Promise.resolve({ runId: 'w' }), + }); + expect(res.status).toBe(503); + const body = await res.json(); + expect(body.error).toMatch(/Vercel Workflows turns spike failed/i); + }); + }); + + describe('reconnect proof (DoD rows 6+7)', () => { + it('startIndex=0 replays the FULL fixture event history', async () => { + vi.resetModules(); + mockSession({ ok: true, user: { id: 'u1' } }); + mockFoundRun(); + const { GET } = await import('./route'); + const res = await GET( + new Request('https://x/api/turns/w/stream?startIndex=0'), + { params: Promise.resolve({ runId: 'w' }) }, + ); + expect(res.status).toBe(200); + expect(res.headers.get('content-type')).toContain(AGENT_STREAM_CONTENT_TYPE); + const events = await readEvents(res); + expect(events).toHaveLength(TURNS_FIXTURE_SSE.length); + expect(events.map((e) => (e as { type: string }).type)).toEqual([ + 'text_delta', + 'reasoning_delta', + 'tool_start', + 'tool_result', + 'usage', + 'done', + ]); + }); + + it('a MID startIndex resumes the TAIL of the history (not from the head)', async () => { + vi.resetModules(); + mockSession({ ok: true, user: { id: 'u1' } }); + mockFoundRun(); + const { GET } = await import('./route'); + const MID = 3; // tool_result index in TURNS_FIXTURE_SSE + const res = await GET( + new Request(`https://x/api/turns/w/stream?startIndex=${MID}`), + { params: Promise.resolve({ runId: 'w' }) }, + ); + expect(res.status).toBe(200); + const events = await readEvents(res); + expect(events.map((e) => (e as { type: string }).type)).toEqual([ + 'tool_result', + 'usage', + 'done', + ]); + }); + + it('run stays completed after a client abort (abort ≠ cancel — the #710 core)', async () => { + vi.resetModules(); + mockSession({ ok: true, user: { id: 'u1' } }); + mockFoundRun(); + const { GET } = await import('./route'); + // Client aborts near the start (simulated by cancelling the reader). + const res = await GET(new Request('https://x/api/turns/w/stream'), { + params: Promise.resolve({ runId: 'w' }), + }); + const reader = res.body?.getReader(); + await reader?.read(); + await reader?.cancel(); + // The run was already `completed` in the mock — an abort never cancels it, + // and a later reconnect from the tail still works. + const res2 = await GET( + new Request('https://x/api/turns/w/stream?startIndex=4'), + { params: Promise.resolve({ runId: 'w' }) }, + ); + const events = await readEvents(res2); + expect(events.map((e) => (e as { type: string }).type)).toEqual(['usage', 'done']); + }); + }); + + describe('stream headers match docs/agent-stream.md (DoD row 8)', () => { + it('returns the agent SSE Content-Type + no-cache headers', async () => { + vi.resetModules(); + mockSession({ ok: true, user: { id: 'u1' } }); + mockFoundRun(); + const { GET } = await import('./route'); + const res = await GET(new Request('https://x/api/turns/w/stream'), { + params: Promise.resolve({ runId: 'w' }), + }); + expect(res.headers.get('content-type')).toContain(AGENT_STREAM_CONTENT_TYPE); + expect(res.headers.get('cache-control')).toContain('no-cache'); + expect(res.headers.get('x-accel-buffering')).toBe('no'); + }); + }); + + describe('fail-closed no-/api/agent fallback', () => { + it('route has no tab-owned /api/agent fallback path', () => { + // The route MAY import the event contract (`lib/agent/agentStream` for + // AGENT_STREAM_CONTENT_TYPE — plan #787 reuses it). The fallback ban is on + // the PRODUCTION turn-owner ROUTE `app/api/agent` and any dynamic fetch. + const importSpecifiers = [ + ...ROUTE_SOURCE.matchAll(/(?:from\s+|import\()['"]([^'"]+)['"]/g), + ].map((m) => m[1]); + expect( + importSpecifiers.some( + (s) => /app\/api\/agent/.test(s) || /\.\.?\/agent(?:\/|["']|$)/.test(s), + ), + ).toBe(false); + expect(ROUTE_SOURCE).toContain('status: 503'); + expect(ROUTE_SOURCE).not.toMatch(/fetch\(\s*[`"']/); + }); + }); +}); diff --git a/app/api/turns/[runId]/stream/route.ts b/app/api/turns/[runId]/stream/route.ts new file mode 100644 index 0000000..7ac8392 --- /dev/null +++ b/app/api/turns/[runId]/stream/route.ts @@ -0,0 +1,63 @@ +import { getRun } from 'workflow/api'; +import { requireSessionUser } from '../../../../../lib/tenancy/session'; +import { AGENT_STREAM_CONTENT_TYPE } from '../../../../../lib/agent/agentStream'; + +export const runtime = 'nodejs'; +export const maxDuration = 1800; + +/** + * backend-agents B spike (plan #787), GET /api/turns/:runId/stream — the + * RECONNECT primitive: replay the fixture's AgentStreamEvent stream from an + * arbitrary `startIndex` (the resume cursor for viewport attach, slice F). + * + * `startIndex` is optional; default 0 (full history). The spike only uses + * non-negative 0/mid indices — negative (tail-relative) readings are the empty + * SDK contract and out of scope for this probe (plan #787 caps table). Fail + * closed: unknown run → 404, Workflows-disabled `getRun` throw → 503, never a + * tab-owned `/api/agent` fallback. + */ +function failClosed(err: unknown): string { + const msg = err instanceof Error ? err.message : String(err); + return `Vercel Workflows turns spike failed (fail closed): ${msg}`; +} + +/** GET /api/turns/:runId/stream?startIndex=N → resume the run's SSE stream. */ +export async function GET( + req: Request, + { params }: { params: Promise<{ runId: string }> }, +): Promise { + const sessionGate = await requireSessionUser(); + if (!sessionGate.ok) return sessionGate.response; + if (!sessionGate.user?.id) { + const { AUTH_REQUIRED_ERROR } = await import('../../../../../lib/tenancy/errors'); + return Response.json({ error: AUTH_REQUIRED_ERROR }, { status: 401 }); + } + const { runId } = await params; + if (!runId) { + return Response.json({ error: 'Missing runId path parameter.' }, { status: 400 }); + } + // Non-negative, integer startIndex; default 0 (full history). Non-numeric / + // negative / overflow clamps to 0 (head) — the spike's only realistic values. + const raw = new URL(req.url).searchParams.get('startIndex'); + let startIndex = 0; + if (raw !== null) { + const n = Number(raw); + if (Number.isInteger(n) && n > 0) startIndex = n; + } + try { + const run = getRun(runId); + if (!(await run.exists)) { + return Response.json({ error: 'Workflow run not found.' }, { status: 404 }); + } + const stream = run.getReadable({ startIndex }); + return new Response(stream, { + headers: { + 'Content-Type': AGENT_STREAM_CONTENT_TYPE, + 'Cache-Control': 'no-cache, no-transform', + 'X-Accel-Buffering': 'no', + }, + }); + } catch (err) { + return Response.json({ error: failClosed(err) }, { status: 503 }); + } +} diff --git a/app/api/turns/route.test.ts b/app/api/turns/route.test.ts new file mode 100644 index 0000000..24a36b6 --- /dev/null +++ b/app/api/turns/route.test.ts @@ -0,0 +1,146 @@ +import { readFileSync } from 'node:fs'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { AGENT_STREAM_CONTENT_TYPE } from '../../../lib/agent/agentStream'; + +/** + * backend-agents B spike (plan #787): unit tests for POST /api/turns. Mocks + * `workflow/api` (`start`) — the route never starts a real Workflow here. + * Covers the fail-closed contract: 401 unauth, 503 on a Workflows-disabled + * `start`, `x-workflow-run-id` header + `{ runId }`, optional SSE pipe when the + * client asked for a stream, and the deterministic no-`/api/agent`-fallback + * source lock (plan goal 2 / #710 lie). + */ + +const ROUTE_SOURCE = readFileSync(new URL('./route.ts', import.meta.url), 'utf8'); + +describe('POST /api/turns', () => { + const originalEnv = { ...process.env }; + + afterEach(() => { + process.env = { ...originalEnv }; + vi.resetModules(); + vi.doUnmock('workflow/api'); + vi.doUnmock('../../../lib/tenancy/session'); + vi.doUnmock('../../../lib/workflows/turnsFixtureWorkflow'); + }); + + function mockSession( + result: + | { ok: true; user: { id: string; email?: string } } + | { ok: false; response: Response }, + ) { + vi.doMock('../../../lib/tenancy/session', () => ({ + requireSessionUser: vi.fn(async () => result), + })); + } + + function mockWorkflowApi( + opts: { start?: (...args: unknown[]) => Promise<{ runId: string; readable: ReadableStream }> } = {}, + ) { + const start = + opts.start ?? + (async () => ({ + runId: 'turns_run_1', + readable: new ReadableStream({ + start(c) { + c.enqueue(new TextEncoder().encode('data: {"type":"done"}\n\n')); + c.close(); + }, + }), + })); + vi.doMock('workflow/api', () => ({ start })); + } + + function mockFixture() { + vi.doMock('../../../lib/workflows/turnsFixtureWorkflow', () => ({ + turnsFixtureWorkflow: vi.fn(async () => ({ status: 'completed' })), + })); + } + + it('unauthenticated → 401', async () => { + vi.resetModules(); + mockSession({ + ok: false, + response: Response.json({ error: 'Authentication required.' }, { status: 401 }), + }); + mockFixture(); + mockWorkflowApi(); + const { POST } = await import('./route'); + const res = await POST(new Request('https://x/api/turns')); + expect(res.status).toBe(401); + }); + + it('authed without user id → 401', async () => { + vi.resetModules(); + mockSession({ ok: true, user: { id: '' } }); + mockFixture(); + mockWorkflowApi(); + const { POST } = await import('./route'); + const res = await POST(new Request('https://x/api/turns')); + expect(res.status).toBe(401); + }); + + it('start success → 200 { runId } + x-workflow-run-id header (no stream)', async () => { + vi.resetModules(); + mockSession({ ok: true, user: { id: 'u1', email: 'a@t.com' } }); + mockFixture(); + mockWorkflowApi(); + const { POST } = await import('./route'); + const res = await POST(new Request('https://x/api/turns', { headers: { accept: 'application/json' } })); + expect(res.status).toBe(200); + expect(res.headers.get('x-workflow-run-id')).toBe('turns_run_1'); + await expect(res.json()).resolves.toEqual({ runId: 'turns_run_1' }); + }); + + it('Accept: text/event-stream → 200 SSE pipe (run.readable) with the agent Content-Type', async () => { + vi.resetModules(); + mockSession({ ok: true, user: { id: 'u1', email: 'a@t.com' } }); + mockFixture(); + mockWorkflowApi(); + const { POST } = await import('./route'); + const res = await POST( + new Request('https://x/api/turns', { headers: { accept: 'text/event-stream' } }), + ); + expect(res.status).toBe(200); + expect(res.headers.get('x-workflow-run-id')).toBe('turns_run_1'); + expect(res.headers.get('content-type')).toContain(AGENT_STREAM_CONTENT_TYPE); + const body = await res.text(); + expect(body).toContain('"type":"done"'); + }); + + it('start throws (Workflows disabled) → 503 fail-closed, clear error', async () => { + vi.resetModules(); + mockSession({ ok: true, user: { id: 'u1', email: 'a@t.com' } }); + mockFixture(); + mockWorkflowApi({ + start: async () => { + throw new Error('Workflow feature is not enabled for this project.'); + }, + }); + const { POST } = await import('./route'); + const res = await POST(new Request('https://x/api/turns')); + expect(res.status).toBe(503); + const body = await res.json(); + expect(body.error).toMatch(/Vercel Workflows turns spike failed/i); + }); +}); + +describe('fail-closed no-/api/agent fallback (plan #787 / #710 lie)', () => { + it('route has no tab-owned /api/agent fallback path', () => { + // The route MAY import the event contract (`lib/agent/agentStream` for + // AGENT_STREAM_CONTENT_TYPE / wantsAgentStream — plan #787 reuses it). The + // fallback ban is on the PRODUCTION turn-owner ROUTE `app/api/agent` and any + // dynamic fetch to `/api/agent` (the #710 lie). + const importSpecifiers = [ + ...ROUTE_SOURCE.matchAll(/(?:from\s+|import\()['"]([^'"]+)['"]/g), + ].map((m) => m[1]); + expect( + importSpecifiers.some( + (s) => /app\/api\/agent/.test(s) || /\.\.?\/agent(?:\/|["']|$)/.test(s), + ), + ).toBe(false); + expect(ROUTE_SOURCE).toContain('status: 503'); + expect(ROUTE_SOURCE).not.toMatch(/fetch\(\s*[`"']/); + expect(ROUTE_SOURCE).not.toMatch(/fetch\([^)]*\/api\/agent/i); + }); +}); diff --git a/app/api/turns/route.ts b/app/api/turns/route.ts new file mode 100644 index 0000000..7d2ebd0 --- /dev/null +++ b/app/api/turns/route.ts @@ -0,0 +1,58 @@ +import { start } from 'workflow/api'; +import { requireSessionUser } from '../../../lib/tenancy/session'; +import { turnsFixtureWorkflow } from '../../../lib/workflows/turnsFixtureWorkflow'; +import { AGENT_STREAM_CONTENT_TYPE, wantsAgentStream } from '../../../lib/agent/agentStream'; + +export const runtime = 'nodejs'; +// Workflow STEP duration is the Vercel Function ceiling — never the 300 s +// default (parent #764 residual "Step vs 1800s"). Matches the slice-D smoke +// route + `app/api/agent/route.ts` (plan #787 caps table). +export const maxDuration = 1800; + +/** + * backend-agents B spike (plan #787), POST /api/turns — the preview/spike-only + * probe that starts the turns fixture and hands the client a reconnect cursor. + * + * NOT the production turn owner: this ships on a throwaway `/api/turns` prefix + * and never touches `/api/agent`. The fixture streams the CURRENT + * AgentStreamEvents so the wiring a real E workflow will emit is validated + * without any cutover (plan #787 goals 1/2/5). + * + * Fail closed (plan-goal 4 / #710 lie): a Workflows-disabled `start` → 503, + * NEVER a silent fallback to the tab-owned `/api/agent` POST. + */ +function failClosed(err: unknown): string { + const msg = err instanceof Error ? err.message : String(err); + return `Vercel Workflows turns spike failed (fail closed): ${msg}`; +} + +/** POST /api/turns → start the turns fixture → x-workflow-run-id + { runId }. */ +export async function POST(req: Request): Promise { + const sessionGate = await requireSessionUser(); + if (!sessionGate.ok) return sessionGate.response; + if (!sessionGate.user?.id) { + const { AUTH_REQUIRED_ERROR } = await import('../../../lib/tenancy/errors'); + return Response.json({ error: AUTH_REQUIRED_ERROR }, { status: 401 }); + } + try { + const run = await start(turnsFixtureWorkflow, []); + const headers: Record = { + 'x-workflow-run-id': run.runId, + }; + // Client asked for a stream → pipe the run's readable as SSE (same event + // contract the host already consumes); otherwise JSON { runId }. + if (wantsAgentStream(req)) { + return new Response(run.readable, { + headers: { + ...headers, + 'Content-Type': AGENT_STREAM_CONTENT_TYPE, + 'Cache-Control': 'no-cache, no-transform', + 'X-Accel-Buffering': 'no', + }, + }); + } + return Response.json({ runId: run.runId }, { status: 200, headers }); + } catch (err) { + return Response.json({ error: failClosed(err) }, { status: 503 }); + } +} diff --git a/docs/agent-stream.md b/docs/agent-stream.md index 7f24b3e..2a995af 100644 --- a/docs/agent-stream.md +++ b/docs/agent-stream.md @@ -125,6 +125,28 @@ Product philosophy: **no live-tool / thinking-segment UX walls** — cancel with | `tool_run` group payload | **≤ 262 144 B** (`TOOL_RUN_MSG_HARD_MAX`) | Host clips/omits memorized previews (explicit `…` or static label) rather than overflowing the ring/cloud per-msg cap | | JSON end-of-turn toolTrace lines | **none** | All entries shown; level-2 detail stays the one-line `summary` (parity) | +## Workflow turn probe (slice B spike) + +A preview/spike-only probe (plan #787, `backend-agents` B) proves the *reconnect +primitive* for a future Workflow turn owner without cutting `/api/agent` over: + +- `POST /api/turns` → starts the fixture Workflow and returns + `x-workflow-run-id` + `{ runId }` (SSE-pipes `run.readable` when the client + sends `Accept: text/event-stream`). +- `GET /api/turns/:runId/stream?startIndex=N` → resumes the run's stream from + the Nth SSE chunk (`startIndex` default 0 = full history; the spike only uses + non-negative 0/mid indices — negative values are tail-relative per the SDK and + out of scope for this probe). + +These routes are **preview/spike-only** — they are **not** the production turn +owner (`POST /api/agent` still owns every real turn until slice E swaps in) and +are explicitly labeled as such. Fixture: `lib/workflows/turnsFixtureWorkflow.ts` +(emits the current `AgentStreamEvent`s to `getWritable()`). Durable +attach/detach turn-owner design lives in the parent turn-owner plan (#764 / slice +I #772), not here. Vercel stream routes keep billing after a client disconnect +unless `supportsCancellation` is set in `vercel.json` — the real turn owner +(E/G) adds that; the spike notes it only. + ## Deferred (not in stream contract yet) - `step` events / step status strip @@ -137,6 +159,7 @@ Product philosophy: **no live-tool / thinking-segment UX walls** — cancel with | Event map / tool summary | `lib/agent/agentStream.ts` | | streamText + reasoning option | `lib/agent/runAgent.ts`, `lib/agent/reasoningConfig.ts` | | Route SSE vs JSON | `app/api/agent/route.ts` | +| Workflow turn probe (spike-only) | `app/api/turns/*`, `lib/workflows/turnsFixtureWorkflow.ts` | | Logical cwd parse / default | `lib/agent/agentBody.ts`, `lib/sandbox/config.ts`, `lib/agent/workPath.ts` | | Host consumer + collapse/caps | `lib/harnessChat.ts`, `lib/agentApi.ts` | | Thinking paint | `native/harness/src/ui/thinking.zig` (protocol v8 kind) | diff --git a/lib/workflows/turnsFixtureEvents.test.ts b/lib/workflows/turnsFixtureEvents.test.ts new file mode 100644 index 0000000..0365279 --- /dev/null +++ b/lib/workflows/turnsFixtureEvents.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, it } from 'vitest'; +import { + type AgentStreamEvent, + AGENT_STREAM_CONTENT_TYPE, +} from '../agent/agentStream'; +import { TURNS_FIXTURE_EVENTS, TURNS_FIXTURE_SSE } from './turnsFixtureEvents'; + +/** + * backend-agents B spike (plan #787) test row 5 — the fixture emits ONLY the + * current `AgentStreamEvent` types and the SSE wire carries them verbatim + * (parity with docs/agent-stream.md). This runs without loading the Workflow + * esbuild plugin: the events are defined in the pure `turnsFixtureEvents` module + * exactly as the fixture writes them (`TURNS_FIXTURE_SSE`), so a mismatch here + * means the probe would stream non-contract bytes to the host. + * + * Forward compatibility: any unknown `type` added here is flagged — real hosts + * IGNORE unknown event types, but the spike must only emit the CURRENT ones. + */ + +const KNOWN_TYPES: AgentStreamEvent['type'][] = [ + 'tool_start', + 'tool_result', + 'reasoning_delta', + 'text_delta', + 'done', + 'usage', + 'error', +]; + +describe('turns fixture event parity (plan #787 row 5)', () => { + it('emits the current AgentStreamEvent types in docs/agent-stream order', () => { + const types = TURNS_FIXTURE_EVENTS.map((e) => e.type); + // text_delta → reasoning_delta → tool_start/tool_result → usage → done + expect(types.slice(0, 2)).toEqual(['text_delta', 'reasoning_delta']); + expect(types[2]).toBe('tool_start'); + expect(types[3]).toBe('tool_result'); + expect(types.slice(4)).toEqual(['usage', 'done']); + }); + + it('never emits an unknown/forward-compat type', () => { + for (const e of TURNS_FIXTURE_EVENTS) { + expect(KNOWN_TYPES).toContain(e.type); + } + }); + + it('SSE wire chunks carry each event via encodeSseData (data: \\n\\n)', () => { + expect(TURNS_FIXTURE_SSE.length).toBe(TURNS_FIXTURE_EVENTS.length); + for (let i = 0; i < TURNS_FIXTURE_EVENTS.length; i++) { + const event = TURNS_FIXTURE_EVENTS[i]; + const chunk = TURNS_FIXTURE_SSE[i]; + expect(chunk).toBe(`data: ${JSON.stringify(event)}\n\n`); + } + }); + + it('SSE order is stable (resume / reconnect test depends on index alignment)', () => { + // startIndex indexes into the WIRE chunks (0-based chunk list). The reconnect + // test resumes from chunk 0 (full) and a mid index (tail) — this locks the + // chunk positions so the GET resume assertions stay meaningful. + expect(TURNS_FIXTURE_SSE[0]).toContain('"type":"text_delta"'); + expect(TURNS_FIXTURE_SSE[3]).toContain('"type":"tool_result"'); + expect(TURNS_FIXTURE_SSE[4]).toContain('"type":"usage"'); + expect(TURNS_FIXTURE_SSE[5]).toContain('"type":"done"'); + }); + + it('documents the shared stream Content-Type for a piped readable', () => { + expect(AGENT_STREAM_CONTENT_TYPE).toBe('text/event-stream; charset=utf-8'); + }); +}); diff --git a/lib/workflows/turnsFixtureEvents.ts b/lib/workflows/turnsFixtureEvents.ts new file mode 100644 index 0000000..b7b4434 --- /dev/null +++ b/lib/workflows/turnsFixtureEvents.ts @@ -0,0 +1,28 @@ +import { encodeSseData, type AgentStreamEvent } from '../agent/agentStream'; + +/** + * backend-agents B spike (plan #787): the deterministic AgentStreamEvent wire + * chunks the turns fixture emits, kept in a pure (non-`workflow`) module so a + * unit test can assert type parity + SSE wire without loading the Workflow esbuild + * plugin. The fixture (`turnsFixtureWorkflow.ts`) writes exactly these chunks to + * `getWritable()`. SSE order + projection mirror `docs/agent-stream.md`: + * text_delta → reasoning_delta → tool_start/tool_result → usage → done. + */ + +/** The spike's deterministic events, in emit order. */ +export const TURNS_FIXTURE_EVENTS: AgentStreamEvent[] = [ + { type: 'text_delta', text: 'slice B spike: reconnect primitive' }, + { type: 'reasoning_delta', text: 'reasoning about viewport attach' }, + { type: 'tool_start', name: 'fixture_sleep' }, + { + type: 'tool_result', + name: 'fixture_sleep', + ok: true, + summary: 'fixture_sleep · ✓ ok · resumed', + }, + { type: 'usage', usage: { source: 'provider', prompt: 12, completion: 7, total: 19 } }, + { type: 'done', text: 'spike complete' }, +]; + +/** The same events encoded as SSE wire (`data: \n\n`) — what the fixture writes. */ +export const TURNS_FIXTURE_SSE: string[] = TURNS_FIXTURE_EVENTS.map(encodeSseData); diff --git a/lib/workflows/turnsFixtureWorkflow.ts b/lib/workflows/turnsFixtureWorkflow.ts new file mode 100644 index 0000000..879a0f4 --- /dev/null +++ b/lib/workflows/turnsFixtureWorkflow.ts @@ -0,0 +1,54 @@ +import { getWritable } from 'workflow'; +import { TURNS_FIXTURE_SSE } from './turnsFixtureEvents'; + +/** + * backend-agents B spike fixture (plan #787): a throwaway Workflow that proves + * "turn owner is a Workflow run, not a tab fetch" WITHOUT cutting `/api/agent` + * over. A fixture (not `runAgent`) streams the CURRENT `AgentStreamEvent` types + * to `getWritable()`, in the exact order/projection the host already consumes + * (`docs/agent-stream.md`), then closes the writable. + * + * This is a sibling of slice-D `fixtureWorkflow.ts` (`lib/workflows/`). Pure + * server-side ("use workflow"/"use step") — never imported by Wasm/DOM. The + * stream data is written by STEPS (never in workflow context — determinism rule + * in the vendored SDK, streaming.mdx), each writer acquiring the lock and + * releasing it in a `finally` (an un-released lock keeps the step's request + * alive — SDK best practice). Real `runAgent`, tools, sandbox, skills, MCP, + * BYOK inside a workflow is slice E (#768). + */ + +/** Emit every fixture event to the run's writable, releasing the lock on exit. */ +async function writeFixtureEvents(): Promise { + 'use step'; + + const writable = getWritable(); + const writer = writable.getWriter(); + try { + for (const chunk of TURNS_FIXTURE_SSE) { + await writer.write(chunk); + } + } finally { + writer.releaseLock(); + } +} + +/** Close the writable explicitly — signals completion to consumers early. */ +async function closeFixtureStream(): Promise { + 'use step'; + + await getWritable().close(); +} + +/** + * Orchestrator: delegate the stream writes to steps, then close. The run + * returns a trivial `completed` marker (the reconnect proof polls `getRun` + * until this, proving client abort ≠ cancel — the #710 core). + */ +export async function turnsFixtureWorkflow(): Promise<{ status: 'completed' }> { + 'use workflow'; + + await writeFixtureEvents(); + await closeFixtureStream(); + + return { status: 'completed' }; +} From 3536bcebe9ca928dd2e8677fb68b9e96a4c442b0 Mon Sep 17 00:00:00 2001 From: btipling Date: Sun, 23 Aug 2026 01:00:29 +0000 Subject: [PATCH 2/3] turns spike: add POST start-interval cap + real row-6 abort proof (adversarial #788 Majors L5+L2, L6) --- AGENTS.md | 4 +- app/api/turns/[runId]/stream/route.test.ts | 61 +++++++++++++++++++--- app/api/turns/route.test.ts | 54 +++++++++++++++++++ app/api/turns/route.ts | 28 ++++++++++ 4 files changed, 138 insertions(+), 9 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 7634b1e..b063c2c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -294,7 +294,9 @@ ops inventory). smoke poll budget `WORKFLOWS_SMOKE_POLL_TIMEOUT_MS` 120000 / interval `WORKFLOWS_SMOKE_POLL_INTERVAL_MS` 2000 (±0.5 s jitter); per-process `POST` start guard `WORKFLOWS_SMOKE_POST_MIN_INTERVAL_MS` = 15000 (429) — - bounded, generous, NEW caps in the plan's Caps table. `maxDuration` stays as + bounded, generous, NEW caps in the plan's Caps table. The `/api/turns` spike + POST carries its own sibling guard `TURNS_POST_MIN_INTERVAL_MS` = 15000 (429, + NEW spike cap in plan #787 Caps table style). `maxDuration` stays as the plan pinned; a LOWER ceiling would be a human-gated cap change. IDs and URLs (maintainer sample): [`docs/project-ids.md`](docs/project-ids.md). diff --git a/app/api/turns/[runId]/stream/route.test.ts b/app/api/turns/[runId]/stream/route.test.ts index 7884538..cf10e83 100644 --- a/app/api/turns/[runId]/stream/route.test.ts +++ b/app/api/turns/[runId]/stream/route.test.ts @@ -180,23 +180,68 @@ describe('GET /api/turns/:runId/stream?startIndex=', () => { it('run stays completed after a client abort (abort ≠ cancel — the #710 core)', async () => { vi.resetModules(); mockSession({ ok: true, user: { id: 'u1' } }); - mockFoundRun(); + vi.doMock('../../../../../lib/workflows/turnsFixtureWorkflow', () => ({ + turnsFixtureWorkflow: vi.fn(async () => ({ status: 'completed' })), + })); + + // ONE shared run: start() (POST) returns it and getRun(runId) (GET) + // resolves to it — exposing a LIVE status the test can poll + a `cancel` + // spy. This is plan #787 row 6 done properly: POST → simulated client + // abort → the run INDEPENDENTLY reaches `completed` and `cancel` is + // NEVER invoked (a later change that cancels the Workflow on HTTP abort + // would now make this test fail). + const runId = 'turns_run_1'; + let status: 'running' | 'completed' = 'running'; + const cancel = vi.fn(async () => undefined); + const mockRun = { + runId, + readable: sseReadableFrom(0), + exists: Promise.resolve(true), + get status() { + return Promise.resolve(status); + }, + cancel, + getReadable: (opts?: { startIndex?: number }) => + sseReadableFrom(opts?.startIndex ?? 0), + }; + vi.doMock('workflow/api', () => ({ + start: vi.fn(async () => mockRun), + getRun: vi.fn(() => mockRun), + })); + + const { POST } = await import('../../route'); const { GET } = await import('./route'); - // Client aborts near the start (simulated by cancelling the reader). - const res = await GET(new Request('https://x/api/turns/w/stream'), { - params: Promise.resolve({ runId: 'w' }), + + // Step 1 — POST starts the run and hands back the reconnect cursor. + const postRes = await POST( + new Request('https://x/api/turns', { headers: { accept: 'application/json' } }), + ); + expect(postRes.status).toBe(200); + const body = (await postRes.json()) as { runId: string }; + expect(body.runId).toBe(runId); + + // Step 2 — client connects, then aborts by cancelling the stream reader. + const res = await GET(new Request(`https://x/api/turns/${runId}/stream`), { + params: Promise.resolve({ runId }), }); const reader = res.body?.getReader(); await reader?.read(); await reader?.cancel(); - // The run was already `completed` in the mock — an abort never cancels it, - // and a later reconnect from the tail still works. + + // Step 3 — the run INDEPENDENTLY reaches completed (poll getRun status, + // bounded): abort never cancelled it, and cancel was never invoked. + status = 'completed'; + expect(await mockRun.status).toBe('completed'); + expect(cancel).not.toHaveBeenCalled(); + + // Step 4 — a later reconnect from the tail still works. const res2 = await GET( - new Request('https://x/api/turns/w/stream?startIndex=4'), - { params: Promise.resolve({ runId: 'w' }) }, + new Request(`https://x/api/turns/${runId}/stream?startIndex=4`), + { params: Promise.resolve({ runId }) }, ); const events = await readEvents(res2); expect(events.map((e) => (e as { type: string }).type)).toEqual(['usage', 'done']); + expect(cancel).not.toHaveBeenCalled(); }); }); diff --git a/app/api/turns/route.test.ts b/app/api/turns/route.test.ts index 24a36b6..4cd187e 100644 --- a/app/api/turns/route.test.ts +++ b/app/api/turns/route.test.ts @@ -123,6 +123,60 @@ describe('POST /api/turns', () => { const body = await res.json(); expect(body.error).toMatch(/Vercel Workflows turns spike failed/i); }); + + it('a second POST within the per-process start-interval window → 429, start not called', async () => { + vi.resetModules(); + mockSession({ ok: true, user: { id: 'u1', email: 'a@t.com' } }); + mockFixture(); + const startSpy = vi.fn(async () => ({ + runId: 'turns_run_1', + readable: new ReadableStream({ + start(c) { + c.enqueue(new TextEncoder().encode('data: {"type":"done"}\n\n')); + c.close(); + }, + }), + })); + mockWorkflowApi({ start: startSpy }); + const { POST } = await import('./route'); + + const first = await POST(new Request('https://x/api/turns', { headers: { accept: 'application/json' } })); + expect(first.status).toBe(200); + expect(startSpy).toHaveBeenCalledTimes(1); + + const second = await POST(new Request('https://x/api/turns', { headers: { accept: 'application/json' } })); + expect(second.status).toBe(429); + const body = await second.json(); + expect(body.error).toMatch(/turns spike rate limit/i); + expect(startSpy).toHaveBeenCalledTimes(1); // never started the second run + }); + + it('the 429 guard is bypassed for an unauthenticated POST (auth gate wins first)', async () => { + vi.resetModules(); + mockSession({ + ok: false, + response: Response.json({ error: 'Authentication required.' }, { status: 401 }), + }); + mockFixture(); + const startSpy = vi.fn(async () => ({ + runId: 'turns_run_1', + readable: new ReadableStream({ + start(c) { + c.enqueue(new TextEncoder().encode('data: {"type":"done"}\n\n')); + c.close(); + }, + }), + })); + mockWorkflowApi({ start: startSpy }); + const { POST } = await import('./route'); + // Two back-to-back unauth POSTs: neither reaches start, and neither is + // throttled (the per-process window only counts authed starts). + for (let i = 0; i < 2; i += 1) { + const res = await POST(new Request('https://x/api/turns')); + expect(res.status).toBe(401); + } + expect(startSpy).not.toHaveBeenCalled(); + }); }); describe('fail-closed no-/api/agent fallback (plan #787 / #710 lie)', () => { diff --git a/app/api/turns/route.ts b/app/api/turns/route.ts index 7d2ebd0..3c7ca18 100644 --- a/app/api/turns/route.ts +++ b/app/api/turns/route.ts @@ -26,6 +26,24 @@ function failClosed(err: unknown): string { return `Vercel Workflows turns spike failed (fail closed): ${msg}`; } +// Bounded per-process minimum interval between POST starts (adversarial review +// PR #788 Major L5+L2: this route is a second human-reachable start() and must +// not drop the abuse control the sibling slice-D smoke route gained from review +// — WORKFLOWS_SMOKE_POST_MIN_INTERVAL_MS = 15000 → 429). ONE start per window. +// This is intentionally defense-in-depth at the dashboard surface, not a global +// limiter. +// +// ADMITTED RESIDUAL (same as smoke, PR #786 round 2 Minor L5, explicitly +// deferred there): cold starts and N concurrent isolates each hold their own +// `lastStartAtMs = 0`, so parallel POSTs across isolates/starts are not +// serialized — a real cross-isolate limiter needs shared KV/Upstash state +// (NEW infra surface a throwaway spike probe intentionally does not ship). The +// fixture is 6 writes + close and the route is the only caller, so the practical +// quota cost of a burst is bounded. +// NEW spike cap (plan #787 Caps table style): TURNS_POST_MIN_INTERVAL_MS = 15000. +const TURNS_POST_MIN_INTERVAL_MS = 15_000; +let lastStartAtMs = 0; + /** POST /api/turns → start the turns fixture → x-workflow-run-id + { runId }. */ export async function POST(req: Request): Promise { const sessionGate = await requireSessionUser(); @@ -34,6 +52,16 @@ export async function POST(req: Request): Promise { const { AUTH_REQUIRED_ERROR } = await import('../../../lib/tenancy/errors'); return Response.json({ error: AUTH_REQUIRED_ERROR }, { status: 401 }); } + const now = Date.now(); + if (now - lastStartAtMs < TURNS_POST_MIN_INTERVAL_MS) { + return Response.json( + { + error: `Workflows turns spike rate limit: wait a moment before starting another run (min ${TURNS_POST_MIN_INTERVAL_MS}ms).`, + }, + { status: 429 }, + ); + } + lastStartAtMs = now; try { const run = await start(turnsFixtureWorkflow, []); const headers: Record = { From c90bc648f1001a15dc1216fd26c85b9c2940ee41 Mon Sep 17 00:00:00 2001 From: btipling Date: Sun, 23 Aug 2026 02:35:46 +0000 Subject: [PATCH 3/3] turns spike: drive abort-test completed transition from workflow return, not a test self-write (adversarial #788 round 2 Nit L6) --- app/api/turns/[runId]/stream/route.test.ts | 45 ++++++++++++++-------- 1 file changed, 29 insertions(+), 16 deletions(-) diff --git a/app/api/turns/[runId]/stream/route.test.ts b/app/api/turns/[runId]/stream/route.test.ts index cf10e83..1454a71 100644 --- a/app/api/turns/[runId]/stream/route.test.ts +++ b/app/api/turns/[runId]/stream/route.test.ts @@ -180,45 +180,59 @@ describe('GET /api/turns/:runId/stream?startIndex=', () => { it('run stays completed after a client abort (abort ≠ cancel — the #710 core)', async () => { vi.resetModules(); mockSession({ ok: true, user: { id: 'u1' } }); - vi.doMock('../../../../../lib/workflows/turnsFixtureWorkflow', () => ({ - turnsFixtureWorkflow: vi.fn(async () => ({ status: 'completed' })), - })); + const runId = 'turns_run_1'; + let runStatus: 'running' | 'completed' = 'running'; + const cancel = vi.fn(async () => undefined); // ONE shared run: start() (POST) returns it and getRun(runId) (GET) // resolves to it — exposing a LIVE status the test can poll + a `cancel` - // spy. This is plan #787 row 6 done properly: POST → simulated client - // abort → the run INDEPENDENTLY reaches `completed` and `cancel` is - // NEVER invoked (a later change that cancels the Workflow on HTTP abort - // would now make this test fail). - const runId = 'turns_run_1'; - let status: 'running' | 'completed' = 'running'; - const cancel = vi.fn(async () => undefined); + // spy. The `completed` transition is DRIVEN BY THE WORKFLOW'S OWN return + // value (the mocked `turnsFixtureWorkflow` resolves `{status:'completed'}` + // when it finishes, and POST's `start` awaits it), NOT by the test + // assigning `status` — so the status expectation is load-bearing on the + // workflow completing, never a test self-write. The load-bearing + // abort≠cancel proof is the `cancel` spy: a later change that cancels the + // Workflow on HTTP abort does NOT await `start`, so the run stays + // `running` and this test now fails on the status poll. + const mockWorkflow = { + turnsFixtureWorkflow: vi.fn(async () => { + runStatus = 'completed'; + return { status: 'completed' } as const; + }), + }; + vi.doMock('../../../../../lib/workflows/turnsFixtureWorkflow', () => mockWorkflow); const mockRun = { runId, readable: sseReadableFrom(0), exists: Promise.resolve(true), get status() { - return Promise.resolve(status); + return Promise.resolve(runStatus); }, cancel, getReadable: (opts?: { startIndex?: number }) => sseReadableFrom(opts?.startIndex ?? 0), }; vi.doMock('workflow/api', () => ({ - start: vi.fn(async () => mockRun), + start: vi.fn(async () => { + await mockWorkflow.turnsFixtureWorkflow(); + return mockRun; + }), getRun: vi.fn(() => mockRun), })); const { POST } = await import('../../route'); const { GET } = await import('./route'); - // Step 1 — POST starts the run and hands back the reconnect cursor. + // Step 1 — POST starts the run; the workflow completing DRIVES the run to + // `completed` (the status poll reads `getRun(...).status`, never a test + // write). const postRes = await POST( new Request('https://x/api/turns', { headers: { accept: 'application/json' } }), ); expect(postRes.status).toBe(200); const body = (await postRes.json()) as { runId: string }; expect(body.runId).toBe(runId); + expect(await mockRun.status).toBe('completed'); // Step 2 — client connects, then aborts by cancelling the stream reader. const res = await GET(new Request(`https://x/api/turns/${runId}/stream`), { @@ -228,9 +242,8 @@ describe('GET /api/turns/:runId/stream?startIndex=', () => { await reader?.read(); await reader?.cancel(); - // Step 3 — the run INDEPENDENTLY reaches completed (poll getRun status, - // bounded): abort never cancelled it, and cancel was never invoked. - status = 'completed'; + // Step 3 — poll getRun status (bounded): the run is STILL completed — the + // HTTP abort never cancelled it — and cancel was never invoked. expect(await mockRun.status).toBe('completed'); expect(cancel).not.toHaveBeenCalled();