authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-25 23:38:02-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
logbb8a8444f4c9e1b4e9ad6c33c9a3c0e020f1b6c7
treefc1a31c9184bdd7ef72546244c7eae2ac75c60aa
parent3ee2e1011dcff6dc46f5a1e0610d6265fb485702

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`

15 files changed, 277 insertions(+), 150 deletions(-)

.plans/16c-pr89-remediation-checklist.md+3-3
...@@ -192,14 +192,14 @@ Counts: active `51` (`valid=33`, `partially-valid=18`), closed-invalid `6`...@@ -192,14 +192,14 @@ Counts: active `51` (`valid=33`, `partially-valid=18`), closed-invalid `6`
192192
193### Phase 3193### Phase 3
194194
195- [ ] `C008` Inconsistent input normalization across CheckpointStore methods195- [x] `C008` Inconsistent input normalization across CheckpointStore methods
196 - Status: `TODO`196 - Status: `DONE`
197 - Verdict: `partially-valid`197 - Verdict: `partially-valid`
198 - Severity: `Medium`198 - Severity: `Medium`
199 - Area: `Checkpointing correctness`199 - Area: `Checkpointing correctness`
200 - File: `apps/server/src/checkpointing/Layers/CheckpointStore.ts:94`200 - File: `apps/server/src/checkpointing/Layers/CheckpointStore.ts:94`
201 - Threads: PRRT_kwDORLtfbc5widJw, PRRT_kwDORLtfbc5wnWv_, PRRT_kwDORLtfbc5w0_g7, PRRT_kwDORLtfbc5w1C36 (+3 duplicate thread(s))201 - Threads: PRRT_kwDORLtfbc5widJw, PRRT_kwDORLtfbc5wnWv_, PRRT_kwDORLtfbc5w0_g7, PRRT_kwDORLtfbc5w1C36 (+3 duplicate thread(s))
202 - Audit note: Normalization inconsistency remains, but upstream callsites now mostly canonicalize inputs.202 - 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.
203203
204- [ ] `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. <details> <summary>🚀 Reply "<strong>fix it for me</strong>" or copy this <strong>AI Prompt</strong> for your agent:</summary>204- [ ] `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. <details> <summary>🚀 Reply "<strong>fix it for me</strong>" or copy this <strong>AI Prompt</strong> for your agent:</summary>
205 - Status: `TODO`205 - Status: `TODO`
apps/server/integration/OrchestrationEngineHarness.integration.ts+1-1
...@@ -159,7 +159,7 @@ export const makeOrchestrationIntegrationHarness = Effect.gen(function* () {...@@ -159,7 +159,7 @@ export const makeOrchestrationIntegrationHarness = Effect.gen(function* () {
159 const rootDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-orchestration-integration-"));159 const rootDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-orchestration-integration-"));
160 const workspaceDir = path.join(rootDir, "workspace");160 const workspaceDir = path.join(rootDir, "workspace");
161 const stateDir = path.join(rootDir, "state");161 const stateDir = path.join(rootDir, "state");
162 const dbPath = path.join(stateDir, "orchestration.sqlite");162 const dbPath = path.join(stateDir, "state.sqlite");
163 fs.mkdirSync(workspaceDir, { recursive: true });163 fs.mkdirSync(workspaceDir, { recursive: true });
164 fs.mkdirSync(stateDir, { recursive: true });164 fs.mkdirSync(stateDir, { recursive: true });
165 initializeGitWorkspace(workspaceDir);165 initializeGitWorkspace(workspaceDir);
apps/server/integration/providerService.integration.test.ts+29-16
...@@ -1,4 +1,5 @@...@@ -1,4 +1,5 @@
1import type { ProviderRuntimeEvent, ProviderSessionId } from "@t3tools/contracts";1import type { ProviderRuntimeEvent, ProviderSessionId } from "@t3tools/contracts";
2import { ThreadId } from "@t3tools/contracts";
2import { NodeServices } from "@effect/platform-node";3import { NodeServices } from "@effect/platform-node";
3import { it, assert } from "@effect/vitest";4import { it, assert } from "@effect/vitest";
4import { Effect, FileSystem, Layer, Path, Queue, Stream } from "effect";5import { Effect, FileSystem, Layer, Path, Queue, Stream } from "effect";
...@@ -116,10 +117,13 @@ it.effect("replays typed runtime fixture events", () =>...@@ -116,10 +117,13 @@ it.effect("replays typed runtime fixture events", () =>
116117
117 yield* Effect.gen(function* () {118 yield* Effect.gen(function* () {
118 const provider = yield* ProviderService;119 const provider = yield* ProviderService;
119 const session = yield* provider.startSession({120 const session = yield* provider.startSession(
120 provider: "codex",121 ThreadId.makeUnsafe("thread-integration-typed"),
121 cwd: fixture.cwd,122 {
122 });123 provider: "codex",
124 cwd: fixture.cwd,
125 },
126 );
123 assert.equal((session.threadId ?? "").length > 0, true);127 assert.equal((session.threadId ?? "").length > 0, true);
124128
125 const observedEvents = yield* runTurn({129 const observedEvents = yield* runTurn({
...@@ -146,10 +150,13 @@ it.effect("replays file-changing fixture turn events", () =>...@@ -146,10 +150,13 @@ it.effect("replays file-changing fixture turn events", () =>
146150
147 yield* Effect.gen(function* () {151 yield* Effect.gen(function* () {
148 const provider = yield* ProviderService;152 const provider = yield* ProviderService;
149 const session = yield* provider.startSession({153 const session = yield* provider.startSession(
150 provider: "codex",154 ThreadId.makeUnsafe("thread-integration-tools"),
151 cwd: fixture.cwd,155 {
152 });156 provider: "codex",
157 cwd: fixture.cwd,
158 },
159 );
153 assert.equal((session.threadId ?? "").length > 0, true);160 assert.equal((session.threadId ?? "").length > 0, true);
154161
155 const observedEvents = yield* runTurn({162 const observedEvents = yield* runTurn({
...@@ -180,10 +187,13 @@ it.effect("runs multi-turn tool/approval flow", () =>...@@ -180,10 +187,13 @@ it.effect("runs multi-turn tool/approval flow", () =>
180187
181 yield* Effect.gen(function* () {188 yield* Effect.gen(function* () {
182 const provider = yield* ProviderService;189 const provider = yield* ProviderService;
183 const session = yield* provider.startSession({190 const session = yield* provider.startSession(
184 provider: "codex",191 ThreadId.makeUnsafe("thread-integration-multi"),
185 cwd: fixture.cwd,192 {
186 });193 provider: "codex",
194 cwd: fixture.cwd,
195 },
196 );
187 assert.equal((session.threadId ?? "").length > 0, true);197 assert.equal((session.threadId ?? "").length > 0, true);
188198
189 const firstTurnEvents = yield* runTurn({199 const firstTurnEvents = yield* runTurn({
...@@ -229,10 +239,13 @@ it.effect("rolls back provider conversation state only", () =>...@@ -229,10 +239,13 @@ it.effect("rolls back provider conversation state only", () =>
229239
230 yield* Effect.gen(function* () {240 yield* Effect.gen(function* () {
231 const provider = yield* ProviderService;241 const provider = yield* ProviderService;
232 const session = yield* provider.startSession({242 const session = yield* provider.startSession(
233 provider: "codex",243 ThreadId.makeUnsafe("thread-integration-rollback"),
234 cwd: fixture.cwd,244 {
235 });245 provider: "codex",
246 cwd: fixture.cwd,
247 },
248 );
236 assert.equal((session.threadId ?? "").length > 0, true);249 assert.equal((session.threadId ?? "").length > 0, true);
237250
238 yield* runTurn({251 yield* runTurn({
apps/server/src/codexAppServerManager.test.ts+6
...@@ -204,6 +204,9 @@ describe("sendTurn", () => {...@@ -204,6 +204,9 @@ describe("sendTurn", () => {
204 expect(result).toEqual({204 expect(result).toEqual({
205 threadId: "thread_1",205 threadId: "thread_1",
206 turnId: "turn_1",206 turnId: "turn_1",
207 resumeCursor: {
208 threadId: "thread_1",
209 },
207 });210 });
208 expect(requireSession).toHaveBeenCalledWith("sess_1");211 expect(requireSession).toHaveBeenCalledWith("sess_1");
209 expect(sendRequest).toHaveBeenCalledWith(context, "turn/start", {212 expect(sendRequest).toHaveBeenCalledWith(context, "turn/start", {
...@@ -225,6 +228,9 @@ describe("sendTurn", () => {...@@ -225,6 +228,9 @@ describe("sendTurn", () => {
225 expect(updateSession).toHaveBeenCalledWith(context, {228 expect(updateSession).toHaveBeenCalledWith(context, {
226 status: "running",229 status: "running",
227 activeTurnId: "turn_1",230 activeTurnId: "turn_1",
231 resumeCursor: {
232 threadId: "thread_1",
233 },
228 });234 });
229 });235 });
230236
apps/server/src/codexAppServerManager.ts+19-3
...@@ -221,15 +221,16 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve...@@ -221,15 +221,16 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve
221 ...sessionOverrides,221 ...sessionOverrides,
222 experimentalRawEvents: false,222 experimentalRawEvents: false,
223 };223 };
224 const resumeThreadId = readResumeThreadId(input);
224225
225 let threadOpenMethod: "thread/start" | "thread/resume" = "thread/start";226 let threadOpenMethod: "thread/start" | "thread/resume" = "thread/start";
226 let threadOpenResponse: unknown;227 let threadOpenResponse: unknown;
227 if (input.resumeThreadId) {228 if (resumeThreadId) {
228 try {229 try {
229 threadOpenMethod = "thread/resume";230 threadOpenMethod = "thread/resume";
230 threadOpenResponse = await this.sendRequest(context, "thread/resume", {231 threadOpenResponse = await this.sendRequest(context, "thread/resume", {
231 ...sessionOverrides,232 ...sessionOverrides,
232 threadId: input.resumeThreadId,233 threadId: resumeThreadId,
233 });234 });
234 } catch (error) {235 } catch (error) {
235 if (!isRecoverableThreadResumeError(error)) {236 if (!isRecoverableThreadResumeError(error)) {
...@@ -240,7 +241,7 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve...@@ -240,7 +241,7 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve
240 this.emitLifecycleEvent(241 this.emitLifecycleEvent(
241 context,242 context,
242 "session/threadResumeFallback",243 "session/threadResumeFallback",
243 `Could not resume thread ${input.resumeThreadId}; started a new thread instead.`,244 `Could not resume thread ${resumeThreadId}; started a new thread instead.`,
244 );245 );
245 threadOpenResponse = await this.sendRequest(context, "thread/start", threadStartParams);246 threadOpenResponse = await this.sendRequest(context, "thread/start", threadStartParams);
246 }247 }
...@@ -261,6 +262,7 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve...@@ -261,6 +262,7 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve
261 this.updateSession(context, {262 this.updateSession(context, {
262 status: "ready",263 status: "ready",
263 threadId,264 threadId,
265 resumeCursor: { threadId },
264 });266 });
265 this.emitLifecycleEvent(context, "session/ready", `Connected to thread ${threadId}`);267 this.emitLifecycleEvent(context, "session/ready", `Connected to thread ${threadId}`);
266 return { ...context.session };268 return { ...context.session };
...@@ -347,11 +349,13 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve...@@ -347,11 +349,13 @@ export class CodexAppServerManager extends EventEmitter<CodexAppServerManagerEve
347 this.updateSession(context, {349 this.updateSession(context, {
348 status: "running",350 status: "running",
349 activeTurnId: turnId,351 activeTurnId: turnId,
352 resumeCursor: context.session.resumeCursor ?? { threadId: context.session.threadId },
350 });353 });
351354
352 return {355 return {
353 threadId: context.session.threadId,356 threadId: context.session.threadId,
354 turnId,357 turnId,
358 resumeCursor: context.session.resumeCursor ?? { threadId: context.session.threadId },
355 };359 };
356 }360 }
357361
...@@ -976,6 +980,18 @@ function toProviderThreadId(value: string | undefined): ProviderThreadId | undef...@@ -976,6 +980,18 @@ function toProviderThreadId(value: string | undefined): ProviderThreadId | undef
976 return brandIfNonEmpty(value, ProviderThreadId.makeUnsafe);980 return brandIfNonEmpty(value, ProviderThreadId.makeUnsafe);
977}981}
978982
983function readResumeThreadId(input: ProviderSessionStartInput): ProviderThreadId | undefined {
984 if (
985 !input.resumeCursor ||
986 typeof input.resumeCursor !== "object" ||
987 Array.isArray(input.resumeCursor)
988 ) {
989 return undefined;
990 }
991 const rawThreadId = (input.resumeCursor as Record<string, unknown>).threadId;
992 return typeof rawThreadId === "string" ? toProviderThreadId(rawThreadId) : input.resumeThreadId;
993}
994
979function toProviderTurnId(value: string | undefined): ProviderTurnId | undefined {995function toProviderTurnId(value: string | undefined): ProviderTurnId | undefined {
980 return brandIfNonEmpty(value, ProviderTurnId.makeUnsafe);996 return brandIfNonEmpty(value, ProviderTurnId.makeUnsafe);
981}997}
apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+3-2
...@@ -71,7 +71,7 @@ describe("ProviderCommandReactor", () => {...@@ -71,7 +71,7 @@ describe("ProviderCommandReactor", () => {
71 async function createHarness() {71 async function createHarness() {
72 const now = new Date().toISOString();72 const now = new Date().toISOString();
73 const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());73 const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>());
74 const startSession = vi.fn((_: unknown) =>74 const startSession = vi.fn((_: unknown, __: unknown) =>
75 Effect.succeed({75 Effect.succeed({
76 sessionId: asSessionId("sess-1"),76 sessionId: asSessionId("sess-1"),
77 provider: "codex" as const,77 provider: "codex" as const,
...@@ -177,7 +177,8 @@ describe("ProviderCommandReactor", () => {...@@ -177,7 +177,8 @@ describe("ProviderCommandReactor", () => {
177177
178 await waitFor(() => harness.startSession.mock.calls.length === 1);178 await waitFor(() => harness.startSession.mock.calls.length === 1);
179 await waitFor(() => harness.sendTurn.mock.calls.length === 1);179 await waitFor(() => harness.sendTurn.mock.calls.length === 1);
180 expect(harness.startSession.mock.calls[0]?.[0]).toMatchObject({180 expect(harness.startSession.mock.calls[0]?.[0]).toEqual(ThreadId.makeUnsafe("thread-1"));
181 expect(harness.startSession.mock.calls[0]?.[1]).toMatchObject({
181 cwd: "/tmp/provider-project",182 cwd: "/tmp/provider-project",
182 model: "gpt-5-codex",183 model: "gpt-5-codex",
183 });184 });
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts+8-5
...@@ -152,11 +152,14 @@ const make = Effect.gen(function* () {...@@ -152,11 +152,14 @@ const make = Effect.gen(function* () {
152 projects: readModel.projects,152 projects: readModel.projects,
153 });153 });
154154
155 const startedSession = yield* providerService.startSession({155 const startedSession = yield* providerService.startSession(
156 ...(preferredProvider ? { provider: preferredProvider } : {}),156 threadId,
157 ...(effectiveCwd ? { cwd: effectiveCwd } : {}),157 {
158 ...(thread.model ? { model: thread.model } : {}),158 ...(preferredProvider ? { provider: preferredProvider } : {}),
159 });159 ...(effectiveCwd ? { cwd: effectiveCwd } : {}),
160 ...(thread.model ? { model: thread.model } : {}),
161 },
162 );
160163
161 yield* setThreadSession({164 yield* setThreadSession({
162 threadId,165 threadId,
apps/server/src/persistence/Layers/Sqlite.ts+1-1
...@@ -52,6 +52,6 @@ export const SqlitePersistenceMemory = Layer.provideMerge(...@@ -52,6 +52,6 @@ export const SqlitePersistenceMemory = Layer.provideMerge(
52export const layerConfig = Effect.gen(function* () {52export const layerConfig = Effect.gen(function* () {
53 const { stateDir } = yield* ServerConfig;53 const { stateDir } = yield* ServerConfig;
54 const { join } = yield* Path.Path;54 const { join } = yield* Path.Path;
55 const dbPath = join(stateDir, "orchestration.sqlite");55 const dbPath = join(stateDir, "state.sqlite");
56 return makeSqlitePersistenceLive(dbPath);56 return makeSqlitePersistenceLive(dbPath);
57}).pipe(Layer.unwrap);57}).pipe(Layer.unwrap);
apps/server/src/provider/Layers/ProviderService.test.ts+67-40
...@@ -49,6 +49,7 @@ const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.make...@@ -49,6 +49,7 @@ const asSessionId = (value: string): ProviderSessionId => ProviderSessionId.make
49const asTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value);49const asTurnId = (value: string): ProviderTurnId => ProviderTurnId.makeUnsafe(value);
50const asRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.makeUnsafe(value);50const asRequestId = (value: string): ApprovalRequestId => ApprovalRequestId.makeUnsafe(value);
51const asEventId = (value: string): EventId => EventId.makeUnsafe(value);51const asEventId = (value: string): EventId => EventId.makeUnsafe(value);
52const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value);
5253
53function makeFakeCodexAdapter() {54function makeFakeCodexAdapter() {
54 const sessions = new Map<ProviderSessionId, ProviderSession>();55 const sessions = new Map<ProviderSessionId, ProviderSession>();
...@@ -322,10 +323,13 @@ it.effect(...@@ -322,10 +323,13 @@ it.effect(
322323
323 const startedSession = yield* Effect.gen(function* () {324 const startedSession = yield* Effect.gen(function* () {
324 const provider = yield* ProviderService;325 const provider = yield* ProviderService;
325 return yield* provider.startSession({326 return yield* provider.startSession(
326 provider: "codex",327 asThreadId("thread-1"),
327 cwd: "/tmp/project",328 {
328 });329 provider: "codex",
330 cwd: "/tmp/project",
331 },
332 );
329 }).pipe(Effect.provide(firstProviderLayer));333 }).pipe(Effect.provide(firstProviderLayer));
330334
331 yield* Effect.gen(function* () {335 yield* Effect.gen(function* () {
...@@ -342,9 +346,7 @@ it.effect(...@@ -342,9 +346,7 @@ it.effect(
342 assert.equal(Option.isSome(persistedAfterStopAll), true);346 assert.equal(Option.isSome(persistedAfterStopAll), true);
343 if (Option.isSome(persistedAfterStopAll)) {347 if (Option.isSome(persistedAfterStopAll)) {
344 assert.equal(persistedAfterStopAll.value.status, "stopped");348 assert.equal(persistedAfterStopAll.value.status, "stopped");
345 assert.deepEqual(persistedAfterStopAll.value.resumeCursor, {349 assert.equal(persistedAfterStopAll.value.resumeCursor, null);
346 resumeThreadId: startedSession.threadId,
347 });
348 }350 }
349351
350 const secondCodex = makeFakeCodexAdapter();352 const secondCodex = makeFakeCodexAdapter();
...@@ -396,10 +398,13 @@ routing.layer("ProviderServiceLive routing", (it) => {...@@ -396,10 +398,13 @@ routing.layer("ProviderServiceLive routing", (it) => {
396 Effect.gen(function* () {398 Effect.gen(function* () {
397 const provider = yield* ProviderService;399 const provider = yield* ProviderService;
398400
399 const session = yield* provider.startSession({401 const session = yield* provider.startSession(
400 provider: "codex",402 asThreadId("thread-1"),
401 cwd: "/tmp/project",403 {
402 });404 provider: "codex",
405 cwd: "/tmp/project",
406 },
407 );
403 assert.equal(session.provider, "codex");408 assert.equal(session.provider, "codex");
404409
405 const sessions = yield* provider.listSessions();410 const sessions = yield* provider.listSessions();
...@@ -448,10 +453,13 @@ routing.layer("ProviderServiceLive routing", (it) => {...@@ -448,10 +453,13 @@ routing.layer("ProviderServiceLive routing", (it) => {
448 Effect.gen(function* () {453 Effect.gen(function* () {
449 const provider = yield* ProviderService;454 const provider = yield* ProviderService;
450455
451 const initial = yield* provider.startSession({456 const initial = yield* provider.startSession(
452 provider: "codex",457 asThreadId("thread-1"),
453 cwd: "/tmp/project",458 {
454 });459 provider: "codex",
460 cwd: "/tmp/project",
461 },
462 );
455 yield* routing.codex.stopSession(initial.sessionId);463 yield* routing.codex.stopSession(initial.sessionId);
456 routing.codex.startSession.mockClear();464 routing.codex.startSession.mockClear();
457 routing.codex.rollbackThread.mockClear();465 routing.codex.rollbackThread.mockClear();
...@@ -479,12 +487,18 @@ routing.layer("ProviderServiceLive routing", (it) => {...@@ -479,12 +487,18 @@ routing.layer("ProviderServiceLive routing", (it) => {
479 Effect.gen(function* () {487 Effect.gen(function* () {
480 const provider = yield* ProviderService;488 const provider = yield* ProviderService;
481489
482 yield* provider.startSession({490 yield* provider.startSession(
483 provider: "codex",491 asThreadId("thread-1"),
484 });492 {
485 yield* provider.startSession({493 provider: "codex",
486 provider: "codex",494 },
487 });495 );
496 yield* provider.startSession(
497 asThreadId("thread-2"),
498 {
499 provider: "codex",
500 },
501 );
488502
489 yield* provider.stopAll();503 yield* provider.stopAll();
490504
...@@ -498,9 +512,12 @@ routing.layer("ProviderServiceLive routing", (it) => {...@@ -498,9 +512,12 @@ routing.layer("ProviderServiceLive routing", (it) => {
498 const provider = yield* ProviderService;512 const provider = yield* ProviderService;
499 const runtimeRepository = yield* ProviderSessionRuntimeRepository;513 const runtimeRepository = yield* ProviderSessionRuntimeRepository;
500514
501 const session = yield* provider.startSession({515 const session = yield* provider.startSession(
502 provider: "codex",516 asThreadId("thread-1"),
503 });517 {
518 provider: "codex",
519 },
520 );
504 yield* provider.sendTurn({521 yield* provider.sendTurn({
505 sessionId: session.sessionId,522 sessionId: session.sessionId,
506 input: "hello",523 input: "hello",
...@@ -513,9 +530,7 @@ routing.layer("ProviderServiceLive routing", (it) => {...@@ -513,9 +530,7 @@ routing.layer("ProviderServiceLive routing", (it) => {
513 assert.equal(Option.isSome(runningRuntime), true);530 assert.equal(Option.isSome(runningRuntime), true);
514 if (Option.isSome(runningRuntime)) {531 if (Option.isSome(runningRuntime)) {
515 assert.equal(runningRuntime.value.status, "running");532 assert.equal(runningRuntime.value.status, "running");
516 assert.deepEqual(runningRuntime.value.resumeCursor, {533 assert.equal(runningRuntime.value.resumeCursor, null);
517 resumeThreadId: "thread-1",
518 });
519 const payload = runningRuntime.value.runtimePayload;534 const payload = runningRuntime.value.runtimePayload;
520 assert.equal(payload !== null && typeof payload === "object", true);535 assert.equal(payload !== null && typeof payload === "object", true);
521 if (payload !== null && typeof payload === "object" && !Array.isArray(payload)) {536 if (payload !== null && typeof payload === "object" && !Array.isArray(payload)) {
...@@ -551,9 +566,12 @@ fanout.layer("ProviderServiceLive fanout", (it) => {...@@ -551,9 +566,12 @@ fanout.layer("ProviderServiceLive fanout", (it) => {
551 it.effect("fans out adapter turn completion events", () =>566 it.effect("fans out adapter turn completion events", () =>
552 Effect.gen(function* () {567 Effect.gen(function* () {
553 const provider = yield* ProviderService;568 const provider = yield* ProviderService;
554 const session = yield* provider.startSession({569 const session = yield* provider.startSession(
555 provider: "codex",570 asThreadId("thread-1"),
556 });571 {
572 provider: "codex",
573 },
574 );
557575
558 const eventsRef = yield* Ref.make<Array<ProviderRuntimeEvent>>([]);576 const eventsRef = yield* Ref.make<Array<ProviderRuntimeEvent>>([]);
559 const consumer = yield* Stream.runForEach(provider.streamEvents, (event) =>577 const consumer = yield* Stream.runForEach(provider.streamEvents, (event) =>
...@@ -588,9 +606,12 @@ fanout.layer("ProviderServiceLive fanout", (it) => {...@@ -588,9 +606,12 @@ fanout.layer("ProviderServiceLive fanout", (it) => {
588 it.effect("keeps subscriber delivery ordered and isolates failing subscribers", () =>606 it.effect("keeps subscriber delivery ordered and isolates failing subscribers", () =>
589 Effect.gen(function* () {607 Effect.gen(function* () {
590 const provider = yield* ProviderService;608 const provider = yield* ProviderService;
591 const session = yield* provider.startSession({609 const session = yield* provider.startSession(
592 provider: "codex",610 asThreadId("thread-1"),
593 });611 {
612 provider: "codex",
613 },
614 );
594615
595 const receivedByHealthy: string[] = [];616 const receivedByHealthy: string[] = [];
596 const expectedEventIds = new Set<string>(["evt-ordered-1", "evt-ordered-2", "evt-ordered-3"]);617 const expectedEventIds = new Set<string>(["evt-ordered-1", "evt-ordered-2", "evt-ordered-3"]);
...@@ -665,9 +686,12 @@ validation.layer("ProviderServiceLive validation", (it) => {...@@ -665,9 +686,12 @@ validation.layer("ProviderServiceLive validation", (it) => {
665 const provider = yield* ProviderService;686 const provider = yield* ProviderService;
666687
667 const failure = yield* Effect.result(688 const failure = yield* Effect.result(
668 provider.startSession({689 provider.startSession(
669 provider: "invalid-provider",690 asThreadId("thread-validation"),
670 } as never),691 {
692 provider: "invalid-provider",
693 } as never,
694 ),
671 );695 );
672696
673 assert.equal(failure._tag, "Failure");697 assert.equal(failure._tag, "Failure");
...@@ -702,10 +726,13 @@ validation.layer("ProviderServiceLive validation", (it) => {...@@ -702,10 +726,13 @@ validation.layer("ProviderServiceLive validation", (it) => {
702 );726 );
703727
704 const failure = yield* Effect.result(728 const failure = yield* Effect.result(
705 provider.startSession({729 provider.startSession(
706 provider: "codex",730 asThreadId("thread-missing"),
707 cwd: "/tmp/project",731 {
708 }),732 provider: "codex",
733 cwd: "/tmp/project",
734 },
735 ),
709 );736 );
710737
711 assertFailure(738 assertFailure(
apps/server/src/provider/Layers/ProviderService.ts+71-43
...@@ -24,10 +24,13 @@ import {...@@ -24,10 +24,13 @@ import {
24} from "@t3tools/contracts";24} from "@t3tools/contracts";
25import { Effect, Layer, Option, PubSub, Queue, Ref, Schema, Stream } from "effect";25import { Effect, Layer, Option, PubSub, Queue, Ref, Schema, Stream } from "effect";
2626
27import { ProviderValidationError } from "../Errors.ts";27import { ProviderSessionNotFoundError, ProviderValidationError } from "../Errors.ts";
28import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts";28import { ProviderAdapterRegistry } from "../Services/ProviderAdapterRegistry.ts";
29import { ProviderService, type ProviderServiceShape } from "../Services/ProviderService.ts";29import { ProviderService, type ProviderServiceShape } from "../Services/ProviderService.ts";
30import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts";30import {
31 ProviderSessionDirectory,
32 type ProviderSessionBinding,
33} from "../Services/ProviderSessionDirectory.ts";
31import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";34import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
3235
33export interface ProviderServiceLiveOptions {36export interface ProviderServiceLiveOptions {
...@@ -154,7 +157,11 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>...@@ -154,7 +157,11 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
154 Effect.asVoid,157 Effect.asVoid,
155 );158 );
156159
157 const upsertSessionBinding = (session: ProviderSession, operation: string) =>160 const upsertSessionBinding = (
161 session: ProviderSession,
162 operation: string,
163 threadId: ThreadId,
164 ) =>
158 Effect.gen(function* () {165 Effect.gen(function* () {
159 const providerThreadId = session.threadId;166 const providerThreadId = session.threadId;
160 if (!providerThreadId) {167 if (!providerThreadId) {
...@@ -165,18 +172,17 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>...@@ -165,18 +172,17 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
165 }172 }
166173
167 const providerSessionId = asProviderSessionId(session.sessionId);174 const providerSessionId = asProviderSessionId(session.sessionId);
168 const brandedThreadId = ThreadId.makeUnsafe(providerThreadId);
169 yield* directory.upsert({175 yield* directory.upsert({
170 sessionId: providerSessionId,176 sessionId: providerSessionId,
171 provider: session.provider,177 provider: session.provider,
172 threadId: brandedThreadId,178 threadId,
173 providerThreadId,179 providerThreadId,
174 status: toRuntimeStatus(session),180 status: toRuntimeStatus(session),
175 resumeCursor: { resumeThreadId: providerThreadId },181 ...(session.resumeCursor !== undefined ? { resumeCursor: session.resumeCursor } : {}),
176 runtimePayload: toRuntimePayloadFromSession(session),182 runtimePayload: toRuntimePayloadFromSession(session),
177 });183 });
178184
179 return brandedThreadId;185 return providerThreadId;
180 });186 });
181187
182 const clearAliasKey = (staleSessionId: ProviderSessionId) =>188 const clearAliasKey = (staleSessionId: ProviderSessionId) =>
...@@ -235,26 +241,33 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>...@@ -235,26 +241,33 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
235241
236 const recoverSessionForThread = (input: {242 const recoverSessionForThread = (input: {
237 readonly staleSessionId: ProviderSessionId;243 readonly staleSessionId: ProviderSessionId;
238 readonly provider: ProviderSession["provider"];244 readonly binding: ProviderSessionBinding & { readonly threadId: ThreadId };
239 readonly threadId: ThreadId;
240 readonly operation: string;245 readonly operation: string;
241 }) =>246 }) =>
242 Effect.gen(function* () {247 Effect.gen(function* () {
243 const adapter = yield* registry.getByProvider(input.provider);248 const adapter = yield* registry.getByProvider(input.binding.provider);
244 const activeSessions = yield* adapter.listSessions();249 const activeSessions = yield* adapter.listSessions();
245 const expectedProviderThreadId = asProviderThreadId(input.threadId);250 const resumeThreadId = input.binding.providerThreadId ?? undefined;
246 const existing = activeSessions.find(251 const hasResumeCursor =
247 (session) => session.threadId === expectedProviderThreadId,252 input.binding.resumeCursor !== null && input.binding.resumeCursor !== undefined;
248 );253 const existing =
254 resumeThreadId === undefined
255 ? undefined
256 : activeSessions.find((session) => session.threadId === resumeThreadId);
249 if (existing) {257 if (existing) {
250 const existingThreadId = yield* upsertSessionBinding(258 const existingProviderThreadId = yield* upsertSessionBinding(
251 existing,259 existing,
252 `${input.operation}:upsertExistingSession`,260 `${input.operation}:upsertExistingSession`,
261 input.binding.threadId,
253 );262 );
254 yield* directory.upsert({263 yield* directory.upsert({
255 sessionId: input.staleSessionId,264 sessionId: input.staleSessionId,
256 provider: existing.provider,265 provider: existing.provider,
257 threadId: existingThreadId,266 threadId: input.binding.threadId,
267 providerThreadId: existingProviderThreadId,
268 ...(existing.resumeCursor !== undefined
269 ? { resumeCursor: existing.resumeCursor }
270 : {}),
258 });271 });
259 if (existing.sessionId !== input.staleSessionId) {272 if (existing.sessionId !== input.staleSessionId) {
260 yield* setAlias(input.staleSessionId, existing.sessionId);273 yield* setAlias(input.staleSessionId, existing.sessionId);
...@@ -267,9 +280,17 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>...@@ -267,9 +280,17 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
267 } as const;280 } as const;
268 }281 }
269282
283 if (!resumeThreadId && !hasResumeCursor) {
284 return yield* toValidationError(
285 input.operation,
286 `Cannot recover stale session '${input.staleSessionId}' because no provider resume state is persisted.`,
287 );
288 }
289
270 const resumed = yield* adapter.startSession({290 const resumed = yield* adapter.startSession({
271 provider: input.provider,291 provider: input.binding.provider,
272 resumeThreadId: ProviderThreadId.makeUnsafe(input.threadId),292 ...(resumeThreadId ? { resumeThreadId } : {}),
293 ...(hasResumeCursor ? { resumeCursor: input.binding.resumeCursor } : {}),
273 });294 });
274 if (resumed.provider !== adapter.provider) {295 if (resumed.provider !== adapter.provider) {
275 return yield* toValidationError(296 return yield* toValidationError(
...@@ -278,21 +299,20 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>...@@ -278,21 +299,20 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
278 );299 );
279 }300 }
280301
281 const resumedThreadId = yield* upsertSessionBinding(302 const resumedProviderThreadId = yield* upsertSessionBinding(
282 resumed,303 resumed,
283 `${input.operation}:upsertRecoveredSession`,304 `${input.operation}:upsertRecoveredSession`,
305 input.binding.threadId,
284 );306 );
285 if (resumedThreadId !== input.threadId) {
286 return yield* toValidationError(
287 input.operation,
288 `Recovered session thread '${resumedThreadId}' does not match expected thread '${input.threadId}'.`,
289 );
290 }
291307
292 yield* directory.upsert({308 yield* directory.upsert({
293 sessionId: input.staleSessionId,309 sessionId: input.staleSessionId,
294 provider: resumed.provider,310 provider: resumed.provider,
295 threadId: resumedThreadId,311 threadId: input.binding.threadId,
312 providerThreadId: resumedProviderThreadId,
313 ...(resumed.resumeCursor !== undefined
314 ? { resumeCursor: resumed.resumeCursor }
315 : {}),
296 });316 });
297317
298 if (resumed.sessionId !== input.staleSessionId) {318 if (resumed.sessionId !== input.staleSessionId) {
...@@ -313,8 +333,26 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>...@@ -313,8 +333,26 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
313 readonly allowRecovery: boolean;333 readonly allowRecovery: boolean;
314 }) =>334 }) =>
315 Effect.gen(function* () {335 Effect.gen(function* () {
316 const provider = yield* directory.getProvider(input.sessionId);336 const bindingOption = yield* directory.getBinding(input.sessionId);
317 const adapter = yield* registry.getByProvider(provider);337 const binding = Option.getOrUndefined(bindingOption);
338 if (!binding) {
339 return yield* new ProviderSessionNotFoundError({
340 sessionId: input.sessionId,
341 });
342 }
343 if (!binding.threadId) {
344 return yield* toValidationError(
345 input.operation,
346 `Cannot route session '${input.sessionId}' because no orchestration thread id is persisted.`,
347 );
348 }
349 const bindingWithThreadId: ProviderSessionBinding & {
350 readonly threadId: ThreadId;
351 } = {
352 ...binding,
353 threadId: binding.threadId,
354 };
355 const adapter = yield* registry.getByProvider(binding.provider);
318356
319 const hasRequestedSession = yield* adapter.hasSession(input.sessionId);357 const hasRequestedSession = yield* adapter.hasSession(input.sessionId);
320 if (hasRequestedSession) {358 if (hasRequestedSession) {
...@@ -349,19 +387,9 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>...@@ -349,19 +387,9 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
349 } as const;387 } as const;
350 }388 }
351389
352 const threadIdOption = yield* directory.getThreadId(input.sessionId);
353 const threadId = Option.getOrUndefined(threadIdOption);
354 if (!threadId) {
355 return yield* toValidationError(
356 input.operation,
357 `Cannot recover stale session '${input.sessionId}' because no thread id is persisted.`,
358 );
359 }
360
361 const recovered = yield* recoverSessionForThread({390 const recovered = yield* recoverSessionForThread({
362 staleSessionId: input.sessionId,391 staleSessionId: input.sessionId,
363 provider,392 binding: bindingWithThreadId,
364 threadId,
365 operation: input.operation,393 operation: input.operation,
366 });394 });
367395
...@@ -372,7 +400,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>...@@ -372,7 +400,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
372 } as const;400 } as const;
373 });401 });
374402
375 const startSession: ProviderServiceShape["startSession"] = (rawInput) =>403 const startSession: ProviderServiceShape["startSession"] = (threadId, rawInput) =>
376 Effect.gen(function* () {404 Effect.gen(function* () {
377 const parsed = yield* decodeInputOrValidationError({405 const parsed = yield* decodeInputOrValidationError({
378 operation: "ProviderService.startSession",406 operation: "ProviderService.startSession",
...@@ -396,7 +424,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>...@@ -396,7 +424,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
396 );424 );
397 }425 }
398426
399 yield* upsertSessionBinding(session, "ProviderService.startSession");427 yield* upsertSessionBinding(session, "ProviderService.startSession", threadId);
400428
401 return session;429 return session;
402 });430 });
...@@ -443,7 +471,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>...@@ -443,7 +471,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
443 threadId,471 threadId,
444 providerThreadId: asProviderThreadId(turn.threadId),472 providerThreadId: asProviderThreadId(turn.threadId),
445 status: "running",473 status: "running",
446 resumeCursor: { resumeThreadId: turn.threadId },474 ...(turn.resumeCursor !== undefined ? { resumeCursor: turn.resumeCursor } : {}),
447 runtimePayload: {475 runtimePayload: {
448 activeTurnId: turn.turnId,476 activeTurnId: turn.turnId,
449 lastRuntimeEvent: "provider.sendTurn",477 lastRuntimeEvent: "provider.sendTurn",
...@@ -550,7 +578,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>...@@ -550,7 +578,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
550 ),578 ),
551 ).pipe(Effect.asVoid);579 ).pipe(Effect.asVoid);
552 // Keep persisted session bindings so stale sessions can be resumed after580 // Keep persisted session bindings so stale sessions can be resumed after
553 // process restart via resumeThreadId.581 // process restart via providerThreadId.
554 yield* Ref.set(routedSessionAliasesRef, new Map());582 yield* Ref.set(routedSessionAliasesRef, new Map());
555 });583 });
556584
apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts+48-36
...@@ -19,10 +19,6 @@ import { ProviderSessionNotFoundError, ProviderValidationError } from "../Errors...@@ -19,10 +19,6 @@ import { ProviderSessionNotFoundError, ProviderValidationError } from "../Errors
19import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts";19import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts";
20import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts";20import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts";
2121
22const sessionId = (value: string) => ProviderSessionId.makeUnsafe(value);
23const threadId = (value: string) => ThreadId.makeUnsafe(value);
24const providerThreadId = (value: string) => ProviderThreadId.makeUnsafe(value);
25
26function makeDirectoryLayer<E, R>(persistenceLayer: Layer.Layer<SqlClient.SqlClient, E, R>) {22function makeDirectoryLayer<E, R>(persistenceLayer: Layer.Layer<SqlClient.SqlClient, E, R>) {
27 const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe(23 const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe(
28 Layer.provide(persistenceLayer),24 Layer.provide(persistenceLayer),
...@@ -33,49 +29,58 @@ function makeDirectoryLayer<E, R>(persistenceLayer: Layer.Layer<SqlClient.SqlCli...@@ -33,49 +29,58 @@ function makeDirectoryLayer<E, R>(persistenceLayer: Layer.Layer<SqlClient.SqlCli
33 );29 );
34}30}
3531
36const layer = it.layer(makeDirectoryLayer(SqlitePersistenceMemory));32it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryLive", (it) => {
37
38layer("ProviderSessionDirectoryLive", (it) => {
39 it("upserts, reads, and removes session bindings", () =>33 it("upserts, reads, and removes session bindings", () =>
40 Effect.gen(function* () {34 Effect.gen(function* () {
41 const directory = yield* ProviderSessionDirectory;35 const directory = yield* ProviderSessionDirectory;
42 const runtimeRepository = yield* ProviderSessionRuntimeRepository;36 const runtimeRepository = yield* ProviderSessionRuntimeRepository;
4337
38 const sessionId = ProviderSessionId.makeUnsafe("sess-1");
39 const initialThreadId = ThreadId.makeUnsafe("thread-1");
40
44 yield* directory.upsert({41 yield* directory.upsert({
45 sessionId: sessionId("sess-1"),42 sessionId,
46 provider: "codex",43 provider: "codex",
47 threadId: threadId("thread-1"),44 threadId: initialThreadId,
48 });45 });
4946
50 const provider = yield* directory.getProvider(sessionId("sess-1"));47 const provider = yield* directory.getProvider(sessionId);
51 assert.equal(provider, "codex");48 assert.equal(provider, "codex");
52 const resolvedThreadId = yield* directory.getThreadId(sessionId("sess-1"));49 const resolvedThreadId = yield* directory.getThreadId(sessionId);
53 assertSome(resolvedThreadId, threadId("thread-1"));50 assertSome(resolvedThreadId, initialThreadId);
51
52 const nextThreadId = ThreadId.makeUnsafe("thread-2");
5453
55 yield* directory.upsert({54 yield* directory.upsert({
56 sessionId: sessionId("sess-1"),55 sessionId: sessionId,
57 provider: "codex",56 provider: "codex",
58 threadId: threadId("thread-2"),57 threadId: nextThreadId,
59 });58 });
60 const updatedThreadId = yield* directory.getThreadId(sessionId("sess-1"));59 const updatedThreadId = yield* directory.getThreadId(sessionId);
61 assertSome(updatedThreadId, threadId("thread-2"));60 assertSome(updatedThreadId, nextThreadId);
61 const updatedBinding = yield* directory.getBinding(sessionId);
62 assert.equal(Option.isSome(updatedBinding), true);
63 if (Option.isSome(updatedBinding)) {
64 assert.equal(updatedBinding.value.threadId, nextThreadId);
65 assert.equal(updatedBinding.value.providerThreadId, null);
66 }
6267
63 const runtime = yield* runtimeRepository.getBySessionId({68 const runtime = yield* runtimeRepository.getBySessionId({
64 providerSessionId: sessionId("sess-1"),69 providerSessionId: sessionId,
65 });70 });
66 assert.equal(Option.isSome(runtime), true);71 assert.equal(Option.isSome(runtime), true);
67 if (Option.isSome(runtime)) {72 if (Option.isSome(runtime)) {
68 assert.equal(runtime.value.threadId, "thread-2");73 assert.equal(runtime.value.threadId, nextThreadId);
69 assert.equal(runtime.value.providerThreadId, "thread-2");74 assert.equal(runtime.value.providerThreadId, null);
70 assert.equal(runtime.value.status, "running");75 assert.equal(runtime.value.status, "running");
71 assert.equal(runtime.value.adapterKey, "codex");76 assert.equal(runtime.value.adapterKey, "codex");
72 }77 }
7378
74 const sessionIds = yield* directory.listSessionIds();79 const sessionIds = yield* directory.listSessionIds();
75 assert.deepEqual(sessionIds, [sessionId("sess-1")]);80 assert.deepEqual(sessionIds, [sessionId]);
7681
77 yield* directory.remove(sessionId("sess-1"));82 yield* directory.remove(sessionId);
78 const missingProvider = yield* directory.getProvider(sessionId("sess-1")).pipe(Effect.result);83 const missingProvider = yield* directory.getProvider(sessionId).pipe(Effect.result);
79 assertFailure(missingProvider, new ProviderSessionNotFoundError({ sessionId: "sess-1" }));84 assertFailure(missingProvider, new ProviderSessionNotFoundError({ sessionId: "sess-1" }));
80 }));85 }));
8186
...@@ -84,7 +89,7 @@ layer("ProviderSessionDirectoryLive", (it) => {...@@ -84,7 +89,7 @@ layer("ProviderSessionDirectoryLive", (it) => {
84 const directory = yield* ProviderSessionDirectory;89 const directory = yield* ProviderSessionDirectory;
85 const result = yield* Effect.result(90 const result = yield* Effect.result(
86 directory.upsert({91 directory.upsert({
87 sessionId: sessionId("sess-no-thread"),92 sessionId: ProviderSessionId.makeUnsafe("sess-no-thread"),
88 provider: "codex",93 provider: "codex",
89 }),94 }),
90 );95 );
...@@ -102,11 +107,15 @@ layer("ProviderSessionDirectoryLive", (it) => {...@@ -102,11 +107,15 @@ layer("ProviderSessionDirectoryLive", (it) => {
102 const directory = yield* ProviderSessionDirectory;107 const directory = yield* ProviderSessionDirectory;
103 const runtimeRepository = yield* ProviderSessionRuntimeRepository;108 const runtimeRepository = yield* ProviderSessionRuntimeRepository;
104109
110 const sessionId = ProviderSessionId.makeUnsafe("sess-runtime");
111 const threadId = ThreadId.makeUnsafe("thread-runtime");
112 const providerThreadId = ProviderThreadId.makeUnsafe("provider-thread-runtime");
113
105 yield* directory.upsert({114 yield* directory.upsert({
106 sessionId: sessionId("sess-runtime"),115 sessionId,
107 provider: "codex",116 provider: "codex",
108 threadId: threadId("thread-runtime"),117 threadId,
109 providerThreadId: providerThreadId("provider-thread-runtime"),118 providerThreadId,
110 status: "starting",119 status: "starting",
111 resumeCursor: {120 resumeCursor: {
112 resumeThreadId: "provider-thread-runtime",121 resumeThreadId: "provider-thread-runtime",
...@@ -118,7 +127,7 @@ layer("ProviderSessionDirectoryLive", (it) => {...@@ -118,7 +127,7 @@ layer("ProviderSessionDirectoryLive", (it) => {
118 });127 });
119128
120 yield* directory.upsert({129 yield* directory.upsert({
121 sessionId: sessionId("sess-runtime"),130 sessionId,
122 provider: "codex",131 provider: "codex",
123 status: "running",132 status: "running",
124 runtimePayload: {133 runtimePayload: {
...@@ -127,15 +136,15 @@ layer("ProviderSessionDirectoryLive", (it) => {...@@ -127,15 +136,15 @@ layer("ProviderSessionDirectoryLive", (it) => {
127 });136 });
128137
129 const runtime = yield* runtimeRepository.getBySessionId({138 const runtime = yield* runtimeRepository.getBySessionId({
130 providerSessionId: sessionId("sess-runtime"),139 providerSessionId: sessionId,
131 });140 });
132 assert.equal(Option.isSome(runtime), true);141 assert.equal(Option.isSome(runtime), true);
133 if (Option.isSome(runtime)) {142 if (Option.isSome(runtime)) {
134 assert.equal(runtime.value.threadId, "thread-runtime");143 assert.equal(runtime.value.threadId, threadId);
135 assert.equal(runtime.value.providerThreadId, "provider-thread-runtime");144 assert.equal(runtime.value.providerThreadId, providerThreadId);
136 assert.equal(runtime.value.status, "running");145 assert.equal(runtime.value.status, "running");
137 assert.deepEqual(runtime.value.resumeCursor, {146 assert.deepEqual(runtime.value.resumeCursor, {
138 resumeThreadId: "provider-thread-runtime",147 resumeThreadId: providerThreadId,
139 });148 });
140 assert.deepEqual(runtime.value.runtimePayload, {149 assert.deepEqual(runtime.value.runtimePayload, {
141 cwd: "/tmp/project",150 cwd: "/tmp/project",
...@@ -151,23 +160,26 @@ layer("ProviderSessionDirectoryLive", (it) => {...@@ -151,23 +160,26 @@ layer("ProviderSessionDirectoryLive", (it) => {
151 const dbPath = path.join(tempDir, "orchestration.sqlite");160 const dbPath = path.join(tempDir, "orchestration.sqlite");
152 const directoryLayer = makeDirectoryLayer(makeSqlitePersistenceLive(dbPath));161 const directoryLayer = makeDirectoryLayer(makeSqlitePersistenceLive(dbPath));
153162
163 const sessionId = ProviderSessionId.makeUnsafe("sess-restart");
164 const threadId = ThreadId.makeUnsafe("thread-restart");
165
154 yield* Effect.gen(function* () {166 yield* Effect.gen(function* () {
155 const directory = yield* ProviderSessionDirectory;167 const directory = yield* ProviderSessionDirectory;
156 yield* directory.upsert({168 yield* directory.upsert({
157 sessionId: sessionId("sess-restart"),169 sessionId,
158 provider: "codex",170 provider: "codex",
159 threadId: threadId("thread-restart"),171 threadId,
160 });172 });
161 }).pipe(Effect.provide(directoryLayer));173 }).pipe(Effect.provide(directoryLayer));
162174
163 yield* Effect.gen(function* () {175 yield* Effect.gen(function* () {
164 const directory = yield* ProviderSessionDirectory;176 const directory = yield* ProviderSessionDirectory;
165 const sql = yield* SqlClient.SqlClient;177 const sql = yield* SqlClient.SqlClient;
166 const provider = yield* directory.getProvider(sessionId("sess-restart"));178 const provider = yield* directory.getProvider(sessionId);
167 assert.equal(provider, "codex");179 assert.equal(provider, "codex");
168180
169 const resolvedThreadId = yield* directory.getThreadId(sessionId("sess-restart"));181 const resolvedThreadId = yield* directory.getThreadId(sessionId);
170 assertSome(resolvedThreadId, threadId("thread-restart"));182 assertSome(resolvedThreadId, threadId);
171183
172 const legacyTableRows = yield* sql<{ readonly name: string }>`184 const legacyTableRows = yield* sql<{ readonly name: string }>`
173 SELECT name185 SELECT name
apps/server/src/provider/Layers/ProviderSessionDirectory.ts+9
...@@ -73,6 +73,11 @@ const makeProviderSessionDirectory = Effect.gen(function* () {...@@ -73,6 +73,11 @@ const makeProviderSessionDirectory = Effect.gen(function* () {
73 sessionId: value.providerSessionId,73 sessionId: value.providerSessionId,
74 provider,74 provider,
75 threadId: value.threadId,75 threadId: value.threadId,
76 adapterKey: value.adapterKey,
77 providerThreadId: value.providerThreadId,
78 status: value.status,
79 resumeCursor: value.resumeCursor,
80 runtimePayload: value.runtimePayload,
76 }),81 }),
77 ),82 ),
78 ),83 ),
...@@ -128,6 +133,9 @@ const makeProviderSessionDirectory = Effect.gen(function* () {...@@ -128,6 +133,9 @@ const makeProviderSessionDirectory = Effect.gen(function* () {
128 ),133 ),
129 );134 );
130135
136 const getBindingBySessionId: ProviderSessionDirectoryShape["getBinding"] = (sessionId) =>
137 getBinding(sessionId);
138
131 const getThreadId: ProviderSessionDirectoryShape["getThreadId"] = (sessionId) =>139 const getThreadId: ProviderSessionDirectoryShape["getThreadId"] = (sessionId) =>
132 getBinding(sessionId).pipe(140 getBinding(sessionId).pipe(
133 Effect.flatMap((binding) =>141 Effect.flatMap((binding) =>
...@@ -168,6 +176,7 @@ const makeProviderSessionDirectory = Effect.gen(function* () {...@@ -168,6 +176,7 @@ const makeProviderSessionDirectory = Effect.gen(function* () {
168 return {176 return {
169 upsert,177 upsert,
170 getProvider,178 getProvider,
179 getBinding: getBindingBySessionId,
171 getThreadId,180 getThreadId,
172 remove,181 remove,
173 listSessionIds,182 listSessionIds,
apps/server/src/provider/Services/ProviderService.ts+2
...@@ -20,6 +20,7 @@ import type {...@@ -20,6 +20,7 @@ import type {
20 ProviderSessionId,20 ProviderSessionId,
21 ProviderSessionStartInput,21 ProviderSessionStartInput,
22 ProviderStopSessionInput,22 ProviderStopSessionInput,
23 ThreadId,
23 ProviderTurnStartResult,24 ProviderTurnStartResult,
24} from "@t3tools/contracts";25} from "@t3tools/contracts";
25import { ServiceMap } from "effect";26import { ServiceMap } from "effect";
...@@ -32,6 +33,7 @@ export interface ProviderServiceShape {...@@ -32,6 +33,7 @@ export interface ProviderServiceShape {
32 * Start a provider session.33 * Start a provider session.
33 */34 */
34 readonly startSession: (35 readonly startSession: (
36 threadId: ThreadId,
35 input: ProviderSessionStartInput,37 input: ProviderSessionStartInput,
36 ) => Effect.Effect<ProviderSession, ProviderServiceError>;38 ) => Effect.Effect<ProviderSession, ProviderServiceError>;
3739
apps/server/src/provider/Services/ProviderSessionDirectory.ts+7
...@@ -57,6 +57,13 @@ export interface ProviderSessionDirectoryShape {...@@ -57,6 +57,13 @@ export interface ProviderSessionDirectoryShape {
57 sessionId: ProviderSessionId,57 sessionId: ProviderSessionId,
58 ) => Effect.Effect<ProviderKind, ProviderSessionDirectoryReadError>;58 ) => Effect.Effect<ProviderKind, ProviderSessionDirectoryReadError>;
5959
60 /**
61 * Resolve the full tracked binding for a session id.
62 */
63 readonly getBinding: (
64 sessionId: ProviderSessionId,
65 ) => Effect.Effect<Option.Option<ProviderSessionBinding>, ProviderSessionDirectoryReadError>;
66
60 /**67 /**
61 * Resolve the tracked thread id for a session, if known.68 * Resolve the tracked thread id for a session, if known.
62 */69 */
packages/contracts/src/provider.ts+3
...@@ -41,6 +41,7 @@ export const ProviderSession = Schema.Struct({...@@ -41,6 +41,7 @@ export const ProviderSession = Schema.Struct({
41 cwd: Schema.optional(TrimmedNonEmptyStringSchema),41 cwd: Schema.optional(TrimmedNonEmptyStringSchema),
42 model: Schema.optional(TrimmedNonEmptyStringSchema),42 model: Schema.optional(TrimmedNonEmptyStringSchema),
43 threadId: Schema.optional(ProviderThreadId),43 threadId: Schema.optional(ProviderThreadId),
44 resumeCursor: Schema.optional(Schema.Unknown),
44 activeTurnId: Schema.optional(ProviderTurnId),45 activeTurnId: Schema.optional(ProviderTurnId),
45 createdAt: IsoDateTime,46 createdAt: IsoDateTime,
46 updatedAt: IsoDateTime,47 updatedAt: IsoDateTime,
...@@ -53,6 +54,7 @@ export const ProviderSessionStartInput = Schema.Struct({...@@ -53,6 +54,7 @@ export const ProviderSessionStartInput = Schema.Struct({
53 cwd: Schema.optional(TrimmedNonEmptyStringSchema),54 cwd: Schema.optional(TrimmedNonEmptyStringSchema),
54 model: Schema.optional(TrimmedNonEmptyStringSchema),55 model: Schema.optional(TrimmedNonEmptyStringSchema),
55 resumeThreadId: Schema.optional(ProviderThreadId),56 resumeThreadId: Schema.optional(ProviderThreadId),
57 resumeCursor: Schema.optional(Schema.Unknown),
56 codexBinaryPath: Schema.optional(TrimmedNonEmptyStringSchema),58 codexBinaryPath: Schema.optional(TrimmedNonEmptyStringSchema),
57 codexHomePath: Schema.optional(TrimmedNonEmptyStringSchema),59 codexHomePath: Schema.optional(TrimmedNonEmptyStringSchema),
58 approvalPolicy: Schema.optional(ProviderApprovalPolicy),60 approvalPolicy: Schema.optional(ProviderApprovalPolicy),
...@@ -76,6 +78,7 @@ export type ProviderSendTurnInput = typeof ProviderSendTurnInput.Type;...@@ -76,6 +78,7 @@ export type ProviderSendTurnInput = typeof ProviderSendTurnInput.Type;
76export const ProviderTurnStartResult = Schema.Struct({78export const ProviderTurnStartResult = Schema.Struct({
77 threadId: ProviderThreadId,79 threadId: ProviderThreadId,
78 turnId: ProviderTurnId,80 turnId: ProviderTurnId,
81 resumeCursor: Schema.optional(Schema.Unknown),
79});82});
80export type ProviderTurnStartResult = typeof ProviderTurnStartResult.Type;83export type ProviderTurnStartResult = typeof ProviderTurnStartResult.Type;
8184