fix(coding-agent): enforce session ownership when joining an in-flight open (#1864)
* fix(coding-agent): wait for worker recovery before session reuse * fix(coding-agent): join overlapping worker starts * fix(coding-agent): enforce session ownership when joining an in-flight open Joining an opening worker via openingWorkers returned the pending worker to any client with the same session path, bypassing the ownership check that the settled path applies in reuseWorkerForCreate. Await the pending open and apply the shared owner check before returning; a rejected open still propagates its original error to joiners. * fix(coding-agent): rejoin an open registered during stale-worker reclaim The reclaimStaleWorkerRegistration await sits between the pending-open check and openingWorkers.set, so a concurrent opener registered during that await was overwritten and a second worker launched for the same saved session (regression from the #1756 reordering). Re-check the map after the reclaim await and join through the shared owner-checked path. * test(coding-agent): cover foreign-owner join at the post-reclaim re-check An intruder client joining the open registered during stale-worker reclaim must get SessionAlreadyActiveError; kills the mutant that returns the pending promise without the owner check.
1- Fixed a race where a concurrent open of a session already being opened by another client bypassed the session ownership check instead of failing with session-already-active.20 unmodified lines21222324252526272829321 unmodified linesExpand all351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460371 unmodified linesExpand all20 unmodified lines refreshWorkerSummaries(worker: WorkerFixture): Promise<void>; findSummaryInWorker(worker: WorkerFixture, selector: string): SessionSummary | undefined; createOrReuseWorker( clientId: string, command: { type: "create"; name?: string; sessionPath?: string }, command: { type: "create"; name?: string; sessionPath?: string; lifecycle?: "client_owned" }, ): Promise<WorkerFixture>; assertSupervisorSavedSessionNameAvailable(sessionPath: string, name: string): Promise<void>; assertSavedSiblingNameAvailable( siblings: Array<Record<string, unknown>>,321 unmodified linesExpand all expect(await Promise.all([first, second])).toEqual([resident, resident]); expect(launchWorker).toHaveBeenCalledOnce(); });
it("enforces session ownership when joining an in-flight open", async () => { const directory = mkdtempSync(join(tmpdir(), "prime-supervisor-pending-owner-")); tempDirs.push(directory); const sessionPath = join(directory, "session.jsonl"); let releaseLaunch!: () => void; const launchGate = new Promise<void>((resolve) => { releaseLaunch = resolve; }); const supervisor = new DaemonSupervisor(join(directory, "daemon.sock"), { defaultSessionConfig: { agentDir: directory, cwd: directory }, descriptorDir: join(directory, "workers"), }) as unknown as SupervisorInternals; const resident = worker("opened"); resident.descriptor.ownerClientId = "owner"; const launchWorker = vi.fn(async () => { await launchGate; return resident; }); Object.assign(supervisor, { launchWorker });
const create = { type: "create" as const, sessionPath, lifecycle: "client_owned" as const }; const first = supervisor.createOrReuseWorker("owner", create); const sameOwner = supervisor.createOrReuseWorker("owner", create); const otherClient = supervisor.createOrReuseWorker("intruder", create); const expectations = Promise.all([ expect(first).resolves.toBe(resident), expect(sameOwner).resolves.toBe(resident), expect(otherClient).rejects.toMatchObject({ code: "session_already_active" }), ]); releaseLaunch(); await expectations; expect(launchWorker).toHaveBeenCalledOnce(); });
it("rejoins an open registered while reclaiming a stale worker registration", async () => { const directory = mkdtempSync(join(tmpdir(), "prime-supervisor-reclaim-rejoin-")); tempDirs.push(directory); const sessionPath = join(directory, "session.jsonl"); let releaseReclaim!: () => void; const reclaimGate = new Promise<void>((resolve) => { releaseReclaim = resolve; }); const supervisor = new DaemonSupervisor(join(directory, "daemon.sock"), { defaultSessionConfig: { agentDir: directory, cwd: directory }, descriptorDir: join(directory, "workers"), }) as unknown as SupervisorInternals; const stale = worker("stale"); stale.descriptor.createCommand = { config: { cwd: directory }, sessionPath }; supervisor.workers.set(stale.descriptor.workerId, stale); const resident = worker("opened"); resident.descriptor.ownerClientId = "owner"; const launchWorker = vi.fn(async () => resident); const reclaimStaleWorkerRegistration = vi.fn(async () => { await reclaimGate; supervisor.workers.delete(stale.descriptor.workerId); return true; }); Object.assign(supervisor, { launchWorker, reclaimStaleWorkerRegistration });
const create = { type: "create" as const, sessionPath, lifecycle: "client_owned" as const }; const first = supervisor.createOrReuseWorker("owner", create); const second = supervisor.createOrReuseWorker("owner", create); const intruder = supervisor.createOrReuseWorker("intruder", create); const expectations = Promise.all([ expect(first).resolves.toBe(resident), expect(second).resolves.toBe(resident), expect(intruder).rejects.toMatchObject({ code: "session_already_active" }), ]); releaseReclaim(); await expectations; expect(launchWorker).toHaveBeenCalledOnce(); });
it("propagates an in-flight open failure to joiners", async () => { const directory = mkdtempSync(join(tmpdir(), "prime-supervisor-pending-failure-")); tempDirs.push(directory); const sessionPath = join(directory, "session.jsonl"); let releaseLaunch!: () => void; const launchGate = new Promise<void>((resolve) => { releaseLaunch = resolve; }); const supervisor = new DaemonSupervisor(join(directory, "daemon.sock"), { defaultSessionConfig: { agentDir: directory, cwd: directory }, descriptorDir: join(directory, "workers"), }) as unknown as SupervisorInternals; const launchWorker = vi.fn(async () => { await launchGate; throw new Error("launch exploded"); }); Object.assign(supervisor, { launchWorker });
const create = { type: "create" as const, sessionPath, lifecycle: "client_owned" as const }; const first = supervisor.createOrReuseWorker("owner", create); const joiner = supervisor.createOrReuseWorker("intruder", create); const expectations = Promise.all([ expect(first).rejects.toThrow("launch exploded"), expect(joiner).rejects.toThrow("launch exploded"), ]); releaseLaunch(); await expectations; });
it("uses injective structural session name reservation keys", () => { expect(sessionNameReservationKey({ name: "b:c", depth: 1, parentSessionPath: "/a" })).not.toBe( sessionNameReservationKey({ name: "c", depth: 1, parentSessionPath: "/a:b" }), );371 unmodified linesExpand all2276 unmodified linesExpand all227722782279228022812281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230329 unmodified lines2333233423352336232623272328233723382339234023413 unmodified lines2345234623472348234023412342234923502351235223534 unmodified lines23582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423853167 unmodified linesExpand all2276 unmodified linesExpand all ? canonicalSessionPath(createCommand.sessionPath) : `new:${command.id ? createCommandIdempotencyKey(clientId, command.id) : createActiveSessionId()}`; const pending = this.openingWorkers.get(key); if (pending) { return pending; return this.joinOpeningWorker(pending, ownerClientId, createCommand.sessionPath ?? key); } if (createCommand.sessionPath) { const existing = this.findWorkerBySessionFile(createCommand.sessionPath); if (existing && !(await this.reclaimStaleWorkerRegistration(existing, command.launchEnv !== undefined))) { return this.reuseWorkerForCreate(existing, ownerClientId, createCommand.sessionPath); } // The reclaim await may have let a concurrent opener register; join it instead of double-launching. const opened = this.openingWorkers.get(key); if (opened) { return this.joinOpeningWorker(opened, ownerClientId, createCommand.sessionPath); } } if (createCommand.sessionPath) { const existing = this.findWorkerBySessionFile(createCommand.sessionPath); if (existing && !(await this.reclaimStaleWorkerRegistration(existing, command.launchEnv !== undefined))) { return this.reuseWorkerForCreate(existing, ownerClientId, createCommand.sessionPath); } } const opening = (async () => { if (!createCommand.name) return this.launchWorker(createCommand, undefined, ownerClientId); const savedSiblings = createCommand.sessionPath ? await this.rlmLedgerSiblings(createCommand.sessionPath) : []; const target = savedSiblings.find(29 unmodified lines throw new Error( `Session "${sessionPath}" is registered to a failed worker that could not be safely reclaimed`, ); } if (worker.descriptor.ownerClientId !== ownerClientId) { throw new SessionAlreadyActiveError(sessionPath, worker.descriptor.rootActiveSessionId); } this.assertWorkerCreateOwner(worker, ownerClientId, sessionPath); if (!this.isWorkerReadyForCreate(worker)) { if (worker.recovery) { await worker.recovery; } else if (this.isWorkerRecoveryEligible(worker)) {3 unmodified lines const current = this.workers.get(worker.descriptor.workerId); if (!current) { throw new Error(`Session "${sessionPath}" worker recovery was interrupted; retry opening the session`); } if (current.descriptor.ownerClientId !== ownerClientId) { throw new SessionAlreadyActiveError(sessionPath, current.descriptor.rootActiveSessionId); } this.assertWorkerCreateOwner(current, ownerClientId, sessionPath); if (!this.isWorkerReadyForCreate(current)) { if (!current.summaries.has(current.descriptor.rootActiveSessionId)) { throw new Error( `Session "${sessionPath}" worker is unavailable for reuse: assigned root session is missing`,4 unmodified lines } return current; }
private async joinOpeningWorker( pending: Promise<ResidentWorker>, ownerClientId: string | undefined, sessionPath: string, ): Promise<ResidentWorker> { const worker = await pending; this.assertWorkerCreateOwner(worker, ownerClientId, sessionPath); return worker; }
private assertWorkerCreateOwner( worker: ResidentWorker, ownerClientId: string | undefined, sessionPath: string, ): void { if (worker.descriptor.ownerClientId !== ownerClientId) { throw new SessionAlreadyActiveError(sessionPath, worker.descriptor.rootActiveSessionId); } }
private isWorkerReadyForCreate(worker: ResidentWorker): boolean { return ( worker.descriptor.lifecycle === "ready" && worker.client !== undefined &&3167 unmodified linesExpand all