| author |  Claudio Vasquez <105945231+crafael23@users.noreply.github.com> 2026-04-16 17:30:56-06:00 |
| committer |  GitHub <noreply@github.com> 2026-04-16 16:30:56-07:00 |
| log | e0117b271a79f66d0ac793da334991e7a2ccb800 |
| tree | f3c00351452bb1fe1bbf950436543b97dfefac3f |
| parent | 188a40c315bceaeec6237f05ae253941ad6bb4dc |
| signature | Signed by PGP key B5690EEEBB952194 |
Fix Claude Process leak[MEMORY INTENSIVE], archiving, and stale claude session monitoring. (#2042)
Co-authored-by: Julius Marminge <julius0216@outlook.com>
Co-authored-by: codex <codex@users.noreply.github.com>19 files changed, 1686 insertions(+), 49 deletions(-)
apps/server/package.json+2-1
| ... | ... | @@ -20,7 +20,8 @@ |
| 20 | 20 | "start": "node dist/bin.mjs", |
| 21 | 21 | "prepare": "effect-language-service patch", |
| 22 | 22 | "typecheck": "tsc --noEmit", |
| 23 | "test": "vitest run" | |
| 23 | "test": "vitest run", | |
| 24 | "test:process-reaper": "vitest run src/server.test.ts src/provider/Layers/ClaudeAdapter.test.ts src/provider/Layers/ProviderSessionDirectory.test.ts src/provider/Layers/ProviderSessionReaper.test.ts src/provider/Layers/CodexAdapter.test.ts" | |
| 24 | 25 | }, |
| 25 | 26 | "dependencies": { |
| 26 | 27 | "@anthropic-ai/claude-agent-sdk": "^0.2.111", |
apps/server/src/codexAppServerManager.test.ts+148
| ... | ... | @@ -470,6 +470,154 @@ describe("startSession", () => { |
| 470 | 470 | manager.stopAll(); |
| 471 | 471 | } |
| 472 | 472 | }); |
| 473 | ||
| 474 | it("disposes an existing session before starting a replacement for the same thread", async () => { | |
| 475 | const manager = new CodexAppServerManager(); | |
| 476 | const existingContext = { | |
| 477 | session: { | |
| 478 | provider: "codex", | |
| 479 | status: "ready", | |
| 480 | threadId: asThreadId("thread-1"), | |
| 481 | runtimeMode: "full-access", | |
| 482 | createdAt: "2026-02-10T00:00:00.000Z", | |
| 483 | updatedAt: "2026-02-10T00:00:00.000Z", | |
| 484 | }, | |
| 485 | }; | |
| 486 | ||
| 487 | ( | |
| 488 | manager as unknown as { | |
| 489 | sessions: Map<ThreadId, typeof existingContext>; | |
| 490 | } | |
| 491 | ).sessions.set(asThreadId("thread-1"), existingContext); | |
| 492 | ||
| 493 | const disposeSession = vi | |
| 494 | .spyOn( | |
| 495 | manager as unknown as { | |
| 496 | disposeSession: ( | |
| 497 | context: typeof existingContext, | |
| 498 | options?: { readonly emitLifecycleEvent?: boolean }, | |
| 499 | ) => void; | |
| 500 | }, | |
| 501 | "disposeSession", | |
| 502 | ) | |
| 503 | .mockImplementation(() => {}); | |
| 504 | const assertSupportedCodexCliVersion = vi | |
| 505 | .spyOn( | |
| 506 | manager as unknown as { | |
| 507 | assertSupportedCodexCliVersion: (input: { | |
| 508 | binaryPath: string; | |
| 509 | cwd: string; | |
| 510 | homePath?: string; | |
| 511 | }) => void; | |
| 512 | }, | |
| 513 | "assertSupportedCodexCliVersion", | |
| 514 | ) | |
| 515 | .mockImplementation(() => {}); | |
| 516 | const processCwd = vi.spyOn(process, "cwd").mockImplementation(() => { | |
| 517 | throw new Error("cwd missing"); | |
| 518 | }); | |
| 519 | ||
| 520 | try { | |
| 521 | await expect( | |
| 522 | manager.startSession({ | |
| 523 | threadId: asThreadId("thread-1"), | |
| 524 | provider: "codex", | |
| 525 | binaryPath: "codex", | |
| 526 | runtimeMode: "full-access", | |
| 527 | }), | |
| 528 | ).rejects.toThrow("cwd missing"); | |
| 529 | ||
| 530 | expect(disposeSession).toHaveBeenCalledWith(existingContext, { | |
| 531 | emitLifecycleEvent: false, | |
| 532 | }); | |
| 533 | expect(assertSupportedCodexCliVersion).not.toHaveBeenCalled(); | |
| 534 | } finally { | |
| 535 | disposeSession.mockRestore(); | |
| 536 | assertSupportedCodexCliVersion.mockRestore(); | |
| 537 | processCwd.mockRestore(); | |
| 538 | ( | |
| 539 | manager as unknown as { | |
| 540 | sessions: Map<ThreadId, typeof existingContext>; | |
| 541 | } | |
| 542 | ).sessions.clear(); | |
| 543 | manager.stopAll(); | |
| 544 | } | |
| 545 | }); | |
| 546 | ||
| 547 | it("continues replacement start when existing session disposal fails", async () => { | |
| 548 | const manager = new CodexAppServerManager(); | |
| 549 | const existingContext = { | |
| 550 | session: { | |
| 551 | provider: "codex", | |
| 552 | status: "ready", | |
| 553 | threadId: asThreadId("thread-1"), | |
| 554 | runtimeMode: "full-access", | |
| 555 | createdAt: "2026-02-10T00:00:00.000Z", | |
| 556 | updatedAt: "2026-02-10T00:00:00.000Z", | |
| 557 | }, | |
| 558 | }; | |
| 559 | ||
| 560 | ( | |
| 561 | manager as unknown as { | |
| 562 | sessions: Map<ThreadId, typeof existingContext>; | |
| 563 | } | |
| 564 | ).sessions.set(asThreadId("thread-1"), existingContext); | |
| 565 | ||
| 566 | const disposeSession = vi | |
| 567 | .spyOn( | |
| 568 | manager as unknown as { | |
| 569 | disposeSession: ( | |
| 570 | context: typeof existingContext, | |
| 571 | options?: { readonly emitLifecycleEvent?: boolean }, | |
| 572 | ) => void; | |
| 573 | }, | |
| 574 | "disposeSession", | |
| 575 | ) | |
| 576 | .mockImplementation(() => { | |
| 577 | throw new Error("dispose failed"); | |
| 578 | }); | |
| 579 | const assertSupportedCodexCliVersion = vi | |
| 580 | .spyOn( | |
| 581 | manager as unknown as { | |
| 582 | assertSupportedCodexCliVersion: (input: { | |
| 583 | binaryPath: string; | |
| 584 | cwd: string; | |
| 585 | homePath?: string; | |
| 586 | }) => void; | |
| 587 | }, | |
| 588 | "assertSupportedCodexCliVersion", | |
| 589 | ) | |
| 590 | .mockImplementation(() => {}); | |
| 591 | const processCwd = vi.spyOn(process, "cwd").mockImplementation(() => { | |
| 592 | throw new Error("cwd missing"); | |
| 593 | }); | |
| 594 | ||
| 595 | try { | |
| 596 | await expect( | |
| 597 | manager.startSession({ | |
| 598 | threadId: asThreadId("thread-1"), | |
| 599 | provider: "codex", | |
| 600 | binaryPath: "codex", | |
| 601 | runtimeMode: "full-access", | |
| 602 | }), | |
| 603 | ).rejects.toThrow("cwd missing"); | |
| 604 | ||
| 605 | expect(disposeSession).toHaveBeenCalledWith(existingContext, { | |
| 606 | emitLifecycleEvent: false, | |
| 607 | }); | |
| 608 | expect(assertSupportedCodexCliVersion).not.toHaveBeenCalled(); | |
| 609 | } finally { | |
| 610 | disposeSession.mockRestore(); | |
| 611 | assertSupportedCodexCliVersion.mockRestore(); | |
| 612 | processCwd.mockRestore(); | |
| 613 | ( | |
| 614 | manager as unknown as { | |
| 615 | sessions: Map<ThreadId, typeof existingContext>; | |
| 616 | } | |
| 617 | ).sessions.clear(); | |
| 618 | manager.stopAll(); | |
| 619 | } | |
| 620 | }); | |
| 473 | 621 | }); |
| 474 | 622 | |
| 475 | 623 | describe("sendTurn", () => { |
apps/server/src/codexAppServerManager.ts+36-2
| ... | ... | @@ -450,6 +450,25 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve |
| 450 | 450 | let context: CodexSessionContext | undefined; |
| 451 | 451 | |
| 452 | 452 | try { |
| 453 | const existingContext = this.sessions.get(threadId); | |
| 454 | if (existingContext) { | |
| 455 | await Effect.logWarning("codex app-server replacing existing session", { | |
| 456 | threadId, | |
| 457 | existingStatus: existingContext.session.status, | |
| 458 | }).pipe(this.runPromise); | |
| 459 | try { | |
| 460 | this.disposeSession(existingContext, { | |
| 461 | emitLifecycleEvent: false, | |
| 462 | }); | |
| 463 | } catch (error) { | |
| 464 | await Effect.logWarning("codex app-server failed to dispose existing session", { | |
| 465 | threadId, | |
| 466 | existingStatus: existingContext.session.status, | |
| 467 | cause: error instanceof Error ? error.message : String(error), | |
| 468 | }).pipe(this.runPromise); | |
| 469 | } | |
| 470 | } | |
| 471 | ||
| 453 | 472 | const resolvedCwd = input.cwd ?? process.cwd(); |
| 454 | 473 | |
| 455 | 474 | const session: ProviderSession = { |
| ... | ... | @@ -901,6 +920,19 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve |
| 901 | 920 | return; |
| 902 | 921 | } |
| 903 | 922 | |
| 923 | this.disposeSession(context, { | |
| 924 | emitLifecycleEvent: true, | |
| 925 | }); | |
| 926 | } | |
| 927 | ||
| 928 | private disposeSession( | |
| 929 | context: CodexSessionContext, | |
| 930 | options?: { readonly emitLifecycleEvent?: boolean }, | |
| 931 | ): void { | |
| 932 | if (context.stopping) { | |
| 933 | return; | |
| 934 | } | |
| 935 | ||
| 904 | 936 | context.stopping = true; |
| 905 | 937 | |
| 906 | 938 | for (const pending of context.pending.values()) { |
| ... | ... | @@ -921,8 +953,10 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve |
| 921 | 953 | status: "closed", |
| 922 | 954 | activeTurnId: undefined, |
| 923 | 955 | }); |
| 924 | this.emitLifecycleEvent(context, "session/closed", "Session stopped"); | |
| 925 | this.sessions.delete(threadId); | |
| 956 | if (options?.emitLifecycleEvent !== false) { | |
| 957 | this.emitLifecycleEvent(context, "session/closed", "Session stopped"); | |
| 958 | } | |
| 959 | this.sessions.delete(context.session.threadId); | |
| 926 | 960 | } |
| 927 | 961 | |
| 928 | 962 | listSessions(): ProviderSession[] { |
apps/server/src/provider/Layers/ClaudeAdapter.test.ts+65
| ... | ... | @@ -1307,6 +1307,71 @@ describe("ClaudeAdapterLive", () => { |
| 1307 | 1307 | ); |
| 1308 | 1308 | }); |
| 1309 | 1309 | |
| 1310 | it.effect("closes the previous session before replacing an existing thread session", () => { | |
| 1311 | const queries: FakeClaudeQuery[] = []; | |
| 1312 | const layer = makeClaudeAdapterLive({ | |
| 1313 | createQuery: () => { | |
| 1314 | const query = new FakeClaudeQuery(); | |
| 1315 | queries.push(query); | |
| 1316 | return query; | |
| 1317 | }, | |
| 1318 | }).pipe( | |
| 1319 | Layer.provideMerge(ServerConfig.layerTest("/tmp/claude-adapter-test", "/tmp")), | |
| 1320 | Layer.provideMerge(ServerSettingsService.layerTest()), | |
| 1321 | Layer.provideMerge(NodeServices.layer), | |
| 1322 | ); | |
| 1323 | ||
| 1324 | return Effect.gen(function* () { | |
| 1325 | const adapter = yield* ClaudeAdapter; | |
| 1326 | ||
| 1327 | const runtimeEventsFiber = yield* Stream.take(adapter.streamEvents, 6).pipe( | |
| 1328 | Stream.runCollect, | |
| 1329 | Effect.forkChild, | |
| 1330 | ); | |
| 1331 | ||
| 1332 | const firstSession = yield* adapter.startSession({ | |
| 1333 | threadId: THREAD_ID, | |
| 1334 | provider: "claudeAgent", | |
| 1335 | runtimeMode: "full-access", | |
| 1336 | }); | |
| 1337 | ||
| 1338 | const secondSession = yield* adapter.startSession({ | |
| 1339 | threadId: THREAD_ID, | |
| 1340 | provider: "claudeAgent", | |
| 1341 | runtimeMode: "full-access", | |
| 1342 | resumeCursor: firstSession.resumeCursor, | |
| 1343 | }); | |
| 1344 | ||
| 1345 | const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); | |
| 1346 | const activeSessions = yield* adapter.listSessions(); | |
| 1347 | ||
| 1348 | assert.equal(queries.length, 2); | |
| 1349 | assert.equal(queries[0]?.closeCalls, 1); | |
| 1350 | assert.equal(queries[1]?.closeCalls, 0); | |
| 1351 | assert.equal(yield* adapter.hasSession(THREAD_ID), true); | |
| 1352 | assert.equal(activeSessions.length, 1); | |
| 1353 | assert.deepEqual(activeSessions[0]?.resumeCursor, secondSession.resumeCursor); | |
| 1354 | assert.deepEqual( | |
| 1355 | runtimeEvents.map((event) => event.type), | |
| 1356 | [ | |
| 1357 | "session.started", | |
| 1358 | "session.configured", | |
| 1359 | "session.state.changed", | |
| 1360 | "session.started", | |
| 1361 | "session.configured", | |
| 1362 | "session.state.changed", | |
| 1363 | ], | |
| 1364 | ); | |
| 1365 | assert.equal( | |
| 1366 | runtimeEvents.some((event) => event.type === "session.exited"), | |
| 1367 | false, | |
| 1368 | ); | |
| 1369 | }).pipe( | |
| 1370 | Effect.provideService(Random.Random, makeDeterministicRandomService()), | |
| 1371 | Effect.provide(layer), | |
| 1372 | ); | |
| 1373 | }); | |
| 1374 | ||
| 1310 | 1375 | it.effect("stopSession does not throw into the SDK prompt consumer", () => { |
| 1311 | 1376 | // The SDK consumes user messages via `for await (... of prompt)`. |
| 1312 | 1377 | // Stopping a session must end that loop cleanly — not throw an error. |
apps/server/src/provider/Layers/ClaudeAdapter.ts+96-20
| ... | ... | @@ -470,7 +470,10 @@ function isTodoTool(toolName: string): boolean { |
| 470 | 470 | return toolName.toLowerCase().includes("todowrite"); |
| 471 | 471 | } |
| 472 | 472 | |
| 473 | type PlanStep = { step: string; status: "pending" | "inProgress" | "completed" }; | |
| 473 | type PlanStep = { | |
| 474 | step: string; | |
| 475 | status: "pending" | "inProgress" | "completed"; | |
| 476 | }; | |
| 474 | 477 | |
| 475 | 478 | function extractPlanStepsFromTodoInput(input: Record<string, unknown>): PlanStep[] | null { |
| 476 | 479 | // TodoWrite format: { todos: [{ content, status, activeForm? }] } |
| ... | ... | @@ -973,7 +976,11 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 973 | 976 | ((input: { |
| 974 | 977 | readonly prompt: AsyncIterable<SDKUserMessage>; |
| 975 | 978 | readonly options: ClaudeQueryOptions; |
| 976 | }) => query({ prompt: input.prompt, options: input.options }) as ClaudeQueryRuntime); | |
| 979 | }) => | |
| 980 | query({ | |
| 981 | prompt: input.prompt, | |
| 982 | options: input.options, | |
| 983 | }) as ClaudeQueryRuntime); | |
| 977 | 984 | |
| 978 | 985 | const sessions = new Map<ThreadId, ClaudeSessionContext>(); |
| 979 | 986 | const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>(); |
| ... | ... | @@ -1012,7 +1019,11 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 1012 | 1019 | ...(typeof message.session_id === "string" |
| 1013 | 1020 | ? { providerThreadId: message.session_id } |
| 1014 | 1021 | : {}), |
| 1015 | ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), | |
| 1022 | ...(context.turnState | |
| 1023 | ? { | |
| 1024 | turnId: asCanonicalTurnId(context.turnState.turnId), | |
| 1025 | } | |
| 1026 | : {}), | |
| 1016 | 1027 | ...(itemId ? { itemId: ProviderItemId.make(itemId) } : {}), |
| 1017 | 1028 | payload: message, |
| 1018 | 1029 | }, |
| ... | ... | @@ -1401,7 +1412,9 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 1401 | 1412 | ...(typeof accumulatedTotalProcessedTokens === "number" && |
| 1402 | 1413 | Number.isFinite(accumulatedTotalProcessedTokens) && |
| 1403 | 1414 | accumulatedTotalProcessedTokens > lastGoodUsage.usedTokens |
| 1404 | ? { totalProcessedTokens: accumulatedTotalProcessedTokens } | |
| 1415 | ? { | |
| 1416 | totalProcessedTokens: accumulatedTotalProcessedTokens, | |
| 1417 | } | |
| 1405 | 1418 | : {}), |
| 1406 | 1419 | } |
| 1407 | 1420 | : accumulatedSnapshot; |
| ... | ... | @@ -1465,7 +1478,9 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 1465 | 1478 | input: tool.input, |
| 1466 | 1479 | }, |
| 1467 | 1480 | }, |
| 1468 | providerRefs: nativeProviderRefs(context, { providerItemId: tool.itemId }), | |
| 1481 | providerRefs: nativeProviderRefs(context, { | |
| 1482 | providerItemId: tool.itemId, | |
| 1483 | }), | |
| 1469 | 1484 | raw: { |
| 1470 | 1485 | source: "claude.sdk.message", |
| 1471 | 1486 | method: "claude/result", |
| ... | ... | @@ -1587,7 +1602,9 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 1587 | 1602 | threadId: context.session.threadId, |
| 1588 | 1603 | turnId: context.turnState.turnId, |
| 1589 | 1604 | ...(assistantBlockEntry?.block |
| 1590 | ? { itemId: asRuntimeItemId(assistantBlockEntry.block.itemId) } | |
| 1605 | ? { | |
| 1606 | itemId: asRuntimeItemId(assistantBlockEntry.block.itemId), | |
| 1607 | } | |
| 1591 | 1608 | : {}), |
| 1592 | 1609 | payload: { |
| 1593 | 1610 | streamKind, |
| ... | ... | @@ -1646,7 +1663,11 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 1646 | 1663 | provider: PROVIDER, |
| 1647 | 1664 | createdAt: stamp.createdAt, |
| 1648 | 1665 | threadId: context.session.threadId, |
| 1649 | ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), | |
| 1666 | ...(context.turnState | |
| 1667 | ? { | |
| 1668 | turnId: asCanonicalTurnId(context.turnState.turnId), | |
| 1669 | } | |
| 1670 | : {}), | |
| 1650 | 1671 | itemId: asRuntimeItemId(nextTool.itemId), |
| 1651 | 1672 | payload: { |
| 1652 | 1673 | itemType: nextTool.itemType, |
| ... | ... | @@ -1658,7 +1679,9 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 1658 | 1679 | input: nextTool.input, |
| 1659 | 1680 | }, |
| 1660 | 1681 | }, |
| 1661 | providerRefs: nativeProviderRefs(context, { providerItemId: nextTool.itemId }), | |
| 1682 | providerRefs: nativeProviderRefs(context, { | |
| 1683 | providerItemId: nextTool.itemId, | |
| 1684 | }), | |
| 1662 | 1685 | raw: { |
| 1663 | 1686 | source: "claude.sdk.message", |
| 1664 | 1687 | method: "claude/stream_event/content_block_delta/input_json_delta", |
| ... | ... | @@ -1677,7 +1700,11 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 1677 | 1700 | provider: PROVIDER, |
| 1678 | 1701 | createdAt: planStamp.createdAt, |
| 1679 | 1702 | threadId: context.session.threadId, |
| 1680 | ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), | |
| 1703 | ...(context.turnState | |
| 1704 | ? { | |
| 1705 | turnId: asCanonicalTurnId(context.turnState.turnId), | |
| 1706 | } | |
| 1707 | : {}), | |
| 1681 | 1708 | payload: { |
| 1682 | 1709 | plan: planSteps, |
| 1683 | 1710 | }, |
| ... | ... | @@ -1747,7 +1774,9 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 1747 | 1774 | input: toolInput, |
| 1748 | 1775 | }, |
| 1749 | 1776 | }, |
| 1750 | providerRefs: nativeProviderRefs(context, { providerItemId: tool.itemId }), | |
| 1777 | providerRefs: nativeProviderRefs(context, { | |
| 1778 | providerItemId: tool.itemId, | |
| 1779 | }), | |
| 1751 | 1780 | raw: { |
| 1752 | 1781 | source: "claude.sdk.message", |
| 1753 | 1782 | method: "claude/stream_event/content_block_start", |
| ... | ... | @@ -1819,7 +1848,9 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 1819 | 1848 | ...(tool.detail ? { detail: tool.detail } : {}), |
| 1820 | 1849 | data: toolData, |
| 1821 | 1850 | }, |
| 1822 | providerRefs: nativeProviderRefs(context, { providerItemId: tool.itemId }), | |
| 1851 | providerRefs: nativeProviderRefs(context, { | |
| 1852 | providerItemId: tool.itemId, | |
| 1853 | }), | |
| 1823 | 1854 | raw: { |
| 1824 | 1855 | source: "claude.sdk.message", |
| 1825 | 1856 | method: "claude/user", |
| ... | ... | @@ -1842,7 +1873,9 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 1842 | 1873 | streamKind, |
| 1843 | 1874 | delta: toolResult.text, |
| 1844 | 1875 | }, |
| 1845 | providerRefs: nativeProviderRefs(context, { providerItemId: tool.itemId }), | |
| 1876 | providerRefs: nativeProviderRefs(context, { | |
| 1877 | providerItemId: tool.itemId, | |
| 1878 | }), | |
| 1846 | 1879 | raw: { |
| 1847 | 1880 | source: "claude.sdk.message", |
| 1848 | 1881 | method: "claude/user", |
| ... | ... | @@ -1867,7 +1900,9 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 1867 | 1900 | ...(tool.detail ? { detail: tool.detail } : {}), |
| 1868 | 1901 | data: toolData, |
| 1869 | 1902 | }, |
| 1870 | providerRefs: nativeProviderRefs(context, { providerItemId: tool.itemId }), | |
| 1903 | providerRefs: nativeProviderRefs(context, { | |
| 1904 | providerItemId: tool.itemId, | |
| 1905 | }), | |
| 1871 | 1906 | raw: { |
| 1872 | 1907 | source: "claude.sdk.message", |
| 1873 | 1908 | method: "claude/user", |
| ... | ... | @@ -2223,7 +2258,9 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 2223 | 2258 | payload: { |
| 2224 | 2259 | summary: message.summary, |
| 2225 | 2260 | ...(message.preceding_tool_use_ids.length > 0 |
| 2226 | ? { precedingToolUseIds: message.preceding_tool_use_ids } | |
| 2261 | ? { | |
| 2262 | precedingToolUseIds: message.preceding_tool_use_ids, | |
| 2263 | } | |
| 2227 | 2264 | : {}), |
| 2228 | 2265 | }, |
| 2229 | 2266 | }); |
| ... | ... | @@ -2440,6 +2477,27 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 2440 | 2477 | }); |
| 2441 | 2478 | } |
| 2442 | 2479 | |
| 2480 | const existingContext = sessions.get(input.threadId); | |
| 2481 | if (existingContext) { | |
| 2482 | yield* Effect.logWarning("claude.session.replacing", { | |
| 2483 | threadId: input.threadId, | |
| 2484 | existingSessionStatus: existingContext.session.status, | |
| 2485 | reason: "startSession called with existing active session", | |
| 2486 | }); | |
| 2487 | yield* stopSessionInternal(existingContext, { | |
| 2488 | emitExitEvent: false, | |
| 2489 | }).pipe( | |
| 2490 | // Replacement cleanup is best-effort: never block the new session on | |
| 2491 | // either typed failures or unexpected defects from tearing down the old one. | |
| 2492 | Effect.catchCause((cause) => | |
| 2493 | Effect.logWarning("claude.session.replace.stop-failed", { | |
| 2494 | threadId: input.threadId, | |
| 2495 | cause, | |
| 2496 | }), | |
| 2497 | ), | |
| 2498 | ); | |
| 2499 | } | |
| 2500 | ||
| 2443 | 2501 | const startedAt = yield* nowIso; |
| 2444 | 2502 | const resumeState = readClaudeResumeState(input.resumeCursor); |
| 2445 | 2503 | const threadId = input.threadId; |
| ... | ... | @@ -2475,7 +2533,10 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 2475 | 2533 | const handleAskUserQuestion = Effect.fn("handleAskUserQuestion")(function* ( |
| 2476 | 2534 | context: ClaudeSessionContext, |
| 2477 | 2535 | toolInput: Record<string, unknown>, |
| 2478 | callbackOptions: { readonly signal: AbortSignal; readonly toolUseID?: string }, | |
| 2536 | callbackOptions: { | |
| 2537 | readonly signal: AbortSignal; | |
| 2538 | readonly toolUseID?: string; | |
| 2539 | }, | |
| 2479 | 2540 | ) { |
| 2480 | 2541 | const requestId = ApprovalRequestId.make(yield* Random.nextUUIDv4); |
| 2481 | 2542 | |
| ... | ... | @@ -2511,7 +2572,11 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 2511 | 2572 | provider: PROVIDER, |
| 2512 | 2573 | createdAt: requestedStamp.createdAt, |
| 2513 | 2574 | threadId: context.session.threadId, |
| 2514 | ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), | |
| 2575 | ...(context.turnState | |
| 2576 | ? { | |
| 2577 | turnId: asCanonicalTurnId(context.turnState.turnId), | |
| 2578 | } | |
| 2579 | : {}), | |
| 2515 | 2580 | requestId: asRuntimeRequestId(requestId), |
| 2516 | 2581 | payload: { questions }, |
| 2517 | 2582 | providerRefs: nativeProviderRefs(context, { |
| ... | ... | @@ -2520,7 +2585,10 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 2520 | 2585 | raw: { |
| 2521 | 2586 | source: "claude.sdk.permission", |
| 2522 | 2587 | method: "canUseTool/AskUserQuestion", |
| 2523 | payload: { toolName: "AskUserQuestion", input: toolInput }, | |
| 2588 | payload: { | |
| 2589 | toolName: "AskUserQuestion", | |
| 2590 | input: toolInput, | |
| 2591 | }, | |
| 2524 | 2592 | }, |
| 2525 | 2593 | }); |
| 2526 | 2594 | |
| ... | ... | @@ -2535,7 +2603,9 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 2535 | 2603 | pendingUserInputs.delete(requestId); |
| 2536 | 2604 | runFork(Deferred.succeed(answersDeferred, {} as ProviderUserInputAnswers)); |
| 2537 | 2605 | }; |
| 2538 | callbackOptions.signal.addEventListener("abort", onAbort, { once: true }); | |
| 2606 | callbackOptions.signal.addEventListener("abort", onAbort, { | |
| 2607 | once: true, | |
| 2608 | }); | |
| 2539 | 2609 | |
| 2540 | 2610 | // Block until the user provides answers. |
| 2541 | 2611 | const answers = yield* Deferred.await(answersDeferred); |
| ... | ... | @@ -2549,7 +2619,11 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 2549 | 2619 | provider: PROVIDER, |
| 2550 | 2620 | createdAt: resolvedStamp.createdAt, |
| 2551 | 2621 | threadId: context.session.threadId, |
| 2552 | ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), | |
| 2622 | ...(context.turnState | |
| 2623 | ? { | |
| 2624 | turnId: asCanonicalTurnId(context.turnState.turnId), | |
| 2625 | } | |
| 2626 | : {}), | |
| 2553 | 2627 | requestId: asRuntimeRequestId(requestId), |
| 2554 | 2628 | payload: { answers }, |
| 2555 | 2629 | providerRefs: nativeProviderRefs(context, { |
| ... | ... | @@ -2719,7 +2793,9 @@ const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( |
| 2719 | 2793 | behavior: "allow", |
| 2720 | 2794 | updatedInput: toolInput, |
| 2721 | 2795 | ...(decision === "acceptForSession" && pendingApproval.suggestions |
| 2722 | ? { updatedPermissions: [...pendingApproval.suggestions] } | |
| 2796 | ? { | |
| 2797 | updatedPermissions: [...pendingApproval.suggestions], | |
| 2798 | } | |
| 2723 | 2799 | : {}), |
| 2724 | 2800 | } satisfies PermissionResult; |
| 2725 | 2801 | } |
apps/server/src/provider/Layers/CodexAdapter.test.ts+1
| ... | ... | @@ -146,6 +146,7 @@ const providerSessionDirectoryTestLayer = Layer.succeed(ProviderSessionDirectory |
| 146 | 146 | getBinding: () => Effect.succeed(Option.none()), |
| 147 | 147 | remove: () => Effect.void, |
| 148 | 148 | listThreadIds: () => Effect.succeed([]), |
| 149 | listBindings: () => Effect.succeed([]), | |
| 149 | 150 | }); |
| 150 | 151 | |
| 151 | 152 | const validationManager = new FakeCodexManager(); |
apps/server/src/provider/Layers/ProviderService.test.ts+37
| ... | ... | @@ -654,6 +654,43 @@ routing.layer("ProviderServiceLive routing", (it) => { |
| 654 | 654 | }), |
| 655 | 655 | ); |
| 656 | 656 | |
| 657 | it.effect("stops stale sessions in other providers after a successful replacement start", () => | |
| 658 | Effect.gen(function* () { | |
| 659 | const provider = yield* ProviderService; | |
| 660 | const threadId = asThreadId("thread-provider-replacement"); | |
| 661 | ||
| 662 | const codexSession = yield* provider.startSession(threadId, { | |
| 663 | provider: "codex", | |
| 664 | threadId, | |
| 665 | cwd: "/tmp/project-provider-replacement", | |
| 666 | runtimeMode: "full-access", | |
| 667 | }); | |
| 668 | ||
| 669 | routing.codex.stopSession.mockClear(); | |
| 670 | routing.claude.stopSession.mockClear(); | |
| 671 | ||
| 672 | const claudeSession = yield* provider.startSession(threadId, { | |
| 673 | provider: "claudeAgent", | |
| 674 | threadId, | |
| 675 | cwd: "/tmp/project-provider-replacement", | |
| 676 | runtimeMode: "full-access", | |
| 677 | }); | |
| 678 | ||
| 679 | assert.equal(codexSession.provider, "codex"); | |
| 680 | assert.equal(claudeSession.provider, "claudeAgent"); | |
| 681 | assert.deepEqual(routing.codex.stopSession.mock.calls, [[threadId]]); | |
| 682 | assert.equal(routing.claude.stopSession.mock.calls.length, 0); | |
| 683 | ||
| 684 | const sessions = yield* provider.listSessions(); | |
| 685 | assert.deepEqual( | |
| 686 | sessions | |
| 687 | .filter((session) => session.threadId === threadId) | |
| 688 | .map((session) => session.provider), | |
| 689 | ["claudeAgent"], | |
| 690 | ); | |
| 691 | }), | |
| 692 | ); | |
| 693 | ||
| 657 | 694 | it.effect("recovers stale sessions for sendTurn using persisted cwd", () => |
| 658 | 695 | Effect.gen(function* () { |
| 659 | 696 | const provider = yield* ProviderService; |
apps/server/src/provider/Layers/ProviderService.ts+38
| ... | ... | @@ -297,6 +297,40 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( |
| 297 | 297 | return { adapter: recovered.adapter, threadId: input.threadId, isActive: true } as const; |
| 298 | 298 | }); |
| 299 | 299 | |
| 300 | const stopStaleSessionsForThread = Effect.fn("stopStaleSessionsForThread")(function* (input: { | |
| 301 | readonly threadId: ThreadId; | |
| 302 | readonly currentProvider: ProviderSession["provider"]; | |
| 303 | }) { | |
| 304 | yield* Effect.forEach( | |
| 305 | adapters, | |
| 306 | (adapter) => | |
| 307 | adapter.provider === input.currentProvider | |
| 308 | ? Effect.void | |
| 309 | : Effect.gen(function* () { | |
| 310 | const hasSession = yield* adapter.hasSession(input.threadId); | |
| 311 | if (!hasSession) { | |
| 312 | return; | |
| 313 | } | |
| 314 | ||
| 315 | yield* adapter.stopSession(input.threadId).pipe( | |
| 316 | Effect.tap(() => | |
| 317 | analytics.record("provider.session.stopped", { | |
| 318 | provider: adapter.provider, | |
| 319 | }), | |
| 320 | ), | |
| 321 | Effect.catchCause((cause) => | |
| 322 | Effect.logWarning("provider.session.stop-stale-failed", { | |
| 323 | threadId: input.threadId, | |
| 324 | provider: adapter.provider, | |
| 325 | cause, | |
| 326 | }), | |
| 327 | ), | |
| 328 | ); | |
| 329 | }), | |
| 330 | { discard: true }, | |
| 331 | ); | |
| 332 | }); | |
| 333 | ||
| 300 | 334 | const startSession: ProviderServiceShape["startSession"] = Effect.fn("startSession")( |
| 301 | 335 | function* (threadId, rawInput) { |
| 302 | 336 | const parsed = yield* decodeInputOrValidationError({ |
| ... | ... | @@ -351,6 +385,10 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( |
| 351 | 385 | ); |
| 352 | 386 | } |
| 353 | 387 | |
| 388 | yield* stopStaleSessionsForThread({ | |
| 389 | threadId, | |
| 390 | currentProvider: adapter.provider, | |
| 391 | }); | |
| 354 | 392 | yield* upsertSessionBinding(session, threadId, { |
| 355 | 393 | modelSelection: input.modelSelection, |
| 356 | 394 | }); |
apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts+72
| ... | ... | @@ -133,6 +133,78 @@ it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryL |
| 133 | 133 | } |
| 134 | 134 | })); |
| 135 | 135 | |
| 136 | it("lists persisted bindings with metadata in oldest-first order", () => | |
| 137 | Effect.gen(function* () { | |
| 138 | const directory = yield* ProviderSessionDirectory; | |
| 139 | const runtimeRepository = yield* ProviderSessionRuntimeRepository; | |
| 140 | ||
| 141 | const olderThreadId = ThreadId.make("thread-runtime-older"); | |
| 142 | const newerThreadId = ThreadId.make("thread-runtime-newer"); | |
| 143 | ||
| 144 | yield* runtimeRepository.upsert({ | |
| 145 | threadId: newerThreadId, | |
| 146 | providerName: "codex", | |
| 147 | adapterKey: "codex", | |
| 148 | runtimeMode: "full-access", | |
| 149 | status: "running", | |
| 150 | lastSeenAt: "2026-04-14T12:05:00.000Z", | |
| 151 | resumeCursor: { | |
| 152 | opaque: "resume-newer", | |
| 153 | }, | |
| 154 | runtimePayload: { | |
| 155 | cwd: "/tmp/newer", | |
| 156 | }, | |
| 157 | }); | |
| 158 | ||
| 159 | yield* runtimeRepository.upsert({ | |
| 160 | threadId: olderThreadId, | |
| 161 | providerName: "claudeAgent", | |
| 162 | adapterKey: "claudeAgent", | |
| 163 | runtimeMode: "approval-required", | |
| 164 | status: "starting", | |
| 165 | lastSeenAt: "2026-04-14T12:00:00.000Z", | |
| 166 | resumeCursor: { | |
| 167 | opaque: "resume-older", | |
| 168 | }, | |
| 169 | runtimePayload: { | |
| 170 | cwd: "/tmp/older", | |
| 171 | }, | |
| 172 | }); | |
| 173 | ||
| 174 | const bindings = yield* directory.listBindings(); | |
| 175 | ||
| 176 | assert.deepEqual(bindings, [ | |
| 177 | { | |
| 178 | threadId: olderThreadId, | |
| 179 | provider: "claudeAgent", | |
| 180 | adapterKey: "claudeAgent", | |
| 181 | runtimeMode: "approval-required", | |
| 182 | status: "starting", | |
| 183 | lastSeenAt: "2026-04-14T12:00:00.000Z", | |
| 184 | resumeCursor: { | |
| 185 | opaque: "resume-older", | |
| 186 | }, | |
| 187 | runtimePayload: { | |
| 188 | cwd: "/tmp/older", | |
| 189 | }, | |
| 190 | }, | |
| 191 | { | |
| 192 | threadId: newerThreadId, | |
| 193 | provider: "codex", | |
| 194 | adapterKey: "codex", | |
| 195 | runtimeMode: "full-access", | |
| 196 | status: "running", | |
| 197 | lastSeenAt: "2026-04-14T12:05:00.000Z", | |
| 198 | resumeCursor: { | |
| 199 | opaque: "resume-newer", | |
| 200 | }, | |
| 201 | runtimePayload: { | |
| 202 | cwd: "/tmp/newer", | |
| 203 | }, | |
| 204 | }, | |
| 205 | ]); | |
| 206 | })); | |
| 207 | ||
| 136 | 208 | it("resets adapterKey to the new provider when provider changes without an explicit adapter key", () => |
| 137 | 209 | Effect.gen(function* () { |
| 138 | 210 | const directory = yield* ProviderSessionDirectory; |
apps/server/src/provider/Layers/ProviderSessionDirectory.ts+38-12
| ... | ... | @@ -1,11 +1,13 @@ |
| 1 | 1 | import { type ProviderKind, type ThreadId } from "@t3tools/contracts"; |
| 2 | 2 | import { Effect, Layer, Option } from "effect"; |
| 3 | 3 | |
| 4 | import type { ProviderSessionRuntime } from "../../persistence/Services/ProviderSessionRuntime.ts"; | |
| 4 | 5 | import { ProviderSessionRuntimeRepository } from "../../persistence/Services/ProviderSessionRuntime.ts"; |
| 5 | 6 | import { ProviderSessionDirectoryPersistenceError, ProviderValidationError } from "../Errors.ts"; |
| 6 | 7 | import { |
| 7 | 8 | ProviderSessionDirectory, |
| 8 | 9 | type ProviderRuntimeBinding, |
| 10 | type ProviderRuntimeBindingWithMetadata, | |
| 9 | 11 | type ProviderSessionDirectoryShape, |
| 10 | 12 | } from "../Services/ProviderSessionDirectory.ts"; |
| 11 | 13 | |
| ... | ... | @@ -50,6 +52,27 @@ function mergeRuntimePayload( |
| 50 | 52 | return next; |
| 51 | 53 | } |
| 52 | 54 | |
| 55 | function toRuntimeBinding( | |
| 56 | runtime: ProviderSessionRuntime, | |
| 57 | operation: string, | |
| 58 | ): Effect.Effect<ProviderRuntimeBindingWithMetadata, ProviderSessionDirectoryPersistenceError> { | |
| 59 | return decodeProviderKind(runtime.providerName, operation).pipe( | |
| 60 | Effect.map( | |
| 61 | (provider) => | |
| 62 | ({ | |
| 63 | threadId: runtime.threadId, | |
| 64 | provider, | |
| 65 | adapterKey: runtime.adapterKey, | |
| 66 | runtimeMode: runtime.runtimeMode, | |
| 67 | status: runtime.status, | |
| 68 | resumeCursor: runtime.resumeCursor, | |
| 69 | runtimePayload: runtime.runtimePayload, | |
| 70 | lastSeenAt: runtime.lastSeenAt, | |
| 71 | }) satisfies ProviderRuntimeBindingWithMetadata, | |
| 72 | ), | |
| 73 | ); | |
| 74 | } | |
| 75 | ||
| 53 | 76 | const makeProviderSessionDirectory = Effect.gen(function* () { |
| 54 | 77 | const repository = yield* ProviderSessionRuntimeRepository; |
| 55 | 78 | |
| ... | ... | @@ -60,18 +83,8 @@ const makeProviderSessionDirectory = Effect.gen(function* () { |
| 60 | 83 | Option.match(runtime, { |
| 61 | 84 | onNone: () => Effect.succeed(Option.none<ProviderRuntimeBinding>()), |
| 62 | 85 | onSome: (value) => |
| 63 | decodeProviderKind(value.providerName, "ProviderSessionDirectory.getBinding").pipe( | |
| 64 | Effect.map((provider) => | |
| 65 | Option.some({ | |
| 66 | threadId: value.threadId, | |
| 67 | provider, | |
| 68 | adapterKey: value.adapterKey, | |
| 69 | runtimeMode: value.runtimeMode, | |
| 70 | status: value.status, | |
| 71 | resumeCursor: value.resumeCursor, | |
| 72 | runtimePayload: value.runtimePayload, | |
| 73 | }), | |
| 74 | ), | |
| 86 | toRuntimeBinding(value, "ProviderSessionDirectory.getBinding").pipe( | |
| 87 | Effect.map((binding) => Option.some(binding)), | |
| 75 | 88 | ), |
| 76 | 89 | }), |
| 77 | 90 | ), |
| ... | ... | @@ -145,12 +158,25 @@ const makeProviderSessionDirectory = Effect.gen(function* () { |
| 145 | 158 | Effect.map((rows) => rows.map((row) => row.threadId)), |
| 146 | 159 | ); |
| 147 | 160 | |
| 161 | const listBindings: ProviderSessionDirectoryShape["listBindings"] = () => | |
| 162 | repository.list().pipe( | |
| 163 | Effect.mapError(toPersistenceError("ProviderSessionDirectory.listBindings:list")), | |
| 164 | Effect.flatMap((rows) => | |
| 165 | Effect.forEach( | |
| 166 | rows, | |
| 167 | (row) => toRuntimeBinding(row, "ProviderSessionDirectory.listBindings"), | |
| 168 | { concurrency: "unbounded" }, | |
| 169 | ), | |
| 170 | ), | |
| 171 | ); | |
| 172 | ||
| 148 | 173 | return { |
| 149 | 174 | upsert, |
| 150 | 175 | getProvider, |
| 151 | 176 | getBinding, |
| 152 | 177 | remove, |
| 153 | 178 | listThreadIds, |
| 179 | listBindings, | |
| 154 | 180 | } satisfies ProviderSessionDirectoryShape; |
| 155 | 181 | }); |
| 156 | 182 |
apps/server/src/provider/Layers/ProviderSessionReaper.test.ts created+522
| ... | ... | @@ -0,0 +1,522 @@ |
| 1 | import * as NodeServices from "@effect/platform-node/NodeServices"; | |
| 2 | import { ProjectId, ThreadId, TurnId } from "@t3tools/contracts"; | |
| 3 | import { Effect, Exit, Layer, ManagedRuntime, Option, Scope, Stream } from "effect"; | |
| 4 | import { afterEach, describe, expect, it, vi } from "vitest"; | |
| 5 | ||
| 6 | import { | |
| 7 | OrchestrationEngineService, | |
| 8 | type OrchestrationEngineShape, | |
| 9 | } from "../../orchestration/Services/OrchestrationEngine.ts"; | |
| 10 | import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; | |
| 11 | import { ProviderSessionRuntimeRepositoryLive } from "../../persistence/Layers/ProviderSessionRuntime.ts"; | |
| 12 | import { ProviderSessionRuntimeRepository } from "../../persistence/Services/ProviderSessionRuntime.ts"; | |
| 13 | import { ProviderValidationError } from "../Errors.ts"; | |
| 14 | import { ProviderSessionReaper } from "../Services/ProviderSessionReaper.ts"; | |
| 15 | import { ProviderService, type ProviderServiceShape } from "../Services/ProviderService.ts"; | |
| 16 | import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts"; | |
| 17 | import { makeProviderSessionReaperLive } from "./ProviderSessionReaper.ts"; | |
| 18 | ||
| 19 | const defaultModelSelection = { | |
| 20 | provider: "codex", | |
| 21 | model: "gpt-5-codex", | |
| 22 | } as const; | |
| 23 | ||
| 24 | async function waitFor( | |
| 25 | predicate: () => boolean | Promise<boolean>, | |
| 26 | timeoutMs = 2_000, | |
| 27 | ): Promise<void> { | |
| 28 | const deadline = Date.now() + timeoutMs; | |
| 29 | const poll = async (): Promise<void> => { | |
| 30 | if (await predicate()) { | |
| 31 | return; | |
| 32 | } | |
| 33 | if (Date.now() >= deadline) { | |
| 34 | throw new Error("Timed out waiting for expectation."); | |
| 35 | } | |
| 36 | await new Promise((resolve) => setTimeout(resolve, 10)); | |
| 37 | return poll(); | |
| 38 | }; | |
| 39 | ||
| 40 | return poll(); | |
| 41 | } | |
| 42 | ||
| 43 | const unsupported = () => Effect.die(new Error("Unsupported provider call in test")) as never; | |
| 44 | ||
| 45 | function makeReadModel( | |
| 46 | threads: ReadonlyArray<{ | |
| 47 | readonly id: ThreadId; | |
| 48 | readonly session: { | |
| 49 | readonly threadId: ThreadId; | |
| 50 | readonly status: "starting" | "running" | "ready" | "interrupted" | "stopped" | "error"; | |
| 51 | readonly providerName: "codex" | "claudeAgent"; | |
| 52 | readonly runtimeMode: "approval-required" | "full-access" | "auto-accept-edits"; | |
| 53 | readonly activeTurnId: TurnId | null; | |
| 54 | readonly lastError: string | null; | |
| 55 | readonly updatedAt: string; | |
| 56 | } | null; | |
| 57 | }>, | |
| 58 | ) { | |
| 59 | const now = new Date().toISOString(); | |
| 60 | const projectId = ProjectId.make("project-provider-session-reaper"); | |
| 61 | ||
| 62 | return { | |
| 63 | snapshotSequence: 0, | |
| 64 | updatedAt: now, | |
| 65 | projects: [ | |
| 66 | { | |
| 67 | id: projectId, | |
| 68 | title: "Provider Reaper Project", | |
| 69 | workspaceRoot: "/tmp/provider-reaper-project", | |
| 70 | defaultModelSelection, | |
| 71 | scripts: [], | |
| 72 | createdAt: now, | |
| 73 | updatedAt: now, | |
| 74 | deletedAt: null, | |
| 75 | }, | |
| 76 | ], | |
| 77 | threads: threads.map((thread) => ({ | |
| 78 | id: thread.id, | |
| 79 | projectId, | |
| 80 | title: `Thread ${thread.id}`, | |
| 81 | modelSelection: defaultModelSelection, | |
| 82 | interactionMode: "default" as const, | |
| 83 | runtimeMode: "full-access" as const, | |
| 84 | branch: null, | |
| 85 | worktreePath: null, | |
| 86 | createdAt: now, | |
| 87 | updatedAt: now, | |
| 88 | archivedAt: null, | |
| 89 | latestTurn: null, | |
| 90 | messages: [], | |
| 91 | session: thread.session, | |
| 92 | activities: [], | |
| 93 | proposedPlans: [], | |
| 94 | checkpoints: [], | |
| 95 | deletedAt: null, | |
| 96 | })), | |
| 97 | }; | |
| 98 | } | |
| 99 | ||
| 100 | describe("ProviderSessionReaper", () => { | |
| 101 | let runtime: ManagedRuntime.ManagedRuntime< | |
| 102 | ProviderSessionReaper | ProviderSessionRuntimeRepository, | |
| 103 | unknown | |
| 104 | > | null = null; | |
| 105 | let scope: Scope.Closeable | null = null; | |
| 106 | ||
| 107 | afterEach(async () => { | |
| 108 | if (scope) { | |
| 109 | await Effect.runPromise(Scope.close(scope, Exit.void)); | |
| 110 | } | |
| 111 | scope = null; | |
| 112 | if (runtime) { | |
| 113 | await runtime.dispose(); | |
| 114 | } | |
| 115 | runtime = null; | |
| 116 | }); | |
| 117 | ||
| 118 | async function createHarness(input: { | |
| 119 | readonly readModel: ReturnType<typeof makeReadModel>; | |
| 120 | readonly stopSessionImplementation?: (input: { | |
| 121 | readonly threadId: ThreadId; | |
| 122 | }) => ReturnType<ProviderServiceShape["stopSession"]>; | |
| 123 | }) { | |
| 124 | const stoppedThreadIds = new Set<ThreadId>(); | |
| 125 | const stopSession = vi.fn<ProviderServiceShape["stopSession"]>( | |
| 126 | (request) => | |
| 127 | (input.stopSessionImplementation | |
| 128 | ? input.stopSessionImplementation(request) | |
| 129 | : Effect.sync(() => { | |
| 130 | stoppedThreadIds.add(request.threadId); | |
| 131 | })) as ReturnType<ProviderServiceShape["stopSession"]>, | |
| 132 | ); | |
| 133 | ||
| 134 | const providerService: ProviderServiceShape = { | |
| 135 | startSession: () => unsupported(), | |
| 136 | sendTurn: () => unsupported(), | |
| 137 | interruptTurn: () => unsupported(), | |
| 138 | respondToRequest: () => unsupported(), | |
| 139 | respondToUserInput: () => unsupported(), | |
| 140 | stopSession, | |
| 141 | listSessions: () => Effect.succeed([]), | |
| 142 | getCapabilities: () => Effect.succeed({ sessionModelSwitch: "in-session" }), | |
| 143 | rollbackConversation: () => unsupported(), | |
| 144 | streamEvents: Stream.empty, | |
| 145 | }; | |
| 146 | ||
| 147 | const orchestrationEngine: OrchestrationEngineShape = { | |
| 148 | getReadModel: () => Effect.succeed(input.readModel), | |
| 149 | readEvents: () => Stream.empty, | |
| 150 | dispatch: () => unsupported(), | |
| 151 | streamDomainEvents: Stream.empty, | |
| 152 | }; | |
| 153 | ||
| 154 | const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( | |
| 155 | Layer.provide(SqlitePersistenceMemory), | |
| 156 | ); | |
| 157 | const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe( | |
| 158 | Layer.provide(runtimeRepositoryLayer), | |
| 159 | ); | |
| 160 | const layer = makeProviderSessionReaperLive({ | |
| 161 | inactivityThresholdMs: 1_000, | |
| 162 | sweepIntervalMs: 60_000, | |
| 163 | }).pipe( | |
| 164 | Layer.provideMerge(providerSessionDirectoryLayer), | |
| 165 | Layer.provideMerge(runtimeRepositoryLayer), | |
| 166 | Layer.provideMerge(Layer.succeed(ProviderService, providerService)), | |
| 167 | Layer.provideMerge(Layer.succeed(OrchestrationEngineService, orchestrationEngine)), | |
| 168 | Layer.provideMerge(NodeServices.layer), | |
| 169 | ); | |
| 170 | ||
| 171 | runtime = ManagedRuntime.make(layer); | |
| 172 | return { stopSession, stoppedThreadIds }; | |
| 173 | } | |
| 174 | ||
| 175 | it("reaps stale persisted sessions without active turns", async () => { | |
| 176 | const threadId = ThreadId.make("thread-reaper-stale"); | |
| 177 | const now = new Date().toISOString(); | |
| 178 | const harness = await createHarness({ | |
| 179 | readModel: makeReadModel([ | |
| 180 | { | |
| 181 | id: threadId, | |
| 182 | session: { | |
| 183 | threadId, | |
| 184 | status: "ready", | |
| 185 | providerName: "claudeAgent", | |
| 186 | runtimeMode: "full-access", | |
| 187 | activeTurnId: null, | |
| 188 | lastError: null, | |
| 189 | updatedAt: now, | |
| 190 | }, | |
| 191 | }, | |
| 192 | ]), | |
| 193 | }); | |
| 194 | const repository = await runtime!.runPromise(Effect.service(ProviderSessionRuntimeRepository)); | |
| 195 | ||
| 196 | await runtime!.runPromise( | |
| 197 | repository.upsert({ | |
| 198 | threadId, | |
| 199 | providerName: "claudeAgent", | |
| 200 | adapterKey: "claudeAgent", | |
| 201 | runtimeMode: "full-access", | |
| 202 | status: "running", | |
| 203 | lastSeenAt: "2026-04-14T00:00:00.000Z", | |
| 204 | resumeCursor: { | |
| 205 | opaque: "resume-stale", | |
| 206 | }, | |
| 207 | runtimePayload: null, | |
| 208 | }), | |
| 209 | ); | |
| 210 | ||
| 211 | const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); | |
| 212 | scope = await Effect.runPromise(Scope.make("sequential")); | |
| 213 | await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); | |
| 214 | ||
| 215 | await waitFor(() => harness.stopSession.mock.calls.length === 1); | |
| 216 | ||
| 217 | expect(harness.stopSession.mock.calls[0]?.[0]).toEqual({ threadId }); | |
| 218 | expect(harness.stoppedThreadIds.has(threadId)).toBe(true); | |
| 219 | }); | |
| 220 | ||
| 221 | it("skips stale sessions when the thread still has an active turn", async () => { | |
| 222 | const threadId = ThreadId.make("thread-reaper-active-turn"); | |
| 223 | const turnId = TurnId.make("turn-reaper-active"); | |
| 224 | const now = new Date().toISOString(); | |
| 225 | const harness = await createHarness({ | |
| 226 | readModel: makeReadModel([ | |
| 227 | { | |
| 228 | id: threadId, | |
| 229 | session: { | |
| 230 | threadId, | |
| 231 | status: "running", | |
| 232 | providerName: "claudeAgent", | |
| 233 | runtimeMode: "full-access", | |
| 234 | activeTurnId: turnId, | |
| 235 | lastError: null, | |
| 236 | updatedAt: now, | |
| 237 | }, | |
| 238 | }, | |
| 239 | ]), | |
| 240 | }); | |
| 241 | const repository = await runtime!.runPromise(Effect.service(ProviderSessionRuntimeRepository)); | |
| 242 | ||
| 243 | await runtime!.runPromise( | |
| 244 | repository.upsert({ | |
| 245 | threadId, | |
| 246 | providerName: "claudeAgent", | |
| 247 | adapterKey: "claudeAgent", | |
| 248 | runtimeMode: "full-access", | |
| 249 | status: "running", | |
| 250 | lastSeenAt: "2026-04-14T00:00:00.000Z", | |
| 251 | resumeCursor: { | |
| 252 | opaque: "resume-active-turn", | |
| 253 | }, | |
| 254 | runtimePayload: null, | |
| 255 | }), | |
| 256 | ); | |
| 257 | ||
| 258 | const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); | |
| 259 | scope = await Effect.runPromise(Scope.make("sequential")); | |
| 260 | await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); | |
| 261 | await new Promise((resolve) => setTimeout(resolve, 50)); | |
| 262 | ||
| 263 | expect(harness.stopSession).not.toHaveBeenCalled(); | |
| 264 | const remaining = await runtime!.runPromise(repository.getByThreadId({ threadId })); | |
| 265 | expect(Option.isSome(remaining)).toBe(true); | |
| 266 | }); | |
| 267 | ||
| 268 | it("does not reap sessions that are still within the inactivity threshold", async () => { | |
| 269 | const threadId = ThreadId.make("thread-reaper-fresh"); | |
| 270 | const now = new Date().toISOString(); | |
| 271 | const harness = await createHarness({ | |
| 272 | readModel: makeReadModel([ | |
| 273 | { | |
| 274 | id: threadId, | |
| 275 | session: { | |
| 276 | threadId, | |
| 277 | status: "ready", | |
| 278 | providerName: "claudeAgent", | |
| 279 | runtimeMode: "full-access", | |
| 280 | activeTurnId: null, | |
| 281 | lastError: null, | |
| 282 | updatedAt: now, | |
| 283 | }, | |
| 284 | }, | |
| 285 | ]), | |
| 286 | }); | |
| 287 | const repository = await runtime!.runPromise(Effect.service(ProviderSessionRuntimeRepository)); | |
| 288 | ||
| 289 | await runtime!.runPromise( | |
| 290 | repository.upsert({ | |
| 291 | threadId, | |
| 292 | providerName: "claudeAgent", | |
| 293 | adapterKey: "claudeAgent", | |
| 294 | runtimeMode: "full-access", | |
| 295 | status: "running", | |
| 296 | lastSeenAt: now, | |
| 297 | resumeCursor: { | |
| 298 | opaque: "resume-fresh", | |
| 299 | }, | |
| 300 | runtimePayload: null, | |
| 301 | }), | |
| 302 | ); | |
| 303 | ||
| 304 | const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); | |
| 305 | scope = await Effect.runPromise(Scope.make("sequential")); | |
| 306 | await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); | |
| 307 | await new Promise((resolve) => setTimeout(resolve, 50)); | |
| 308 | ||
| 309 | expect(harness.stopSession).not.toHaveBeenCalled(); | |
| 310 | const remaining = await runtime!.runPromise(repository.getByThreadId({ threadId })); | |
| 311 | expect(Option.isSome(remaining)).toBe(true); | |
| 312 | }); | |
| 313 | ||
| 314 | it("skips persisted sessions that are already marked stopped", async () => { | |
| 315 | const threadId = ThreadId.make("thread-reaper-stopped"); | |
| 316 | const now = new Date().toISOString(); | |
| 317 | const harness = await createHarness({ | |
| 318 | readModel: makeReadModel([ | |
| 319 | { | |
| 320 | id: threadId, | |
| 321 | session: { | |
| 322 | threadId, | |
| 323 | status: "stopped", | |
| 324 | providerName: "claudeAgent", | |
| 325 | runtimeMode: "full-access", | |
| 326 | activeTurnId: null, | |
| 327 | lastError: null, | |
| 328 | updatedAt: now, | |
| 329 | }, | |
| 330 | }, | |
| 331 | ]), | |
| 332 | }); | |
| 333 | const repository = await runtime!.runPromise(Effect.service(ProviderSessionRuntimeRepository)); | |
| 334 | ||
| 335 | await runtime!.runPromise( | |
| 336 | repository.upsert({ | |
| 337 | threadId, | |
| 338 | providerName: "claudeAgent", | |
| 339 | adapterKey: "claudeAgent", | |
| 340 | runtimeMode: "full-access", | |
| 341 | status: "stopped", | |
| 342 | lastSeenAt: "2026-04-14T00:00:00.000Z", | |
| 343 | resumeCursor: { | |
| 344 | opaque: "resume-stopped", | |
| 345 | }, | |
| 346 | runtimePayload: null, | |
| 347 | }), | |
| 348 | ); | |
| 349 | ||
| 350 | const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); | |
| 351 | scope = await Effect.runPromise(Scope.make("sequential")); | |
| 352 | await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); | |
| 353 | await new Promise((resolve) => setTimeout(resolve, 50)); | |
| 354 | ||
| 355 | expect(harness.stopSession).not.toHaveBeenCalled(); | |
| 356 | const remaining = await runtime!.runPromise(repository.getByThreadId({ threadId })); | |
| 357 | expect(Option.isSome(remaining)).toBe(true); | |
| 358 | }); | |
| 359 | ||
| 360 | it("continues reaping other sessions when one stop attempt fails", async () => { | |
| 361 | const failedThreadId = ThreadId.make("thread-reaper-stop-failure"); | |
| 362 | const reapedThreadId = ThreadId.make("thread-reaper-stop-success"); | |
| 363 | const now = new Date().toISOString(); | |
| 364 | const harness = await createHarness({ | |
| 365 | readModel: makeReadModel([ | |
| 366 | { | |
| 367 | id: failedThreadId, | |
| 368 | session: { | |
| 369 | threadId: failedThreadId, | |
| 370 | status: "ready", | |
| 371 | providerName: "claudeAgent", | |
| 372 | runtimeMode: "full-access", | |
| 373 | activeTurnId: null, | |
| 374 | lastError: null, | |
| 375 | updatedAt: now, | |
| 376 | }, | |
| 377 | }, | |
| 378 | { | |
| 379 | id: reapedThreadId, | |
| 380 | session: { | |
| 381 | threadId: reapedThreadId, | |
| 382 | status: "ready", | |
| 383 | providerName: "codex", | |
| 384 | runtimeMode: "full-access", | |
| 385 | activeTurnId: null, | |
| 386 | lastError: null, | |
| 387 | updatedAt: now, | |
| 388 | }, | |
| 389 | }, | |
| 390 | ]), | |
| 391 | stopSessionImplementation: (request) => | |
| 392 | request.threadId === failedThreadId | |
| 393 | ? Effect.fail( | |
| 394 | new ProviderValidationError({ | |
| 395 | operation: "ProviderSessionReaper.test", | |
| 396 | issue: "simulated stop failure", | |
| 397 | }), | |
| 398 | ) | |
| 399 | : Effect.void, | |
| 400 | }); | |
| 401 | const repository = await runtime!.runPromise(Effect.service(ProviderSessionRuntimeRepository)); | |
| 402 | ||
| 403 | await runtime!.runPromise( | |
| 404 | repository.upsert({ | |
| 405 | threadId: failedThreadId, | |
| 406 | providerName: "claudeAgent", | |
| 407 | adapterKey: "claudeAgent", | |
| 408 | runtimeMode: "full-access", | |
| 409 | status: "running", | |
| 410 | lastSeenAt: "2026-04-14T00:00:00.000Z", | |
| 411 | resumeCursor: { | |
| 412 | opaque: "resume-failure", | |
| 413 | }, | |
| 414 | runtimePayload: null, | |
| 415 | }), | |
| 416 | ); | |
| 417 | await runtime!.runPromise( | |
| 418 | repository.upsert({ | |
| 419 | threadId: reapedThreadId, | |
| 420 | providerName: "codex", | |
| 421 | adapterKey: "codex", | |
| 422 | runtimeMode: "full-access", | |
| 423 | status: "running", | |
| 424 | lastSeenAt: "2026-04-14T00:01:00.000Z", | |
| 425 | resumeCursor: { | |
| 426 | opaque: "resume-success", | |
| 427 | }, | |
| 428 | runtimePayload: null, | |
| 429 | }), | |
| 430 | ); | |
| 431 | ||
| 432 | const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); | |
| 433 | scope = await Effect.runPromise(Scope.make("sequential")); | |
| 434 | await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); | |
| 435 | ||
| 436 | await waitFor(() => harness.stopSession.mock.calls.length === 2); | |
| 437 | ||
| 438 | expect(harness.stopSession.mock.calls.map(([request]) => request.threadId)).toEqual([ | |
| 439 | failedThreadId, | |
| 440 | reapedThreadId, | |
| 441 | ]); | |
| 442 | }); | |
| 443 | ||
| 444 | it("continues reaping other sessions when one stop attempt defects", async () => { | |
| 445 | const defectThreadId = ThreadId.make("thread-reaper-stop-defect"); | |
| 446 | const reapedThreadId = ThreadId.make("thread-reaper-stop-after-defect"); | |
| 447 | const now = new Date().toISOString(); | |
| 448 | const harness = await createHarness({ | |
| 449 | readModel: makeReadModel([ | |
| 450 | { | |
| 451 | id: defectThreadId, | |
| 452 | session: { | |
| 453 | threadId: defectThreadId, | |
| 454 | status: "ready", | |
| 455 | providerName: "claudeAgent", | |
| 456 | runtimeMode: "full-access", | |
| 457 | activeTurnId: null, | |
| 458 | lastError: null, | |
| 459 | updatedAt: now, | |
| 460 | }, | |
| 461 | }, | |
| 462 | { | |
| 463 | id: reapedThreadId, | |
| 464 | session: { | |
| 465 | threadId: reapedThreadId, | |
| 466 | status: "ready", | |
| 467 | providerName: "codex", | |
| 468 | runtimeMode: "full-access", | |
| 469 | activeTurnId: null, | |
| 470 | lastError: null, | |
| 471 | updatedAt: now, | |
| 472 | }, | |
| 473 | }, | |
| 474 | ]), | |
| 475 | stopSessionImplementation: (request) => | |
| 476 | request.threadId === defectThreadId | |
| 477 | ? Effect.die(new Error("simulated stop defect")) | |
| 478 | : Effect.void, | |
| 479 | }); | |
| 480 | const repository = await runtime!.runPromise(Effect.service(ProviderSessionRuntimeRepository)); | |
| 481 | ||
| 482 | await runtime!.runPromise( | |
| 483 | repository.upsert({ | |
| 484 | threadId: defectThreadId, | |
| 485 | providerName: "claudeAgent", | |
| 486 | adapterKey: "claudeAgent", | |
| 487 | runtimeMode: "full-access", | |
| 488 | status: "running", | |
| 489 | lastSeenAt: "2026-04-14T00:00:00.000Z", | |
| 490 | resumeCursor: { | |
| 491 | opaque: "resume-defect", | |
| 492 | }, | |
| 493 | runtimePayload: null, | |
| 494 | }), | |
| 495 | ); | |
| 496 | await runtime!.runPromise( | |
| 497 | repository.upsert({ | |
| 498 | threadId: reapedThreadId, | |
| 499 | providerName: "codex", | |
| 500 | adapterKey: "codex", | |
| 501 | runtimeMode: "full-access", | |
| 502 | status: "running", | |
| 503 | lastSeenAt: "2026-04-14T00:01:00.000Z", | |
| 504 | resumeCursor: { | |
| 505 | opaque: "resume-after-defect", | |
| 506 | }, | |
| 507 | runtimePayload: null, | |
| 508 | }), | |
| 509 | ); | |
| 510 | ||
| 511 | const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); | |
| 512 | scope = await Effect.runPromise(Scope.make("sequential")); | |
| 513 | await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); | |
| 514 | ||
| 515 | await waitFor(() => harness.stopSession.mock.calls.length === 2); | |
| 516 | ||
| 517 | expect(harness.stopSession.mock.calls.map(([request]) => request.threadId)).toEqual([ | |
| 518 | defectThreadId, | |
| 519 | reapedThreadId, | |
| 520 | ]); | |
| 521 | }); | |
| 522 | }); |
apps/server/src/provider/Layers/ProviderSessionReaper.ts created+133
| ... | ... | @@ -0,0 +1,133 @@ |
| 1 | import { Duration, Effect, Layer, Schedule } from "effect"; | |
| 2 | ||
| 3 | import { OrchestrationEngineService } from "../../orchestration/Services/OrchestrationEngine.ts"; | |
| 4 | import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; | |
| 5 | import { | |
| 6 | ProviderSessionReaper, | |
| 7 | type ProviderSessionReaperShape, | |
| 8 | } from "../Services/ProviderSessionReaper.ts"; | |
| 9 | import { ProviderService } from "../Services/ProviderService.ts"; | |
| 10 | ||
| 11 | const DEFAULT_INACTIVITY_THRESHOLD_MS = 30 * 60 * 1000; | |
| 12 | const DEFAULT_SWEEP_INTERVAL_MS = 5 * 60 * 1000; | |
| 13 | ||
| 14 | export interface ProviderSessionReaperLiveOptions { | |
| 15 | readonly inactivityThresholdMs?: number; | |
| 16 | readonly sweepIntervalMs?: number; | |
| 17 | } | |
| 18 | ||
| 19 | const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) => | |
| 20 | Effect.gen(function* () { | |
| 21 | const providerService = yield* ProviderService; | |
| 22 | const directory = yield* ProviderSessionDirectory; | |
| 23 | const orchestrationEngine = yield* OrchestrationEngineService; | |
| 24 | ||
| 25 | const inactivityThresholdMs = Math.max( | |
| 26 | 1, | |
| 27 | options?.inactivityThresholdMs ?? DEFAULT_INACTIVITY_THRESHOLD_MS, | |
| 28 | ); | |
| 29 | const sweepIntervalMs = Math.max(1, options?.sweepIntervalMs ?? DEFAULT_SWEEP_INTERVAL_MS); | |
| 30 | ||
| 31 | const sweep = Effect.gen(function* () { | |
| 32 | const readModel = yield* orchestrationEngine.getReadModel(); | |
| 33 | const threadsById = new Map(readModel.threads.map((thread) => [thread.id, thread] as const)); | |
| 34 | const bindings = yield* directory.listBindings(); | |
| 35 | const now = Date.now(); | |
| 36 | let reapedCount = 0; | |
| 37 | ||
| 38 | for (const binding of bindings) { | |
| 39 | if (binding.status === "stopped") { | |
| 40 | continue; | |
| 41 | } | |
| 42 | ||
| 43 | const lastSeenMs = Date.parse(binding.lastSeenAt); | |
| 44 | if (Number.isNaN(lastSeenMs)) { | |
| 45 | yield* Effect.logWarning("provider.session.reaper.invalid-last-seen", { | |
| 46 | threadId: binding.threadId, | |
| 47 | provider: binding.provider, | |
| 48 | lastSeenAt: binding.lastSeenAt, | |
| 49 | }); | |
| 50 | continue; | |
| 51 | } | |
| 52 | ||
| 53 | const idleDurationMs = now - lastSeenMs; | |
| 54 | if (idleDurationMs < inactivityThresholdMs) { | |
| 55 | continue; | |
| 56 | } | |
| 57 | ||
| 58 | const thread = threadsById.get(binding.threadId); | |
| 59 | if (thread?.session?.activeTurnId != null) { | |
| 60 | yield* Effect.logDebug("provider.session.reaper.skipped-active-turn", { | |
| 61 | threadId: binding.threadId, | |
| 62 | activeTurnId: thread.session.activeTurnId, | |
| 63 | idleDurationMs, | |
| 64 | }); | |
| 65 | continue; | |
| 66 | } | |
| 67 | ||
| 68 | const reaped = yield* providerService.stopSession({ threadId: binding.threadId }).pipe( | |
| 69 | Effect.tap(() => | |
| 70 | Effect.logInfo("provider.session.reaped", { | |
| 71 | threadId: binding.threadId, | |
| 72 | provider: binding.provider, | |
| 73 | idleDurationMs, | |
| 74 | reason: "inactivity_threshold", | |
| 75 | }), | |
| 76 | ), | |
| 77 | Effect.as(true), | |
| 78 | Effect.catchCause((cause) => | |
| 79 | Effect.logWarning("provider.session.reaper.stop-failed", { | |
| 80 | threadId: binding.threadId, | |
| 81 | provider: binding.provider, | |
| 82 | idleDurationMs, | |
| 83 | cause, | |
| 84 | }).pipe(Effect.as(false)), | |
| 85 | ), | |
| 86 | ); | |
| 87 | ||
| 88 | if (reaped) { | |
| 89 | reapedCount += 1; | |
| 90 | } | |
| 91 | } | |
| 92 | ||
| 93 | if (reapedCount > 0) { | |
| 94 | yield* Effect.logInfo("provider.session.reaper.sweep-complete", { | |
| 95 | reapedCount, | |
| 96 | totalBindings: bindings.length, | |
| 97 | }); | |
| 98 | } | |
| 99 | }); | |
| 100 | ||
| 101 | const start: ProviderSessionReaperShape["start"] = () => | |
| 102 | Effect.gen(function* () { | |
| 103 | yield* Effect.forkScoped( | |
| 104 | sweep.pipe( | |
| 105 | Effect.catch((error: unknown) => | |
| 106 | Effect.logWarning("provider.session.reaper.sweep-failed", { | |
| 107 | error, | |
| 108 | }), | |
| 109 | ), | |
| 110 | Effect.catchDefect((defect: unknown) => | |
| 111 | Effect.logWarning("provider.session.reaper.sweep-defect", { | |
| 112 | defect, | |
| 113 | }), | |
| 114 | ), | |
| 115 | Effect.repeat(Schedule.spaced(Duration.millis(sweepIntervalMs))), | |
| 116 | ), | |
| 117 | ); | |
| 118 | ||
| 119 | yield* Effect.logInfo("provider.session.reaper.started", { | |
| 120 | inactivityThresholdMs, | |
| 121 | sweepIntervalMs, | |
| 122 | }); | |
| 123 | }); | |
| 124 | ||
| 125 | return { | |
| 126 | start, | |
| 127 | } satisfies ProviderSessionReaperShape; | |
| 128 | }); | |
| 129 | ||
| 130 | export const makeProviderSessionReaperLive = (options?: ProviderSessionReaperLiveOptions) => | |
| 131 | Layer.effect(ProviderSessionReaper, makeProviderSessionReaper(options)); | |
| 132 | ||
| 133 | export const ProviderSessionReaperLive = makeProviderSessionReaperLive(); |
apps/server/src/provider/Services/ProviderSessionDirectory.ts+9
| ... | ... | @@ -22,6 +22,10 @@ export interface ProviderRuntimeBinding { |
| 22 | 22 | readonly runtimeMode?: RuntimeMode; |
| 23 | 23 | } |
| 24 | 24 | |
| 25 | export interface ProviderRuntimeBindingWithMetadata extends ProviderRuntimeBinding { | |
| 26 | readonly lastSeenAt: string; | |
| 27 | } | |
| 28 | ||
| 25 | 29 | export type ProviderSessionDirectoryReadError = ProviderSessionDirectoryPersistenceError; |
| 26 | 30 | |
| 27 | 31 | export type ProviderSessionDirectoryWriteError = |
| ... | ... | @@ -49,6 +53,11 @@ export interface ProviderSessionDirectoryShape { |
| 49 | 53 | ReadonlyArray<ThreadId>, |
| 50 | 54 | ProviderSessionDirectoryPersistenceError |
| 51 | 55 | >; |
| 56 | ||
| 57 | readonly listBindings: () => Effect.Effect< | |
| 58 | ReadonlyArray<ProviderRuntimeBindingWithMetadata>, | |
| 59 | ProviderSessionDirectoryPersistenceError | |
| 60 | >; | |
| 52 | 61 | } |
| 53 | 62 | |
| 54 | 63 | export class ProviderSessionDirectory extends Context.Service< |
apps/server/src/provider/Services/ProviderSessionReaper.ts created+14
| ... | ... | @@ -0,0 +1,14 @@ |
| 1 | import { Context } from "effect"; | |
| 2 | import type { Effect, Scope } from "effect"; | |
| 3 | ||
| 4 | export interface ProviderSessionReaperShape { | |
| 5 | /** | |
| 6 | * Start the background provider session reaper within the provided scope. | |
| 7 | */ | |
| 8 | readonly start: () => Effect.Effect<void, never, Scope.Scope>; | |
| 9 | } | |
| 10 | ||
| 11 | export class ProviderSessionReaper extends Context.Service< | |
| 12 | ProviderSessionReaper, | |
| 13 | ProviderSessionReaperShape | |
| 14 | >()("t3/provider/Services/ProviderSessionReaper") {} |
apps/server/src/server.test.ts+415-6
| ... | ... | @@ -11,6 +11,7 @@ import { |
| 11 | 11 | KeybindingRule, |
| 12 | 12 | MessageId, |
| 13 | 13 | OpenError, |
| 14 | type OrchestrationThreadShell, | |
| 14 | 15 | TerminalNotRunningError, |
| 15 | 16 | type OrchestrationCommand, |
| 16 | 17 | type OrchestrationEvent, |
| ... | ... | @@ -166,6 +167,32 @@ const makeDefaultOrchestrationReadModel = () => { |
| 166 | 167 | }; |
| 167 | 168 | }; |
| 168 | 169 | |
| 170 | const makeDefaultOrchestrationThreadShell = ( | |
| 171 | overrides: Partial<OrchestrationThreadShell> = {}, | |
| 172 | ): OrchestrationThreadShell => { | |
| 173 | const now = new Date().toISOString(); | |
| 174 | return { | |
| 175 | id: defaultThreadId, | |
| 176 | projectId: defaultProjectId, | |
| 177 | title: "Default Thread", | |
| 178 | modelSelection: defaultModelSelection, | |
| 179 | runtimeMode: "full-access", | |
| 180 | interactionMode: "default", | |
| 181 | branch: null, | |
| 182 | worktreePath: null, | |
| 183 | latestTurn: null, | |
| 184 | createdAt: now, | |
| 185 | updatedAt: now, | |
| 186 | archivedAt: null, | |
| 187 | session: null, | |
| 188 | latestUserMessageAt: null, | |
| 189 | hasPendingApprovals: false, | |
| 190 | hasPendingUserInput: false, | |
| 191 | hasActionableProposedPlan: false, | |
| 192 | ...overrides, | |
| 193 | }; | |
| 194 | }; | |
| 195 | ||
| 169 | 196 | const workspaceAndProjectServicesLayer = Layer.mergeAll( |
| 170 | 197 | WorkspacePathsLive, |
| 171 | 198 | WorkspaceEntriesLive.pipe(Layer.provide(WorkspacePathsLive)), |
| ... | ... | @@ -2945,21 +2972,48 @@ it.layer(NodeServices.layer)("server router seam", (it) => { |
| 2945 | 2972 | }).pipe(Effect.provide(NodeHttpServer.layerTest)), |
| 2946 | 2973 | ); |
| 2947 | 2974 | |
| 2948 | it.effect("closes thread terminals after a successful archive command", () => | |
| 2975 | it.effect("stops the provider session and closes thread terminals after archive", () => | |
| 2949 | 2976 | Effect.gen(function* () { |
| 2950 | 2977 | const threadId = ThreadId.make("thread-archive"); |
| 2951 | const closeInputs: Array<Parameters<TerminalManagerShape["close"]>[0]> = []; | |
| 2978 | const effects: string[] = []; | |
| 2979 | const dispatchedCommands: Array<OrchestrationCommand> = []; | |
| 2980 | const now = new Date().toISOString(); | |
| 2952 | 2981 | |
| 2953 | 2982 | yield* buildAppUnderTest({ |
| 2954 | 2983 | layers: { |
| 2955 | 2984 | terminalManager: { |
| 2956 | 2985 | close: (input) => |
| 2957 | 2986 | Effect.sync(() => { |
| 2958 | closeInputs.push(input); | |
| 2987 | effects.push(`terminal.close:${input.threadId}`); | |
| 2959 | 2988 | }), |
| 2960 | 2989 | }, |
| 2961 | 2990 | orchestrationEngine: { |
| 2962 | dispatch: () => Effect.succeed({ sequence: 8 }), | |
| 2991 | dispatch: (command) => | |
| 2992 | Effect.sync(() => { | |
| 2993 | dispatchedCommands.push(command); | |
| 2994 | effects.push(`dispatch:${command.type}`); | |
| 2995 | return { sequence: dispatchedCommands.length }; | |
| 2996 | }), | |
| 2997 | }, | |
| 2998 | projectionSnapshotQuery: { | |
| 2999 | getThreadShellById: () => | |
| 3000 | Effect.succeed( | |
| 3001 | Option.some( | |
| 3002 | makeDefaultOrchestrationThreadShell({ | |
| 3003 | id: threadId, | |
| 3004 | updatedAt: now, | |
| 3005 | session: { | |
| 3006 | threadId, | |
| 3007 | status: "ready", | |
| 3008 | providerName: "claudeAgent", | |
| 3009 | runtimeMode: "full-access", | |
| 3010 | activeTurnId: null, | |
| 3011 | lastError: null, | |
| 3012 | updatedAt: now, | |
| 3013 | }, | |
| 3014 | }), | |
| 3015 | ), | |
| 3016 | ), | |
| 2963 | 3017 | }, |
| 2964 | 3018 | }, |
| 2965 | 3019 | }); |
| ... | ... | @@ -2975,8 +3029,363 @@ it.layer(NodeServices.layer)("server router seam", (it) => { |
| 2975 | 3029 | ), |
| 2976 | 3030 | ); |
| 2977 | 3031 | |
| 2978 | assert.equal(dispatchResult.sequence, 8); | |
| 2979 | assert.deepEqual(closeInputs, [{ threadId }]); | |
| 3032 | assert.equal(dispatchResult.sequence, 1); | |
| 3033 | assert.deepEqual(effects, [ | |
| 3034 | "dispatch:thread.archive", | |
| 3035 | "dispatch:thread.session.stop", | |
| 3036 | `terminal.close:${threadId}`, | |
| 3037 | ]); | |
| 3038 | const sessionStopCommand = dispatchedCommands[1]; | |
| 3039 | assert.equal(sessionStopCommand?.type, "thread.session.stop"); | |
| 3040 | if (sessionStopCommand?.type === "thread.session.stop") { | |
| 3041 | assert.equal(sessionStopCommand.threadId, threadId); | |
| 3042 | } | |
| 3043 | }).pipe(Effect.provide(NodeHttpServer.layerTest)), | |
| 3044 | ); | |
| 3045 | ||
| 3046 | it.effect("checks session status before archiving removes the thread from active lookups", () => | |
| 3047 | Effect.gen(function* () { | |
| 3048 | const threadId = ThreadId.make("thread-archive-precheck"); | |
| 3049 | const effects: string[] = []; | |
| 3050 | const dispatchedCommands: Array<OrchestrationCommand> = []; | |
| 3051 | const now = new Date().toISOString(); | |
| 3052 | let archived = false; | |
| 3053 | ||
| 3054 | yield* buildAppUnderTest({ | |
| 3055 | layers: { | |
| 3056 | terminalManager: { | |
| 3057 | close: (input) => | |
| 3058 | Effect.sync(() => { | |
| 3059 | effects.push(`terminal.close:${input.threadId}`); | |
| 3060 | }), | |
| 3061 | }, | |
| 3062 | orchestrationEngine: { | |
| 3063 | dispatch: (command) => | |
| 3064 | Effect.sync(() => { | |
| 3065 | dispatchedCommands.push(command); | |
| 3066 | effects.push(`dispatch:${command.type}`); | |
| 3067 | if (command.type === "thread.archive") { | |
| 3068 | archived = true; | |
| 3069 | } | |
| 3070 | return { sequence: dispatchedCommands.length }; | |
| 3071 | }), | |
| 3072 | }, | |
| 3073 | projectionSnapshotQuery: { | |
| 3074 | getThreadShellById: () => | |
| 3075 | Effect.sync(() => { | |
| 3076 | effects.push(`query:thread-shell:${archived ? "archived" : "active"}`); | |
| 3077 | return archived | |
| 3078 | ? Option.none() | |
| 3079 | : Option.some( | |
| 3080 | makeDefaultOrchestrationThreadShell({ | |
| 3081 | id: threadId, | |
| 3082 | updatedAt: now, | |
| 3083 | session: { | |
| 3084 | threadId, | |
| 3085 | status: "ready", | |
| 3086 | providerName: "claudeAgent", | |
| 3087 | runtimeMode: "full-access", | |
| 3088 | activeTurnId: null, | |
| 3089 | lastError: null, | |
| 3090 | updatedAt: now, | |
| 3091 | }, | |
| 3092 | }), | |
| 3093 | ); | |
| 3094 | }), | |
| 3095 | }, | |
| 3096 | }, | |
| 3097 | }); | |
| 3098 | ||
| 3099 | const wsUrl = yield* getWsServerUrl("/ws"); | |
| 3100 | const dispatchResult = yield* Effect.scoped( | |
| 3101 | withWsRpcClient(wsUrl, (client) => | |
| 3102 | client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ | |
| 3103 | type: "thread.archive", | |
| 3104 | commandId: CommandId.make("cmd-thread-archive-precheck"), | |
| 3105 | threadId, | |
| 3106 | }), | |
| 3107 | ), | |
| 3108 | ); | |
| 3109 | ||
| 3110 | assert.equal(dispatchResult.sequence, 1); | |
| 3111 | assert.deepEqual(effects, [ | |
| 3112 | "query:thread-shell:active", | |
| 3113 | "dispatch:thread.archive", | |
| 3114 | "dispatch:thread.session.stop", | |
| 3115 | `terminal.close:${threadId}`, | |
| 3116 | ]); | |
| 3117 | assert.deepEqual( | |
| 3118 | dispatchedCommands.map((command) => command.type), | |
| 3119 | ["thread.archive", "thread.session.stop"], | |
| 3120 | ); | |
| 3121 | }).pipe(Effect.provide(NodeHttpServer.layerTest)), | |
| 3122 | ); | |
| 3123 | ||
| 3124 | it.effect("archives without dispatching session stop when the thread has no session", () => | |
| 3125 | Effect.gen(function* () { | |
| 3126 | const threadId = ThreadId.make("thread-archive-no-session"); | |
| 3127 | const effects: string[] = []; | |
| 3128 | const dispatchedCommands: Array<OrchestrationCommand> = []; | |
| 3129 | ||
| 3130 | yield* buildAppUnderTest({ | |
| 3131 | layers: { | |
| 3132 | terminalManager: { | |
| 3133 | close: (input) => | |
| 3134 | Effect.sync(() => { | |
| 3135 | effects.push(`terminal.close:${input.threadId}`); | |
| 3136 | }), | |
| 3137 | }, | |
| 3138 | orchestrationEngine: { | |
| 3139 | dispatch: (command) => | |
| 3140 | Effect.sync(() => { | |
| 3141 | dispatchedCommands.push(command); | |
| 3142 | effects.push(`dispatch:${command.type}`); | |
| 3143 | return { sequence: dispatchedCommands.length }; | |
| 3144 | }), | |
| 3145 | }, | |
| 3146 | projectionSnapshotQuery: { | |
| 3147 | getThreadShellById: () => | |
| 3148 | Effect.succeed( | |
| 3149 | Option.some(makeDefaultOrchestrationThreadShell({ id: threadId, session: null })), | |
| 3150 | ), | |
| 3151 | }, | |
| 3152 | }, | |
| 3153 | }); | |
| 3154 | ||
| 3155 | const wsUrl = yield* getWsServerUrl("/ws"); | |
| 3156 | const dispatchResult = yield* Effect.scoped( | |
| 3157 | withWsRpcClient(wsUrl, (client) => | |
| 3158 | client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ | |
| 3159 | type: "thread.archive", | |
| 3160 | commandId: CommandId.make("cmd-thread-archive-no-session"), | |
| 3161 | threadId, | |
| 3162 | }), | |
| 3163 | ), | |
| 3164 | ); | |
| 3165 | ||
| 3166 | assert.equal(dispatchResult.sequence, 1); | |
| 3167 | assert.deepEqual(effects, ["dispatch:thread.archive", `terminal.close:${threadId}`]); | |
| 3168 | assert.deepEqual( | |
| 3169 | dispatchedCommands.map((command) => command.type), | |
| 3170 | ["thread.archive"], | |
| 3171 | ); | |
| 3172 | }).pipe(Effect.provide(NodeHttpServer.layerTest)), | |
| 3173 | ); | |
| 3174 | ||
| 3175 | it.effect( | |
| 3176 | "archives without dispatching session stop when the thread session is already stopped", | |
| 3177 | () => | |
| 3178 | Effect.gen(function* () { | |
| 3179 | const threadId = ThreadId.make("thread-archive-stopped-session"); | |
| 3180 | const effects: string[] = []; | |
| 3181 | const dispatchedCommands: Array<OrchestrationCommand> = []; | |
| 3182 | const now = new Date().toISOString(); | |
| 3183 | ||
| 3184 | yield* buildAppUnderTest({ | |
| 3185 | layers: { | |
| 3186 | terminalManager: { | |
| 3187 | close: (input) => | |
| 3188 | Effect.sync(() => { | |
| 3189 | effects.push(`terminal.close:${input.threadId}`); | |
| 3190 | }), | |
| 3191 | }, | |
| 3192 | orchestrationEngine: { | |
| 3193 | dispatch: (command) => | |
| 3194 | Effect.sync(() => { | |
| 3195 | dispatchedCommands.push(command); | |
| 3196 | effects.push(`dispatch:${command.type}`); | |
| 3197 | return { sequence: dispatchedCommands.length }; | |
| 3198 | }), | |
| 3199 | }, | |
| 3200 | projectionSnapshotQuery: { | |
| 3201 | getThreadShellById: () => | |
| 3202 | Effect.succeed( | |
| 3203 | Option.some( | |
| 3204 | makeDefaultOrchestrationThreadShell({ | |
| 3205 | id: threadId, | |
| 3206 | updatedAt: now, | |
| 3207 | session: { | |
| 3208 | threadId, | |
| 3209 | status: "stopped", | |
| 3210 | providerName: "claudeAgent", | |
| 3211 | runtimeMode: "full-access", | |
| 3212 | activeTurnId: null, | |
| 3213 | lastError: null, | |
| 3214 | updatedAt: now, | |
| 3215 | }, | |
| 3216 | }), | |
| 3217 | ), | |
| 3218 | ), | |
| 3219 | }, | |
| 3220 | }, | |
| 3221 | }); | |
| 3222 | ||
| 3223 | const wsUrl = yield* getWsServerUrl("/ws"); | |
| 3224 | const dispatchResult = yield* Effect.scoped( | |
| 3225 | withWsRpcClient(wsUrl, (client) => | |
| 3226 | client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ | |
| 3227 | type: "thread.archive", | |
| 3228 | commandId: CommandId.make("cmd-thread-archive-stopped-session"), | |
| 3229 | threadId, | |
| 3230 | }), | |
| 3231 | ), | |
| 3232 | ); | |
| 3233 | ||
| 3234 | assert.equal(dispatchResult.sequence, 1); | |
| 3235 | assert.deepEqual(effects, ["dispatch:thread.archive", `terminal.close:${threadId}`]); | |
| 3236 | assert.deepEqual( | |
| 3237 | dispatchedCommands.map((command) => command.type), | |
| 3238 | ["thread.archive"], | |
| 3239 | ); | |
| 3240 | }).pipe(Effect.provide(NodeHttpServer.layerTest)), | |
| 3241 | ); | |
| 3242 | ||
| 3243 | it.effect("archives and still closes terminals when session stop fails", () => | |
| 3244 | Effect.gen(function* () { | |
| 3245 | const threadId = ThreadId.make("thread-archive-stop-failure"); | |
| 3246 | const effects: string[] = []; | |
| 3247 | const dispatchedCommands: Array<OrchestrationCommand> = []; | |
| 3248 | const now = new Date().toISOString(); | |
| 3249 | ||
| 3250 | yield* buildAppUnderTest({ | |
| 3251 | layers: { | |
| 3252 | terminalManager: { | |
| 3253 | close: (input) => | |
| 3254 | Effect.sync(() => { | |
| 3255 | effects.push(`terminal.close:${input.threadId}`); | |
| 3256 | }), | |
| 3257 | }, | |
| 3258 | orchestrationEngine: { | |
| 3259 | dispatch: (command) => { | |
| 3260 | dispatchedCommands.push(command); | |
| 3261 | effects.push(`dispatch:${command.type}`); | |
| 3262 | if (command.type === "thread.session.stop") { | |
| 3263 | return Effect.fail( | |
| 3264 | new OrchestrationListenerCallbackError({ | |
| 3265 | listener: "domain-event", | |
| 3266 | detail: "simulated archive stop failure", | |
| 3267 | }), | |
| 3268 | ); | |
| 3269 | } | |
| 3270 | return Effect.succeed({ sequence: dispatchedCommands.length }); | |
| 3271 | }, | |
| 3272 | }, | |
| 3273 | projectionSnapshotQuery: { | |
| 3274 | getThreadShellById: () => | |
| 3275 | Effect.succeed( | |
| 3276 | Option.some( | |
| 3277 | makeDefaultOrchestrationThreadShell({ | |
| 3278 | id: threadId, | |
| 3279 | updatedAt: now, | |
| 3280 | session: { | |
| 3281 | threadId, | |
| 3282 | status: "ready", | |
| 3283 | providerName: "claudeAgent", | |
| 3284 | runtimeMode: "full-access", | |
| 3285 | activeTurnId: null, | |
| 3286 | lastError: null, | |
| 3287 | updatedAt: now, | |
| 3288 | }, | |
| 3289 | }), | |
| 3290 | ), | |
| 3291 | ), | |
| 3292 | }, | |
| 3293 | }, | |
| 3294 | }); | |
| 3295 | ||
| 3296 | const wsUrl = yield* getWsServerUrl("/ws"); | |
| 3297 | const dispatchResult = yield* Effect.scoped( | |
| 3298 | withWsRpcClient(wsUrl, (client) => | |
| 3299 | client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ | |
| 3300 | type: "thread.archive", | |
| 3301 | commandId: CommandId.make("cmd-thread-archive-stop-failure"), | |
| 3302 | threadId, | |
| 3303 | }), | |
| 3304 | ), | |
| 3305 | ); | |
| 3306 | ||
| 3307 | assert.equal(dispatchResult.sequence, 1); | |
| 3308 | assert.deepEqual(effects, [ | |
| 3309 | "dispatch:thread.archive", | |
| 3310 | "dispatch:thread.session.stop", | |
| 3311 | `terminal.close:${threadId}`, | |
| 3312 | ]); | |
| 3313 | assert.deepEqual( | |
| 3314 | dispatchedCommands.map((command) => command.type), | |
| 3315 | ["thread.archive", "thread.session.stop"], | |
| 3316 | ); | |
| 3317 | }).pipe(Effect.provide(NodeHttpServer.layerTest)), | |
| 3318 | ); | |
| 3319 | ||
| 3320 | it.effect("archives and still closes terminals when session stop defects", () => | |
| 3321 | Effect.gen(function* () { | |
| 3322 | const threadId = ThreadId.make("thread-archive-stop-defect"); | |
| 3323 | const effects: string[] = []; | |
| 3324 | const dispatchedCommands: Array<OrchestrationCommand> = []; | |
| 3325 | const now = new Date().toISOString(); | |
| 3326 | ||
| 3327 | yield* buildAppUnderTest({ | |
| 3328 | layers: { | |
| 3329 | terminalManager: { | |
| 3330 | close: (input) => | |
| 3331 | Effect.sync(() => { | |
| 3332 | effects.push(`terminal.close:${input.threadId}`); | |
| 3333 | }), | |
| 3334 | }, | |
| 3335 | orchestrationEngine: { | |
| 3336 | dispatch: (command) => { | |
| 3337 | dispatchedCommands.push(command); | |
| 3338 | effects.push(`dispatch:${command.type}`); | |
| 3339 | if (command.type === "thread.session.stop") { | |
| 3340 | return Effect.die(new Error("simulated archive stop defect")); | |
| 3341 | } | |
| 3342 | return Effect.succeed({ sequence: dispatchedCommands.length }); | |
| 3343 | }, | |
| 3344 | }, | |
| 3345 | projectionSnapshotQuery: { | |
| 3346 | getThreadShellById: () => | |
| 3347 | Effect.succeed( | |
| 3348 | Option.some( | |
| 3349 | makeDefaultOrchestrationThreadShell({ | |
| 3350 | id: threadId, | |
| 3351 | updatedAt: now, | |
| 3352 | session: { | |
| 3353 | threadId, | |
| 3354 | status: "ready", | |
| 3355 | providerName: "claudeAgent", | |
| 3356 | runtimeMode: "full-access", | |
| 3357 | activeTurnId: null, | |
| 3358 | lastError: null, | |
| 3359 | updatedAt: now, | |
| 3360 | }, | |
| 3361 | }), | |
| 3362 | ), | |
| 3363 | ), | |
| 3364 | }, | |
| 3365 | }, | |
| 3366 | }); | |
| 3367 | ||
| 3368 | const wsUrl = yield* getWsServerUrl("/ws"); | |
| 3369 | const dispatchResult = yield* Effect.scoped( | |
| 3370 | withWsRpcClient(wsUrl, (client) => | |
| 3371 | client[ORCHESTRATION_WS_METHODS.dispatchCommand]({ | |
| 3372 | type: "thread.archive", | |
| 3373 | commandId: CommandId.make("cmd-thread-archive-stop-defect"), | |
| 3374 | threadId, | |
| 3375 | }), | |
| 3376 | ), | |
| 3377 | ); | |
| 3378 | ||
| 3379 | assert.equal(dispatchResult.sequence, 1); | |
| 3380 | assert.deepEqual(effects, [ | |
| 3381 | "dispatch:thread.archive", | |
| 3382 | "dispatch:thread.session.stop", | |
| 3383 | `terminal.close:${threadId}`, | |
| 3384 | ]); | |
| 3385 | assert.deepEqual( | |
| 3386 | dispatchedCommands.map((command) => command.type), | |
| 3387 | ["thread.archive", "thread.session.stop"], | |
| 3388 | ); | |
| 2980 | 3389 | }).pipe(Effect.provide(NodeHttpServer.layerTest)), |
| 2981 | 3390 | ); |
| 2982 | 3391 |
apps/server/src/server.ts+16-7
| ... | ... | @@ -23,6 +23,7 @@ import { makeCodexAdapterLive } from "./provider/Layers/CodexAdapter"; |
| 23 | 23 | import { makeClaudeAdapterLive } from "./provider/Layers/ClaudeAdapter"; |
| 24 | 24 | import { ProviderAdapterRegistryLive } from "./provider/Layers/ProviderAdapterRegistry"; |
| 25 | 25 | import { makeProviderServiceLive } from "./provider/Layers/ProviderService"; |
| 26 | import { ProviderSessionReaperLive } from "./provider/Layers/ProviderSessionReaper"; | |
| 26 | 27 | import { CheckpointDiffQueryLive } from "./checkpointing/Layers/CheckpointDiffQuery"; |
| 27 | 28 | import { CheckpointStoreLive } from "./checkpointing/Layers/CheckpointStore"; |
| 28 | 29 | import { GitCoreLive } from "./git/Layers/GitCore"; |
| ... | ... | @@ -134,6 +135,10 @@ const CheckpointingLayerLive = Layer.empty.pipe( |
| 134 | 135 | Layer.provideMerge(CheckpointStoreLive), |
| 135 | 136 | ); |
| 136 | 137 | |
| 138 | const ProviderSessionDirectoryLayerLive = ProviderSessionDirectoryLive.pipe( | |
| 139 | Layer.provide(ProviderSessionRuntimeRepositoryLive), | |
| 140 | ); | |
| 141 | ||
| 137 | 142 | const ProviderLayerLive = Layer.unwrap( |
| 138 | 143 | Effect.gen(function* () { |
| 139 | 144 | const { providerEventLogPath } = yield* ServerConfig; |
| ... | ... | @@ -143,9 +148,6 @@ const ProviderLayerLive = Layer.unwrap( |
| 143 | 148 | const canonicalEventLogger = yield* makeEventNdjsonLogger(providerEventLogPath, { |
| 144 | 149 | stream: "canonical", |
| 145 | 150 | }); |
| 146 | const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe( | |
| 147 | Layer.provide(ProviderSessionRuntimeRepositoryLive), | |
| 148 | ); | |
| 149 | 151 | const codexAdapterLayer = makeCodexAdapterLive( |
| 150 | 152 | nativeEventLogger ? { nativeEventLogger } : undefined, |
| 151 | 153 | ); |
| ... | ... | @@ -155,11 +157,14 @@ const ProviderLayerLive = Layer.unwrap( |
| 155 | 157 | const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe( |
| 156 | 158 | Layer.provide(codexAdapterLayer), |
| 157 | 159 | Layer.provide(claudeAdapterLayer), |
| 158 | Layer.provideMerge(providerSessionDirectoryLayer), | |
| 160 | Layer.provideMerge(ProviderSessionDirectoryLayerLive), | |
| 159 | 161 | ); |
| 160 | 162 | return makeProviderServiceLive( |
| 161 | 163 | canonicalEventLogger ? { canonicalEventLogger } : undefined, |
| 162 | ).pipe(Layer.provide(adapterRegistryLayer), Layer.provide(providerSessionDirectoryLayer)); | |
| 164 | ).pipe( | |
| 165 | Layer.provide(adapterRegistryLayer), | |
| 166 | Layer.provideMerge(ProviderSessionDirectoryLayerLive), | |
| 167 | ); | |
| 163 | 168 | }), |
| 164 | 169 | ); |
| 165 | 170 | |
| ... | ... | @@ -194,12 +199,16 @@ const AuthLayerLive = ServerAuthLive.pipe( |
| 194 | 199 | Layer.provide(ServerSecretStoreLive), |
| 195 | 200 | ); |
| 196 | 201 | |
| 202 | const ProviderRuntimeLayerLive = ProviderSessionReaperLive.pipe( | |
| 203 | Layer.provideMerge(ProviderLayerLive), | |
| 204 | Layer.provideMerge(OrchestrationLayerLive), | |
| 205 | ); | |
| 206 | ||
| 197 | 207 | const RuntimeDependenciesLive = ReactorLayerLive.pipe( |
| 198 | 208 | // Core Services |
| 199 | 209 | Layer.provideMerge(CheckpointingLayerLive), |
| 200 | 210 | Layer.provideMerge(GitLayerLive), |
| 201 | Layer.provideMerge(OrchestrationLayerLive), | |
| 202 | Layer.provideMerge(ProviderLayerLive), | |
| 211 | Layer.provideMerge(ProviderRuntimeLayerLive), | |
| 203 | 212 | Layer.provideMerge(TerminalLayerLive), |
| 204 | 213 | Layer.provideMerge(PersistenceLayerLive), |
| 205 | 214 | Layer.provideMerge(KeybindingsLive), |
apps/server/src/serverRuntimeStartup.ts+6-1
| ... | ... | @@ -32,6 +32,7 @@ import { ServerSettingsService } from "./serverSettings"; |
| 32 | 32 | import { ServerEnvironment } from "./environment/Services/ServerEnvironment"; |
| 33 | 33 | import { AnalyticsService } from "./telemetry/Services/AnalyticsService"; |
| 34 | 34 | import { ServerAuth } from "./auth/Services/ServerAuth"; |
| 35 | import { ProviderSessionReaper } from "./provider/Services/ProviderSessionReaper"; | |
| 35 | 36 | import { |
| 36 | 37 | formatHeadlessServeOutput, |
| 37 | 38 | formatHostForUrl, |
| ... | ... | @@ -281,6 +282,7 @@ export const makeServerRuntimeStartup = Effect.gen(function* () { |
| 281 | 282 | const serverConfig = yield* ServerConfig; |
| 282 | 283 | const keybindings = yield* Keybindings; |
| 283 | 284 | const orchestrationReactor = yield* OrchestrationReactor; |
| 285 | const providerSessionReaper = yield* ProviderSessionReaper; | |
| 284 | 286 | const lifecycleEvents = yield* ServerLifecycleEvents; |
| 285 | 287 | const serverSettings = yield* ServerSettingsService; |
| 286 | 288 | const serverEnvironment = yield* ServerEnvironment; |
| ... | ... | @@ -325,7 +327,10 @@ export const makeServerRuntimeStartup = Effect.gen(function* () { |
| 325 | 327 | yield* Effect.logDebug("startup phase: starting orchestration reactors"); |
| 326 | 328 | yield* runStartupPhase( |
| 327 | 329 | "reactors.start", |
| 328 | orchestrationReactor.start().pipe(Scope.provide(reactorScope)), | |
| 330 | Effect.gen(function* () { | |
| 331 | yield* orchestrationReactor.start().pipe(Scope.provide(reactorScope)); | |
| 332 | yield* providerSessionReaper.start().pipe(Scope.provide(reactorScope)); | |
| 333 | }), | |
| 329 | 334 | ); |
| 330 | 335 | |
| 331 | 336 | const welcomeBase = yield* resolveWelcomeBase; |
apps/server/src/ws.ts+37
| ... | ... | @@ -550,8 +550,45 @@ const makeWsRpcLayer = (currentSessionId: AuthSessionId) => |
| 550 | 550 | ORCHESTRATION_WS_METHODS.dispatchCommand, |
| 551 | 551 | Effect.gen(function* () { |
| 552 | 552 | const normalizedCommand = yield* normalizeDispatchCommand(command); |
| 553 | const shouldStopSessionAfterArchive = | |
| 554 | normalizedCommand.type === "thread.archive" | |
| 555 | ? yield* projectionSnapshotQuery | |
| 556 | .getThreadShellById(normalizedCommand.threadId) | |
| 557 | .pipe( | |
| 558 | Effect.map( | |
| 559 | Option.match({ | |
| 560 | onNone: () => false, | |
| 561 | onSome: (thread) => | |
| 562 | thread.session !== null && thread.session.status !== "stopped", | |
| 563 | }), | |
| 564 | ), | |
| 565 | Effect.catch(() => Effect.succeed(false)), | |
| 566 | ) | |
| 567 | : false; | |
| 553 | 568 | const result = yield* dispatchNormalizedCommand(normalizedCommand); |
| 554 | 569 | if (normalizedCommand.type === "thread.archive") { |
| 570 | if (shouldStopSessionAfterArchive) { | |
| 571 | yield* Effect.gen(function* () { | |
| 572 | const stopCommand = yield* normalizeDispatchCommand({ | |
| 573 | type: "thread.session.stop", | |
| 574 | commandId: CommandId.make( | |
| 575 | `session-stop-for-archive:${normalizedCommand.commandId}`, | |
| 576 | ), | |
| 577 | threadId: normalizedCommand.threadId, | |
| 578 | createdAt: new Date().toISOString(), | |
| 579 | }); | |
| 580 | ||
| 581 | yield* dispatchNormalizedCommand(stopCommand); | |
| 582 | }).pipe( | |
| 583 | Effect.catchCause((cause) => | |
| 584 | Effect.logWarning("failed to stop provider session during archive", { | |
| 585 | threadId: normalizedCommand.threadId, | |
| 586 | cause, | |
| 587 | }), | |
| 588 | ), | |
| 589 | ); | |
| 590 | } | |
| 591 | ||
| 555 | 592 | yield* terminalManager.close({ threadId: normalizedCommand.threadId }).pipe( |
| 556 | 593 | Effect.catch((error) => |
| 557 | 594 | Effect.logWarning("failed to close thread terminals after archive", { |
package.json+1
| ... | ... | @@ -40,6 +40,7 @@ |
| 40 | 40 | "typecheck": "turbo run typecheck", |
| 41 | 41 | "lint": "oxlint --report-unused-disable-directives", |
| 42 | 42 | "test": "turbo run test", |
| 43 | "test:process-reaper": "bun run --cwd apps/server test:process-reaper", | |
| 43 | 44 | "test:desktop-smoke": "turbo run smoke-test --filter=@t3tools/desktop", |
| 44 | 45 | "fmt": "oxfmt", |
| 45 | 46 | "fmt:check": "oxfmt --check", |