diff --git a/devlog/_plan/260826_session_lane_bounds/020_reconnect_followup.md b/devlog/_plan/260826_session_lane_bounds/020_reconnect_followup.md new file mode 100644 index 0000000000..25ef96e0e7 --- /dev/null +++ b/devlog/_plan/260826_session_lane_bounds/020_reconnect_followup.md @@ -0,0 +1,97 @@ +# Same-session reconnect follow-up + +## Loop spec + +- Archetype: repair. +- Trigger: a Codex retry for a trivial prompt receives local HTTP 503 + `active turns capacity reached` while the previous request for the same logical + session is still settling. +- Goal: allow overlapping/reconnecting requests from one identified session without + removing the 64-unique-session memory bound or the 256-active-turn process bound. +- Non-goals: no queue, scheduler, timeout change, retry loop, error-envelope change, + Lab dependency, or broader lifecycle refactor. +- Verifier: `bun test tests/session-lane-recall-harness.test.ts` exercises the same-lane + HTTP boundary and unique-lane cap; baseline exit 0 and it imports the changed lifecycle + module directly. `bun run typecheck` baseline exit 0 and `tsconfig.json` includes `src`; + Bun independently compiles the changed test file in the focused test command. +- Stop condition: same-lane overlap is admitted, the lane stays retained until its last + lease releases, the 65th unique lane remains rejected, and all required PR gates pass. +- Memory artifact: this document and the focused regression test. +- Expected terminal outcomes: DONE with a focused PR, or BLOCKED if ref-counted cleanup + cannot preserve the existing unique-lane memory oracle. +- Escalation: the main agent keeps the tightly coupled lifecycle/test change; an + independent reviewer audits the plan and final diff. Any scope delegated later requires + a P-phase amendment. + +## Root cause and rejected hypotheses + +- H1, genuine 256-turn exhaustion: rejected for the reported event because a fresh runtime + snapshot showed `activeTurnCount: 0`, and the error is also emitted before the global + gate when a duplicate lane is present. +- H2, permanent active-turn leak: not supported by the post-event snapshot because active + turns returned to zero. The failure is transient while an earlier stream settles. +- H3, same-session admission collision: supported by `tryAdmitTurn`, which returns `null` + whenever `activeSessionLanes.has(lane)` and maps that condition to the exact 503 shown by + Codex. The existing HTTP test codifies that rejection. + +## Diff-level plan + +### `src/server/lifecycle.ts` + +- Replace the unique-lane `Set` with a fixed-key reference-count map. +- Reject only a previously unseen lane when 64 unique lanes are already active. +- Increment a lane's lease count on admission; decrement on idempotent lease release and + delete only when the last same-lane lease settles. +- Keep `sessionLaneMetrics().active` and retained-byte accounting defined as unique lanes, + preserving the #820 memory envelope. +- Keep `sessionLaneMetrics().admitted` defined as first admission of a previously inactive + unique lane; a same-lane overlap increments only the lane's lease count. Define + `rejected` narrowly as a new unique lane refused at the 64-lane cap. This intentionally + removes same-lane reconnects from rejection accounting; no new metric field is added + because the metric is test-only and no production consumer currently exposes it. + +### `tests/session-lane-recall-harness.test.ts` + +- Hold a real HTTP request at the mocked upstream boundary, release the manually held + same-lane lease, and prove the HTTP lease keeps that lane active until its response + settles. Keep a separate invalid-JSON assertion for the stable downstream 400 envelope. +- Add direct lease assertions proving two same-lane leases share one retained lane and that + releasing either lease first cannot prematurely free it. +- Add a cross-gate assertion proving 256 active turns on one lane still make the next + same-lane request fail at the process-wide turn gate. +- Keep the 65th-unique-lane rejection and parent-plus-child lane-isolation coverage. + +### This record + +- Record the observed production symptom, chosen correction, rejected alternatives, and + fresh verification evidence. Move the owning unit to `_fin` only if the original #820 + unit is otherwise terminal; this follow-up does not silently close unrelated work. + +## Acceptance criteria and activation + +| Condition | Activation | Observable proof | +| --- | --- | --- | +| Same-session reconnect | Hold one lease, POST `/v1/responses` with the same `session_id`, and pause the mocked upstream | Releasing the manual lease leaves one active lane until the HTTP response settles; the request returns 200, not admission 503 | +| Same-lane cleanup ordering | Admit two leases for one lane and release them in both orders | Unique active lanes stay 1 until the final release, then become 0 | +| Unique-lane cap | Hold 64 distinct lane leases and request a 65th | 65th returns null and rejection metric increments | +| Global cap with same lane | Hold 256 leases for one lane, then request that lane again | 257th total turn is rejected by the process-wide gate while unique-lane metrics stay bounded | + +## Alternatives rejected + +- Delete all session-lane accounting: would undo #820's fixed 64-unique-session memory + envelope. +- Queue same-session retries: adds scheduler state and latency outside this repair. +- Increase the lane cap: does not fix collisions on an already-active logical session. +- Hide the 503 with client retries: preserves the faulty server admission decision and + creates avoidable reconnect churn. + +## Verification policy + +- Final verification runs only on the `lidge-ai` SSH host in an isolated checkout. Local + verification was stopped after the operator clarified this repository's execution policy; + local results are not PR evidence. +- The remote gate is the focused session/lifecycle slice, full typecheck, full test suite, + and privacy scan. The PR records only the remote commands and results. +- Test change classification: required behavior regressions only. No assertion, threshold, + coverage rule, or test is skipped or deleted; the old same-session 503 assertion is + replaced because that response is the reported defect. diff --git a/src/server/lifecycle.ts b/src/server/lifecycle.ts index eb26c46079..b81d9bf46f 100644 --- a/src/server/lifecycle.ts +++ b/src/server/lifecycle.ts @@ -41,7 +41,7 @@ export interface ActiveTurnLease extends AdmissionLease { } const activeTurns = new Map(); const admittedTurns = new Set(); -const activeSessionLanes = new Set(); +const activeSessionLaneRefCounts = new Map(); let sessionLanePeak = 0; let sessionLaneAdmitted = 0; let sessionLaneRejected = 0; @@ -156,7 +156,7 @@ export function resetLifecycleDrainStateForTests(): void { temporaryDrainOwners.clear(); nativeMainDrainOwners.clear(); nativeMainTurns.clear(); - activeSessionLanes.clear(); + activeSessionLaneRefCounts.clear(); sessionLanePeak = 0; sessionLaneAdmitted = 0; sessionLaneRejected = 0; @@ -173,16 +173,21 @@ export function tryAdmitTurn(sessionLaneId?: string): ActiveTurnLease | null { const opaqueSessionLaneId = sessionLaneId ? createHash("sha256").update(sessionLaneId).digest("hex").slice(0, SESSION_LANE_ID_BYTES) : undefined; - if (opaqueSessionLaneId && (activeSessionLanes.has(opaqueSessionLaneId) || activeSessionLanes.size >= MAX_ACTIVE_SESSION_LANES)) { + const sessionLaneRefCount = opaqueSessionLaneId + ? activeSessionLaneRefCounts.get(opaqueSessionLaneId) ?? 0 + : 0; + if (opaqueSessionLaneId && sessionLaneRefCount === 0 && activeSessionLaneRefCounts.size >= MAX_ACTIVE_SESSION_LANES) { sessionLaneRejected += 1; return null; } const gateLease = turnGate.tryAcquire(); if (!gateLease) return null; if (opaqueSessionLaneId) { - activeSessionLanes.add(opaqueSessionLaneId); - sessionLaneAdmitted += 1; - sessionLanePeak = Math.max(sessionLanePeak, activeSessionLanes.size); + activeSessionLaneRefCounts.set(opaqueSessionLaneId, sessionLaneRefCount + 1); + if (sessionLaneRefCount === 0) { + sessionLaneAdmitted += 1; + sessionLanePeak = Math.max(sessionLanePeak, activeSessionLaneRefCounts.size); + } } const controllers = new Set(); let active = true; @@ -234,7 +239,13 @@ export function tryAdmitTurn(sessionLaneId?: string): ActiveTurnLease | null { } controllers.clear(); nativeMainTurns.delete(lease); - if (opaqueSessionLaneId) activeSessionLanes.delete(opaqueSessionLaneId); + if (opaqueSessionLaneId) { + const currentRefCount = activeSessionLaneRefCounts.get(opaqueSessionLaneId); + if (currentRefCount === 1) activeSessionLaneRefCounts.delete(opaqueSessionLaneId); + else if (currentRefCount && currentRefCount > 1) { + activeSessionLaneRefCounts.set(opaqueSessionLaneId, currentRefCount - 1); + } + } gateLease.release(); }, }; @@ -294,11 +305,11 @@ export interface SessionLaneMetrics { } export function sessionLaneMetrics(): SessionLaneMetrics { return { - active: activeSessionLanes.size, + active: activeSessionLaneRefCounts.size, peak: sessionLanePeak, admitted: sessionLaneAdmitted, rejected: sessionLaneRejected, - retainedBytes: activeSessionLanes.size * SESSION_LANE_ID_BYTES, + retainedBytes: activeSessionLaneRefCounts.size * SESSION_LANE_ID_BYTES, }; } export function getNativeMainProfileRequestCount(): number { diff --git a/tests/session-lane-recall-harness.test.ts b/tests/session-lane-recall-harness.test.ts index a2bc5044a9..f01b3c3cb9 100644 --- a/tests/session-lane-recall-harness.test.ts +++ b/tests/session-lane-recall-harness.test.ts @@ -5,6 +5,7 @@ import { join } from "node:path"; import { createOpenAIChatAdapter as createOpenAIChatAdapterProduction } from "../src/adapters/openai-chat"; import { saveConfig } from "../src/config"; import { + MAX_ACTIVE_TURNS, MAX_ACTIVE_SESSION_LANES, SESSION_LANE_ID_BYTES, resetLifecycleDrainStateForTests, @@ -103,9 +104,19 @@ async function runRecallWave(sessionCount: 32 | 64) { expect(sessionLaneMetrics()).toMatchObject({ active: sessionCount, peak: sessionCount, + admitted: sessionCount, + rejected: 0, retainedBytes: sessionCount * SESSION_LANE_ID_BYTES, }); - expect(tryAdmitTurn("logical-session-0")).toBeNull(); + const overlappingLease = tryAdmitTurn("logical-session-0"); + expect(overlappingLease).not.toBeNull(); + expect(sessionLaneMetrics()).toMatchObject({ + active: sessionCount, + admitted: sessionCount, + rejected: 0, + retainedBytes: sessionCount * SESSION_LANE_ID_BYTES, + }); + overlappingLease?.release(); const firstCalls = await Promise.all(Array.from({ length: sessionCount }, (_, session) => parseCalls(session, 1))); for (const lease of leases) lease.release(); @@ -139,17 +150,44 @@ async function runRecallWave(sessionCount: 32 | 64) { } describe("#820 concurrent tool-recall session harness", () => { - test("the HTTP boundary rejects an overlapping recall on the same logical session", async () => { + test("the HTTP boundary admits a reconnect while the same logical session is settling", async () => { resetLifecycleDrainStateForTests(); const previousHome = process.env.OPENCODEX_HOME; + const originalFetch = globalThis.fetch; const home = mkdtempSync(join(tmpdir(), "ocx-session-lane-")); process.env.OPENCODEX_HOME = home; + let markUpstreamStarted!: () => void; + const upstreamStarted = new Promise(resolve => { markUpstreamStarted = resolve; }); + let finishUpstream!: () => void; + const upstreamResponse = new Promise(resolve => { + finishUpstream = () => resolve(Response.json({ + id: "resp_reconnect", + object: "response", + status: "completed", + model: "test-model", + output: [], + usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 }, + })); + }); + globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const url = input instanceof Request ? input.url : String(input); + if (url === "https://reconnect.example.test/v1/responses") { + markUpstreamStarted(); + return upstreamResponse; + } + return originalFetch(input, init); + }) as typeof fetch; saveConfig({ port: 0, hostname: "127.0.0.1", - defaultProvider: "openai", + defaultProvider: "reconnect", providers: { - openai: { adapter: "openai-responses", baseUrl: "https://api.openai.com/v1", authMode: "forward" }, + reconnect: { + adapter: "openai-responses", + baseUrl: "https://reconnect.example.test/v1", + authMode: "key", + apiKey: "test-key", + }, }, } as OcxConfig); const headers = new Headers({ "content-type": "application/json", session_id: "recall-session" }); @@ -157,23 +195,35 @@ describe("#820 concurrent tool-recall session harness", () => { const server = startServer(0); try { expect(held).not.toBeNull(); - const overlapping = await fetch(new URL("/v1/responses", server.url), { + const overlappingResponse = originalFetch(new URL("/v1/responses", server.url), { method: "POST", headers, - body: "not-json", + body: JSON.stringify({ model: "reconnect/test-model", input: "hello", stream: false }), }); - expect(overlapping.status).toBe(503); - expect(await overlapping.json()).toMatchObject({ error: { code: "server_busy" } }); + await upstreamStarted; + expect(sessionLaneMetrics()).toMatchObject({ active: 1, admitted: 1, rejected: 0 }); held?.release(); - const afterRelease = await fetch(new URL("/v1/responses", server.url), { + expect(sessionLaneMetrics()).toMatchObject({ active: 1, retainedBytes: SESSION_LANE_ID_BYTES }); + finishUpstream(); + const overlapping = await overlappingResponse; + expect(overlapping.status).toBe(200); + await overlapping.text(); + expect(sessionLaneMetrics()).toMatchObject({ active: 0, retainedBytes: 0 }); + + const invalid = await originalFetch(new URL("/v1/responses", server.url), { method: "POST", headers, body: "not-json", }); - expect(afterRelease.status).toBe(400); + expect(invalid.status).toBe(400); + expect(await invalid.json()).toMatchObject({ + error: { type: "invalid_request_error", message: "Invalid JSON body" }, + }); } finally { + finishUpstream(); held?.release(); await server.stop(true); + globalThis.fetch = originalFetch; if (previousHome === undefined) delete process.env.OPENCODEX_HOME; else process.env.OPENCODEX_HOME = previousHome; rmSync(home, { recursive: true, force: true }); @@ -201,6 +251,46 @@ describe("#820 concurrent tool-recall session harness", () => { expect(sessionLaneMetrics().retainedBytes).toBe(0); }); + test("same-lane reconnect leases retain one lane until the final release", () => { + resetLifecycleDrainStateForTests(); + const first = tryAdmitTurn("reconnect-lane"); + const second = tryAdmitTurn("reconnect-lane"); + expect(first).not.toBeNull(); + expect(second).not.toBeNull(); + expect(sessionLaneMetrics()).toMatchObject({ + active: 1, + peak: 1, + admitted: 1, + rejected: 0, + retainedBytes: SESSION_LANE_ID_BYTES, + }); + + first?.release(); + expect(sessionLaneMetrics()).toMatchObject({ active: 1, retainedBytes: SESSION_LANE_ID_BYTES }); + second?.release(); + expect(sessionLaneMetrics()).toMatchObject({ active: 0, retainedBytes: 0 }); + + const third = tryAdmitTurn("reconnect-lane"); + const fourth = tryAdmitTurn("reconnect-lane"); + expect(third).not.toBeNull(); + expect(fourth).not.toBeNull(); + fourth?.release(); + expect(sessionLaneMetrics()).toMatchObject({ active: 1, retainedBytes: SESSION_LANE_ID_BYTES }); + third?.release(); + expect(sessionLaneMetrics()).toMatchObject({ active: 0, retainedBytes: 0 }); + }); + + test("same-lane reconnects remain bounded by the global active-turn cap", () => { + resetLifecycleDrainStateForTests(); + const leases = Array.from({ length: MAX_ACTIVE_TURNS }, () => tryAdmitTurn("global-cap-lane")); + expect(leases.every(Boolean)).toBe(true); + expect(sessionLaneMetrics()).toMatchObject({ active: 1, admitted: 1, rejected: 0 }); + expect(tryAdmitTurn("global-cap-lane")).toBeNull(); + expect(sessionLaneMetrics()).toMatchObject({ active: 1, admitted: 1, rejected: 0 }); + for (const lease of leases) lease?.release(); + expect(sessionLaneMetrics()).toMatchObject({ active: 0, retainedBytes: 0 }); + }); + /** * The regression this lane derivation exists to avoid (#820). * @@ -227,9 +317,12 @@ describe("#820 concurrent tool-recall session harness", () => { const siblingLeases = siblingLanes.map(lane => tryAdmitTurn(lane)); expect(siblingLeases.every(Boolean)).toBe(true); - // Same parent AND same child thread: one logical conversation, so the second overlapping - // turn is refused rather than admitted alongside the first. - expect(tryAdmitTurn(sessionLaneIdFromRequest(spawn("child-a")))).toBeNull(); + // Same parent AND same child thread still shares one fixed-size lane, while a reconnect + // gets its own process-wide turn lease instead of a local 503. + const overlappingSibling = tryAdmitTurn(sessionLaneIdFromRequest(spawn("child-a"))); + expect(overlappingSibling).not.toBeNull(); + expect(sessionLaneMetrics()).toMatchObject({ active: 3, admitted: 3, rejected: 0 }); + overlappingSibling?.release(); for (const lease of siblingLeases) lease?.release(); expect(sessionLaneMetrics().retainedBytes).toBe(0);