fix(lifecycle): admit same-session reconnect turns

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.

src/server/lifecycle.ts

@@ -41,7 +41,7 @@ export interface ActiveTurnLease extends AdmissionLease {
 }
 const activeTurns = new Map<AbortController, ActiveTurnLease>();
 const admittedTurns = new Set<ActiveTurnLease>();
-const activeSessionLanes = new Set<string>();
+const activeSessionLaneRefCounts = new Map<string, number>();
 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<AbortController>();
   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 {

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,41 +150,80 @@ 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<void>(resolve => { markUpstreamStarted = resolve; });
+    let finishUpstream!: () => void;
+    const upstreamResponse = new Promise<Response>(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" });
     const held = tryAdmitTurn(sessionLaneIdFromRequest(headers));
     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);