From bb8a8444f4c9e1b4e9ad6c33c9a3c0e020f1b6c7 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 25 Feb 2026 23:38:02 -0800 Subject: [PATCH] Bind provider sessions to thread IDs and preserve resume cursors - Require `threadId` when starting provider sessions and route recovery via persisted bindings - Persist and propagate provider `resumeCursor`/`providerThreadId` instead of synthesizing values - Expose full session bindings in `ProviderSessionDirectory` and update tests/integration fixtures - Rename sqlite state file from `orchestration.sqlite` to `state.sqlite` --- .plans/16c-pr89-remediation-checklist.md | 6 +- .../OrchestrationEngineHarness.integration.ts | 2 +- .../providerService.integration.test.ts | 45 ++++--- apps/server/src/codexAppServerManager.test.ts | 6 + apps/server/src/codexAppServerManager.ts | 22 +++- .../Layers/ProviderCommandReactor.test.ts | 5 +- .../Layers/ProviderCommandReactor.ts | 13 +- apps/server/src/persistence/Layers/Sqlite.ts | 2 +- .../provider/Layers/ProviderService.test.ts | 107 ++++++++++------ .../src/provider/Layers/ProviderService.ts | 114 +++++++++++------- .../Layers/ProviderSessionDirectory.test.ts | 84 +++++++------ .../Layers/ProviderSessionDirectory.ts | 9 ++ .../src/provider/Services/ProviderService.ts | 2 + .../Services/ProviderSessionDirectory.ts | 7 ++ packages/contracts/src/provider.ts | 3 + 15 files changed, 277 insertions(+), 150 deletions(-) diff --git a/.plans/16c-pr89-remediation-checklist.md b/.plans/16c-pr89-remediation-checklist.md index 8d30b2dd408d98e75c8e44c1b6bb9089ee93f7ff..19046e4ee80f1a85aa47299810b88f0dc6b76a83 100644 --- a/.plans/16c-pr89-remediation-checklist.md +++ b/.plans/16c-pr89-remediation-checklist.md @@ -192,14 +192,14 @@ Counts: active `51` (`valid=33`, `partially-valid=18`), closed-invalid `6` ### Phase 3 -- [ ] `C008` Inconsistent input normalization across CheckpointStore methods - - Status: `TODO` +- [x] `C008` Inconsistent input normalization across CheckpointStore methods + - Status: `DONE` - Verdict: `partially-valid` - Severity: `Medium` - Area: `Checkpointing correctness` - File: `apps/server/src/checkpointing/Layers/CheckpointStore.ts:94` - Threads: PRRT_kwDORLtfbc5widJw, PRRT_kwDORLtfbc5wnWv_, PRRT_kwDORLtfbc5w0_g7, PRRT_kwDORLtfbc5w1C36 (+3 duplicate thread(s)) - - Audit note: Normalization inconsistency remains, but upstream callsites now mostly canonicalize inputs. + - Audit note: Edge schema strategy is in place across contracts/consumers (trim/normalize via schemas and decode at boundaries); CheckpointStore remains an internal repository boundary. - [ ] `C017` `REQUIRED_SNAPSHOT_PROJECTORS` includes `pending-approvals` and `thread-turns`, but `getSnapshot` doesn't query their data. If these projectors lag behind, the returned `snapshotSequence` will be lower than what the included data actually reflects, causing clients to replay already-applied events. Consider filtering `REQUIRED_SNAPSHOT_PROJECTORS` to only include projectors whose data is actually fetched in the snapshot.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent: - Status: `TODO` diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts index 5fa6fead39193dcfb9c2fcc22613828778ba3fd4..077b62f89863781fc2c01f4159c680076bea34fa 100644 --- a/apps/server/integration/OrchestrationEngineHarness.integration.ts +++ b/apps/server/integration/OrchestrationEngineHarness.integration.ts @@ -159,7 +159,7 @@ export const makeOrchestrationIntegrationHarness = Effect.gen(function* () { const rootDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-orchestration-integration-")); const workspaceDir = path.join(rootDir, "workspace"); const stateDir = path.join(rootDir, "state"); - const dbPath = path.join(stateDir, "orchestration.sqlite"); + const dbPath = path.join(stateDir, "state.sqlite"); fs.mkdirSync(workspaceDir, { recursive: true }); fs.mkdirSync(stateDir, { recursive: true }); initializeGitWorkspace(workspaceDir); diff --git a/apps/server/integration/providerService.integration.test.ts b/apps/server/integration/providerService.integration.test.ts index 353b7d984ceca4036ceb159ec8f50b26448f73f7..9212a72b2cfaefd976ac4c9a9ad7d68fb218261e 100644 --- a/apps/server/integration/providerService.integration.test.ts +++ b/apps/server/integration/providerService.integration.test.ts @@ -1,4 +1,5 @@ import type { ProviderRuntimeEvent, ProviderSessionId } from "@t3tools/contracts"; +import { ThreadId } from "@t3tools/contracts"; import { NodeServices } from "@effect/platform-node"; import { it, assert } from "@effect/vitest"; import { Effect, FileSystem, Layer, Path, Queue, Stream } from "effect"; @@ -116,10 +117,13 @@ it.effect("replays typed runtime fixture events", () => yield* Effect.gen(function* () { const provider = yield* ProviderService; - const session = yield* provider.startSession({ - provider: "codex", - cwd: fixture.cwd, - }); + const session = yield* provider.startSession( + ThreadId.makeUnsafe("thread-integration-typed"), + { + provider: "codex", + cwd: fixture.cwd, + }, + ); assert.equal((session.threadId ?? "").length > 0, true); const observedEvents = yield* runTurn({ @@ -146,10 +150,13 @@ it.effect("replays file-changing fixture turn events", () => yield* Effect.gen(function* () { const provider = yield* ProviderService; - const session = yield* provider.startSession({ - provider: "codex", - cwd: fixture.cwd, - }); + const session = yield* provider.startSession( + ThreadId.makeUnsafe("thread-integration-tools"), + { + provider: "codex", + cwd: fixture.cwd, + }, + ); assert.equal((session.threadId ?? "").length > 0, true); const observedEvents = yield* runTurn({ @@ -180,10 +187,13 @@ it.effect("runs multi-turn tool/approval flow", () => yield* Effect.gen(function* () { const provider = yield* ProviderService; - const session = yield* provider.startSession({ - provider: "codex", - cwd: fixture.cwd, - }); + const session = yield* provider.startSession( + ThreadId.makeUnsafe("thread-integration-multi"), + { + provider: "codex", + cwd: fixture.cwd, + }, + ); assert.equal((session.threadId ?? "").length > 0, true); const firstTurnEvents = yield* runTurn({ @@ -229,10 +239,13 @@ it.effect("rolls back provider conversation state only", () => yield* Effect.gen(function* () { const provider = yield* ProviderService; - const session = yield* provider.startSession({ - provider: "codex", - cwd: fixture.cwd, - }); + const session = yield* provider.startSession( + ThreadId.makeUnsafe("thread-integration-rollback"), + { + provider: "codex", + cwd: fixture.cwd, + }, + ); assert.equal((session.threadId ?? "").length > 0, true); yield* runTurn({ diff --git a/apps/server/src/codexAppServerManager.test.ts b/apps/server/src/codexAppServerManager.test.ts index f0af874337229dfe373ece3efd84e5702cd3a2f4..b7b6c91237c89fe924e1cca5c21a4c04493ab1b8 100644 --- a/apps/server/src/codexAppServerManager.test.ts +++ b/apps/server/src/codexAppServerManager.test.ts @@ -204,6 +204,9 @@ describe("sendTurn", () => { expect(result).toEqual({ threadId: "thread_1", turnId: "turn_1", + resumeCursor: { + threadId: "thread_1", + }, }); expect(requireSession).toHaveBeenCalledWith("sess_1"); expect(sendRequest).toHaveBeenCalledWith(context, "turn/start", { @@ -225,6 +228,9 @@ describe("sendTurn", () => { expect(updateSession).toHaveBeenCalledWith(context, { status: "running", activeTurnId: "turn_1", + resumeCursor: { + threadId: "thread_1", + }, }); }); diff --git a/apps/server/src/codexAppServerManager.ts b/apps/server/src/codexAppServerManager.ts index e94e8d650e83cc2a6c0c24812c93b05836122390..f388baef17b0e684d08523b51ffc3cdf229b72b7 100644 --- a/apps/server/src/codexAppServerManager.ts +++ b/apps/server/src/codexAppServerManager.ts @@ -221,15 +221,16 @@ export class CodexAppServerManager extends EventEmitter).threadId; + return typeof rawThreadId === "string" ? toProviderThreadId(rawThreadId) : input.resumeThreadId; +} + function toProviderTurnId(value: string | undefined): ProviderTurnId | undefined { return brandIfNonEmpty(value, ProviderTurnId.makeUnsafe); } diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index f028349f58752b8cab4321c8e078419db6af12cc..3c59e54cba866c2f25cede4f0974e09485214cb5 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -71,7 +71,7 @@ describe("ProviderCommandReactor", () => { async function createHarness() { const now = new Date().toISOString(); const runtimeEventPubSub = Effect.runSync(PubSub.unbounded()); - const startSession = vi.fn((_: unknown) => + const startSession = vi.fn((_: unknown, __: unknown) => Effect.succeed({ sessionId: asSessionId("sess-1"), provider: "codex" as const, @@ -177,7 +177,8 @@ describe("ProviderCommandReactor", () => { await waitFor(() => harness.startSession.mock.calls.length === 1); await waitFor(() => harness.sendTurn.mock.calls.length === 1); - expect(harness.startSession.mock.calls[0]?.[0]).toMatchObject({ + expect(harness.startSession.mock.calls[0]?.[0]).toEqual(ThreadId.makeUnsafe("thread-1")); + expect(harness.startSession.mock.calls[0]?.[1]).toMatchObject({ cwd: "/tmp/provider-project", model: "gpt-5-codex", }); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 68fcc6644fa0e3782a0c9f24d84ebfc2569e8ee3..1b09cc0a409606fc666f90173f767c6e434f1ffc 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -152,11 +152,14 @@ const make = Effect.gen(function* () { projects: readModel.projects, }); - const startedSession = yield* providerService.startSession({ - ...(preferredProvider ? { provider: preferredProvider } : {}), - ...(effectiveCwd ? { cwd: effectiveCwd } : {}), - ...(thread.model ? { model: thread.model } : {}), - }); + const startedSession = yield* providerService.startSession( + threadId, + { + ...(preferredProvider ? { provider: preferredProvider } : {}), + ...(effectiveCwd ? { cwd: effectiveCwd } : {}), + ...(thread.model ? { model: thread.model } : {}), + }, + ); yield* setThreadSession({ threadId, diff --git a/apps/server/src/persistence/Layers/Sqlite.ts b/apps/server/src/persistence/Layers/Sqlite.ts index c2396a247cf3409ac96531c65d50111a12df5b11..33f99482d9b3dc0c8f3cdd7c74c42713607b5d29 100644 --- a/apps/server/src/persistence/Layers/Sqlite.ts +++ b/apps/server/src/persistence/Layers/Sqlite.ts @@ -52,6 +52,6 @@ export const SqlitePersistenceMemory = Layer.provideMerge( export const layerConfig = Effect.gen(function* () { const { stateDir } = yield* ServerConfig; const { join } = yield* Path.Path; - const dbPath = join(stateDir, "orchestration.sqlite"); + const dbPath = join(stateDir, "state.sqlite"); return makeSqlitePersistenceLive(dbPath); }).pipe(Layer.unwrap); diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index ff8cdd0ea9c1d733aa331a2c04ff51fa48374b92..7dcd739619ec19b9037c105a58047aec8174a3d5 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -49,6 +49,7 @@ const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.make const asTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value); const asRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.makeUnsafe(value); const asEventId = (value: string): EventId => EventId.makeUnsafe(value); +const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); function makeFakeCodexAdapter() { const sessions = new Map(); @@ -322,10 +323,13 @@ it.effect( const startedSession = yield* Effect.gen(function* () { const provider = yield* ProviderService; - return yield* provider.startSession({ - provider: "codex", - cwd: "/tmp/project", - }); + return yield* provider.startSession( + asThreadId("thread-1"), + { + provider: "codex", + cwd: "/tmp/project", + }, + ); }).pipe(Effect.provide(firstProviderLayer)); yield* Effect.gen(function* () { @@ -342,9 +346,7 @@ it.effect( assert.equal(Option.isSome(persistedAfterStopAll), true); if (Option.isSome(persistedAfterStopAll)) { assert.equal(persistedAfterStopAll.value.status, "stopped"); - assert.deepEqual(persistedAfterStopAll.value.resumeCursor, { - resumeThreadId: startedSession.threadId, - }); + assert.equal(persistedAfterStopAll.value.resumeCursor, null); } const secondCodex = makeFakeCodexAdapter(); @@ -396,10 +398,13 @@ routing.layer("ProviderServiceLive routing", (it) => { Effect.gen(function* () { const provider = yield* ProviderService; - const session = yield* provider.startSession({ - provider: "codex", - cwd: "/tmp/project", - }); + const session = yield* provider.startSession( + asThreadId("thread-1"), + { + provider: "codex", + cwd: "/tmp/project", + }, + ); assert.equal(session.provider, "codex"); const sessions = yield* provider.listSessions(); @@ -448,10 +453,13 @@ routing.layer("ProviderServiceLive routing", (it) => { Effect.gen(function* () { const provider = yield* ProviderService; - const initial = yield* provider.startSession({ - provider: "codex", - cwd: "/tmp/project", - }); + const initial = yield* provider.startSession( + asThreadId("thread-1"), + { + provider: "codex", + cwd: "/tmp/project", + }, + ); yield* routing.codex.stopSession(initial.sessionId); routing.codex.startSession.mockClear(); routing.codex.rollbackThread.mockClear(); @@ -479,12 +487,18 @@ routing.layer("ProviderServiceLive routing", (it) => { Effect.gen(function* () { const provider = yield* ProviderService; - yield* provider.startSession({ - provider: "codex", - }); - yield* provider.startSession({ - provider: "codex", - }); + yield* provider.startSession( + asThreadId("thread-1"), + { + provider: "codex", + }, + ); + yield* provider.startSession( + asThreadId("thread-2"), + { + provider: "codex", + }, + ); yield* provider.stopAll(); @@ -498,9 +512,12 @@ routing.layer("ProviderServiceLive routing", (it) => { const provider = yield* ProviderService; const runtimeRepository = yield* ProviderSessionRuntimeRepository; - const session = yield* provider.startSession({ - provider: "codex", - }); + const session = yield* provider.startSession( + asThreadId("thread-1"), + { + provider: "codex", + }, + ); yield* provider.sendTurn({ sessionId: session.sessionId, input: "hello", @@ -513,9 +530,7 @@ routing.layer("ProviderServiceLive routing", (it) => { assert.equal(Option.isSome(runningRuntime), true); if (Option.isSome(runningRuntime)) { assert.equal(runningRuntime.value.status, "running"); - assert.deepEqual(runningRuntime.value.resumeCursor, { - resumeThreadId: "thread-1", - }); + assert.equal(runningRuntime.value.resumeCursor, null); const payload = runningRuntime.value.runtimePayload; assert.equal(payload !== null && typeof payload === "object", true); if (payload !== null && typeof payload === "object" && !Array.isArray(payload)) { @@ -551,9 +566,12 @@ fanout.layer("ProviderServiceLive fanout", (it) => { it.effect("fans out adapter turn completion events", () => Effect.gen(function* () { const provider = yield* ProviderService; - const session = yield* provider.startSession({ - provider: "codex", - }); + const session = yield* provider.startSession( + asThreadId("thread-1"), + { + provider: "codex", + }, + ); const eventsRef = yield* Ref.make>([]); const consumer = yield* Stream.runForEach(provider.streamEvents, (event) => @@ -588,9 +606,12 @@ fanout.layer("ProviderServiceLive fanout", (it) => { it.effect("keeps subscriber delivery ordered and isolates failing subscribers", () => Effect.gen(function* () { const provider = yield* ProviderService; - const session = yield* provider.startSession({ - provider: "codex", - }); + const session = yield* provider.startSession( + asThreadId("thread-1"), + { + provider: "codex", + }, + ); const receivedByHealthy: string[] = []; const expectedEventIds = new Set(["evt-ordered-1", "evt-ordered-2", "evt-ordered-3"]); @@ -665,9 +686,12 @@ validation.layer("ProviderServiceLive validation", (it) => { const provider = yield* ProviderService; const failure = yield* Effect.result( - provider.startSession({ - provider: "invalid-provider", - } as never), + provider.startSession( + asThreadId("thread-validation"), + { + provider: "invalid-provider", + } as never, + ), ); assert.equal(failure._tag, "Failure"); @@ -702,10 +726,13 @@ validation.layer("ProviderServiceLive validation", (it) => { ); const failure = yield* Effect.result( - provider.startSession({ - provider: "codex", - cwd: "/tmp/project", - }), + provider.startSession( + asThreadId("thread-missing"), + { + provider: "codex", + cwd: "/tmp/project", + }, + ), ); assertFailure( diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index d01ee4404b01cc010d59540b1df4c7220bd31f9e..46fa47045a68d555bebdc068993d9300e3682a9b 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -24,10 +24,13 @@ import { } from "@t3tools/contracts"; import { Effect, Layer, Option, PubSub, Queue, Ref, Schema, Stream } from "effect"; -import { ProviderValidationError } from "../Errors.ts"; +import { ProviderSessionNotFoundError, ProviderValidationError } from "../Errors.ts"; import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts"; import { ProviderService, type ProviderServiceShape } from "../Services/ProviderService.ts"; -import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; +import { + ProviderSessionDirectory, + type ProviderSessionBinding, +} from "../Services/ProviderSessionDirectory.ts"; import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; export interface ProviderServiceLiveOptions { @@ -154,7 +157,11 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => Effect.asVoid, ); - const upsertSessionBinding = (session: ProviderSession, operation: string) => + const upsertSessionBinding = ( + session: ProviderSession, + operation: string, + threadId: ThreadId, + ) => Effect.gen(function* () { const providerThreadId = session.threadId; if (!providerThreadId) { @@ -165,18 +172,17 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => } const providerSessionId = asProviderSessionId(session.sessionId); - const brandedThreadId = ThreadId.makeUnsafe(providerThreadId); yield* directory.upsert({ sessionId: providerSessionId, provider: session.provider, - threadId: brandedThreadId, + threadId, providerThreadId, status: toRuntimeStatus(session), - resumeCursor: { resumeThreadId: providerThreadId }, + ...(session.resumeCursor !== undefined ? { resumeCursor: session.resumeCursor } : {}), runtimePayload: toRuntimePayloadFromSession(session), }); - return brandedThreadId; + return providerThreadId; }); const clearAliasKey = (staleSessionId: ProviderSessionId) => @@ -235,26 +241,33 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => const recoverSessionForThread = (input: { readonly staleSessionId: ProviderSessionId; - readonly provider: ProviderSession["provider"]; - readonly threadId: ThreadId; + readonly binding: ProviderSessionBinding & { readonly threadId: ThreadId }; readonly operation: string; }) => Effect.gen(function* () { - const adapter = yield* registry.getByProvider(input.provider); + const adapter = yield* registry.getByProvider(input.binding.provider); const activeSessions = yield* adapter.listSessions(); - const expectedProviderThreadId = asProviderThreadId(input.threadId); - const existing = activeSessions.find( - (session) => session.threadId === expectedProviderThreadId, - ); + const resumeThreadId = input.binding.providerThreadId ?? undefined; + const hasResumeCursor = + input.binding.resumeCursor !== null && input.binding.resumeCursor !== undefined; + const existing = + resumeThreadId === undefined + ? undefined + : activeSessions.find((session) => session.threadId === resumeThreadId); if (existing) { - const existingThreadId = yield* upsertSessionBinding( + const existingProviderThreadId = yield* upsertSessionBinding( existing, `${input.operation}:upsertExistingSession`, + input.binding.threadId, ); yield* directory.upsert({ sessionId: input.staleSessionId, provider: existing.provider, - threadId: existingThreadId, + threadId: input.binding.threadId, + providerThreadId: existingProviderThreadId, + ...(existing.resumeCursor !== undefined + ? { resumeCursor: existing.resumeCursor } + : {}), }); if (existing.sessionId !== input.staleSessionId) { yield* setAlias(input.staleSessionId, existing.sessionId); @@ -267,9 +280,17 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => } as const; } + if (!resumeThreadId && !hasResumeCursor) { + return yield* toValidationError( + input.operation, + `Cannot recover stale session '${input.staleSessionId}' because no provider resume state is persisted.`, + ); + } + const resumed = yield* adapter.startSession({ - provider: input.provider, - resumeThreadId: ProviderThreadId.makeUnsafe(input.threadId), + provider: input.binding.provider, + ...(resumeThreadId ? { resumeThreadId } : {}), + ...(hasResumeCursor ? { resumeCursor: input.binding.resumeCursor } : {}), }); if (resumed.provider !== adapter.provider) { return yield* toValidationError( @@ -278,21 +299,20 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => ); } - const resumedThreadId = yield* upsertSessionBinding( + const resumedProviderThreadId = yield* upsertSessionBinding( resumed, `${input.operation}:upsertRecoveredSession`, + input.binding.threadId, ); - if (resumedThreadId !== input.threadId) { - return yield* toValidationError( - input.operation, - `Recovered session thread '${resumedThreadId}' does not match expected thread '${input.threadId}'.`, - ); - } yield* directory.upsert({ sessionId: input.staleSessionId, provider: resumed.provider, - threadId: resumedThreadId, + threadId: input.binding.threadId, + providerThreadId: resumedProviderThreadId, + ...(resumed.resumeCursor !== undefined + ? { resumeCursor: resumed.resumeCursor } + : {}), }); if (resumed.sessionId !== input.staleSessionId) { @@ -313,8 +333,26 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => readonly allowRecovery: boolean; }) => Effect.gen(function* () { - const provider = yield* directory.getProvider(input.sessionId); - const adapter = yield* registry.getByProvider(provider); + const bindingOption = yield* directory.getBinding(input.sessionId); + const binding = Option.getOrUndefined(bindingOption); + if (!binding) { + return yield* new ProviderSessionNotFoundError({ + sessionId: input.sessionId, + }); + } + if (!binding.threadId) { + return yield* toValidationError( + input.operation, + `Cannot route session '${input.sessionId}' because no orchestration thread id is persisted.`, + ); + } + const bindingWithThreadId: ProviderSessionBinding & { + readonly threadId: ThreadId; + } = { + ...binding, + threadId: binding.threadId, + }; + const adapter = yield* registry.getByProvider(binding.provider); const hasRequestedSession = yield* adapter.hasSession(input.sessionId); if (hasRequestedSession) { @@ -349,19 +387,9 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => } as const; } - const threadIdOption = yield* directory.getThreadId(input.sessionId); - const threadId = Option.getOrUndefined(threadIdOption); - if (!threadId) { - return yield* toValidationError( - input.operation, - `Cannot recover stale session '${input.sessionId}' because no thread id is persisted.`, - ); - } - const recovered = yield* recoverSessionForThread({ staleSessionId: input.sessionId, - provider, - threadId, + binding: bindingWithThreadId, operation: input.operation, }); @@ -372,7 +400,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => } as const; }); - const startSession: ProviderServiceShape["startSession"] = (rawInput) => + const startSession: ProviderServiceShape["startSession"] = (threadId, rawInput) => Effect.gen(function* () { const parsed = yield* decodeInputOrValidationError({ operation: "ProviderService.startSession", @@ -396,7 +424,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => ); } - yield* upsertSessionBinding(session, "ProviderService.startSession"); + yield* upsertSessionBinding(session, "ProviderService.startSession", threadId); return session; }); @@ -443,7 +471,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => threadId, providerThreadId: asProviderThreadId(turn.threadId), status: "running", - resumeCursor: { resumeThreadId: turn.threadId }, + ...(turn.resumeCursor !== undefined ? { resumeCursor: turn.resumeCursor } : {}), runtimePayload: { activeTurnId: turn.turnId, lastRuntimeEvent: "provider.sendTurn", @@ -550,7 +578,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => ), ).pipe(Effect.asVoid); // Keep persisted session bindings so stale sessions can be resumed after - // process restart via resumeThreadId. + // process restart via providerThreadId. yield* Ref.set(routedSessionAliasesRef, new Map()); }); diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts index be1c33d38a71bd9c72ad927def2208750e5d8d4b..90fa94032b1fc22789fb3b3055e8c0029eb9c83e 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts @@ -19,10 +19,6 @@ import { ProviderSessionNotFoundError, ProviderValidationError } from "../Errors import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts"; -const sessionId = (value: string) => ProviderSessionId.makeUnsafe(value); -const threadId = (value: string) => ThreadId.makeUnsafe(value); -const providerThreadId = (value: string) => ProviderThreadId.makeUnsafe(value); - function makeDirectoryLayer(persistenceLayer: Layer.Layer) { const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( Layer.provide(persistenceLayer), @@ -33,49 +29,58 @@ function makeDirectoryLayer(persistenceLayer: Layer.Layer { +it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryLive", (it) => { it("upserts, reads, and removes session bindings", () => Effect.gen(function* () { const directory = yield* ProviderSessionDirectory; const runtimeRepository = yield* ProviderSessionRuntimeRepository; + const sessionId = ProviderSessionId.makeUnsafe("sess-1"); + const initialThreadId = ThreadId.makeUnsafe("thread-1"); + yield* directory.upsert({ - sessionId: sessionId("sess-1"), + sessionId, provider: "codex", - threadId: threadId("thread-1"), + threadId: initialThreadId, }); - const provider = yield* directory.getProvider(sessionId("sess-1")); + const provider = yield* directory.getProvider(sessionId); assert.equal(provider, "codex"); - const resolvedThreadId = yield* directory.getThreadId(sessionId("sess-1")); - assertSome(resolvedThreadId, threadId("thread-1")); + const resolvedThreadId = yield* directory.getThreadId(sessionId); + assertSome(resolvedThreadId, initialThreadId); + + const nextThreadId = ThreadId.makeUnsafe("thread-2"); yield* directory.upsert({ - sessionId: sessionId("sess-1"), + sessionId: sessionId, provider: "codex", - threadId: threadId("thread-2"), + threadId: nextThreadId, }); - const updatedThreadId = yield* directory.getThreadId(sessionId("sess-1")); - assertSome(updatedThreadId, threadId("thread-2")); + const updatedThreadId = yield* directory.getThreadId(sessionId); + assertSome(updatedThreadId, nextThreadId); + const updatedBinding = yield* directory.getBinding(sessionId); + assert.equal(Option.isSome(updatedBinding), true); + if (Option.isSome(updatedBinding)) { + assert.equal(updatedBinding.value.threadId, nextThreadId); + assert.equal(updatedBinding.value.providerThreadId, null); + } const runtime = yield* runtimeRepository.getBySessionId({ - providerSessionId: sessionId("sess-1"), + providerSessionId: sessionId, }); assert.equal(Option.isSome(runtime), true); if (Option.isSome(runtime)) { - assert.equal(runtime.value.threadId, "thread-2"); - assert.equal(runtime.value.providerThreadId, "thread-2"); + assert.equal(runtime.value.threadId, nextThreadId); + assert.equal(runtime.value.providerThreadId, null); assert.equal(runtime.value.status, "running"); assert.equal(runtime.value.adapterKey, "codex"); } const sessionIds = yield* directory.listSessionIds(); - assert.deepEqual(sessionIds, [sessionId("sess-1")]); + assert.deepEqual(sessionIds, [sessionId]); - yield* directory.remove(sessionId("sess-1")); - const missingProvider = yield* directory.getProvider(sessionId("sess-1")).pipe(Effect.result); + yield* directory.remove(sessionId); + const missingProvider = yield* directory.getProvider(sessionId).pipe(Effect.result); assertFailure(missingProvider, new ProviderSessionNotFoundError({ sessionId: "sess-1" })); })); @@ -84,7 +89,7 @@ layer("ProviderSessionDirectoryLive", (it) => { const directory = yield* ProviderSessionDirectory; const result = yield* Effect.result( directory.upsert({ - sessionId: sessionId("sess-no-thread"), + sessionId: ProviderSessionId.makeUnsafe("sess-no-thread"), provider: "codex", }), ); @@ -102,11 +107,15 @@ layer("ProviderSessionDirectoryLive", (it) => { const directory = yield* ProviderSessionDirectory; const runtimeRepository = yield* ProviderSessionRuntimeRepository; + const sessionId = ProviderSessionId.makeUnsafe("sess-runtime"); + const threadId = ThreadId.makeUnsafe("thread-runtime"); + const providerThreadId = ProviderThreadId.makeUnsafe("provider-thread-runtime"); + yield* directory.upsert({ - sessionId: sessionId("sess-runtime"), + sessionId, provider: "codex", - threadId: threadId("thread-runtime"), - providerThreadId: providerThreadId("provider-thread-runtime"), + threadId, + providerThreadId, status: "starting", resumeCursor: { resumeThreadId: "provider-thread-runtime", @@ -118,7 +127,7 @@ layer("ProviderSessionDirectoryLive", (it) => { }); yield* directory.upsert({ - sessionId: sessionId("sess-runtime"), + sessionId, provider: "codex", status: "running", runtimePayload: { @@ -127,15 +136,15 @@ layer("ProviderSessionDirectoryLive", (it) => { }); const runtime = yield* runtimeRepository.getBySessionId({ - providerSessionId: sessionId("sess-runtime"), + providerSessionId: sessionId, }); assert.equal(Option.isSome(runtime), true); if (Option.isSome(runtime)) { - assert.equal(runtime.value.threadId, "thread-runtime"); - assert.equal(runtime.value.providerThreadId, "provider-thread-runtime"); + assert.equal(runtime.value.threadId, threadId); + assert.equal(runtime.value.providerThreadId, providerThreadId); assert.equal(runtime.value.status, "running"); assert.deepEqual(runtime.value.resumeCursor, { - resumeThreadId: "provider-thread-runtime", + resumeThreadId: providerThreadId, }); assert.deepEqual(runtime.value.runtimePayload, { cwd: "/tmp/project", @@ -151,23 +160,26 @@ layer("ProviderSessionDirectoryLive", (it) => { const dbPath = path.join(tempDir, "orchestration.sqlite"); const directoryLayer = makeDirectoryLayer(makeSqlitePersistenceLive(dbPath)); + const sessionId = ProviderSessionId.makeUnsafe("sess-restart"); + const threadId = ThreadId.makeUnsafe("thread-restart"); + yield* Effect.gen(function* () { const directory = yield* ProviderSessionDirectory; yield* directory.upsert({ - sessionId: sessionId("sess-restart"), + sessionId, provider: "codex", - threadId: threadId("thread-restart"), + threadId, }); }).pipe(Effect.provide(directoryLayer)); yield* Effect.gen(function* () { const directory = yield* ProviderSessionDirectory; const sql = yield* SqlClient.SqlClient; - const provider = yield* directory.getProvider(sessionId("sess-restart")); + const provider = yield* directory.getProvider(sessionId); assert.equal(provider, "codex"); - const resolvedThreadId = yield* directory.getThreadId(sessionId("sess-restart")); - assertSome(resolvedThreadId, threadId("thread-restart")); + const resolvedThreadId = yield* directory.getThreadId(sessionId); + assertSome(resolvedThreadId, threadId); const legacyTableRows = yield* sql<{ readonly name: string }>` SELECT name diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts index dabfe9384f6334860f1cd4e030e7eb565f119cd7..700f8b50914af68602a6046b5e463918e1d39dc3 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts @@ -73,6 +73,11 @@ const makeProviderSessionDirectory = Effect.gen(function* () { sessionId: value.providerSessionId, provider, threadId: value.threadId, + adapterKey: value.adapterKey, + providerThreadId: value.providerThreadId, + status: value.status, + resumeCursor: value.resumeCursor, + runtimePayload: value.runtimePayload, }), ), ), @@ -128,6 +133,9 @@ const makeProviderSessionDirectory = Effect.gen(function* () { ), ); + const getBindingBySessionId: ProviderSessionDirectoryShape["getBinding"] = (sessionId) => + getBinding(sessionId); + const getThreadId: ProviderSessionDirectoryShape["getThreadId"] = (sessionId) => getBinding(sessionId).pipe( Effect.flatMap((binding) => @@ -168,6 +176,7 @@ const makeProviderSessionDirectory = Effect.gen(function* () { return { upsert, getProvider, + getBinding: getBindingBySessionId, getThreadId, remove, listSessionIds, diff --git a/apps/server/src/provider/Services/ProviderService.ts b/apps/server/src/provider/Services/ProviderService.ts index d85010b5aa43fc42f707508852c79191cc5475e7..d704b7be1e2e230393bedd500e3cb357b52397b3 100644 --- a/apps/server/src/provider/Services/ProviderService.ts +++ b/apps/server/src/provider/Services/ProviderService.ts @@ -20,6 +20,7 @@ import type { ProviderSessionId, ProviderSessionStartInput, ProviderStopSessionInput, + ThreadId, ProviderTurnStartResult, } from "@t3tools/contracts"; import { ServiceMap } from "effect"; @@ -32,6 +33,7 @@ export interface ProviderServiceShape { * Start a provider session. */ readonly startSession: ( + threadId: ThreadId, input: ProviderSessionStartInput, ) => Effect.Effect; diff --git a/apps/server/src/provider/Services/ProviderSessionDirectory.ts b/apps/server/src/provider/Services/ProviderSessionDirectory.ts index b3b68f48971a18abe2edc72b3c2b8d8825d1a5f6..af04a38aa17f8f38c9de7a8136650da61580f1e6 100644 --- a/apps/server/src/provider/Services/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Services/ProviderSessionDirectory.ts @@ -57,6 +57,13 @@ export interface ProviderSessionDirectoryShape { sessionId: ProviderSessionId, ) => Effect.Effect; + /** + * Resolve the full tracked binding for a session id. + */ + readonly getBinding: ( + sessionId: ProviderSessionId, + ) => Effect.Effect, ProviderSessionDirectoryReadError>; + /** * Resolve the tracked thread id for a session, if known. */ diff --git a/packages/contracts/src/provider.ts b/packages/contracts/src/provider.ts index eb5f021320ea8464a4a35afd714c5d23aa981ab4..639d5dcb3f9432946c5d32480ab9031d7dd35395 100644 --- a/packages/contracts/src/provider.ts +++ b/packages/contracts/src/provider.ts @@ -41,6 +41,7 @@ export const ProviderSession = Schema.Struct({ cwd: Schema.optional(TrimmedNonEmptyStringSchema), model: Schema.optional(TrimmedNonEmptyStringSchema), threadId: Schema.optional(ProviderThreadId), + resumeCursor: Schema.optional(Schema.Unknown), activeTurnId: Schema.optional(ProviderTurnId), createdAt: IsoDateTime, updatedAt: IsoDateTime, @@ -53,6 +54,7 @@ export const ProviderSessionStartInput = Schema.Struct({ cwd: Schema.optional(TrimmedNonEmptyStringSchema), model: Schema.optional(TrimmedNonEmptyStringSchema), resumeThreadId: Schema.optional(ProviderThreadId), + resumeCursor: Schema.optional(Schema.Unknown), codexBinaryPath: Schema.optional(TrimmedNonEmptyStringSchema), codexHomePath: Schema.optional(TrimmedNonEmptyStringSchema), approvalPolicy: Schema.optional(ProviderApprovalPolicy), @@ -76,6 +78,7 @@ export type ProviderSendTurnInput = typeof ProviderSendTurnInput.Type; export const ProviderTurnStartResult = Schema.Struct({ threadId: ProviderThreadId, turnId: ProviderTurnId, + resumeCursor: Schema.optional(Schema.Unknown), }); export type ProviderTurnStartResult = typeof ProviderTurnStartResult.Type; -- 2.54.0