diff --git a/app/api/turns/[runId]/cancel/route.test.ts b/app/api/turns/[runId]/cancel/route.test.ts new file mode 100644 index 00000000..e0361f09 --- /dev/null +++ b/app/api/turns/[runId]/cancel/route.test.ts @@ -0,0 +1,629 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { AUTH_REQUIRED_ERROR } from '../../../../../lib/tenancy/errors'; + +/** + * Route tests for backend-agents G22 (#816) — `POST /api/turns/:runId/cancel` + * durable-turn server cancel seam. + * + * Mocks `getRun`, `sanitizeTurnRunId`, `isRedisSafeOpaqueId`, + * `requireSessionUser`, `resolveSessionStore`, `sessionKeyFor`, + * `isEnvelopeStore`, `overlayWorkerMeta`, and `createProdServices` so the + * route never opens a real DB/Redis connection or reaches the Workflows API. + * + * Covers the locked test matrix (plan #816 Testing §Route unit): + * 1. 401 unauth (requireSessionUser gate fires first) + * 2. 400 invalid runId (sanitizeTurnRunId → undefined) + * 3. 400 missing sessionId + * 3b. 400 invalid (non-opaque) sessionId + * 4. 404 ownership mismatch (envelope.turnRunId !== runId) + * 4b. 404 absent run (run.exists === false) + * 5. 409 terminal-with-status no-op (run.cancel NOT called, no overlay) + * 6. live run → cancel() called exactly once + 'cancelling' overlay PATCH + * with strictly-newer updatedAt (LWW: Math.max(now, stored+1)) + * 7. cancel throw → 503 fail-closed and NO overlay write + * 8. store unavailable → 503 (resolve not-ok / resolve throws / read throws) + * 9. 429 min-interval soft guard: second accepted cancel of the SAME run + * inside the window → 429 + Retry-After; a terminal 409 / 404 / 503 never + * burns the window; an accepted cancel of wr_a does NOT 429 wr_b on the + * same session (pass 8: window is sessionId:runId) + * 10. accepted cancel with overlay {ok:false} → still 200 + warning (PATCH + * failure is non-fatal — the run's own terminal persist owns the truth) + * 11. tenant resolve failure → 503 + */ + +/** Distant-future stored updatedAt so the LWW clock assertion proves stored+1. */ +const FUTURE_UPDATED_AT = 9_000_000_000_000; + +describe('POST /api/turns/:runId/cancel', () => { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + let getRunMock: any; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + let sanitizeTurnRunIdMock: any; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + let isRedisSafeOpaqueIdMock: any; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + let overlayWorkerMetaMock: any; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + let readEnvelopeMock: any; + + /** Spy on run.cancel — asserted exactly-once on accept, never on terminal. */ + let runCancelSpy: ReturnType; + + // Tenancy mock state + let envelopeTurnRunId: string; + let envelopePresent: boolean; + let tenantResolveOk: boolean; + let storeAvailable: boolean; + let storeResolveThrows: boolean; + let readEnvelopeThrows: boolean; + let runStatus: string; + let runExists: boolean; + + function resetState() { + runCancelSpy = vi.fn(async () => {}); + overlayWorkerMetaMock = vi.fn(async () => ({ ok: true as const, meta: {} })); + readEnvelopeMock = vi.fn(async () => { + if (readEnvelopeThrows) throw new Error('read-envelope-crash'); + return envelopePresent + ? { meta: { turnRunId: envelopeTurnRunId }, updatedAt: FUTURE_UPDATED_AT } + : null; + }); + + envelopeTurnRunId = 'wf_turn_123'; + envelopePresent = true; + tenantResolveOk = true; + storeAvailable = true; + storeResolveThrows = false; + readEnvelopeThrows = false; + runStatus = 'running'; + runExists = true; + } + + function mockGetRun() { + getRunMock = vi.fn(() => ({ + runId: 'wf_turn_123', + exists: Promise.resolve(runExists), + status: Promise.resolve(runStatus), + cancel: runCancelSpy, + })); + vi.doMock('workflow/api', () => ({ + getRun: getRunMock, + start: vi.fn(), + })); + } + + function mockSessionCaps() { + sanitizeTurnRunIdMock = vi.fn((v: unknown) => { + if (typeof v !== 'string') return undefined; + const s = v.trim(); + if (!s || s.length > 512) return undefined; + return /^[A-Za-z0-9_-]{1,512}$/.test(s) ? s : undefined; + }); + isRedisSafeOpaqueIdMock = vi.fn((s: unknown) => { + if (typeof s !== 'string') return false; + return /^[A-Za-z0-9_-]{1,512}$/.test(s); + }); + vi.doMock('../../../../../lib/sessionCloudCaps', () => ({ + sanitizeTurnRunId: sanitizeTurnRunIdMock, + isRedisSafeOpaqueId: isRedisSafeOpaqueIdMock, + TURN_CANCEL_MIN_INTERVAL_MS: 1000, + })); + } + + function mockAuthedSession(userId = 'u1') { + vi.doMock('../../../../../lib/tenancy/session', () => ({ + requireSessionUser: vi.fn(async () => ({ + ok: true as const, + user: { id: userId, email: 'a@b.c' }, + })), + })); + } + + function mockUnauthed() { + vi.doMock('../../../../../lib/tenancy/session', () => ({ + requireSessionUser: vi.fn(async () => ({ + ok: false as const, + response: Response.json( + { error: AUTH_REQUIRED_ERROR }, + { status: 401 }, + ), + })), + })); + } + + function mockTenancyOk(sessionId = 's1', turnRunId?: string) { + const resolvedTurnRunId = turnRunId ?? envelopeTurnRunId; + envelopeTurnRunId = resolvedTurnRunId; + + vi.doMock('../../../../../lib/di', () => ({ + createProdServices: vi.fn(() => ({ + harnessSessionsRedis: { + resolveTenantIdForUser: vi.fn(async () => + tenantResolveOk + ? { ok: true as const, value: 't1' } + : { ok: false as const, code: 'SESSION_STORE_UNAVAILABLE' as const, error: 'db-down' }, + ), + }, + })), + })); + + vi.doMock('../../../../../lib/tenancy/harnessSessionsRedis', () => ({ + resolveSessionStore: vi.fn(async () => { + if (storeResolveThrows) throw new Error('store-resolve-crash'); + if (!storeAvailable) + return { ok: false as const, code: 'SESSION_STORE_UNAVAILABLE' as const, error: 'down' }; + return { + ok: true as const, + value: { readEnvelope: readEnvelopeMock }, + }; + }), + sessionKeyFor: vi.fn( + (_tenantId: string, _userId: string, sid: string) => + ({ tenantId: 't1', userId: 'u1', sessionId: sid }), + ), + })); + + vi.doMock('../../../../../lib/sessions/sessionStore', () => ({ + isEnvelopeStore: vi.fn(() => true), + })); + + vi.doMock('../../../../../lib/agent/workerMetaOverlay', () => ({ + overlayWorkerMeta: overlayWorkerMetaMock, + })); + } + + function standardHarness(sessionId = 's1') { + mockSessionCaps(); + mockGetRun(); + mockTenancyOk(sessionId); + } + + function postCancel(runId: string, sessionId?: string): Promise { + const url = new URL(`https://x/api/turns/${runId}/cancel`); + if (sessionId !== undefined) url.searchParams.set('sessionId', sessionId); + return POST( + new Request(url, { method: 'POST' }), + { params: Promise.resolve({ runId }) }, + ); + } + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + let POST: (req: Request, ctx: any) => Promise; + + beforeEach(() => { + resetState(); + }); + + afterEach(() => { + vi.resetModules(); + vi.doUnmock('workflow/api'); + vi.doUnmock('../../../../../lib/sessionCloudCaps'); + vi.doUnmock('../../../../../lib/tenancy/session'); + vi.doUnmock('../../../../../lib/di'); + vi.doUnmock('../../../../../lib/tenancy/harnessSessionsRedis'); + vi.doUnmock('../../../../../lib/sessions/sessionStore'); + vi.doUnmock('../../../../../lib/agent/workerMetaOverlay'); + }); + + // ── Row 1 — 401 unauth ── + it('row 1 — auth failure → 401 before any gate', async () => { + standardHarness(); + mockUnauthed(); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', 's1'); + + expect(res.status).toBe(401); + expect(await res.json()).toEqual({ error: AUTH_REQUIRED_ERROR }); + expect(getRunMock).not.toHaveBeenCalled(); + expect(runCancelSpy).not.toHaveBeenCalled(); + expect(overlayWorkerMetaMock).not.toHaveBeenCalled(); + }); + + // ── Row 2 — 400 invalid runId ── + it('row 2 — invalid runId (sanitizeTurnRunId → undefined) → 400, getRun NOT called', async () => { + mockGetRun(); + mockAuthedSession(); + + sanitizeTurnRunIdMock = vi.fn((_v: unknown) => undefined); + vi.doMock('../../../../../lib/sessionCloudCaps', () => ({ + sanitizeTurnRunId: sanitizeTurnRunIdMock, + isRedisSafeOpaqueId: vi.fn(() => true), + TURN_CANCEL_MIN_INTERVAL_MS: 1000, + })); + + mockTenancyOk(); + ({ POST } = await import('./route')); + + const res = await postCancel('bad:id!', 's1'); + + expect(res.status).toBe(400); + const body = (await res.json()) as { error: string }; + expect(body.error).toBe('Invalid runId'); + expect(getRunMock).not.toHaveBeenCalled(); + expect(runCancelSpy).not.toHaveBeenCalled(); + }); + + // ── Row 3 — 400 missing sessionId ── + it('row 3 — missing sessionId query param → 400', async () => { + standardHarness(); + mockAuthedSession(); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123'); + + expect(res.status).toBe(400); + const body = (await res.json()) as { error: string }; + expect(body.error).toMatch(/sessionId/); + expect(getRunMock).not.toHaveBeenCalled(); + expect(runCancelSpy).not.toHaveBeenCalled(); + }); + + // ── Row 3b — 400 invalid (non-opaque) sessionId ── + it('row 3b — invalid (non-opaque) sessionId → 400', async () => { + standardHarness(); + mockAuthedSession(); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', '*'); + + expect(res.status).toBe(400); + const body = (await res.json()) as { error: string }; + expect(body.error).toMatch(/Invalid sessionId/); + expect(getRunMock).not.toHaveBeenCalled(); + expect(runCancelSpy).not.toHaveBeenCalled(); + }); + + // ── Row 4 — 404 ownership mismatch ── + it('row 4 — envelope.turnRunId !== runId → 404 (tenancy guard), getRun NOT called', async () => { + mockSessionCaps(); + mockGetRun(); + mockAuthedSession(); + mockTenancyOk('s1', 'wf_other_run'); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', 's1'); + + expect(res.status).toBe(404); + const body = (await res.json()) as { error: string }; + expect(body.error).toMatch(/Run not found/); + expect(body.error).toContain('wf_turn_123'); + expect(getRunMock).not.toHaveBeenCalled(); + expect(runCancelSpy).not.toHaveBeenCalled(); + expect(overlayWorkerMetaMock).not.toHaveBeenCalled(); + }); + + // ── Row 4b — 404 absent run ── + it('row 4b — run.exists === false → 404, cancel NOT called', async () => { + standardHarness(); + mockAuthedSession(); + runExists = false; + mockGetRun(); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', 's1'); + + expect(res.status).toBe(404); + const body = (await res.json()) as { error: string }; + expect(body.error).toMatch(/Run not found/); + expect(runCancelSpy).not.toHaveBeenCalled(); + expect(overlayWorkerMetaMock).not.toHaveBeenCalled(); + }); + + // ── Row 5 — 409 terminal no-op ── + it('row 5 — terminal run (completed) → 409 with status in body, cancel NOT called, no overlay', async () => { + standardHarness(); + mockAuthedSession(); + runStatus = 'completed'; + mockGetRun(); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', 's1'); + + expect(res.status).toBe(409); + const body = (await res.json()) as { runId: string; status: string }; + expect(body.runId).toBe('wf_turn_123'); + expect(body.status).toBe('completed'); + expect(runCancelSpy).not.toHaveBeenCalled(); + expect(overlayWorkerMetaMock).not.toHaveBeenCalled(); + }); + + it('row 5b — terminal run (failed) → 409 no-op', async () => { + standardHarness(); + mockAuthedSession(); + runStatus = 'failed'; + mockGetRun(); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', 's1'); + expect(res.status).toBe(409); + expect(runCancelSpy).not.toHaveBeenCalled(); + }); + + it('row 5c — terminal run (cancelled) → 409 no-op', async () => { + standardHarness(); + mockAuthedSession(); + runStatus = 'cancelled'; + mockGetRun(); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', 's1'); + expect(res.status).toBe(409); + expect(runCancelSpy).not.toHaveBeenCalled(); + }); + + // ── Row 6 — live run → cancel + 'cancelling' overlay ── + it('row 6 — live run → 200, cancel() called exactly once, cancelling overlay PATCH with strictly-newer updatedAt', async () => { + standardHarness(); + mockAuthedSession(); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', 's1'); + + expect(res.status).toBe(200); + const body = (await res.json()) as { runId: string; turnStatus: string; warning?: string }; + expect(body.runId).toBe('wf_turn_123'); + expect(body.turnStatus).toBe('cancelling'); + expect(body.warning).toBeUndefined(); + expect(res.headers.get('x-workflow-run-id')).toBe('wf_turn_123'); + expect(res.headers.get('x-workflow-run-warning')).toBeNull(); + + expect(runCancelSpy).toHaveBeenCalledTimes(1); + + // 'cancelling' overlay PATCH — worker-owned key, copy-forward, LWW clock. + expect(overlayWorkerMetaMock).toHaveBeenCalledTimes(1); + const patchCall = overlayWorkerMetaMock.mock.calls[0][0]; + expect(patchCall.patch).toEqual({ turnStatus: 'cancelling' }); + expect(patchCall.envelopeStore).toBeTruthy(); + // LWW: Math.max(Date.now(), stored+1). stored+1 (9e12+1) > Date.now() + // (~1.76e12) → clock MUST pick stored+1. A bare Date.now() would fail. + expect(typeof patchCall.updatedAt).toBe('number'); + expect(patchCall.updatedAt).toBeGreaterThanOrEqual(FUTURE_UPDATED_AT + 1); + expect(patchCall.key).toEqual({ tenantId: 't1', userId: 'u1', sessionId: 's1' }); + }); + + it('row 6b — live run (pending) → cancel() called', async () => { + standardHarness(); + mockAuthedSession(); + runStatus = 'pending'; + mockGetRun(); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', 's1'); + expect(res.status).toBe(200); + expect(runCancelSpy).toHaveBeenCalledTimes(1); + }); + + // ── Row 7 — cancel throw → 503 fail-closed, NO overlay ── + it('row 7 — run.cancel() throws → 503 fail-closed, NO overlay write', async () => { + standardHarness(); + mockAuthedSession(); + runCancelSpy = vi.fn(async () => { + throw new Error('lost race: run already terminal'); + }); + mockGetRun(); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', 's1'); + + expect(res.status).toBe(503); + const body = (await res.json()) as { error: string }; + expect(body.error).toMatch(/fail closed/); + expect(runCancelSpy).toHaveBeenCalledTimes(1); + // The 'cancelling' marker must NOT persist on a failed cancel — never a + // partial cancel claim. + expect(overlayWorkerMetaMock).not.toHaveBeenCalled(); + }); + + it('row 7b — getRun status rejects (infra) → 503 fail-closed, cancel NOT called', async () => { + standardHarness(); + mockAuthedSession(); + getRunMock = vi.fn(() => ({ + runId: 'wf_turn_123', + exists: Promise.resolve(true), + status: Promise.reject(new Error('Workflows world unavailable')), + cancel: runCancelSpy, + })); + vi.doMock('workflow/api', () => ({ getRun: getRunMock, start: vi.fn() })); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', 's1'); + + expect(res.status).toBe(503); + const body = (await res.json()) as { error: string }; + expect(body.error).toMatch(/fail closed/); + expect(runCancelSpy).not.toHaveBeenCalled(); + expect(overlayWorkerMetaMock).not.toHaveBeenCalled(); + }); + + // ── Row 8 — store unavailable → 503 ── + it('row 8 — store resolve not-ok → 503 fail-closed, getRun NOT called', async () => { + mockSessionCaps(); + mockGetRun(); + mockAuthedSession(); + storeAvailable = false; + mockTenancyOk(); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', 's1'); + + expect(res.status).toBe(503); + const body = (await res.json()) as { error: string }; + expect(body.error).toMatch(/store unavailable/); + expect(getRunMock).not.toHaveBeenCalled(); + expect(runCancelSpy).not.toHaveBeenCalled(); + }); + + it('row 8b — store resolve throws → 503 fail-closed', async () => { + mockSessionCaps(); + mockGetRun(); + mockAuthedSession(); + storeResolveThrows = true; + mockTenancyOk(); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', 's1'); + + expect(res.status).toBe(503); + const body = (await res.json()) as { error: string }; + expect(body.error).toMatch(/store unavailable/); + expect(runCancelSpy).not.toHaveBeenCalled(); + }); + + it('row 8c — readEnvelope throws → 503 fail-closed', async () => { + mockSessionCaps(); + mockGetRun(); + mockAuthedSession(); + readEnvelopeThrows = true; + mockTenancyOk(); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', 's1'); + + expect(res.status).toBe(503); + const body = (await res.json()) as { error: string }; + expect(body.error).toMatch(/store unavailable/); + expect(runCancelSpy).not.toHaveBeenCalled(); + }); + + // ── Row 9 — 429 min-interval soft guard ── + it('row 9 — second accepted cancel inside window → 429 + Retry-After; window advances only on accepted cancel', async () => { + standardHarness(); + mockAuthedSession(); + ({ POST } = await import('./route')); + + // First cancel: accepted (live run), advances the window. + const res1 = await postCancel('wf_turn_123', 's1'); + expect(res1.status).toBe(200); + expect(runCancelSpy).toHaveBeenCalledTimes(1); + + // Second cancel immediately: 429 (window not elapsed). + const res2 = await postCancel('wf_turn_123', 's1'); + expect(res2.status).toBe(429); + const body2 = (await res2.json()) as { error: string }; + expect(body2.error).toMatch(/too many cancel requests/i); + expect(res2.headers.get('Retry-After')).toBe('1'); + // Still only one cancel() — the 429 never reached the run. + expect(runCancelSpy).toHaveBeenCalledTimes(1); + expect(overlayWorkerMetaMock).toHaveBeenCalledTimes(1); + }); + + it('row 9d — accepted cancel of wr_a does NOT 429 wr_b on the same session (adversarial-review #927 pass 8)', async () => { + standardHarness(); + mockAuthedSession(); + mockTenancyOk('s1', 'wf_a'); + ({ POST } = await import('./route')); + + const res1 = await postCancel('wf_a', 's1'); + expect(res1.status).toBe(200); + expect(runCancelSpy).toHaveBeenCalledTimes(1); + + // Next turn on the same session — Stop must not 429 from wr_a's window. + envelopeTurnRunId = 'wf_b'; + const res2 = await postCancel('wf_b', 's1'); + expect(res2.status).toBe(200); + expect(runCancelSpy).toHaveBeenCalledTimes(2); + }); + + it('row 9b — a terminal 409 does NOT burn the window (follow-up live cancel accepted)', async () => { + standardHarness(); + mockAuthedSession(); + // First: terminal run → 409 (no cancel, no window advance). + runStatus = 'completed'; + mockGetRun(); + ({ POST } = await import('./route')); + + const res1 = await postCancel('wf_turn_123', 's1'); + expect(res1.status).toBe(409); + expect(runCancelSpy).not.toHaveBeenCalled(); + + // Second: live run → 200 (the 409 never advanced the window). + runStatus = 'running'; + mockGetRun(); + const res2 = await postCancel('wf_turn_123', 's1'); + expect(res2.status).toBe(200); + expect(runCancelSpy).toHaveBeenCalledTimes(1); + }); + + it('row 9c — a 404 (ownership) does NOT burn the window', async () => { + mockSessionCaps(); + mockGetRun(); + mockAuthedSession(); + mockTenancyOk('s1', 'wf_other_run'); + ({ POST } = await import('./route')); + + const res1 = await postCancel('wf_turn_123', 's1'); + expect(res1.status).toBe(404); + + // Fix ownership → live cancel accepted (the 404 never advanced the window). + mockTenancyOk('s1', 'wf_turn_123'); + ({ POST } = await import('./route')); + const res2 = await postCancel('wf_turn_123', 's1'); + expect(res2.status).toBe(200); + expect(runCancelSpy).toHaveBeenCalledTimes(1); + }); + + // ── Row 10 — accepted cancel with overlay {ok:false} → still 200 + warning ── + it('row 10 — accepted cancel + overlay {ok:false} → 200 with warning (PATCH failure non-fatal)', async () => { + standardHarness(); + mockAuthedSession(); + overlayWorkerMetaMock = vi.fn(async () => ({ + ok: false as const, + code: 'lww_conflict', + error: 'worker PATCH updatedAt <= stored envelope updatedAt.', + })); + mockTenancyOk(); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', 's1'); + + // Still 200 — the cancel WAS accepted; the marker PATCH is best-effort. + expect(res.status).toBe(200); + const body = (await res.json()) as { runId: string; turnStatus: string; warning?: string }; + expect(body.runId).toBe('wf_turn_123'); + expect(body.turnStatus).toBe('cancelling'); + expect(body.warning).toBe('Cancelling PATCH did not persist (lww_conflict)'); + expect(res.headers.get('x-workflow-run-warning')).toBe( + 'Cancelling PATCH did not persist (lww_conflict)', + ); + expect(runCancelSpy).toHaveBeenCalledTimes(1); + }); + + it('row 10b — accepted cancel + overlay throws → 200 with warning', async () => { + standardHarness(); + mockAuthedSession(); + overlayWorkerMetaMock = vi.fn(async () => { + throw new Error('Redis write timeout.'); + }); + mockTenancyOk(); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', 's1'); + + expect(res.status).toBe(200); + const body = (await res.json()) as { warning?: string }; + expect(body.warning).toBe('Cancelling PATCH failed to persist'); + expect(res.headers.get('x-workflow-run-warning')).toBe( + 'Cancelling PATCH failed to persist', + ); + expect(runCancelSpy).toHaveBeenCalledTimes(1); + }); + + // ── Row 11 — tenant resolve failure → 503 ── + it('row 11 — tenant resolve fails → 503, getRun NOT called', async () => { + mockSessionCaps(); + mockGetRun(); + mockAuthedSession(); + tenantResolveOk = false; + mockTenancyOk(); + ({ POST } = await import('./route')); + + const res = await postCancel('wf_turn_123', 's1'); + + expect(res.status).toBe(503); + const body = (await res.json()) as { error: string }; + expect(body.error).toMatch(/Unable to resolve tenant/); + expect(getRunMock).not.toHaveBeenCalled(); + expect(runCancelSpy).not.toHaveBeenCalled(); + }); +}); diff --git a/app/api/turns/[runId]/cancel/route.ts b/app/api/turns/[runId]/cancel/route.ts new file mode 100644 index 00000000..fc2fae48 --- /dev/null +++ b/app/api/turns/[runId]/cancel/route.ts @@ -0,0 +1,285 @@ +/** + * backend-agents G22 (#816) — `POST /api/turns/:runId/cancel`: the durable-turn + * **server cancel** seam. Stop/Esc on an attached durable run cancels the + * Workflow run route-side via `getRun(runId).cancel()` — this route never + * enters the workflow, never adds a step, and never passes a signal/closure + * into `start()`. The loop's existing `'cancelled'` status path (abort-signal + * → `executeTool` → `'cancelled'` value → `fail('cancelled', …)` → terminal + * persist + writable close) is the in-loop half and is unchanged. + * + * Mirrors `[runId]/stream` gate-for-gate: + * 1. Auth (`requireSessionUser`) → 401 (middleware matcher + in-route dual). + * 2. `sessionId` query REQUIRED + `isRedisSafeOpaqueId` → 400. + * 3. Envelope read; `meta.turnRunId === runId` else **404** (no existence + * leak across sessions; never 403). Store unavailable / read throw → 503 + * FAIL-CLOSED (the tenancy gate is the only owner check). + * 4. `getRun(cleanRunId)` — absent → 404; `run.status` terminal + * (`completed|failed|cancelled`) → **409** with the terminal status in the + * body (idempotent no-op: the turn already ended; `run.cancel` NOT called). + * 5. Live (`pending|running`) → `run.cancel()`. Cancel throw → fail-closed + * 503 (never a partial cancel claim; the lost-race terminalizes-between- + * status-read-and-cancel case lands here and the host re-resolves via the + * attach/terminal event — never a false "cancelled"). + * 6. Accepted cancel → `overlayWorkerMeta` PATCH `turnStatus: 'cancelling'` + * (worker-owned key, copy-then-override, `updatedAt` strictly newer — the + * same LWW contract as the start route's running PATCH). `turnRunId` rides + * unchanged in the envelope per the B8 copy-forward contract (the PATCH + * only overrides `turnStatus`). The run's own terminal persist then owns + * the terminal status (`persistOverlayStatus` unchanged — a cancelled run + * still persists `'completed'`; `'cancelling'` is a host-held liveness + * state only, always superseded). + * + * C15 interplay: the start route's live-only 409 requires BOTH + * `turnStatus ∈ {running, cancelling}` AND a non-terminal `getRun` status, so + * a stale `'cancelling'` with a terminal/absent run never blocks the next + * prompt. + * + * Per-run min-interval soft guard (`TURN_CANCEL_MIN_INTERVAL_MS`, NEW cap, + * plan #816 Caps table): same Map+boundedSet shape as C15's start guard — + * per-process, zero-I/O, keyed by `sessionId:runId` so an accepted cancel of + * wr_1 cannot 429 Stop on wr_2 (adversarial-review #927 pass 8). The window + * advances ONLY on an **accepted** cancel so a terminal 409 / ownership 404 / + * 503 never burns it. Bounds `getRun`+PATCH write amplification from a hostile + * repeat-Stop client on the **same** run. + * + * No body. The route never reads or writes the transcript, checkpoint, or + * queue (Wasm FIFO + F21 mirror untouched). + */ +import { getRun } from 'workflow/api'; +import { overlayWorkerMeta } from '../../../../../lib/agent/workerMetaOverlay'; +import { createProdServices } from '../../../../../lib/di'; +import { + isRedisSafeOpaqueId, + sanitizeTurnRunId, + TURN_CANCEL_MIN_INTERVAL_MS, +} from '../../../../../lib/sessionCloudCaps'; +import { isEnvelopeStore } from '../../../../../lib/sessions/sessionStore'; +import { + resolveSessionStore, + sessionKeyFor, +} from '../../../../../lib/tenancy/harnessSessionsRedis'; +import { requireSessionUser } from '../../../../../lib/tenancy/session'; + +export const runtime = 'nodejs'; +export const maxDuration = 1800; + +/** Composition root — all wiring constructed here, never in route body. */ +const services = createProdServices(); + +/** + * G22 per-process soft abuse guard — per-run (`sessionId:runId`), NOT global. + * Same Map+boundedSet shape as the C15 start guard on `app/api/turns/route.ts`. + * Key includes `runId` so an accepted cancel of wr_1 cannot 429 Stop on wr_2 + * (adversarial-review #927 pass 8). Same-run repeat-Stop still 429s. The window + * advances ONLY on an accepted cancel — a terminal 409 / ownership 404 / + * store-or-cancel 503 never burns it. + */ +const lastCancelAtMs = new Map(); +const TURN_CANCEL_CACHE_MAX = 256; + +function boundedSet(m: Map, key: string, value: T): Map { + m.set(key, value); + if (m.size > TURN_CANCEL_CACHE_MAX) { + const oldest = m.keys().next().value; + if (oldest !== undefined) m.delete(oldest); + } + return m; +} + +/** Workflow run statuses after which a cancel is an idempotent no-op. */ +const TERMINAL_RUN_STATUSES = new Set(['completed', 'failed', 'cancelled']); + +/** + * POST /api/turns/:runId/cancel?sessionId=... + * + * Server-cancel one live durable run. Response: + * - 200 `{ runId, turnStatus: 'cancelling' }` on an accepted cancel + * - 400 invalid runId / missing-or-invalid sessionId + * - 401 auth failure + * - 404 run not found OR ownership mismatch (tenancy guard) + * - 409 `{ runId, status }` when the run is already terminal (idempotent no-op) + * - 429 repeat-Stop soft guard (window advances only on accepted cancel) + * - 503 fail-closed for tenant resolve / store unavailable / cancel throw + */ +export async function POST( + req: Request, + { params }: { params: Promise<{ runId: string }> }, +): Promise { + // Auth gate — same requireSessionUser as the stream route (dual gate with + // the middleware matcher). + const sessionGate = await requireSessionUser(); + if (!sessionGate.ok) return sessionGate.response; + const userId = sessionGate.user?.id; + if (!userId) { + const { AUTH_REQUIRED_ERROR } = await import('../../../../../lib/tenancy/errors'); + return Response.json({ error: AUTH_REQUIRED_ERROR }, { status: 401 }); + } + + const { runId } = await params; + + // Validate runId against TURN_RUN_ID_MAX (A1 #795) — a bad URL param is a + // client error (400), same as the stream route. + const cleanRunId = sanitizeTurnRunId(runId); + if (cleanRunId === undefined) { + return Response.json({ error: 'Invalid runId' }, { status: 400 }); + } + + // Parse and sanitize sessionId — REQUIRED for tenancy-bound ownership + // verification (same contract as the stream route). + const rawSessionId = new URL(req.url).searchParams.get('sessionId'); + if (!rawSessionId) { + return Response.json( + { error: 'sessionId query parameter is required.' }, + { status: 400 }, + ); + } + if (!isRedisSafeOpaqueId(rawSessionId)) { + return Response.json( + { error: 'Invalid sessionId.' }, + { status: 400 }, + ); + } + const sessionId = rawSessionId; // type-narrowed by isRedisSafeOpaqueId + + // G22 429 min-interval guard — per-run soft abuse gate, zero I/O, + // BEFORE any gate that may await (tenant resolve, envelope read, getRun). + // Keyed by sessionId:runId so an accepted cancel of wr_1 cannot 429 Stop + // on wr_2 (adversarial-review #927 pass 8). Advances only on an accepted + // cancel below. + const now = Date.now(); + const cancelWindowKey = `${sessionId}:${cleanRunId}`; + const last = lastCancelAtMs.get(cancelWindowKey); + if (last != null && now - last < TURN_CANCEL_MIN_INTERVAL_MS) { + return Response.json( + { + error: + 'Too many cancel requests. Please wait before cancelling again.', + }, + { + status: 429, + headers: { + 'Retry-After': String(Math.ceil(TURN_CANCEL_MIN_INTERVAL_MS / 1000)), + }, + }, + ); + } + + // Tenancy check — resolve tenant, read the session envelope, verify + // envelope.meta.turnRunId matches the requested runId. FAIL-CLOSED (503) + // when the store is unavailable or the read throws; mismatch/miss → 404 + // (never 403 — no existence leak across sessions). + const tenantRes = + await services.harnessSessionsRedis.resolveTenantIdForUser(userId); + if (!tenantRes.ok) { + return Response.json( + { error: 'Unable to resolve tenant for run cancel.' }, + { status: 503 }, + ); + } + const sessionKey = sessionKeyFor(tenantRes.value, userId, sessionId); + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + let envelopeStore: any; + try { + const storeRes = await resolveSessionStore(); + if (!storeRes.ok || !isEnvelopeStore(storeRes.value)) { + return Response.json( + { error: 'Unable to cancel run (store unavailable).' }, + { status: 503 }, + ); + } + envelopeStore = storeRes.value; + } catch { + return Response.json( + { error: 'Unable to cancel run (store unavailable).' }, + { status: 503 }, + ); + } + + let storedUpdatedAt = 0; + try { + const envelope = await envelopeStore.readEnvelope(sessionKey); + if (!envelope || envelope.meta?.turnRunId !== cleanRunId) { + return Response.json( + { error: `Run not found: ${cleanRunId}` }, + { status: 404 }, + ); + } + storedUpdatedAt = + typeof envelope.updatedAt === 'number' ? envelope.updatedAt : 0; + } catch { + return Response.json( + { error: 'Unable to cancel run (store unavailable).' }, + { status: 503 }, + ); + } + + // Status truth gate — `getRun` absent → 404; terminal → 409 idempotent + // no-op (run.cancel NOT called); live → cancel. Infra throw → 503 + // fail-closed (never a partial cancel claim). + try { + const run = getRun(cleanRunId); + if (!(await run.exists)) { + return Response.json( + { error: `Run not found: ${cleanRunId}` }, + { status: 404 }, + ); + } + const status = await run.status; + if (TERMINAL_RUN_STATUSES.has(status)) { + return Response.json( + { runId: cleanRunId, status }, + { status: 409 }, + ); + } + await run.cancel(); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + return Response.json( + { error: `Unable to cancel run (fail closed): ${msg}` }, + { status: 503 }, + ); + } + + // Accepted cancel — advance the per-run soft-guard window (only an + // accepted cancel burns it), then persist the host-held liveness marker. + boundedSet(lastCancelAtMs, cancelWindowKey, Date.now()); + + // G22 host-held `'cancelling'` marker: worker-owned `turnStatus` PATCH via + // the B8 overlay seam. `turnRunId` rides unchanged (copy-forward). The + // strictly-newer clock mirrors the start route's running PATCH so a host + // PUT in the same millisecond can never self-conflict. A PATCH failure is + // non-fatal: the run's own terminal persist owns the terminal status, and + // C15's live-only 409 requires a non-terminal `getRun` status too, so a + // stale marker can never block the next prompt. + let cancelWarning: string | null = null; + try { + const cancellingRes = await overlayWorkerMeta({ + envelopeStore, + key: sessionKey, + patch: { turnStatus: 'cancelling' as const }, + updatedAt: Math.max(Date.now(), storedUpdatedAt + 1), + }); + if (!cancellingRes.ok) { + // Stable warning — code only, never the raw error (can carry Redis + // host/port/connect strings via overlayWorkerMeta's toMessage paths). + cancelWarning = `Cancelling PATCH did not persist (${cancellingRes.code})`; + } + } catch { + // Stable warning — never interpolate err.message (can carry Redis details). + cancelWarning = 'Cancelling PATCH failed to persist'; + } + + const headers: Record = { + 'x-workflow-run-id': cleanRunId, + }; + if (cancelWarning) { + headers['x-workflow-run-warning'] = cancelWarning; + } + const body: { runId: string; turnStatus: 'cancelling'; warning?: string } = { + runId: cleanRunId, + turnStatus: 'cancelling', + }; + if (cancelWarning) body.warning = cancelWarning; + return Response.json(body, { headers }); +} diff --git a/app/harness/HarnessHost.tsx b/app/harness/HarnessHost.tsx index b0bad1c1..0a960ab6 100644 --- a/app/harness/HarnessHost.tsx +++ b/app/harness/HarnessHost.tsx @@ -13,7 +13,30 @@ import { } from '../../lib/harnessChat'; import { resetHarnessImageSession } from '../../lib/harnessImages'; import { resetHarnessMathSession } from '../../lib/harnessMath'; -import { decideDetach, shouldAbortReader, abortReasonFor, decideDetachPersist, putPreservedTurn, shouldApplyMintBind, shouldSetHostTurnNote, isDetachAbort, releaseBusyViewport } from '../../lib/detachTurn'; +import { + decideDetach, + shouldAbortReader, + abortReasonFor, + decideDetachPersist, + putPreservedTurn, + shouldApplyMintBind, + shouldSetHostTurnNote, + isDetachAbort, + releaseBusyViewport, + decideStopFoldPre, + decideStopFoldPost, + shouldSkipCancelPost, + applyStopFoldToSession, + decideCancelAckApply, + shouldKickCancelRetryAttach, + shouldAbortReaderOnCancelAck, + abortReasonForCancelAck, + decideCancelRetryKickWhen, + CANCEL_RETRY_NOTE, + CANCEL_FAILED_NOTE, + type StopFoldAction, +} from '../../lib/detachTurn'; +import { cancelTurn } from '../../lib/turnApi'; import { decideHotResume, decideSendAttach, shouldPaintAttachFollowUpNote, shouldPaintAttachFollowUpDetachNote, shouldRepostAttachFollowUp, shouldSkipAttachHotResume, shouldKickHotResume, ATTACH_FOLLOW_UP_NOTE, ATTACH_FOLLOW_UP_DETACH_NOTE, isAttachFollowUpHostNote, coldAttachFromSnapshot, type HeapApplied } from '../../lib/turnAttach'; import { HarnessBridge, @@ -248,6 +271,44 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) { const repoRef = useRef(null); const pollRef = useRef(null); const abortRef = useRef(null); + /** + * Plan #816 (G22) — run ids we have already POSTed cancel for (in-flight or + * accepted). Failed ack removes the id so Stop can retry. `shouldSkipCancelPost` + * is a named always-false seam (posted-id set is the once-per-run skip). + */ + const cancelPostedRunIdsRef = useRef(new Set()); + /** + * Latest Stop-fold action for a run id, set only **after** cancelTurn + * returns. `persistTurn` applies this so an abort-fold cannot beat a failed + * ack. + */ + const pendingStopFoldRef = useRef<{ runId: string; fold: StopFoldAction } | null>( + null, + ); + /** + * Item 5: failed-ack kick armed while the Stop-aborted `runPrompt` is still + * in try/finally. `finally` (same generation) performs `kickColdAttach`. + */ + const pendingCancelRetryAttachRef = useRef<{ + sessionId: string; + runId: string; + } | null>(null); + /** + * Item 5: generation of the in-flight `runPrompt`. `finally` only clears + * Busy/inflight when this still matches. 0 = no prompt in try/finally. + */ + const promptGenerationRef = useRef(0); + const activePromptGenerationRef = useRef(0); + /** + * Adversarial-review #927 pass 5 — last persistTurn snapshot that carried a + * G22 fold. `persist-detached` folds onto this (abort snapshot with + * Turn-ended) instead of the Stop-fire capture (pre-abort, stale messages). + */ + const lastStopPersistRef = useRef<{ + sessionId: string; + runId: string; + snapshot: SessionSnapshot; + } | null>(null); const inflightRef = useRef(false); /** * Plan #887 — session-scoped one-shot auto-continue flag. Clears on the next @@ -503,7 +564,7 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) { /** * Plan #813 — cold attach after the ring was rebuilt from Blob/local * (boot / adopt / switch-back). Always `startIndex=0` + dedup. No-ops when - * not `running` or a turn is already inflight. + * not live (`running`/`cancelling`) or a turn is already inflight. */ const kickColdAttach = useCallback(() => { if (inflightRef.current) return; @@ -539,7 +600,8 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) { * Clear): close THIS reader only, never classify the turn as stopped. * Durable detach aborts with DETACH_ABORT_REASON so classifyTurnFailure * returns `'detach'` (not `'stop'`) and the fail fold keeps turnRunId/running. - * Stop/Esc is NEVER routed here — the poll's takePendingCancel stays a raw abort. + * Stop/Esc is NEVER routed here — the poll POSTs server cancel and aborts + * only after ack (legacy `/api/agent` still raw-aborts). */ const detachTurn = useCallback(() => { const s = sessionRef.current; @@ -573,6 +635,16 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) { async (prompt: string, opts?: RunPromptOpts) => { const bridge = bridgeRef.current; if (!bridge || inflightRef.current) return; + // Adversarial-review #927: a leftover Stop fold from the previous + // turn must not ride persistTurn of this prompt (pre-headers snapshot + // may still carry the old id, or have cleared it). Null only THIS + // session's abort snapshot — a destination runPrompt (switch + Send / + // kickColdAttach) must not drop the abandoned session's slot + // (pass 6: persist-detached would fall back to Stop-fire cancelSnapshot). + pendingStopFoldRef.current = null; + if (lastStopPersistRef.current?.sessionId === sessionRef.current.id) { + lastStopPersistRef.current = null; + } // Plan #887: next operator submit (not attach, not auto-continue) clears // the one-shot flag so a later recoverable can fire again. @@ -636,27 +708,52 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) { const controller = new AbortController(); abortRef.current = controller; inflightRef.current = true; + const myGeneration = ++promptGenerationRef.current; + activePromptGenerationRef.current = myGeneration; const epoch = turnEpochRef.current; const startedId = sessionRef.current.id; // Adversarial #844: capture the repo object NOW. Unmount cleanup nulls // `repoRef` before the abort microtask reaches persistTurn/finally. const repo = repoRef.current; - const persistTurn = (snapshot: SessionSnapshot, paintQuota = true) => { + const persistTurn = (snapshot: SessionSnapshot, paintQuota = true): SessionSnapshot => { + const pendingFold = pendingStopFoldRef.current; + const foldedSnapshot = + pendingFold != null + ? applyStopFoldToSession(snapshot, pendingFold.runId, pendingFold.fold) + : snapshot; + const persistRunId = pendingFold?.runId ?? foldedSnapshot.turnRunId; + if ( + persistRunId !== undefined && + (pendingFold != null || + (lastStopPersistRef.current != null && + lastStopPersistRef.current.sessionId === foldedSnapshot.id && + lastStopPersistRef.current.runId === persistRunId)) + ) { + // pendingFold set: record the abort snapshot. pendingFold null: + // refresh the existing slot so a destination runPrompt that wiped + // pendingStopFoldRef cannot freeze a pre-abort onSessionPatch + // (adversarial-review #927 pass 6). + lastStopPersistRef.current = { + sessionId: foldedSnapshot.id, + runId: persistRunId, + snapshot: foldedSnapshot, + }; + } const pendingMintId = pendingMintBindRef.current; const action = decideDetachPersist({ detached: turnEpochRef.current !== epoch, discarded: discardedSessionIdsRef.current.has(startedId) || - discardedSessionIdsRef.current.has(snapshot.id) || + discardedSessionIdsRef.current.has(foldedSnapshot.id) || (pendingMintId != null && discardedSessionIdsRef.current.has(pendingMintId)), - turnRunId: snapshot.turnRunId, - turnStatus: snapshot.turnStatus, + turnRunId: foldedSnapshot.turnRunId, + turnStatus: foldedSnapshot.turnStatus, }); - if (action === 'drop') return; + if (action === 'drop') return foldedSnapshot; if (action === 'preserve') { // Adversarial #844: first-turn unmount must PUT the deferred mint UUID, // not local sess_*. Switch must not writeLocal (generation token). - const { preserved } = putPreservedTurn(repo, snapshot, startedId, pendingMintId); + const { preserved } = putPreservedTurn(repo, foldedSnapshot, startedId, pendingMintId); if ( shouldApplyMintBind({ sessionId: sessionRef.current.id, @@ -667,9 +764,10 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) { ) { writeLocalSession(preserved); } - return; + return preserved; } - persist(snapshot, { paintQuota }); + persist(foldedSnapshot, { paintQuota }); + return foldedSnapshot; }; setBusy(true); setHostNote(null); @@ -781,7 +879,7 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) { // Always persist — including user Stop/cancel (and late abort after a finished // stream). Dropping session on signal.aborted left SessionStore behind Wasm: // Load earlier / refresh could wipe the cancelled turn from the ring. - persistTurn(folded, folded.turnStatus !== 'running'); + const persisted = persistTurn(folded, folded.turnStatus !== 'running'); const operatorStop = shouldSkipAttachHotResume({ attaching, aborted: controller.signal.aborted, @@ -791,15 +889,15 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) { attaching, streamOpened, operatorStop, - turnStatus: folded.turnStatus, - turnRunId: folded.turnRunId, + turnStatus: persisted.turnStatus, + turnRunId: persisted.turnRunId, }); - if (kickHot && folded.turnRunId) { + if (kickHot && persisted.turnRunId) { heapAppliedRef.current = { - runId: folded.turnRunId, - count: folded.turnStreamCursor ?? 0, + runId: persisted.turnRunId, + count: persisted.turnStreamCursor ?? 0, }; - } else if (folded.turnStatus !== 'running') { + } else if (persisted.turnStatus !== 'running') { heapAppliedRef.current = null; } @@ -808,13 +906,14 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) { // no longer applies. Wasm follow-up was stripped; pushUser paints it. const repostFollowUp = shouldRepostAttachFollowUp({ sendWhileRunning, - turnStatus: folded.turnStatus, + turnStatus: persisted.turnStatus, + operatorStop, }); if (repostFollowUp) { queueMicrotask(() => { void runPromptRef.current(prompt, { pushUser: true }); }); - } else if (!result.ok && shouldSetHostTurnNote(folded.turnStatus)) { + } else if (!result.ok && shouldSetHostTurnNote(persisted.turnStatus)) { setHostNote(result.error); } else if ( shouldPaintAttachFollowUpNote({ @@ -847,8 +946,8 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) { // Plan #813: SSE drop while still mounted → hot resume at this-heap C. // Empty-EOF GET (applied == startIndex) must not reconnect (spin). // F5 is never this path (heapApplied was nulled; activateSession is cold). - // Operator Stop during attach: skip auto-resume this tick (D18 reader - // close, not G22 cancel — adversarial #857). + // Operator Stop during attach: skip auto-resume this tick (G22 cancel + // already fired; shouldKickHotResume is false via operatorStop). if (kickHot) { const resume = decideHotResume({ turnRunId: folded.turnRunId, @@ -928,12 +1027,27 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) { } } if (!detached) { - inflightRef.current = false; - setBusy(false); + if (promptGenerationRef.current === myGeneration) { + inflightRef.current = false; + setBusy(false); + } + } + if (activePromptGenerationRef.current === myGeneration) { + activePromptGenerationRef.current = 0; + } + const pendingRetry = pendingCancelRetryAttachRef.current; + if ( + pendingRetry != null && + !detached && + sessionRef.current.id === pendingRetry.sessionId && + promptGenerationRef.current === myGeneration + ) { + pendingCancelRetryAttachRef.current = null; + queueMicrotask(kickColdAttach); } } }, - [persist, setUrlSessionId, writeLocalSession], + [persist, setUrlSessionId, writeLocalSession, kickColdAttach], ); runPromptRef.current = runPrompt; @@ -1166,9 +1280,9 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) { // or refresh can recover — don't permanently strand local-only this page load. if (result.kind === 'local') setCloudEnabled(r.enabled); void refreshSessions(); - // Plan #813: after Blob/local hydrate, cold-attach a still-running - // turn. activateSession also kicks; inflightRef de-dupes the pair. - // Do not auto-attach completed sessions (`turnStatus !== 'running'`). + // Plan #813: after Blob/local hydrate, cold-attach a still-live + // turn (`running` or `cancelling`). activateSession also kicks; + // inflightRef de-dupes the pair. Do not auto-attach completed. if (!cancelled) { queueMicrotask(kickColdAttach); } @@ -1178,15 +1292,152 @@ export default function HarnessHost({ authNav }: { authNav?: ReactNode } = {}) { if (cancelled) return; const b = bridgeRef.current; if (b) { - // Protocol v9: Stop first — abort inflight and skip starting a turn this tick. + // Protocol v9: Stop first — consume pending cancel this tick + // (durable: POST cancel, abort only after ack; legacy: abort now). if (b.takePendingCancel()) { - abortRef.current?.abort(); - releaseBusyViewport({ - inflightRef, - setBusy, - setQueuePromoteAllowed: (allowed) => b.setQueuePromoteAllowed(allowed), - setLifecycleReady: () => b.setLifecycle(Lifecycle.Ready), + // Plan #816 (G22): Stop on an attached durable run fires the + // server cancel ONCE per run id. Do not abort, release Busy, or + // persist `'cancelling'` until `cancelTurn` returns (punch-list + // items 1–2). keepalive so Stop then F5 does not drop the POST. + // Failed ack keeps `running` so Stop can retry. Terminal (409) / + // gone (404) clears the id + folds `completed` (orphan-unstick) + // and must not paint "you stopped" over a finished run. + const stopSession = sessionRef.current; + const stopRunId = stopSession.turnRunId; + const stopFoldPre = decideStopFoldPre({ + turnRunId: stopRunId, + turnStatus: stopSession.turnStatus, }); + if ( + stopFoldPre.kind === 'cancelling' && + stopRunId !== undefined && + !cancelPostedRunIdsRef.current.has(stopRunId) && + !shouldSkipCancelPost({ + turnRunId: stopRunId, + turnStatus: stopSession.turnStatus, + }) + ) { + const cancelSessionId = stopSession.id; + // Capture repo + snapshot at Stop fire: unmount cleanup nulls + // `repoRef`, and switch replaces `sessionRef`. Failed ack must + // PUT keep-running onto THIS session (adversarial-review #927 + // pass 4) — never onto liveNow. + const cancelRepo = repoRef.current; + const cancelSnapshot = stopSession; + const stopGeneration = activePromptGenerationRef.current; + cancelPostedRunIdsRef.current.add(stopRunId); + void cancelTurn(stopRunId, { + sessionId: cancelSessionId, + keepalive: true, + }).then( + (outcome) => { + const liveNow = sessionRef.current; + const fold = decideStopFoldPost({ + pre: stopFoldPre, + outcome, + }); + const activeGen = activePromptGenerationRef.current; + // `inflightRef` is the original turn until we abort — not a + // newer prompt. Generation mismatch is the real "new prompt + // owns the tab" signal (item 5; do not reuse inflightRef). + const newerPrompt = + activeGen !== 0 && activeGen !== stopGeneration; + const originalPromptActive = + stopGeneration !== 0 && activeGen === stopGeneration; + const apply = decideCancelAckApply({ + unmounted: cancelled, + liveSessionId: liveNow.id, + cancelSessionId, + liveTurnRunId: liveNow.turnRunId, + stopRunId, + discarded: discardedSessionIdsRef.current.has(cancelSessionId), + inflight: newerPrompt, + fold, + }); + if (apply.dropPostedId) { + cancelPostedRunIdsRef.current.delete(stopRunId); + } + if (apply.commit === 'drop') return; + if (apply.commit === 'persist-detached') { + const preserved = lastStopPersistRef.current; + const base = + preserved != null && + preserved.sessionId === cancelSessionId && + preserved.runId === stopRunId + ? preserved.snapshot + : cancelSnapshot; + const folded = applyStopFoldToSession( + base, + stopRunId, + apply.fold, + ); + const detached = { ...folded, id: cancelSessionId }; + cancelRepo?.put(cancelSessionId, detached); + pendingStopFoldRef.current = { runId: stopRunId, fold: apply.fold }; + if (sessionRef.current.id === cancelSessionId) { + persist(detached); + } + } else { + pendingStopFoldRef.current = { runId: stopRunId, fold: apply.fold }; + if (apply.commit === 'persist') { + persist(applyStopFoldToSession(liveNow, stopRunId, apply.fold)); + } + } + if ( + shouldAbortReaderOnCancelAck({ + fold: apply.fold, + commit: apply.commit, + }) + ) { + abortRef.current?.abort(abortReasonForCancelAck(apply.fold)); + releaseBusyViewport({ + inflightRef, + setBusy, + setQueuePromoteAllowed: (allowed) => + b.setQueuePromoteAllowed(allowed), + setLifecycleReady: () => b.setLifecycle(Lifecycle.Ready), + }); + } + const kickWhen = decideCancelRetryKickWhen({ + shouldKick: shouldKickCancelRetryAttach({ + fold: apply.fold, + commit: apply.commit, + unmounted: cancelled, + liveSessionId: sessionRef.current.id, + cancelSessionId, + inflight: originalPromptActive || newerPrompt, + }), + promptActive: originalPromptActive, + }); + if (kickWhen !== 'none') { + setHostNote(CANCEL_RETRY_NOTE); + if (kickWhen === 'pending') { + pendingCancelRetryAttachRef.current = { + sessionId: cancelSessionId, + runId: stopRunId, + }; + } else { + pendingCancelRetryAttachRef.current = null; + queueMicrotask(kickColdAttach); + } + } else if ( + apply.fold.kind === 'keep-running' && + !cancelled && + sessionRef.current.id === cancelSessionId + ) { + setHostNote(CANCEL_FAILED_NOTE); + } + }, + ); + } else if (stopFoldPre.kind === 'legacy-clear') { + abortRef.current?.abort(); + releaseBusyViewport({ + inflightRef, + setBusy, + setQueuePromoteAllowed: (allowed) => b.setQueuePromoteAllowed(allowed), + setLifecycleReady: () => b.setLifecycle(Lifecycle.Ready), + }); + } } else if (inflightRef.current || switchInFlightRef.current) { foldPendingSessionSwitch(true, () => b.takePendingSessionSwitch(), () => {}); } else { diff --git a/docs/feature-divide.md b/docs/feature-divide.md index 50b0679d..25c58ffe 100644 --- a/docs/feature-divide.md +++ b/docs/feature-divide.md @@ -46,9 +46,9 @@ optional login chrome). | Image bytes (fetch/decode) | **DOM host** | Browser fetch → RGBA → `inv_image_cache_put`; paint stays Wasm | | Math pixels (TeX raster) | **DOM host** | Host MathJax SVG → RGBA → `inv_math_cache_put`; paint stays Wasm | | **Composer + Send** | **Wasm** | Primary input; dynamic absolute-rect from previous-frame measured height: idle hugs one line (~44 px), grows up to cap (124 px), scrolls internally past 120 px content; glyphs inset 5 px from the field border; Send/Stop icon bottom-pinned (`gravity_y = 1.0`) stays on field baseline at all heights (plan #579) | -| **Stop / cancel turn** | **Wasm** control + **DOM** abort | Canvas **Stop** (icon-only ■, plan #457) while busy → pending cancel (protocol v9); host aborts `AbortController` **and** clears Busy / Ready / clocks this tick (`releaseBusyViewport`). Server cancel of the Workflow run is a separate seam (not this host abort). | +| **Stop / cancel turn** | **Wasm** control + **DOM** host | Canvas **Stop** (icon-only ■, plan #457) while busy → pending cancel (protocol v9). On an attached durable run the host POSTs a **server cancel** (`POST /api/turns/:runId/cancel` → `getRun(runId).cancel()`) **before** aborting the reader or clearing Busy. Accepted / terminal / gone: fold (`'cancelling'` keeping `turnRunId`, or clear+`completed` on 409/404) then abort + `releaseBusyViewport`. Failed cancel keeps `running` and Busy so Stop can retry. Legacy `/api/agent` (no run id) still aborts this tick. Unmount / switch / New / Clear / logout only ever **detach** (close this reader, `DETACH_ABORT_REASON`) — they never cancel the run. The run's own terminal persist owns the terminal status. | | Busy / error presentation for turns | **Wasm** | EMBER for errors | -| Whole-turn `mm:ss` clock (Busy) | **Wasm** (busy row) fed by the **DOM** host | The host owns the only reliable wall-clock (no WASI clock in Wasm) and ticks it ~1 Hz, pushing the elapsed seconds into the Wasm busy row via protocol **v14** `inv_set_turn_elapsed` (plan #567). The canvas appends `Waiting for model… · 0:42` in-canvas while a turn runs; reset to 0 when host `busy` clears (Stop this tick, terminal stream, or error) so no `0:00` lingers. Composer/Stop stay **Wasm** | +| Whole-turn `mm:ss` clock (Busy) | **Wasm** (busy row) fed by the **DOM** host | The host owns the only reliable wall-clock (no WASI clock in Wasm) and ticks it ~1 Hz, pushing the elapsed seconds into the Wasm busy row via protocol **v14** `inv_set_turn_elapsed` (plan #567). The canvas appends `Waiting for model… · 0:42` in-canvas while a turn runs; reset to 0 when host `busy` clears (accepted Stop ack, terminal stream, or error) so no `0:00` lingers. Composer/Stop stay **Wasm** | | 2×4 busy spinner (plan #574) | **Wasm paint** fed by the **DOM** host | **Wasm** paints a 2×4 WARM rectangle grid left of `Waiting for model…` (clockwise pulse; pure LUT `busy_spinner.zig`, zero I/O/alloc in the frame path). **DOM host** drives the pulse phase on the same Busy ticker at **`HARNESS_BUSY_TICK_HZ` = 10 Hz** (`HarnessBridge.setBusyTick` → additive `inv_set_busy_tick`; the v14 `mm:ss` clock is fed every 10th tick ≈ 1 Hz). **Reduced motion** (read fresh at each busy start): the host skips only the per-tick pulse push, grid static at phase 0 — the `mm:ss` **clock keeps ticking** (no reduced-motion clock regression). Idle/Stop/error clears both to 0. Old host + new Wasm degrades to a static grid (busy_tick stays 0) | | Keyboard (keymap table, leader, help overlay) | **Wasm** | All chords in `native/harness/src/keymap.zig` + one dispatcher `ui/keymap_dispatch.zig`; in-canvas help panel `ui/help_overlay.zig`. **DOM adds no shortcut UI / React cheatsheet / `window` keydown** | | Empty / onboarding copy for agent | **Wasm** | | diff --git a/lib/detachTurn.test.ts b/lib/detachTurn.test.ts index cfddb23c..f64e756f 100644 --- a/lib/detachTurn.test.ts +++ b/lib/detachTurn.test.ts @@ -23,16 +23,29 @@ import { resolve } from 'node:path'; import { describe, expect, it, vi } from 'vitest'; import { abortReasonFor, + abortReasonForCancelAck, + applyStopFoldToSession, + CANCEL_FAILED_NOTE, + CANCEL_RETRY_NOTE, + decideCancelAckApply, + decideCancelRetryKickWhen, decideDetach, decideDetachPersist, + decideStopFoldPost, + decideStopFoldPre, DETACH_ABORT_REASON, + G22_ACCEPTED_ABORT_REASON, isDetachAbort, + isG22AcceptedAbort, preserveTargetId, putPreservedTurn, releaseBusyViewport, shouldAbortReader, + shouldAbortReaderOnCancelAck, shouldApplyMintBind, + shouldKickCancelRetryAttach, shouldSetHostTurnNote, + shouldSkipCancelPost, type DetachTurnInput, } from './detachTurn'; @@ -139,6 +152,19 @@ describe('decideDetach (plan #812 D18 contract)', () => { expect(isDetachAbort(stop.signal)).toBe(false); expect(isDetachAbort(undefined)).toBe(false); }); + + it('isG22AcceptedAbort: only abort(G22_ACCEPTED_ABORT_REASON)', () => { + const accepted = new AbortController(); + accepted.abort(G22_ACCEPTED_ABORT_REASON); + expect(isG22AcceptedAbort(accepted.signal)).toBe(true); + const raw = new AbortController(); + raw.abort(); + expect(isG22AcceptedAbort(raw.signal)).toBe(false); + const detach = new AbortController(); + detach.abort(DETACH_ABORT_REASON); + expect(isG22AcceptedAbort(detach.signal)).toBe(false); + expect(isG22AcceptedAbort(undefined)).toBe(false); + }); }); describe('decideDetachPersist (adversarial #844 Clear-vs-PUT / late persist)', () => { @@ -161,7 +187,7 @@ describe('decideDetachPersist (adversarial #844 Clear-vs-PUT / late persist)', ( ).toBe('drop'); }); - it('detached + running + turnRunId → preserve (Switch/New/unmount)', () => { + it('detached + running/cancelling + turnRunId → preserve (Switch/New/unmount; G22 liveness)', () => { expect( decideDetachPersist({ detached: true, @@ -170,6 +196,14 @@ describe('decideDetachPersist (adversarial #844 Clear-vs-PUT / late persist)', ( turnStatus: 'running', }), ).toBe('preserve'); + expect( + decideDetachPersist({ + detached: true, + discarded: false, + turnRunId: 'wr_live', + turnStatus: 'cancelling', + }), + ).toBe('preserve'); }); it('detached without a durable running id → drop (no omit-clear PUT)', () => { @@ -189,7 +223,7 @@ describe('decideDetachPersist (adversarial #844 Clear-vs-PUT / late persist)', ( detached: true, discarded: false, turnRunId: 'wr_old', - turnStatus: 'cancelling', + turnStatus: 'idle', }), ).toBe('drop'); }); @@ -356,18 +390,19 @@ describe('HarnessHost detach wiring source-lock (plan #812 D18)', () => { expect(helper).toContain('turnEpochRef.current += 1'); }); - it('raw abort() sites: helper (reasoned) + runPrompt supersede + poll Stop', () => { + it('raw abort() sites: helper (reasoned) + runPrompt supersede + poll ack + poll legacy', () => { const aborts = host.match(/abortRef\.current\?\.abort\(/g) ?? []; - expect(aborts.length).toBe(3); + expect(aborts.length).toBe(4); const poll = host.slice( host.indexOf('const poll = () =>'), host.indexOf('pollRef.current = window.setTimeout(poll, 150)'), ); expect(poll).toContain('takePendingCancel()'); expect(poll).toContain('abortRef.current?.abort()'); + expect(poll).toContain('abortReasonForCancelAck'); expect(poll).toContain('releaseBusyViewport('); expect(poll).not.toContain('decideDetach'); - expect(poll).not.toContain('abortReasonFor'); + expect(poll).not.toContain('abortReasonFor('); expect(poll).not.toContain('turnEpochRef'); }); @@ -418,7 +453,7 @@ describe('HarnessHost detach wiring source-lock (plan #812 D18)', () => { const runStart = host.indexOf('const runPrompt = useCallback'); const run = host.slice(runStart, host.indexOf('useEffect(() => {', runStart)); expect(run).toContain('const repo = repoRef.current'); - expect(run).toContain('putPreservedTurn(repo, snapshot, startedId, pendingMintId)'); + expect(run).toContain('putPreservedTurn(repo, foldedSnapshot, startedId, pendingMintId)'); expect(run).toContain('shouldApplyMintBind('); expect(run).toContain('pendingMintBindRef.current'); expect(run).toContain('switchInFlightRef.current'); @@ -448,9 +483,114 @@ describe('HarnessHost detach wiring source-lock (plan #812 D18)', () => { it('runPrompt skips ember hostNote on same-tab running detach (adversarial #853)', () => { const runStart = host.indexOf('const runPrompt = useCallback'); const run = host.slice(runStart, host.indexOf('useEffect(() => {', runStart)); - expect(run).toContain('shouldSetHostTurnNote(folded.turnStatus)'); + expect(run).toContain('shouldSetHostTurnNote(persisted.turnStatus)'); expect(run).not.toMatch(/if \(!result\.ok\) \{\s*setHostNote\(result\.error\);/); }); + + it('poll Stop wires the G22 server cancel (plan #816)', () => { + const poll = host.slice( + host.indexOf('const poll = () =>'), + host.indexOf('pollRef.current = window.setTimeout(poll, 150)'), + ); + // Cancel POST fires through the turnApi client + the pre/post fold planner. + expect(poll).toContain('cancelTurn('); + expect(poll).toContain('keepalive: true'); + expect(poll).toContain('decideStopFoldPre('); + expect(poll).toContain('decideStopFoldPost('); + expect(poll).toContain('shouldSkipCancelPost('); + expect(poll).toContain('applyStopFoldToSession('); + expect(poll).toContain('cancelPostedRunIdsRef'); + expect(poll).toContain('pendingStopFoldRef'); + expect(poll).toContain("apply.fold.kind === 'keep-running'"); + expect(poll).toContain('shouldKickCancelRetryAttach('); + expect(poll).toContain('decideCancelRetryKickWhen('); + expect(poll).toContain('kickColdAttach'); + expect(poll).toContain('CANCEL_RETRY_NOTE'); + expect(poll).toContain('CANCEL_FAILED_NOTE'); + expect(poll).toContain('setHostNote('); + expect(poll).not.toContain('will end on its own'); + expect(poll).toContain('persist(applyStopFoldToSession('); + expect(poll).toContain('cancelPostedRunIdsRef.current.delete'); + expect(poll).toContain('decideCancelAckApply('); + expect(poll).toContain('apply.dropPostedId'); + expect(poll).toContain("apply.commit === 'drop'"); + expect(poll).toContain("apply.commit === 'persist'"); + expect(poll).toContain("apply.commit === 'persist-detached'"); + expect(poll).toContain('cancelRepo?.put(cancelSessionId, detached)'); + expect(poll).toContain('discardedSessionIdsRef.current.has(cancelSessionId)'); + expect(poll).not.toContain('if (inflightRef.current) return'); + expect(poll).toContain('lastStopPersistRef'); + expect(poll).toContain('preserved.sessionId === cancelSessionId'); + expect(poll).toContain('sessionRef.current.id === cancelSessionId'); + expect(poll).toContain('shouldAbortReaderOnCancelAck('); + expect(poll).toContain('abortReasonForCancelAck'); + expect(poll).toContain('pendingCancelRetryAttachRef'); + expect(poll).toContain('activePromptGenerationRef'); + }); + + it('poll does not abort, release Busy, or persist cancelling before cancelTurn ack', () => { + const poll = host.slice( + host.indexOf('const poll = () =>'), + host.indexOf('pollRef.current = window.setTimeout(poll, 150)'), + ); + const thenAt = poll.indexOf('.then('); + expect(thenAt).toBeGreaterThan(0); + const beforeThen = poll.slice(0, thenAt); + expect(beforeThen).not.toContain('abortRef.current?.abort'); + expect(beforeThen).not.toContain('releaseBusyViewport'); + expect(beforeThen).not.toContain('pendingStopFoldRef.current ='); + expect(beforeThen).toContain('keepalive: true'); + const thenBody = poll.slice(thenAt); + expect(thenBody).toContain('shouldAbortReaderOnCancelAck'); + expect(thenBody).toContain('releaseBusyViewport'); + expect(thenBody).toContain('pendingStopFoldRef.current ='); + }); + + it('feature-divide Stop row matches wait-for-ack (adversarial-review #927)', () => { + const divide = readFileSync(resolve(process.cwd(), 'docs/feature-divide.md'), 'utf8'); + const stopRow = divide.split('\n').find((l) => l.includes('**Stop / cancel turn**')) ?? ''; + expect(stopRow).toContain('POST /api/turns/:runId/cancel'); + expect(stopRow).toContain('**before** aborting'); + expect(stopRow).toContain('releaseBusyViewport'); + expect(stopRow).toContain('DETACH_ABORT_REASON'); + expect(stopRow).not.toMatch(/clears Busy \/ Ready \/ clocks this tick/); + const clockRow = + divide.split('\n').find((l) => l.includes('Whole-turn `mm:ss` clock')) ?? ''; + expect(clockRow).not.toContain('Stop this tick'); + }); + + it('runPrompt generation-gates finally Busy clear and deferred cancel-retry kick', () => { + const runStart = host.indexOf('const runPrompt = useCallback'); + const run = host.slice(runStart, host.indexOf('useEffect(() => {', runStart)); + expect(run).toContain('promptGenerationRef.current'); + expect(run).toContain('activePromptGenerationRef.current'); + expect(run).toContain('pendingCancelRetryAttachRef.current'); + expect(run).toContain('promptGenerationRef.current === myGeneration'); + expect(run).toContain('queueMicrotask(kickColdAttach)'); + }); + + it('runPrompt nulls pendingStopFoldRef so a leftover fold cannot ride the next persistTurn (adversarial-review #927)', () => { + const runStart = host.indexOf('const runPrompt = useCallback'); + const run = host.slice(runStart, host.indexOf('useEffect(() => {', runStart)); + expect(run).toContain('pendingStopFoldRef.current = null'); + // Pass 6: wipe THIS session's abort snapshot only. A destination + // runPrompt (switch + Send / kickColdAttach) must not drop A's slot or + // persist-detached falls back to Stop-fire cancelSnapshot. + expect(run).toContain( + 'lastStopPersistRef.current?.sessionId === sessionRef.current.id', + ); + expect(run).toContain('lastStopPersistRef.current = null'); + }); + + it('persistTurn applies pendingStopFold so harnessChat cannot beat a failed ack (adversarial-review #927)', () => { + expect(host).toContain('pendingStopFoldRef.current'); + expect(host).toContain('applyStopFoldToSession(snapshot, pendingFold.runId, pendingFold.fold)'); + expect(host).toContain('lastStopPersistRef.current'); + // Pass 6: persistTurn refreshes the abort snapshot even when pendingFold + // was nulled by a destination runPrompt (sessionId+runId match). + expect(host).toContain('lastStopPersistRef.current.sessionId === foldedSnapshot.id'); + expect(host).toContain('lastStopPersistRef.current.runId === persistRunId'); + }); }); describe('shouldSetHostTurnNote (adversarial #853 same-tab detach)', () => { @@ -458,10 +598,456 @@ describe('shouldSetHostTurnNote (adversarial #853 same-tab detach)', () => { expect(shouldSetHostTurnNote('running')).toBe(false); }); - it('completed / cancelling / unset still surface the note', () => { + it('cancelling (G22 Stop success) does not surface host error chrome (adversarial-review #927 pass 6)', () => { + expect(shouldSetHostTurnNote('cancelling')).toBe(false); + }); + + it('completed / unset still surface the note', () => { expect(shouldSetHostTurnNote('completed')).toBe(true); - expect(shouldSetHostTurnNote('cancelling')).toBe(true); expect(shouldSetHostTurnNote(undefined)).toBe(true); }); }); +describe('G22 Stop/Esc server-cancel fold planner (plan #816)', () => { + describe('decideStopFoldPre', () => { + it('live durable run (turnRunId + running) → cancelling (route to server cancel)', () => { + expect( + decideStopFoldPre({ turnRunId: 'wr_live', turnStatus: 'running' }), + ).toEqual({ kind: 'cancelling' }); + }); + + it('no run id (legacy /api/agent path) → legacy-clear', () => { + expect(decideStopFoldPre({ turnStatus: 'running' })).toEqual({ + kind: 'legacy-clear', + }); + expect(decideStopFoldPre({})).toEqual({ kind: 'legacy-clear' }); + }); + + it('run id but completed / idle → legacy-clear; cancelling still routes to server cancel (pass 4)', () => { + // Pass 4: `'cancelling'` is a POST candidate so a poisoned optimistic + // marker (failed ack + switch/unmount) can retry Stop. Posted-id set + // is the once-per-run skip. + expect( + decideStopFoldPre({ turnRunId: 'wr_1', turnStatus: 'cancelling' }), + ).toEqual({ kind: 'cancelling' }); + expect( + decideStopFoldPre({ turnRunId: 'wr_1', turnStatus: 'completed' }), + ).toEqual({ kind: 'legacy-clear' }); + expect(decideStopFoldPre({ turnRunId: 'wr_1', turnStatus: 'idle' })).toEqual({ + kind: 'legacy-clear', + }); + }); + }); + + describe('decideStopFoldPost', () => { + const pre = { kind: 'cancelling' as const }; + + it('accepted → cancelling (KEEP turnRunId, fold cancelling)', () => { + expect(decideStopFoldPost({ pre, outcome: { kind: 'accepted' } })).toEqual({ + kind: 'cancelling', + }); + }); + + it('terminal (409) → clear-terminal (orphan-unstick: clear id + completed)', () => { + expect(decideStopFoldPost({ pre, outcome: { kind: 'terminal' } })).toEqual({ + kind: 'clear-terminal', + }); + }); + + it('gone (404) → clear-terminal (orphan-unstick)', () => { + expect(decideStopFoldPost({ pre, outcome: { kind: 'gone' } })).toEqual({ + kind: 'clear-terminal', + }); + }); + + it('failed (429/5xx/network) → keep-running (never a fake cancel)', () => { + expect(decideStopFoldPost({ pre, outcome: { kind: 'failed' } })).toEqual({ + kind: 'keep-running', + }); + }); + + it('legacy-clear pre never reaches the outcome mapping', () => { + expect( + decideStopFoldPost({ + pre: { kind: 'legacy-clear' }, + outcome: { kind: 'accepted' }, + }), + ).toEqual({ kind: 'legacy-clear' }); + expect( + decideStopFoldPost({ + pre: { kind: 'legacy-clear' }, + outcome: { kind: 'failed' }, + }), + ).toEqual({ kind: 'legacy-clear' }); + }); + }); + + describe('shouldSkipCancelPost', () => { + it('never skips on session status — posted-id set is the once-per-run skip (pass 4)', () => { + expect( + shouldSkipCancelPost({ turnRunId: 'wr_live', turnStatus: 'cancelling' }), + ).toBe(false); + expect( + shouldSkipCancelPost({ turnRunId: 'wr_live', turnStatus: 'running' }), + ).toBe(false); + expect(shouldSkipCancelPost({ turnStatus: 'cancelling' })).toBe(false); + expect( + shouldSkipCancelPost({ turnRunId: 'wr_1', turnStatus: 'completed' }), + ).toBe(false); + expect(shouldSkipCancelPost({})).toBe(false); + }); + }); + + describe('applyStopFoldToSession (adversarial-review #927)', () => { + const live = { turnRunId: 'wr_live', turnStatus: 'cancelling' as const }; + + it('cancelling keeps the id and folds cancelling', () => { + expect( + applyStopFoldToSession( + { turnRunId: 'wr_live', turnStatus: 'running' as const }, + 'wr_live', + { kind: 'cancelling' }, + ), + ).toEqual({ turnRunId: 'wr_live', turnStatus: 'cancelling' }); + }); + + it('keep-running reverts optimistic cancelling so Stop can retry', () => { + expect( + applyStopFoldToSession(live, 'wr_live', { kind: 'keep-running' }), + ).toEqual({ turnRunId: 'wr_live', turnStatus: 'running' }); + }); + + it('clear-terminal drops the id and folds completed', () => { + expect( + applyStopFoldToSession(live, 'wr_live', { kind: 'clear-terminal' }), + ).toEqual({ turnRunId: undefined, turnStatus: 'completed' }); + }); + + it('never clobbers a newer run id', () => { + expect( + applyStopFoldToSession( + { turnRunId: 'wr_newer', turnStatus: 'running' as const }, + 'wr_live', + { kind: 'keep-running' }, + ), + ).toEqual({ turnRunId: 'wr_newer', turnStatus: 'running' }); + }); + + it('never plants a cleared id (adversarial-review #927)', () => { + const cleared = { turnRunId: undefined, turnStatus: 'completed' as const }; + expect( + applyStopFoldToSession(cleared, 'wr_live', { kind: 'keep-running' }), + ).toEqual(cleared); + expect( + applyStopFoldToSession(cleared, 'wr_live', { kind: 'cancelling' }), + ).toEqual(cleared); + expect( + applyStopFoldToSession(cleared, 'wr_live', { kind: 'clear-terminal' }), + ).toEqual(cleared); + }); + + it('legacy-clear is a no-op on the snapshot', () => { + expect(applyStopFoldToSession(live, 'wr_live', { kind: 'legacy-clear' })).toEqual( + live, + ); + }); + }); + + describe('decideCancelAckApply (adversarial-review #927 pass 4)', () => { + const keep = { kind: 'keep-running' as const }; + const accepted = { kind: 'cancelling' as const }; + const base = { + unmounted: false, + liveSessionId: 's1', + cancelSessionId: 's1', + liveTurnRunId: 'wr_live', + stopRunId: 'wr_live', + discarded: false, + inflight: false, + }; + + it('failed ack idle → persist + dropPostedId (Stop can retry)', () => { + expect(decideCancelAckApply({ ...base, fold: keep })).toEqual({ + fold: keep, + dropPostedId: true, + commit: 'persist', + }); + }); + + it('failed ack while inflight → pending-only + dropPostedId (no snapshot persist)', () => { + expect( + decideCancelAckApply({ ...base, inflight: true, fold: keep }), + ).toEqual({ + fold: keep, + dropPostedId: true, + commit: 'pending-only', + }); + }); + + it('accepted ack idle → persist, keep posted-id (once per run id)', () => { + expect(decideCancelAckApply({ ...base, fold: accepted })).toEqual({ + fold: accepted, + dropPostedId: false, + commit: 'persist', + }); + }); + + it('discarded → drop (still dropPostedId on failed; never resurrect Clear)', () => { + expect( + decideCancelAckApply({ ...base, discarded: true, fold: keep }), + ).toEqual({ fold: keep, dropPostedId: true, commit: 'drop' }); + }); + + it('newer turnRunId on the same session → drop (do not clobber the new run)', () => { + expect( + decideCancelAckApply({ + ...base, + liveTurnRunId: 'wr_newer', + fold: keep, + }), + ).toEqual({ fold: keep, dropPostedId: true, commit: 'drop' }); + }); + + it('unmount / switch → persist-detached (keep-running onto cancelSessionId)', () => { + expect( + decideCancelAckApply({ ...base, unmounted: true, fold: keep }), + ).toEqual({ + fold: keep, + dropPostedId: true, + commit: 'persist-detached', + }); + expect( + decideCancelAckApply({ + ...base, + liveSessionId: 's2', + fold: keep, + }), + ).toEqual({ + fold: keep, + dropPostedId: true, + commit: 'persist-detached', + }); + }); + + it('switch + inflight on the destination still persist-detached (not pending-only on liveNow)', () => { + expect( + decideCancelAckApply({ + ...base, + liveSessionId: 's2', + inflight: true, + fold: keep, + }), + ).toEqual({ + fold: keep, + dropPostedId: true, + commit: 'persist-detached', + }); + }); + + it('cleared turnRunId (undefined) is not a newer id — still persist keep-running', () => { + expect( + decideCancelAckApply({ + ...base, + liveTurnRunId: undefined, + fold: keep, + }), + ).toEqual({ fold: keep, dropPostedId: true, commit: 'persist' }); + }); + + it('terminal/gone fold drops posted-id', () => { + const clear = { kind: 'clear-terminal' as const }; + expect(decideCancelAckApply({ ...base, fold: clear }).dropPostedId).toBe( + true, + ); + }); + + it('accepted ack + switch still persist-detached (do not keep posted-id off the abandoned session)', () => { + expect( + decideCancelAckApply({ + ...base, + liveSessionId: 's2', + fold: accepted, + }), + ).toEqual({ + fold: accepted, + dropPostedId: false, + commit: 'persist-detached', + }); + }); + }); + + describe('shouldKickCancelRetryAttach (adversarial-review #927 pass 7)', () => { + const keep = { kind: 'keep-running' as const }; + const accepted = { kind: 'cancelling' as const }; + const kickBase = { + fold: keep, + commit: 'persist' as const, + unmounted: false, + liveSessionId: 's1', + cancelSessionId: 's1', + inflight: false, + }; + + it('same-session idle keep-running persist → kick (restore Busy / Stop)', () => { + expect(shouldKickCancelRetryAttach(kickBase)).toBe(true); + }); + + it('pending-only (inflight) → no kick', () => { + expect( + shouldKickCancelRetryAttach({ + ...kickBase, + commit: 'pending-only', + inflight: true, + }), + ).toBe(false); + }); + + it('persist-detached (switch / unmount) → no kick', () => { + expect( + shouldKickCancelRetryAttach({ + ...kickBase, + commit: 'persist-detached', + liveSessionId: 's2', + }), + ).toBe(false); + expect( + shouldKickCancelRetryAttach({ + ...kickBase, + commit: 'persist-detached', + unmounted: true, + }), + ).toBe(false); + }); + + it('drop (Clear / newer id) → no kick', () => { + expect( + shouldKickCancelRetryAttach({ ...kickBase, commit: 'drop' }), + ).toBe(false); + }); + + it('accepted cancelling persist → no kick', () => { + expect( + shouldKickCancelRetryAttach({ ...kickBase, fold: accepted }), + ).toBe(false); + }); + + it('retry note never tells the operator the run will end on its own', () => { + expect(CANCEL_RETRY_NOTE).toContain('re-attaching so you can Stop again'); + expect(CANCEL_RETRY_NOTE).not.toContain('will end on its own'); + expect(CANCEL_FAILED_NOTE).toContain('the run is still live'); + expect(CANCEL_FAILED_NOTE).not.toContain('will end on its own'); + expect(CANCEL_FAILED_NOTE).not.toContain('re-attaching'); + }); + }); + + describe('shouldAbortReaderOnCancelAck + abortReasonForCancelAck', () => { + it('accepted persist → abort with G22 accepted reason + release', () => { + expect( + shouldAbortReaderOnCancelAck({ + fold: { kind: 'cancelling' }, + commit: 'persist', + }), + ).toBe(true); + expect(abortReasonForCancelAck({ kind: 'cancelling' })).toBe( + G22_ACCEPTED_ABORT_REASON, + ); + }); + + it('409/404 persist → abort without accepted reason (no "you stopped")', () => { + expect( + shouldAbortReaderOnCancelAck({ + fold: { kind: 'clear-terminal' }, + commit: 'persist', + }), + ).toBe(true); + expect(abortReasonForCancelAck({ kind: 'clear-terminal' })).toBeUndefined(); + }); + + it('failed keep-running never aborts the live reader', () => { + expect( + shouldAbortReaderOnCancelAck({ + fold: { kind: 'keep-running' }, + commit: 'persist', + }), + ).toBe(false); + }); + + it('persist-detached / pending-only / drop never abort the destination reader', () => { + expect( + shouldAbortReaderOnCancelAck({ + fold: { kind: 'cancelling' }, + commit: 'persist-detached', + }), + ).toBe(false); + expect( + shouldAbortReaderOnCancelAck({ + fold: { kind: 'cancelling' }, + commit: 'pending-only', + }), + ).toBe(false); + expect( + shouldAbortReaderOnCancelAck({ + fold: { kind: 'clear-terminal' }, + commit: 'drop', + }), + ).toBe(false); + }); + }); + + describe('decideCancelRetryKickWhen (item 5)', () => { + it('defers kick while the aborted prompt is still in try/finally', () => { + expect( + decideCancelRetryKickWhen({ shouldKick: true, promptActive: true }), + ).toBe('pending'); + }); + + it('kicks now when the aborted invocation already finished', () => { + expect( + decideCancelRetryKickWhen({ shouldKick: true, promptActive: false }), + ).toBe('now'); + }); + + it('pending-only / persist-detached / drop stay none (pass 7)', () => { + expect( + decideCancelRetryKickWhen({ shouldKick: false, promptActive: true }), + ).toBe('none'); + expect( + decideCancelRetryKickWhen({ shouldKick: false, promptActive: false }), + ).toBe('none'); + }); + }); + + describe('G22 cancel ack race rows (punch-list items 1–4)', () => { + const live = { turnRunId: 'wr_live', turnStatus: 'running' as const }; + + it('row 1: Stop + unload before 200 does not persist cancelling', () => { + // No outcome yet — applyStopFold is not called. Envelope stays running. + expect(live.turnStatus).toBe('running'); + expect(live.turnRunId).toBe('wr_live'); + // Identity keep-running (what a dropped POST would fold if anything). + expect(applyStopFoldToSession(live, 'wr_live', { kind: 'keep-running' })).toEqual( + live, + ); + }); + + it('row 2: 503 ack never leaves a persisted stop line (keep-running)', () => { + const folded = applyStopFoldToSession(live, 'wr_live', { kind: 'keep-running' }); + expect(folded.turnStatus).toBe('running'); + expect(folded.turnRunId).toBe('wr_live'); + expect( + shouldAbortReaderOnCancelAck({ fold: { kind: 'keep-running' }, commit: 'persist' }), + ).toBe(false); + }); + + it('row 3: 409 terminal does not abort with the accepted reason (no "you stopped")', () => { + const folded = applyStopFoldToSession(live, 'wr_live', { kind: 'clear-terminal' }); + expect(folded.turnStatus).toBe('completed'); + expect(folded.turnRunId).toBeUndefined(); + expect(abortReasonForCancelAck({ kind: 'clear-terminal' })).toBeUndefined(); + expect(abortReasonForCancelAck({ kind: 'cancelling' })).toBe( + G22_ACCEPTED_ABORT_REASON, + ); + }); + }); +}); + diff --git a/lib/detachTurn.ts b/lib/detachTurn.ts index e9c7b5ce..6a1b6576 100644 --- a/lib/detachTurn.ts +++ b/lib/detachTurn.ts @@ -17,6 +17,13 @@ import type { TurnStatus } from './sessionCloudCaps'; /** `AbortController.abort` reason for a durable detach (not a user Stop). */ export const DETACH_ABORT_REASON = 'detach'; +/** + * `AbortController.abort` reason after a G22 cancel POST was **accepted**. + * `runHarnessTurn` folds `'cancelling'` + Turn-ended only for this reason. + * A raw abort (unload / abort-before-ack) keeps `running` and paints no stop line. + */ +export const G22_ACCEPTED_ABORT_REASON = 'g22-accepted'; + /** Decision a detach/cancel site should act on. */ export type DetachDecision = 'detach' | 'detach-close' | 'noop' | 'cancel'; @@ -91,6 +98,11 @@ export function isDetachAbort(signal?: AbortSignal): boolean { return signal?.aborted === true && signal.reason === DETACH_ABORT_REASON; } +/** True when Stop aborted the reader after the cancel POST returned `accepted`. */ +export function isG22AcceptedAbort(signal?: AbortSignal): boolean { + return signal?.aborted === true && signal.reason === G22_ACCEPTED_ABORT_REASON; +} + /** * What the host should do with a turn snapshot after a leave-turn site * (adversarial #844 re-review). @@ -99,7 +111,7 @@ export function isDetachAbort(signal?: AbortSignal): boolean { * |-------|--------| * | Clear/remove discarded the started id | `drop` — never PUT (LWW upsert would resurrect) | * | Still on this turn (epoch match) | `live` — writeLocal + put as today | - * | Detached + running + turnRunId | `preserve` — PUT onto `preserveTargetId` (pending mint UUID, else startedId); never clobber a switched live session | + * | Detached + running/cancelling + turnRunId | `preserve` — PUT onto `preserveTargetId` (pending mint UUID, else startedId); never clobber a switched live session. `'cancelling'` is host-held **liveness** (G22 / C15), not terminal. | * | Detached without a durable running id | `drop` — skip PUT so we cannot omit-clear C14d | */ export type DetachPersistAction = 'live' | 'preserve' | 'drop'; @@ -121,7 +133,9 @@ export interface DetachPersistInput { export function decideDetachPersist(input: DetachPersistInput): DetachPersistAction { if (input.discarded) return 'drop'; if (!input.detached) return 'live'; - if (input.turnRunId && input.turnStatus === 'running') return 'preserve'; + if (input.turnRunId && (input.turnStatus === 'running' || input.turnStatus === 'cancelling')) { + return 'preserve'; + } return 'drop'; } @@ -193,12 +207,14 @@ export function shouldApplyMintBind(input: { * same-epoch + `{ ok: false }` + `turnStatus: 'running'` — canvas Ready, run * still live. A host error string would lie that the turn failed. * - * `running` is the persist contract for detach (D18 fold). Other fail - * outcomes write `completed` (or leave a leftover terminal status) and - * still surface the note. + * `running` is the persist contract for detach (D18 fold). G22 Stop's + * success state is `'cancelling'` (host-held liveness) — the canvas already + * has `Turn ended · you stopped`; ember `host: Request cancelled.` would be + * dual chrome (adversarial-review #927 pass 6). Other fail outcomes write + * `completed` (or leave a leftover terminal status) and still surface the note. */ export function shouldSetHostTurnNote(turnStatus?: TurnStatus): boolean { - return turnStatus !== 'running'; + return turnStatus !== 'running' && turnStatus !== 'cancelling'; } /** @@ -220,4 +236,272 @@ export function releaseBusyViewport(hooks: BusyViewportHooks): void { hooks.setLifecycleReady(); } +// ── Plan #816 (G22) — Stop/Esc server-cancel fold planner ── + +/** Outcome of the G22 cancel POST, mapped 1:1 from `lib/turnApi.cancelTurn`. */ +export type CancelPostOutcome = + | { kind: 'accepted' } + | { kind: 'terminal' } + | { kind: 'gone' } + | { kind: 'failed' }; + +/** + * What the host Stop fold should do with the session after one Stop/Esc tick + * (plan #816 Host Stop fold + Cancel race table). + * + * | Input | Fold | + * |-------|------| + * | No live `turnRunId` (legacy `/api/agent` path) | `legacy-clear` — old `turnRunId: undefined` + `completed` fold | + * | Live run + cancel accepted | `cancelling` — KEEP `turnRunId`, fold `turnStatus: 'cancelling'` | + * | Cancel POST failed (429/5xx/network) | `keep-running` — keep `turnRunId` + `running`, paint a soft note | + * | Run terminal (409) or gone (404) | `clear-terminal` — clear `turnRunId` + fold `completed` (orphan-unstick) | + */ +export type StopFoldAction = + | { kind: 'legacy-clear' } + | { kind: 'cancelling' } + | { kind: 'keep-running' } + | { kind: 'clear-terminal' }; + +/** + * Fold the session-side Stop decision BEFORE knowing the cancel POST outcome. + * A live durable id in `running` **or** `cancelling` routes to the server + * cancel. `'cancelling'` is the accepted overlay (and a leftover poisoned + * marker) — it must still be a POST candidate so a failed ack + switch/unmount + * can retry Stop after the posted-id is dropped. The host does **not** persist + * `'cancelling'` until the POST returns. The in-memory `cancelPostedRunIdsRef` + * is the once-per-run skip. Everything else keeps today's legacy clear fold. + */ +export function decideStopFoldPre(input: { + turnRunId?: string; + turnStatus?: TurnStatus; +}): Extract { + if ( + input.turnRunId !== undefined && + (input.turnStatus === 'running' || input.turnStatus === 'cancelling') + ) { + return { kind: 'cancelling' }; + } + return { kind: 'legacy-clear' }; +} + +/** + * Fold the session-side Stop decision AFTER the cancel POST resolves + * (plan #816 Cancel race & failure semantics). A `cancelling` pre-fold is + * re-resolved by the server truth; `legacy-clear` never reaches here. + */ +export function decideStopFoldPost(input: { + pre: Extract; + outcome: CancelPostOutcome; +}): StopFoldAction { + if (input.pre.kind === 'legacy-clear') return { kind: 'legacy-clear' }; + switch (input.outcome.kind) { + case 'accepted': + return { kind: 'cancelling' }; + case 'terminal': + case 'gone': + return { kind: 'clear-terminal' }; + case 'failed': + return { kind: 'keep-running' }; + } +} + +/** + * True when a second Stop/Esc must not re-POST for this run. + * + * Pass 4 (adversarial-review #927): session `'cancelling'` is the optimistic + * pre-ack marker, **not** the accepted-ack skip. A poisoned `'cancelling'` + * (failed ack + switch/unmount/F5) must be able to retry Stop. The in-memory + * `cancelPostedRunIdsRef` is the once-per-run skip (in-flight / accepted). + * Always false — kept as a named seam so the host poll source-lock still + * names it; the posted-id set is the real skip. + */ +export function shouldSkipCancelPost(_input: { + turnRunId?: string; + turnStatus?: TurnStatus; +}): boolean { + return false; +} + +/** + * Apply a G22 Stop-fold action onto a session snapshot. + * + * Used by the host poll *and* by `persistTurn` so a late abort-fold cannot + * win a race against a failed cancel POST (plan #816: failed ack keeps + * `running` so Stop can retry). The host does not persist `'cancelling'` + * until `cancelTurn` returns `accepted`. + * A newer `turnRunId` on the snapshot is never clobbered. + * A snapshot that already cleared `turnRunId` is never planted back + * (adversarial-review #927: leftover pendingFold must not resurrect a + * cleared id onto the next prompt's pre-headers persist). + */ +export function applyStopFoldToSession< + T extends { turnRunId?: string; turnStatus?: TurnStatus }, +>(session: T, runId: string, fold: StopFoldAction): T { + if (fold.kind === 'legacy-clear') return session; + if (session.turnRunId === undefined || session.turnRunId !== runId) { + return session; + } + switch (fold.kind) { + case 'cancelling': + return { ...session, turnRunId: runId, turnStatus: 'cancelling' }; + case 'keep-running': + return { ...session, turnRunId: runId, turnStatus: 'running' }; + case 'clear-terminal': + return { ...session, turnRunId: undefined, turnStatus: 'completed' }; + } +} + +/** + * What the G22 cancel-ack `then()` should do (adversarial-review #927 pass 4). + * + * Pass 1: a failed POST must drop the posted-id and fold `keep-running` so + * Stop can retry. Pass 2 skipped the *entire* ack when `inflight` so a slow + * persist could not stomp the next prompt — and that return also skipped the + * posted-id delete, restoring ghost spend. Pass 3 split the commit but used + * `drop` for switch/unmount, which left optimistic `'cancelling'` on the + * abandoned session. Pass 4 persists onto `cancelSessionId` for those leave + * sites (captured snapshot + captured repo) and only `drop`s Clear. + * + * | Condition | `commit` | posted-id | + * |-----------|----------|-----------| + * | discarded (Clear) / newer `turnRunId` on the **same** session | `drop` | still drop on failed/terminal | + * | unmount / switched session | `persist-detached` onto `cancelSessionId` | drop on failed/terminal | + * | new `runPrompt` inflight (same session) | `pending-only` (set pendingFold, no persist) | drop on failed/terminal | + * | same session, idle | `persist` the fold onto `liveNow` | drop on failed/terminal | + * + * `dropPostedId` is true for `keep-running` / `clear-terminal` (retry / orphan + * unstick) and false for `accepted` `'cancelling'` (once per run id). + */ +export type CancelAckCommit = 'drop' | 'pending-only' | 'persist' | 'persist-detached'; + +export type CancelAckApply = { + fold: StopFoldAction; + dropPostedId: boolean; + commit: CancelAckCommit; +}; + +export function decideCancelAckApply(input: { + unmounted: boolean; + liveSessionId: string; + cancelSessionId: string; + liveTurnRunId?: string; + stopRunId: string; + discarded: boolean; + inflight: boolean; + fold: StopFoldAction; +}): CancelAckApply { + const dropPostedId = + input.fold.kind === 'keep-running' || input.fold.kind === 'clear-terminal'; + // Clear resurrection — never PUT a deleted row. + if (input.discarded) { + return { fold: input.fold, dropPostedId, commit: 'drop' }; + } + const switched = input.liveSessionId !== input.cancelSessionId; + const newerOnSameSession = + !switched && + input.liveTurnRunId !== input.stopRunId && + input.liveTurnRunId !== undefined; + // A newer run id on the same session must not be clobbered by keep-running. + if (newerOnSameSession) { + return { fold: input.fold, dropPostedId, commit: 'drop' }; + } + // Switch / unmount: persist onto cancelSessionId (captured snapshot + repo), + // never onto liveNow. Unmount cleanup nulls repoRef — the host captures the + // repo object at Stop fire (same pattern as adversarial #844). + if (input.unmounted || switched) { + return { fold: input.fold, dropPostedId, commit: 'persist-detached' }; + } + if (input.inflight) { + return { fold: input.fold, dropPostedId, commit: 'pending-only' }; + } + return { fold: input.fold, dropPostedId, commit: 'persist' }; +} + +/** + * Host note when a G22 cancel POST failed (keep-running) **and** this tab + * re-attaches so Wasm Busy / ■ Stop return (adversarial-review #927 pass 7). + * Must never say the run "will end on its own" — that is the 1h-wall lie. + */ +export const CANCEL_RETRY_NOTE = + 'Stop signal did not reach the server — the run is still live; re-attaching so you can Stop again.'; + +/** + * Host note when a G22 cancel POST failed but this tab cannot re-attach + * (pending-only / a follow-up `runPrompt` is already inflight). Honest: the + * run is live. Does not claim Stop retry is armed. + */ +export const CANCEL_FAILED_NOTE = + 'Stop signal did not reach the server — the run is still live.'; + +/** + * True when the keep-running ack should cold-attach this tab so Wasm Busy + * (and ■ Stop / Esc) return. Stop is Busy-only (`composer_chrome.zig`); + * Busy is restored only after the Stop-aborted `runPrompt` `finally` (or + * immediately when that invocation already finished). Posted-id is dropped + * on keep-running so a later Stop can re-POST — but only if Busy is restored + * (adversarial-review #927 pass 7 / punch-list item 5). + * + * | Condition | Kick? | + * |-----------|-------| + * | `persist` + keep-running + same session + idle | yes | + * | `pending-only` (`runPrompt` inflight) | no — the next prompt owns the tab | + * | `persist-detached` (switch/unmount) | no — switch-back cold-attaches | + * | `drop` (Clear / newer id) | no | + */ +export function shouldKickCancelRetryAttach(input: { + fold: StopFoldAction; + commit: CancelAckCommit; + unmounted: boolean; + liveSessionId: string; + cancelSessionId: string; + inflight: boolean; +}): boolean { + return ( + input.fold.kind === 'keep-running' && + input.commit === 'persist' && + !input.unmounted && + !input.inflight && + input.liveSessionId === input.cancelSessionId + ); +} + +/** + * Abort + release Busy only after a cancel POST that this tab still owns + * (`persist`). Failed keep-running must not abort the live reader (item 1). + * Switch/unmount/Clear already detached — never abort the destination reader. + */ +export function shouldAbortReaderOnCancelAck(input: { + fold: StopFoldAction; + commit: CancelAckCommit; +}): boolean { + if (input.commit !== 'persist') return false; + return input.fold.kind === 'cancelling' || input.fold.kind === 'clear-terminal'; +} + +/** Abort reason after an owned cancel ack. Accepted only; 409/404 stay raw. */ +export function abortReasonForCancelAck(fold: StopFoldAction): string | undefined { + return fold.kind === 'cancelling' ? G22_ACCEPTED_ABORT_REASON : undefined; +} + +/** + * When the failed-ack kick should run (item 5). Never use `inflightRef` after + * `releaseBusyViewport` — that flag is already false while `runPrompt` unwinds. + * + * | `shouldKick` | prompt still in try/finally | Result | + * |--------------|-----------------------------|--------| + * | false | * | `none` | + * | true | yes | `pending` — `finally` kicks | + * | true | no | `now` — invocation already finished | + */ +export type CancelRetryKickWhen = 'none' | 'pending' | 'now'; + +export function decideCancelRetryKickWhen(input: { + shouldKick: boolean; + promptActive: boolean; +}): CancelRetryKickWhen { + if (!input.shouldKick) return 'none'; + return input.promptActive ? 'pending' : 'now'; +} + + diff --git a/lib/harnessChat.test.ts b/lib/harnessChat.test.ts index 441ecae1..d32516d6 100644 --- a/lib/harnessChat.test.ts +++ b/lib/harnessChat.test.ts @@ -1669,7 +1669,7 @@ describe('runHarnessTurn stream agent (phase 1)', () => { ).toBe(false); }); - it('Stop after onTurnStarted clears this-turn running (adversarial #844)', async () => { + it('abort + Request cancelled. without accepted cancel keeps running, no stop line', async () => { const exp = makeMockExports(); const bridge = new HarnessBridge(exp); const { runHarnessTurn } = await import('./harnessChat'); @@ -1680,24 +1680,23 @@ describe('runHarnessTurn stream agent (phase 1)', () => { sendAgentStream: async (_prompt, init) => { await init?.onTurnStarted?.({ turnRunId: 'wr_live' }); controller.abort(); - // Production abort-after-headers now carries turnRunId; also prove the - // omit shape still clears via this-turn running. return { ok: false, error: 'Request cancelled.' }; }, }); - expect(next.turnRunId).toBeUndefined(); - expect(next.turnStatus).toBe('completed'); + expect(next.turnRunId).toBe('wr_live'); + expect(next.turnStatus).toBe('running'); expect( next.messages.some( (m) => m.role === 'system' && m.text === describeTurnEnd('stop'), ), - ).toBe(true); + ).toBe(false); }); - it('Stop after onTurnStarted with abort result id still clears (adversarial #844)', async () => { + it('row 2: failed cancel overlay on abort-before-ack keeps running, no stop line', async () => { const exp = makeMockExports(); const bridge = new HarnessBridge(exp); const { runHarnessTurn } = await import('./harnessChat'); + const { applyStopFoldToSession } = await import('./detachTurn'); const controller = new AbortController(); const { session: next } = await runHarnessTurn(bridge, createEmptySession(), 'work', { streamAgent: true, @@ -1708,6 +1707,115 @@ describe('runHarnessTurn stream agent (phase 1)', () => { return { ok: false, error: 'Request cancelled.', turnRunId: 'wr_live' }; }, }); + expect(next.turnStatus).toBe('running'); + expect( + next.messages.some( + (m) => m.role === 'system' && m.text === describeTurnEnd('stop'), + ), + ).toBe(false); + const folded = applyStopFoldToSession(next, 'wr_live', { kind: 'keep-running' }); + expect(folded.turnStatus).toBe('running'); + expect(folded.turnRunId).toBe('wr_live'); + expect( + folded.messages.some( + (m) => m.role === 'system' && m.text === describeTurnEnd('stop'), + ), + ).toBe(false); + }); + + it('row 3: 409 terminal overlay after raw abort does not claim you stopped', async () => { + const exp = makeMockExports(); + const bridge = new HarnessBridge(exp); + const { runHarnessTurn } = await import('./harnessChat'); + const { applyStopFoldToSession } = await import('./detachTurn'); + const controller = new AbortController(); + const { session: next } = await runHarnessTurn(bridge, createEmptySession(), 'work', { + streamAgent: true, + signal: controller.signal, + sendAgentStream: async (_prompt, init) => { + await init?.onTurnStarted?.({ turnRunId: 'wr_live' }); + controller.abort(); + return { ok: false, error: 'Request cancelled.', turnRunId: 'wr_live' }; + }, + }); + expect( + next.messages.some( + (m) => m.role === 'system' && m.text === describeTurnEnd('stop'), + ), + ).toBe(false); + const folded = applyStopFoldToSession(next, 'wr_live', { kind: 'clear-terminal' }); + expect(folded.turnStatus).toBe('completed'); + expect(folded.turnRunId).toBeUndefined(); + expect( + folded.messages.some( + (m) => m.role === 'system' && m.text === describeTurnEnd('stop'), + ), + ).toBe(false); + }); + + it('accepted cancel abort folds cancelling KEEPING this-turn id (G22 plan #816)', async () => { + const exp = makeMockExports(); + const bridge = new HarnessBridge(exp); + const { runHarnessTurn } = await import('./harnessChat'); + const { G22_ACCEPTED_ABORT_REASON } = await import('./detachTurn'); + const controller = new AbortController(); + const { session: next } = await runHarnessTurn(bridge, createEmptySession(), 'work', { + streamAgent: true, + signal: controller.signal, + sendAgentStream: async (_prompt, init) => { + await init?.onTurnStarted?.({ turnRunId: 'wr_live' }); + controller.abort(G22_ACCEPTED_ABORT_REASON); + return { ok: false, error: 'Request cancelled.' }; + }, + }); + expect(next.turnRunId).toBe('wr_live'); + expect(next.turnStatus).toBe('cancelling'); + expect( + next.messages.some( + (m) => m.role === 'system' && m.text === describeTurnEnd('stop'), + ), + ).toBe(true); + }); + + it('accepted cancel abort with result id folds cancelling KEEPING id (G22 plan #816)', async () => { + const exp = makeMockExports(); + const bridge = new HarnessBridge(exp); + const { runHarnessTurn } = await import('./harnessChat'); + const { G22_ACCEPTED_ABORT_REASON } = await import('./detachTurn'); + const controller = new AbortController(); + const { session: next } = await runHarnessTurn(bridge, createEmptySession(), 'work', { + streamAgent: true, + signal: controller.signal, + sendAgentStream: async (_prompt, init) => { + await init?.onTurnStarted?.({ turnRunId: 'wr_live' }); + controller.abort(G22_ACCEPTED_ABORT_REASON); + return { ok: false, error: 'Request cancelled.', turnRunId: 'wr_live' }; + }, + }); + expect(next.turnRunId).toBe('wr_live'); + expect(next.turnStatus).toBe('cancelling'); + expect( + next.messages.some( + (m) => m.role === 'system' && m.text === describeTurnEnd('stop'), + ), + ).toBe(true); + }); + + it('G22: operator Stop with NO live run id (legacy /api/agent path) keeps the old clear fold', async () => { + const exp = makeMockExports(); + const bridge = new HarnessBridge(exp); + const { runHarnessTurn } = await import('./harnessChat'); + const controller = new AbortController(); + const { session: next } = await runHarnessTurn(bridge, createEmptySession(), 'work', { + streamAgent: true, + signal: controller.signal, + sendAgentStream: async () => { + // Legacy path: no onTurnStarted, no turnRunId on the result. + controller.abort(); + return { ok: false, error: 'Request cancelled.' }; + }, + }); + // No live run id → the legacy `turnRunId: undefined` + `completed` fold. expect(next.turnRunId).toBeUndefined(); expect(next.turnStatus).toBe('completed'); expect( @@ -5674,13 +5782,14 @@ describe('runHarnessTurn attach handshake (plan #813 / E19)', () => { ).toBe(true); }); - it('test 6i: attach Stop after onTurnStarted keeps running, no you-stopped (adversarial #857)', async () => { + it('test 6i: attach accepted-cancel abort folds cancelling + Turn-ended (G22)', async () => { const exp = makeMockExports(); const bridge = new HarnessBridge(exp); const session = runningSession(); const sendAgent = vi.fn(async () => { throw new Error('must not POST /api/agent'); }); + const { G22_ACCEPTED_ABORT_REASON } = await import('./detachTurn'); const controller = new AbortController(); const { result, session: next } = await runHarnessTurn(bridge, session, '', { sendAgent, @@ -5692,7 +5801,7 @@ describe('runHarnessTurn attach handshake (plan #813 / E19)', () => { attachStream: async (runId, opts: AttachInit) => { await opts.onTurnStarted?.({ turnRunId: runId }); await opts.onEvent?.({ type: 'reasoning_delta', text: 'hmm' }); - controller.abort(); + controller.abort(G22_ACCEPTED_ABORT_REASON); return { ok: false as const, error: 'Request cancelled.', turnRunId: runId }; }, }, @@ -5700,20 +5809,19 @@ describe('runHarnessTurn attach handshake (plan #813 / E19)', () => { expect(result.ok).toBe(false); expect(sendAgent).not.toHaveBeenCalled(); expect(next.turnRunId).toBe('wr_live'); - expect(next.turnStatus).toBe('running'); + expect(next.turnStatus).toBe('cancelling'); expect( next.messages.some( (m) => m.role === 'system' && m.text === describeTurnEnd('stop'), ), - ).toBe(false); - expect(exp.__messages.some((m) => isTurnEndLine(m.text))).toBe(false); + ).toBe(true); expect(exp.__lifecycle()).toBe(Lifecycle.Ready); expect(exp.__messages.some((m) => m.kind === MessageKind.Thinking && m.text === 'hmm')).toBe( true, ); }); - it('test 6j: attach Stop before onTurnStarted keeps running, no subscribe-fail EMBER (adversarial #857)', async () => { + it('test 6j: attach abort before ack keeps running, no stop line, no subscribe-fail EMBER', async () => { const exp = makeMockExports(); const bridge = new HarnessBridge(exp); const session = runningSession(); @@ -5734,7 +5842,11 @@ describe('runHarnessTurn attach handshake (plan #813 / E19)', () => { expect(next.turnStatus).toBe('running'); expect(next.turnRunId).toBe('wr_1'); expect(exp.__lifecycle()).toBe(Lifecycle.Ready); - expect(exp.__messages.some((m) => isTurnEndLine(m.text))).toBe(false); + expect( + next.messages.some( + (m) => m.role === 'system' && m.text === describeTurnEnd('stop'), + ), + ).toBe(false); expect( exp.__messages.some((m) => m.kind === MessageKind.Error), ).toBe(false); @@ -5768,7 +5880,7 @@ describe('runHarnessTurn attach handshake (plan #813 / E19)', () => { expect(exp.__lifecycle()).toBe(Lifecycle.Ready); }); - it('test 6k: Send-while-running attach Stop keeps running, strips follow-up, no still-attached note (adversarial #857)', async () => { + it('test 6k: Send-while-running attach Stop folds cancelling + Turn-ended, strips follow-up (G22 / adversarial-review #927)', async () => { const exp = makeMockExports(); const bridge = new HarnessBridge(exp); bridge.pushMessage(MessageKind.User, 'hello'); @@ -5777,6 +5889,7 @@ describe('runHarnessTurn attach handshake (plan #813 / E19)', () => { const sendAgent = vi.fn(async () => { throw new Error('must not POST /api/agent'); }); + const { G22_ACCEPTED_ABORT_REASON } = await import('./detachTurn'); const controller = new AbortController(); const { result, session: next } = await runHarnessTurn(bridge, session, 'follow-up', { sendAgent, @@ -5788,7 +5901,7 @@ describe('runHarnessTurn attach handshake (plan #813 / E19)', () => { attachStream: async (runId, opts: AttachInit) => { await opts.onTurnStarted?.({ turnRunId: runId }); await opts.onEvent?.({ type: 'reasoning_delta', text: 'hmm' }); - controller.abort(); + controller.abort(G22_ACCEPTED_ABORT_REASON); return { ok: false as const, error: 'Request cancelled.', turnRunId: runId }; }, }, @@ -5796,13 +5909,12 @@ describe('runHarnessTurn attach handshake (plan #813 / E19)', () => { expect(result.ok).toBe(false); expect(sendAgent).not.toHaveBeenCalled(); expect(next.turnRunId).toBe('wr_live'); - expect(next.turnStatus).toBe('running'); + expect(next.turnStatus).toBe('cancelling'); expect( next.messages.some( (m) => m.role === 'system' && m.text === describeTurnEnd('stop'), ), - ).toBe(false); - expect(exp.__messages.some((m) => isTurnEndLine(m.text))).toBe(false); + ).toBe(true); expect(exp.__lifecycle()).toBe(Lifecycle.Ready); expect(exp.__messages.filter((m) => m.kind === MessageKind.User).map((m) => m.text)).toEqual([ 'hello', diff --git a/lib/harnessChat.ts b/lib/harnessChat.ts index af681b0c..deb0eea3 100644 --- a/lib/harnessChat.ts +++ b/lib/harnessChat.ts @@ -19,7 +19,7 @@ import { type ToolTraceEntry, } from './agentApi'; import { sendTurn, sendTurnStream, attachTurnStream } from './turnApi'; -import { isDetachAbort } from './detachTurn'; +import { isDetachAbort, isG22AcceptedAbort } from './detachTurn'; import { type AgentStreamEvent } from './agent/agentStream'; import { isProviderRefusalFinish, truncatedFinishError } from './agent/modelFinish'; import { @@ -1753,18 +1753,27 @@ export async function runHarnessTurn( // non-terminal EMBER. After onTurnStarted, producer SSE error / 5xx // reuses the POST give-up fold. EOF without terminal stays D18 via // durableIncomplete. - // Attach Stop/Esc: reader-only abort (D18), not G22 server cancel — - // keep `running`, no Turn ended · you stopped (adversarial #857). - // Producer cancelled SSE (`Request cancelled.` without abort) is a - // **terminal** Stop fold — clear `running` (plan #919 / source #918). - const attachOperatorStop = - attaching && fail.kind === 'stop' && opts?.signal?.aborted === true; + // Attach Stop/Esc is G22 server cancel (plan #816 / punch-list): fold + // `'cancelling'` + Turn-ended only after an accepted cancel abort + // (`G22_ACCEPTED_ABORT_REASON`). A raw abort before ack keeps `running` + // with no stop line so F5 can attach. Producer cancelled SSE + // (`Request cancelled.` without abort) is a **terminal** Stop fold — + // clear `running` (plan #919 / source #918). const attachSubscribeFail = attaching && !sawDurableStart && fail.kind !== 'stop' && fail.kind !== 'detach' && !isAttachRunGone(agentResult.ok ? undefined : agentResult.status); + // Raw abort on a live durable id before the cancel POST was accepted. + // Keep `running`, no Turn-ended line (punch-list items 1–2). + const abortBeforeAck = + fail.kind === 'stop' && + opts?.signal?.aborted === true && + !isG22AcceptedAbort(opts?.signal) && + failedSession.turnRunId !== undefined && + (failedSession.turnStatus === 'running' || + failedSession.turnStatus === 'cancelling'); // Cold-attach strip: persist the Blob suffix only when nothing was // painted (503/404 before events) so we do not LWW a user-only // transcript. Thinking-only incomplete GET must keep the stripped @@ -1790,7 +1799,7 @@ export async function runHarnessTurn( : agentResult.error || 'Unable to attach to run stream.' ).trim(); failedSession = paintSubscribeFail(bridge, failedSession, line); - } else if (fail.kind !== 'detach' && !attachOperatorStop) { + } else if (fail.kind !== 'detach' && !abortBeforeAck) { failedSession = pushTurnEnd(bridge, failedSession, fail.kind, fail.detail); } // Phase 2 (#465): a cancel/timeout/hard-error turn still persists the last @@ -1845,10 +1854,10 @@ export async function runHarnessTurn( // onTurnStarted, producer SSE error reuses the POST give-up fold // (clear `running`). Attach 404 (run gone) falls through and clears // so C15 does not 409 a dead id. - // Attach Stop/Esc (adversarial #857): same keep-running as detach — - // abort closes this reader only (D18); G22 owns server cancel. POST - // Stop still clears (this branch is attach-only). - if (fail.kind === 'detach' || attachSubscribeFail || attachOperatorStop) { + // Attach Stop/Esc is G22 (plan #816 / punch-list): keep-running on + // abort-before-ack. Accepted cancel (`G22_ACCEPTED_ABORT_REASON`) falls + // through to the cancelling / Turn-ended fold. + if (fail.kind === 'detach' || attachSubscribeFail || abortBeforeAck) { const id = agentResult.turnRunId ?? (failedSession.turnStatus === 'running' @@ -1863,13 +1872,34 @@ export async function runHarnessTurn( } } else if ( agentResult.turnRunId !== undefined || - (fail.kind === 'stop' && failedSession.turnStatus === 'running') + (fail.kind === 'stop' && + // Enter the fold for a this-turn live stop (`running`) OR a legacy + // stop with no durable id at all (fold `completed`). A leftover + // TERMINAL id (`completed`/`cancelling`) from a PRIOR turn is NOT + // this turn's — pre-headers Stop must not clear it (adversarial + // #844), so skip the fold and leave it as-is. + (failedSession.turnStatus === 'running' || + failedSession.turnRunId === undefined)) ) { - failedSession = { - ...failedSession, - turnRunId: undefined, - turnStatus: 'completed', - }; + // Plan #816 (G22) — fold `'cancelling'` only after the cancel POST + // was accepted (host aborts with `G22_ACCEPTED_ABORT_REASON`). A raw + // abort before ack keeps `running` (handled above). Producer + // `Request cancelled.` with no abort still clears the id (the run is + // already terminal). The old `turnRunId: undefined` + `completed` + // fold survives for the legacy `/api/agent` path (no live run id). + const keepCancelling = + fail.kind === 'stop' && + isG22AcceptedAbort(opts?.signal) && + (failedSession.turnStatus === 'running' || + failedSession.turnStatus === 'cancelling') && + failedSession.turnRunId !== undefined; + failedSession = keepCancelling + ? { ...failedSession, turnStatus: 'cancelling' } + : { + ...failedSession, + turnRunId: undefined, + turnStatus: 'completed', + }; } lastUiKind = attachSubscribeFail || @@ -1906,7 +1936,7 @@ export async function runHarnessTurn( // on Error (never consumes the queue head; Continue inserted at head when // non-empty) unless this was an operator Stop, which stays Ready (queue // untouched, drains only on a later success). - setFailLifecycle(bridge, attachSubscribeFail || attachOperatorStop ? 'detach' : fail.kind); + setFailLifecycle(bridge, attachSubscribeFail ? 'detach' : fail.kind); return { result: { ok: false, diff --git a/lib/sessionCloudCaps.test.ts b/lib/sessionCloudCaps.test.ts index deccebc7..0cd2038a 100644 --- a/lib/sessionCloudCaps.test.ts +++ b/lib/sessionCloudCaps.test.ts @@ -18,6 +18,7 @@ import { TURN_STREAM_CURSOR_MAX, TURN_STREAM_STATUS_POLL_MS, TURN_START_MIN_INTERVAL_MS, + TURN_CANCEL_MIN_INTERVAL_MS, TURN_STATUS_MAX_BYTES, TURN_STATUS_VALUES, TURN_WALL_CLOCK_MAX_MS, @@ -342,6 +343,13 @@ describe('TURN_STREAM_STATUS_POLL_MS (start/attach status poll)', () => { }); }); +describe('TURN_CANCEL_MIN_INTERVAL_MS (plan #816 G22 server cancel)', () => { + it('is a NEW 1000 ms cap matching TURN_START_MIN_INTERVAL_MS', () => { + expect(TURN_CANCEL_MIN_INTERVAL_MS).toBe(1000); + expect(TURN_CANCEL_MIN_INTERVAL_MS).toBe(TURN_START_MIN_INTERVAL_MS); + }); +}); + // Plan #923 (backend-agents): hard 1-hour wall-clock cap on a durable turn. // `TURN_WALL_CLOCK_MAX_MS` is a NEW cap whose value is the human-authorized // product lock (1h). The two `_TTL_`/`_PROBE_EVERY_` seams are NEW cache/probe diff --git a/lib/sessionCloudCaps.ts b/lib/sessionCloudCaps.ts index 3d280bdf..2da31574 100644 --- a/lib/sessionCloudCaps.ts +++ b/lib/sessionCloudCaps.ts @@ -374,6 +374,21 @@ export function sanitizeTurnStreamCursor(value: unknown): number | undefined { */ export const TURN_START_MIN_INTERVAL_MS = 1000; +/** + * Min-interval between accepted cancels of the **same** run on + * `POST /api/turns/:runId/cancel` (plan #816, backend-agents G22). A + * per-process `Map` keyed by `sessionId:runId` advances ONLY + * on an **accepted** cancel — a terminal 409 / ownership 404 / store-or-cancel + * 503 never burns the window. Same Map+boundedSet shape as the C15 start + * guard: a **soft** abuse guard (survives one Vercel Function invocation), + * not a durable rate limit; it bounds `getRun`+PATCH write amplification from + * a hostile repeat-Stop client on one run. Key includes `runId` so an accepted + * cancel of wr_1 cannot 429 Stop on wr_2 (adversarial-review #927 pass 8). + * **NEW generous cap**: 1 second matches `TURN_START_MIN_INTERVAL_MS`. No + * existing cap value changed → **no human gate**. + */ +export const TURN_CANCEL_MIN_INTERVAL_MS = 1000; + /** * How often a start/attach SSE wrapper re-reads `getRun().status` while the * client readable is open. `status` is a snapshot (same as the live-only 409 diff --git a/lib/turnApi.test.ts b/lib/turnApi.test.ts index fb0afe40..9f06ed64 100644 --- a/lib/turnApi.test.ts +++ b/lib/turnApi.test.ts @@ -6,7 +6,7 @@ * `sessionId`/`personaId`/`cwd` on the body, JSON 4xx, SSE success + failure. */ import { afterEach, describe, expect, it, vi } from 'vitest'; -import { attachTurnStream, sendTurn, sendTurnStream } from './turnApi'; +import { attachTurnStream, cancelTurn, sendTurn, sendTurnStream } from './turnApi'; function sseResponse(chunks: string[], header?: { 'x-workflow-run-id': string }): Response { const body = new ReadableStream({ @@ -587,3 +587,117 @@ describe('attachTurnStream (GET attach — plan #813 E19)', () => { expect(fetchMock).not.toHaveBeenCalled(); }); }); + +describe('cancelTurn (POST cancel — plan #816 G22)', () => { + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it('POSTs /api/turns/:runId/cancel?sessionId= → accepted on 200', async () => { + const fetchMock = vi.fn(async (url: string, init?: RequestInit) => { + expect(init?.method).toBe('POST'); + expect(init?.keepalive).toBe(true); + expect(String(url)).toBe('/api/turns/wr_live/cancel?sessionId=s_1'); + return Response.json( + { runId: 'wr_live', turnStatus: 'cancelling' }, + { status: 200, headers: { 'Content-Type': 'application/json' } }, + ); + }); + vi.stubGlobal('fetch', fetchMock); + const result = await cancelTurn('wr_live', { sessionId: 's_1' }); + expect(result).toEqual({ kind: 'accepted' }); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it('404 → gone (run expired / ownership mismatch)', async () => { + vi.stubGlobal( + 'fetch', + vi.fn(async () => + Response.json( + { error: 'Run not found: wr_gone' }, + { status: 404, headers: { 'Content-Type': 'application/json' } }, + ), + ), + ); + const result = await cancelTurn('wr_gone', { sessionId: 's_1' }); + expect(result).toEqual({ kind: 'gone' }); + }); + + it('409 → terminal with the run status from the body', async () => { + vi.stubGlobal( + 'fetch', + vi.fn(async () => + Response.json( + { runId: 'wr_done', status: 'completed' }, + { status: 409, headers: { 'Content-Type': 'application/json' } }, + ), + ), + ); + const result = await cancelTurn('wr_done', { sessionId: 's_1' }); + expect(result).toEqual({ kind: 'terminal', status: 'completed' }); + }); + + it('409 with a non-JSON body → terminal with the generic status', async () => { + vi.stubGlobal( + 'fetch', + vi.fn(async () => new Response('conflict', { status: 409 })), + ); + const result = await cancelTurn('wr_done', { sessionId: 's_1' }); + expect(result).toEqual({ kind: 'terminal', status: 'terminal' }); + }); + + it('503 → failed with the server error string', async () => { + vi.stubGlobal( + 'fetch', + vi.fn(async () => + Response.json( + { error: 'Unable to cancel run (fail closed): lost race' }, + { status: 503, headers: { 'Content-Type': 'application/json' } }, + ), + ), + ); + const result = await cancelTurn('wr_live', { sessionId: 's_1' }); + expect(result).toEqual({ + kind: 'failed', + status: 503, + error: 'Unable to cancel run (fail closed): lost race', + }); + }); + + it('429 → failed (soft guard) with status', async () => { + vi.stubGlobal( + 'fetch', + vi.fn(async () => + Response.json( + { error: 'Too many cancel requests. Please wait before cancelling again.' }, + { status: 429, headers: { 'Content-Type': 'application/json' } }, + ), + ), + ); + const result = await cancelTurn('wr_live', { sessionId: 's_1' }); + expect(result.kind).toBe('failed'); + if (result.kind === 'failed') expect(result.status).toBe(429); + }); + + it('network throw → failed (never throws)', async () => { + vi.stubGlobal( + 'fetch', + vi.fn(async () => { + throw new TypeError('Failed to fetch'); + }), + ); + const result = await cancelTurn('wr_live', { sessionId: 's_1' }); + expect(result.kind).toBe('failed'); + if (result.kind === 'failed') expect(result.error).toBe('Failed to fetch'); + }); + + it('invalid runId / sessionId fail closed before fetch', async () => { + const fetchMock = vi.fn(); + vi.stubGlobal('fetch', fetchMock); + const badRun = await cancelTurn('bad:id!', { sessionId: 's_1' }); + expect(badRun).toEqual({ kind: 'failed', status: 400, error: 'Invalid runId' }); + const badSession = await cancelTurn('wr_1', { sessionId: 'not opaque!' }); + expect(badSession).toEqual({ kind: 'failed', status: 400, error: 'Invalid sessionId.' }); + expect(fetchMock).not.toHaveBeenCalled(); + }); +}); diff --git a/lib/turnApi.ts b/lib/turnApi.ts index 25a46b0d..0eae329a 100644 --- a/lib/turnApi.ts +++ b/lib/turnApi.ts @@ -1,6 +1,7 @@ /** * Plan #811 (D17) — host client for POST /api/turns (durable-turn transport). * Plan #813 (E19) — GET attach client `attachTurnStream`. + * Plan #816 (G22) — POST cancel client `cancelTurn`. * Replaces the legacy `/api/agent` transport for production `runPrompt`. * `/api/agent` stays reachable via the legacy `sendAgent`/`sendAgentStream` * exports — tests inject those via `RunHarnessTurnOptions`. @@ -389,6 +390,79 @@ export type AttachTurnStreamOpts = { onTurnStarted?: (info: { turnRunId: string }) => void | Promise; }; +/** Plan #816 (G22) — outcome of a `POST /api/turns/:runId/cancel` call. */ +export type CancelTurnResult = + | { kind: 'accepted' } + | { kind: 'terminal'; status: string } + | { kind: 'gone' } + | { kind: 'failed'; status?: number; error: string }; + +/** + * Plan #816 (G22) — POST `/api/turns/:runId/cancel?sessionId=`. + * + * Server-cancel one live durable run. Never throws: every failure path is a + * typed result the host Stop fold maps onto its race table — + * - `accepted` (200) → fold `turnStatus: 'cancelling'` KEEPING `turnRunId`. + * - `terminal` (409, body `{status}`) → the run already ended; host clears + * `turnRunId` + folds `completed` (idempotent no-op, not a client error). + * - `gone` (404) → run expired/ownership mismatch; host clears `turnRunId` + + * folds `completed` (the orphan-unstick path). + * - `failed` (429/5xx/network/abort) → host keeps `turnRunId` + `running`, + * paints a soft note; the run continues to its own terminal (never a + * silent fake cancel). + */ +export async function cancelTurn( + runId: string, + opts: { sessionId: string; keepalive?: boolean }, +): Promise { + const cleanRunId = sanitizeTurnRunId(runId); + if (cleanRunId === undefined) { + return { kind: 'failed', status: 400, error: 'Invalid runId' }; + } + if (!isRedisSafeOpaqueId(opts.sessionId)) { + return { kind: 'failed', status: 400, error: 'Invalid sessionId.' }; + } + + const params = new URLSearchParams(); + params.set('sessionId', opts.sessionId); + const path = `/api/turns/${encodeURIComponent(cleanRunId)}/cancel?${params.toString()}`; + + let res: Response; + try { + // keepalive: unload must not drop the POST (Stop then F5 before 200). + res = await fetch(path, { method: 'POST', keepalive: opts.keepalive !== false }); + } catch (err) { + return { + kind: 'failed', + error: err instanceof Error ? err.message : 'Network request failed.', + }; + } + + if (res.status === 200) return { kind: 'accepted' }; + if (res.status === 404) return { kind: 'gone' }; + if (res.status === 409) { + let status = 'terminal'; + try { + const data = (await res.json()) as { status?: unknown }; + if (typeof data.status === 'string' && data.status) status = data.status; + } catch { + // Body parse failure keeps the generic 'terminal' status. + } + return { kind: 'terminal', status }; + } + + let error = `Request failed (${res.status}).`; + try { + const data = (await res.json()) as { error?: unknown }; + if (typeof data.error === 'string' && data.error.trim()) { + error = data.error; + } + } catch { + // Non-JSON error body keeps the status-derived message. + } + return { kind: 'failed', status: res.status, error }; +} + /** * Plan #813 (E19) — GET `/api/turns/:runId/stream?sessionId=&startIndex=`. * Reuses `readAgentStream`. Abort closes **this reader only** (D18: never a diff --git a/lib/turnAttach.test.ts b/lib/turnAttach.test.ts index cabfb667..ef837e51 100644 --- a/lib/turnAttach.test.ts +++ b/lib/turnAttach.test.ts @@ -32,7 +32,7 @@ import { import { TURN_STREAM_CURSOR_MAX } from './sessionCloudCaps'; describe('decideAttachClass', () => { - it('none when not running / no run id', () => { + it('none when not live / no run id', () => { expect( decideAttachClass({ turnStatus: 'completed', @@ -48,6 +48,25 @@ describe('decideAttachClass', () => { ).toEqual({ kind: 'none' }); }); + it('cancelling with a run id is live (F5 after accepted G22 cancel)', () => { + expect( + decideAttachClass({ + turnRunId: 'wr_1', + turnStatus: 'cancelling', + envelopeCursor: 4, + heapApplied: null, + }), + ).toEqual({ kind: 'cold', startIndex: 0 }); + expect( + decideAttachClass({ + turnRunId: 'wr_1', + turnStatus: 'cancelling', + envelopeCursor: 4, + heapApplied: { runId: 'wr_1', count: 4 }, + }), + ).toEqual({ kind: 'hot', startIndex: 4 }); + }); + it('F5 / boot (no heap applied) is cold at 0 even when envelope C is large', () => { expect( decideAttachClass({ @@ -128,6 +147,14 @@ describe('coldAttachFromSnapshot', () => { dedup: true, }); }); + + it('cold spec when restored snapshot is cancelling (F5 + live getRun)', () => { + expect(coldAttachFromSnapshot({ turnStatus: 'cancelling', turnRunId: 'wr_1' })).toEqual({ + runId: 'wr_1', + startIndex: 0, + dedup: true, + }); + }); }); describe('isAttachRunGone (adversarial #857)', () => { @@ -246,7 +273,7 @@ describe('shouldKickHotResume (plan #919)', () => { attaching: false, streamOpened: true, operatorStop: false, - turnStatus: 'completed', + turnStatus: 'cancelling', turnRunId: 'wr_1', }), ).toBe(false); @@ -360,12 +387,12 @@ describe('harnessChat attach hydrate source-lock (adversarial #857)', () => { expect(src).not.toMatch(/if \(coldBackup\) \{\s*opts\?\.onSessionPatch\?\(\{ \.\.\.s, messages: coldBackup \}\)/); }); - it('attach Stop keep-running is D18-shaped (adversarial #857)', () => { - expect(src).toContain( - 'attaching && fail.kind === \'stop\' && opts?.signal?.aborted === true', - ); - expect(src).toContain('fail.kind === \'detach\' || attachSubscribeFail || attachOperatorStop'); - expect(src).toContain('fail.kind !== \'detach\' && !attachOperatorStop'); + it('attach Stop is G22 cancelling + Turn-ended only after accepted cancel abort', () => { + expect(src).not.toContain('attachOperatorStop'); + expect(src).toContain('fail.kind === \'detach\' || attachSubscribeFail || abortBeforeAck'); + expect(src).toContain('} else if (fail.kind !== \'detach\' && !abortBeforeAck) {'); + expect(src).toContain('failedSession = pushTurnEnd(bridge, failedSession, fail.kind, fail.detail)'); + expect(src).toContain('isG22AcceptedAbort'); }); it('ATTACH_FOLLOW_UP_NOTE is not a Turn-ended line (canvas System + TEAL host mirror)', () => { @@ -510,6 +537,29 @@ describe('shouldRepostAttachFollowUp (adversarial #857 remapped prompt)', () => ).toBe(false); }); + it('does not re-POST on operator Stop even when fold is cancelling (G22 / adversarial-review #927)', () => { + expect( + shouldRepostAttachFollowUp({ + sendWhileRunning: true, + turnStatus: 'cancelling', + operatorStop: true, + }), + ).toBe(false); + expect( + shouldRepostAttachFollowUp({ + sendWhileRunning: true, + turnStatus: 'cancelling', + }), + ).toBe(false); + expect( + shouldRepostAttachFollowUp({ + sendWhileRunning: true, + turnStatus: 'completed', + operatorStop: true, + }), + ).toBe(false); + }); + it('is mutually exclusive with the follow-up note', () => { const running = { sendWhileRunning: true, resultOk: false, turnStatus: 'running' as const }; const done = { sendWhileRunning: true, resultOk: true, turnStatus: 'completed' as const }; diff --git a/lib/turnAttach.ts b/lib/turnAttach.ts index a75a76af..d0896e28 100644 --- a/lib/turnAttach.ts +++ b/lib/turnAttach.ts @@ -18,6 +18,11 @@ import { export type HeapApplied = { runId: string; count: number }; +/** Envelope liveness that may still have a live Workflow run (F5 attach). */ +export function isLiveTurnStatus(status?: TurnStatus): boolean { + return status === 'running' || status === 'cancelling'; +} + /** * Host + in-canvas note when operator Send is remapped to attach * (adversarial #857). Not a Turn-ended line; not EMBER. Composer text was @@ -82,14 +87,20 @@ export function shouldPaintAttachFollowUpDetachNote(input: { /** * After Send-while-running attach returns: POST the remapped prompt when the * run is no longer live (`done` / 404 / post-start SSE error). C15 409 no - * longer applies. Never while `running` (EOF / 503 — note path). Never for - * kickColdAttach / hot-resume (empty prompt, `sendWhileRunning` false). + * longer applies. Never while `running` or `'cancelling'` (still live). Never + * for kickColdAttach / hot-resume (empty prompt, `sendWhileRunning` false). + * Never on operator Stop. */ export function shouldRepostAttachFollowUp(input: { sendWhileRunning: boolean; turnStatus?: TurnStatus; + operatorStop?: boolean; }): boolean { - return input.sendWhileRunning && input.turnStatus !== 'running'; + return ( + input.sendWhileRunning && + !isLiveTurnStatus(input.turnStatus) && + !input.operatorStop + ); } /** @@ -115,6 +126,8 @@ export type AttachDecision = * Classify hot resume vs cold attach by **this heap's ring**, not envelope `C`. * * - No live run → none (do not attach completed sessions on boot). + * - `'cancelling'` is still live until the workflow actually ends (accepted + * G22 cancel, or a leftover marker) — F5 must cold-attach so Stop returns. * - Heap has not applied this `turnRunId` (F5 / login / new tab / switch) → cold * at `startIndex=0`, even if envelope `C` is large. * - Same-heap live ring → hot resume at **heap-applied** count. Envelope `C` is @@ -127,7 +140,7 @@ export function decideAttachClass(input: { envelopeCursor?: number; heapApplied: HeapApplied | null; }): AttachDecision { - if (input.turnStatus !== 'running' || !input.turnRunId) { + if (!isLiveTurnStatus(input.turnStatus) || !input.turnRunId) { return { kind: 'none' }; } const heap = input.heapApplied; @@ -143,15 +156,16 @@ export function decideAttachClass(input: { export type ColdAttachSpec = { runId: string; startIndex: 0; dedup: true }; /** - * Predicate of `kickColdAttach`: only a restored snapshot that is still - * `running` with a `turnRunId` attaches. Envelope liveness is **not** consulted - * here (F5-stale-local miss). + * Predicate of `kickColdAttach`: a restored snapshot that is still + * `running` or `cancelling` with a `turnRunId` attaches. Envelope liveness is + * **not** consulted here (F5-stale-local miss). `'cancelling'` stays attachable + * so F5 after an accepted cancel (run still winding down) restores Busy + Stop. */ export function coldAttachFromSnapshot(s: { turnStatus?: TurnStatus; turnRunId?: string; }): ColdAttachSpec | null { - if (s.turnStatus !== 'running' || !s.turnRunId) return null; + if (!isLiveTurnStatus(s.turnStatus) || !s.turnRunId) return null; return { runId: s.turnRunId, startIndex: 0, dedup: true }; } @@ -214,7 +228,8 @@ export type SendAttachSpec = | { kind: 'cold'; runId: string; startIndex: 0; dedup: true }; /** - * Classify operator Send while a durable run is still `running`. + * Classify operator Send while a durable run is still live (`running` or + * `'cancelling'`). * * Never POST (C15 409 mixes Turn ended + Error with keep-running). Heap with * applied frames (`count > 0`) → hot resume at `C`. Count 0 or a different / diff --git a/middleware.test.ts b/middleware.test.ts index 54f02b4f..08441af5 100644 --- a/middleware.test.ts +++ b/middleware.test.ts @@ -321,6 +321,26 @@ describe('middleware auth gate', () => { vi.resetModules(); const { config } = await import('./middleware'); expect(config.matcher).toContain('/api/turns'); + expect(config.matcher).toContain('/api/turns/:path*'); + }); + + it('401 JSON on unauth POST /api/turns/:runId/cancel when tenancy on', async () => { + process.env.DATABASE_URL = 'postgres://x'; + process.env.AUTH_SECRET = 'test-secret-value-for-jwt-middleware!!'; + process.env.CREDENTIALS_ENCRYPTION_KEY = 'key-material'; + vi.resetModules(); + vi.doMock('next-auth/jwt', () => ({ + getToken: vi.fn(async () => null), + })); + const { middleware } = await loadMw(); + const res = await middleware( + new Request('http://localhost/api/turns/wr_live/cancel?sessionId=s1', { + method: 'POST', + }) as never, + ); + expect(res.status).toBe(401); + const body = (await res.json()) as { error: string }; + expect(body.error).toBe(AUTH_REQUIRED_ERROR); }); it('allows GET /api/skills when JWT sub present', async () => { diff --git a/middleware.ts b/middleware.ts index 29d0c5bf..c7d894d7 100644 --- a/middleware.ts +++ b/middleware.ts @@ -16,10 +16,13 @@ function isApiProtected(pathname: string): boolean { if (pathname === '/api/settings/personas' || pathname.startsWith('/api/settings/personas/')) { return true; } + // durable-turn start + nested stream/cancel (plan #816 dual gate; adversarial-review #927). + if (pathname === '/api/turns' || pathname.startsWith('/api/turns/')) { + return true; + } return ( pathname === '/api/chat' || pathname === '/api/agent' || - pathname === '/api/turns' || pathname === '/api/models' || pathname === '/api/sandboxes' || pathname === '/api/personas' || @@ -111,6 +114,7 @@ export const config = { '/api/chat', '/api/agent', '/api/turns', + '/api/turns/:path*', '/api/models', '/api/sandboxes', '/api/personas',