fix(coding-agent): wait for worker recovery before session reuse (#1756)
* fix(coding-agent): wait for worker recovery before session reuse * fix(coding-agent): join overlapping worker starts
1- Fixed saved-session resume when its resident worker is still recovering after a daemon restart.35 unmodified lines3637383940404142434445464647484950290 unmodified linesExpand all341342343344345346347348349350351352377 unmodified linesExpand all35 unmodified lines
interface WorkerFixture { descriptor: { workerId: string; lifecycle: "ready"; lifecycle: "ready" | "starting"; rootActiveSessionId: string; rootSessionId: string; pid: number; authenticationToken: string; ownerClientId?: string; createCommand: { config: { cwd: string } }; createCommand: { config: { cwd: string }; sessionPath?: string }; }; client: { request: ReturnType<typeof vi.fn>; requestWorker: ReturnType<typeof vi.fn>;290 unmodified linesExpand all launchWorker, });
const first = supervisor.createOrReuseWorker("client", { type: "create", name: "named", sessionPath }); const starting = worker("starting"); starting.descriptor.lifecycle = "starting"; starting.descriptor.createCommand = { config: { cwd: "/tmp/project" }, sessionPath }; supervisor.workers.set(starting.descriptor.workerId, starting); const second = supervisor.createOrReuseWorker("client", { type: "create", sessionPath }); releaseSiblings(); expect(await Promise.all([first, second])).toEqual([resident, resident]); expect(launchWorker).toHaveBeenCalledOnce();377 unmodified linesExpand all1611 unmodified linesExpand all1612161316141615161616161617161816191620162116221623162416251625162616271628162916301631162816321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617972143 unmodified linesExpand all1611 unmodified linesExpand all expect(worker.descriptor.lifecycle).toBe("failed"); expect(supervisor.persistWorker).toHaveBeenCalledWith(worker); });
it("rejects create reuse when a failed worker cannot be safely reclaimed", () => { it("rejects create reuse when a failed worker cannot be safely reclaimed", async () => { const worker = { descriptor: { workerId: "failed-unreclaimed", rootActiveSessionId: "active-failed", lifecycle: "failed", }, }; const supervisor = Object.create(DaemonSupervisor.prototype) as { reuseWorkerForCreate(target: typeof worker, ownerClientId: undefined, sessionPath: string): typeof worker; reuseWorkerForCreate( target: typeof worker, ownerClientId: undefined, sessionPath: string, ): Promise<typeof worker>; };
expect(() => supervisor.reuseWorkerForCreate(worker, undefined, "/tmp/failed.jsonl")).toThrow(/failed worker/); await expect(supervisor.reuseWorkerForCreate(worker, undefined, "/tmp/failed.jsonl")).rejects.toThrow( /failed worker/, ); });
it("waits for worker recovery before reusing a saved session", async () => { const root = { id: "active-root", activeSessionId: "active-root", sessionId: "session-root", cwd: "/tmp" }; const recovery = createDeferred<void>(); const worker = { descriptor: { workerId: "recovering-worker", rootActiveSessionId: root.activeSessionId, lifecycle: "recovering", }, client: undefined as object | undefined, summaries: new Map<string, SessionSummary>(), recovery: recovery.promise, intentionalStop: false, }; const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { workers: new Map([[worker.descriptor.workerId, worker]]), }) as { reuseWorkerForCreate( target: typeof worker, ownerClientId: undefined, sessionPath: string, ): Promise<typeof worker>; };
let settled = false; const reused = supervisor.reuseWorkerForCreate(worker, undefined, "/tmp/session.jsonl").finally(() => { settled = true; }); await Promise.resolve(); expect(settled).toBe(false);
worker.descriptor.lifecycle = "ready"; worker.client = {}; worker.summaries.set(root.activeSessionId, root as SessionSummary); recovery.resolve();
await expect(reused).resolves.toBe(worker); });
it("starts recovery before reusing a persisted recovering worker", async () => { const root = { id: "active-root", activeSessionId: "active-root", sessionId: "session-root", cwd: "/tmp" }; const worker = { descriptor: { workerId: "persisted-recovering-worker", rootActiveSessionId: root.activeSessionId, lifecycle: "recovering", }, client: undefined as object | undefined, summaries: new Map<string, SessionSummary>(), intentionalStop: false, }; const recoverWorker = vi.fn(async () => { worker.descriptor.lifecycle = "ready"; worker.client = {}; worker.summaries.set(root.activeSessionId, root as SessionSummary); }); const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { workers: new Map([[worker.descriptor.workerId, worker]]), shuttingDown: false, recoverWorker, }) as { reuseWorkerForCreate( target: typeof worker, ownerClientId: undefined, sessionPath: string, ): Promise<typeof worker>; };
await expect(supervisor.reuseWorkerForCreate(worker, undefined, "/tmp/session.jsonl")).resolves.toBe(worker); expect(recoverWorker).toHaveBeenCalledOnce(); });
it("starts recovery for a disconnected worker still marked ready", async () => { const root = { id: "active-root", activeSessionId: "active-root", sessionId: "session-root", cwd: "/tmp" }; const worker = { descriptor: { workerId: "disconnected-ready-worker", rootActiveSessionId: root.activeSessionId, lifecycle: "ready", }, client: undefined as object | undefined, summaries: new Map([[root.activeSessionId, root as SessionSummary]]), intentionalStop: false, }; const recoverWorker = vi.fn(async () => { worker.client = {}; }); const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { workers: new Map([[worker.descriptor.workerId, worker]]), shuttingDown: false, recoverWorker, }) as { reuseWorkerForCreate( target: typeof worker, ownerClientId: undefined, sessionPath: string, ): Promise<typeof worker>; };
await expect(supervisor.reuseWorkerForCreate(worker, undefined, "/tmp/session.jsonl")).resolves.toBe(worker); expect(recoverWorker).toHaveBeenCalledOnce(); });
it("rejects recovered workers whose assigned root is still missing", async () => { const worker = { descriptor: { workerId: "rootless-worker", rootActiveSessionId: "active-root", lifecycle: "ready", }, client: {}, summaries: new Map<string, SessionSummary>(), recovery: Promise.resolve(), intentionalStop: false, }; const supervisor = Object.assign(Object.create(DaemonSupervisor.prototype), { workers: new Map([[worker.descriptor.workerId, worker]]), }) as { reuseWorkerForCreate( target: typeof worker, ownerClientId: undefined, sessionPath: string, ): Promise<typeof worker>; };
await expect(supervisor.reuseWorkerForCreate(worker, undefined, "/tmp/session.jsonl")).rejects.toThrow( "assigned root session is missing", ); });
it("preserves cached summaries when recovery omits the assigned root", async () => { const root = { id: "active-root", activeSessionId: "active-root", sessionId: "session-root", cwd: "/tmp" }; const worker = { descriptor: { workerId: "root-omitting-worker", rootActiveSessionId: root.activeSessionId, }, client: { request: vi.fn(async () => success(undefined, "list", { sessions: [{ id: "other", activeSessionId: "other", sessionId: "session-other", cwd: "/tmp" }], }), ), }, summaries: new Map([[root.activeSessionId, root as SessionSummary]]), intentionalStop: false, }; const supervisor = Object.create(DaemonSupervisor.prototype) as { refreshWorkerSummaries(target: typeof worker, recovery: boolean): Promise<void>; };
await expect(supervisor.refreshWorkerSummaries(worker, true)).rejects.toThrow( "Session worker omitted its root session during recovery", ); expect(worker.summaries.get(root.activeSessionId)).toBe(root); });
it("ignores conflicting paths on workers unrelated to a session lookup", () => { const unrelated = { descriptor: { workerId: "unrelated",2143 unmodified linesExpand all