| author |  Julius Marminge <julius0216@outlook.com> 2026-02-26 19:12:09-08:00 |
| committer |  Julius Marminge <jmarminge@gmail.com> 2026-02-26 21:26:35-08:00 |
| log | 718084065e27f51101d9273c9956353a241cb3a5 |
| tree | 58193a551feb6398e1d5bfb732af1cc159fbe5b5 |
| parent | 57dd091e05904dfe07721254863249305932c0b1 |
Restart provider sessions when runtime mode changes
- Persist `approvalPolicy` and `sandboxMode` on orchestration sessions
- Reuse existing sessions when runtime mode is unchanged
- Stop and recreate provider sessions when runtime mode settings change
- Add reactor tests for reuse vs restart behavior4 files changed, 181 insertions(+), 32 deletions(-)
apps
server
src
packages
contracts
apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts+118-6
| ... | ... | @@ -71,16 +71,18 @@ describe("ProviderCommandReactor", () => { |
| 71 | 71 | async function createHarness() { |
| 72 | 72 | const now = new Date().toISOString(); |
| 73 | 73 | const runtimeEventPubSub = Effect.runSync(PubSub.unbounded<ProviderRuntimeEvent>()); |
| 74 | const startSession = vi.fn((_: unknown, __: unknown) => | |
| 75 | Effect.succeed({ | |
| 76 | sessionId: asSessionId("sess-1"), | |
| 74 | let nextSessionIndex = 1; | |
| 75 | const startSession = vi.fn((_: unknown, __: unknown) => { | |
| 76 | const sessionIndex = nextSessionIndex++; | |
| 77 | return Effect.succeed({ | |
| 78 | sessionId: asSessionId(`sess-${sessionIndex}`), | |
| 77 | 79 | provider: "codex" as const, |
| 78 | 80 | status: "ready" as const, |
| 79 | threadId: ProviderThreadId.makeUnsafe("provider-thread-1"), | |
| 81 | threadId: ProviderThreadId.makeUnsafe(`provider-thread-${sessionIndex}`), | |
| 80 | 82 | createdAt: now, |
| 81 | 83 | updatedAt: now, |
| 82 | }), | |
| 83 | ); | |
| 84 | }); | |
| 85 | }); | |
| 84 | 86 | const sendTurn = vi.fn((_: unknown) => |
| 85 | 87 | Effect.succeed({ |
| 86 | 88 | threadId: ProviderThreadId.makeUnsafe("provider-thread-1"), |
| ... | ... | @@ -190,6 +192,116 @@ describe("ProviderCommandReactor", () => { |
| 190 | 192 | const readModel = await Effect.runPromise(harness.engine.getReadModel()); |
| 191 | 193 | const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); |
| 192 | 194 | expect(thread?.session?.providerSessionId).toBe("sess-1"); |
| 195 | expect(thread?.session?.approvalPolicy).toBe("on-request"); | |
| 196 | expect(thread?.session?.sandboxMode).toBe("workspace-write"); | |
| 197 | }); | |
| 198 | ||
| 199 | it("reuses the same provider session when runtime mode is unchanged", async () => { | |
| 200 | const harness = await createHarness(); | |
| 201 | const now = new Date().toISOString(); | |
| 202 | ||
| 203 | await Effect.runPromise( | |
| 204 | harness.engine.dispatch({ | |
| 205 | type: "thread.turn.start", | |
| 206 | commandId: CommandId.makeUnsafe("cmd-turn-start-unchanged-1"), | |
| 207 | threadId: ThreadId.makeUnsafe("thread-1"), | |
| 208 | message: { | |
| 209 | messageId: asMessageId("user-message-unchanged-1"), | |
| 210 | role: "user", | |
| 211 | text: "first", | |
| 212 | attachments: [], | |
| 213 | }, | |
| 214 | approvalPolicy: "on-request", | |
| 215 | sandboxMode: "workspace-write", | |
| 216 | createdAt: now, | |
| 217 | }), | |
| 218 | ); | |
| 219 | ||
| 220 | await waitFor(() => harness.startSession.mock.calls.length === 1); | |
| 221 | await waitFor(() => harness.sendTurn.mock.calls.length === 1); | |
| 222 | ||
| 223 | await Effect.runPromise( | |
| 224 | harness.engine.dispatch({ | |
| 225 | type: "thread.turn.start", | |
| 226 | commandId: CommandId.makeUnsafe("cmd-turn-start-unchanged-2"), | |
| 227 | threadId: ThreadId.makeUnsafe("thread-1"), | |
| 228 | message: { | |
| 229 | messageId: asMessageId("user-message-unchanged-2"), | |
| 230 | role: "user", | |
| 231 | text: "second", | |
| 232 | attachments: [], | |
| 233 | }, | |
| 234 | approvalPolicy: "on-request", | |
| 235 | sandboxMode: "workspace-write", | |
| 236 | createdAt: now, | |
| 237 | }), | |
| 238 | ); | |
| 239 | ||
| 240 | await waitFor(() => harness.sendTurn.mock.calls.length === 2); | |
| 241 | expect(harness.startSession.mock.calls.length).toBe(1); | |
| 242 | expect(harness.stopSession.mock.calls.length).toBe(0); | |
| 243 | }); | |
| 244 | ||
| 245 | it("restarts the provider session when runtime mode changes", async () => { | |
| 246 | const harness = await createHarness(); | |
| 247 | const now = new Date().toISOString(); | |
| 248 | ||
| 249 | await Effect.runPromise( | |
| 250 | harness.engine.dispatch({ | |
| 251 | type: "thread.turn.start", | |
| 252 | commandId: CommandId.makeUnsafe("cmd-turn-start-runtime-mode-1"), | |
| 253 | threadId: ThreadId.makeUnsafe("thread-1"), | |
| 254 | message: { | |
| 255 | messageId: asMessageId("user-message-runtime-mode-1"), | |
| 256 | role: "user", | |
| 257 | text: "first", | |
| 258 | attachments: [], | |
| 259 | }, | |
| 260 | approvalPolicy: "never", | |
| 261 | sandboxMode: "danger-full-access", | |
| 262 | createdAt: now, | |
| 263 | }), | |
| 264 | ); | |
| 265 | ||
| 266 | await waitFor(() => harness.startSession.mock.calls.length === 1); | |
| 267 | await waitFor(() => harness.sendTurn.mock.calls.length === 1); | |
| 268 | ||
| 269 | await Effect.runPromise( | |
| 270 | harness.engine.dispatch({ | |
| 271 | type: "thread.turn.start", | |
| 272 | commandId: CommandId.makeUnsafe("cmd-turn-start-runtime-mode-2"), | |
| 273 | threadId: ThreadId.makeUnsafe("thread-1"), | |
| 274 | message: { | |
| 275 | messageId: asMessageId("user-message-runtime-mode-2"), | |
| 276 | role: "user", | |
| 277 | text: "second", | |
| 278 | attachments: [], | |
| 279 | }, | |
| 280 | approvalPolicy: "on-request", | |
| 281 | sandboxMode: "workspace-write", | |
| 282 | createdAt: now, | |
| 283 | }), | |
| 284 | ); | |
| 285 | ||
| 286 | await waitFor(() => harness.stopSession.mock.calls.length === 1); | |
| 287 | await waitFor(() => harness.startSession.mock.calls.length === 2); | |
| 288 | await waitFor(() => harness.sendTurn.mock.calls.length === 2); | |
| 289 | ||
| 290 | expect(harness.stopSession.mock.calls[0]?.[0]).toEqual({ sessionId: asSessionId("sess-1") }); | |
| 291 | expect(harness.startSession.mock.calls[1]?.[1]).toMatchObject({ | |
| 292 | resumeThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"), | |
| 293 | approvalPolicy: "on-request", | |
| 294 | sandboxMode: "workspace-write", | |
| 295 | }); | |
| 296 | expect(harness.sendTurn.mock.calls[1]?.[0]).toMatchObject({ | |
| 297 | sessionId: asSessionId("sess-2"), | |
| 298 | }); | |
| 299 | ||
| 300 | const readModel = await Effect.runPromise(harness.engine.getReadModel()); | |
| 301 | const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); | |
| 302 | expect(thread?.session?.providerSessionId).toBe("sess-2"); | |
| 303 | expect(thread?.session?.approvalPolicy).toBe("on-request"); | |
| 304 | expect(thread?.session?.sandboxMode).toBe("workspace-write"); | |
| 193 | 305 | }); |
| 194 | 306 | |
| 195 | 307 | it("reacts to thread.turn.interrupt-requested by calling provider interrupt", async () => { |
apps/server/src/orchestration/Layers/ProviderCommandReactor.ts+49-26
| ... | ... | @@ -8,6 +8,8 @@ import { |
| 8 | 8 | type ProviderSandboxMode, |
| 9 | 9 | type OrchestrationSession, |
| 10 | 10 | type ThreadId, |
| 11 | type ProviderSession, | |
| 12 | type ProviderThreadId, | |
| 11 | 13 | type TurnId, |
| 12 | 14 | } from "@t3tools/contracts"; |
| 13 | 15 | import { Cache, Cause, Duration, Effect, Layer, Option, Queue, Stream } from "effect"; |
| ... | ... | @@ -141,11 +143,8 @@ const make = Effect.gen(function* () { |
| 141 | 143 | return yield* Effect.die(new Error(`Thread '${threadId}' was not found in read model.`)); |
| 142 | 144 | } |
| 143 | 145 | |
| 144 | const existingSessionId = thread.session?.providerSessionId; | |
| 145 | if (existingSessionId) { | |
| 146 | return existingSessionId; | |
| 147 | } | |
| 148 | ||
| 146 | const desiredApprovalPolicy = options?.approvalPolicy ?? thread.session?.approvalPolicy; | |
| 147 | const desiredSandboxMode = options?.sandboxMode ?? thread.session?.sandboxMode; | |
| 149 | 148 | const preferredProvider: ProviderKind | undefined = |
| 150 | 149 | thread.session?.providerName === "codex" || thread.session?.providerName === "claudeCode" |
| 151 | 150 | ? thread.session.providerName |
| ... | ... | @@ -155,30 +154,54 @@ const make = Effect.gen(function* () { |
| 155 | 154 | projects: readModel.projects, |
| 156 | 155 | }); |
| 157 | 156 | |
| 158 | const startedSession = yield* providerService.startSession(threadId, { | |
| 159 | ...(preferredProvider ? { provider: preferredProvider } : {}), | |
| 160 | ...(effectiveCwd ? { cwd: effectiveCwd } : {}), | |
| 161 | ...(thread.model ? { model: thread.model } : {}), | |
| 162 | ...(options?.approvalPolicy !== undefined ? { approvalPolicy: options.approvalPolicy } : {}), | |
| 163 | ...(options?.sandboxMode !== undefined ? { sandboxMode: options.sandboxMode } : {}), | |
| 164 | }); | |
| 157 | const startProviderSession = (resumeThreadId?: ProviderThreadId | null) => | |
| 158 | providerService.startSession(threadId, { | |
| 159 | ...(preferredProvider ? { provider: preferredProvider } : {}), | |
| 160 | ...(effectiveCwd ? { cwd: effectiveCwd } : {}), | |
| 161 | ...(thread.model ? { model: thread.model } : {}), | |
| 162 | ...(resumeThreadId ? { resumeThreadId } : {}), | |
| 163 | ...(desiredApprovalPolicy !== undefined ? { approvalPolicy: desiredApprovalPolicy } : {}), | |
| 164 | ...(desiredSandboxMode !== undefined ? { sandboxMode: desiredSandboxMode } : {}), | |
| 165 | }); | |
| 165 | 166 | |
| 166 | yield* setThreadSession({ | |
| 167 | threadId, | |
| 168 | session: { | |
| 167 | const bindSessionToThread = (session: ProviderSession) => | |
| 168 | setThreadSession({ | |
| 169 | 169 | threadId, |
| 170 | status: mapProviderSessionStatusToOrchestrationStatus(startedSession.status), | |
| 171 | providerName: startedSession.provider, | |
| 172 | providerSessionId: startedSession.sessionId, | |
| 173 | providerThreadId: startedSession.threadId ?? null, | |
| 174 | // Provider turn ids are not orchestration turn ids. | |
| 175 | activeTurnId: null, | |
| 176 | lastError: startedSession.lastError ?? null, | |
| 177 | updatedAt: startedSession.updatedAt, | |
| 178 | }, | |
| 179 | createdAt, | |
| 180 | }); | |
| 170 | session: { | |
| 171 | threadId, | |
| 172 | status: mapProviderSessionStatusToOrchestrationStatus(session.status), | |
| 173 | providerName: session.provider, | |
| 174 | providerSessionId: session.sessionId, | |
| 175 | providerThreadId: session.threadId ?? null, | |
| 176 | ...(desiredApprovalPolicy !== undefined ? { approvalPolicy: desiredApprovalPolicy } : {}), | |
| 177 | ...(desiredSandboxMode !== undefined ? { sandboxMode: desiredSandboxMode } : {}), | |
| 178 | // Provider turn ids are not orchestration turn ids. | |
| 179 | activeTurnId: null, | |
| 180 | lastError: session.lastError ?? null, | |
| 181 | updatedAt: session.updatedAt, | |
| 182 | }, | |
| 183 | createdAt, | |
| 184 | }); | |
| 185 | ||
| 186 | const existingSessionId = thread.session?.providerSessionId; | |
| 187 | if (existingSessionId) { | |
| 188 | const approvalPolicyChanged = | |
| 189 | options?.approvalPolicy !== undefined && options.approvalPolicy !== thread.session?.approvalPolicy; | |
| 190 | const sandboxModeChanged = | |
| 191 | options?.sandboxMode !== undefined && options.sandboxMode !== thread.session?.sandboxMode; | |
| 192 | ||
| 193 | if (!approvalPolicyChanged && !sandboxModeChanged) { | |
| 194 | return existingSessionId; | |
| 195 | } | |
| 196 | ||
| 197 | yield* providerService.stopSession({ sessionId: existingSessionId }); | |
| 198 | const restartedSession = yield* startProviderSession(thread.session?.providerThreadId ?? null); | |
| 199 | yield* bindSessionToThread(restartedSession); | |
| 200 | return restartedSession.sessionId; | |
| 201 | } | |
| 181 | 202 | |
| 203 | const startedSession = yield* startProviderSession(); | |
| 204 | yield* bindSessionToThread(startedSession); | |
| 182 | 205 | return startedSession.sessionId; |
| 183 | 206 | }); |
| 184 | 207 |
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+12
| ... | ... | @@ -371,6 +371,12 @@ const make = Effect.gen(function* () { |
| 371 | 371 | providerName: event.provider, |
| 372 | 372 | providerSessionId: event.sessionId, |
| 373 | 373 | providerThreadId, |
| 374 | ...(thread.session?.approvalPolicy !== undefined | |
| 375 | ? { approvalPolicy: thread.session.approvalPolicy } | |
| 376 | : {}), | |
| 377 | ...(thread.session?.sandboxMode !== undefined | |
| 378 | ? { sandboxMode: thread.session.sandboxMode } | |
| 379 | : {}), | |
| 374 | 380 | activeTurnId, |
| 375 | 381 | lastError, |
| 376 | 382 | updatedAt: now, |
| ... | ... | @@ -479,6 +485,12 @@ const make = Effect.gen(function* () { |
| 479 | 485 | providerName: event.provider, |
| 480 | 486 | providerSessionId: event.sessionId, |
| 481 | 487 | providerThreadId, |
| 488 | ...(thread.session?.approvalPolicy !== undefined | |
| 489 | ? { approvalPolicy: thread.session.approvalPolicy } | |
| 490 | : {}), | |
| 491 | ...(thread.session?.sandboxMode !== undefined | |
| 492 | ? { sandboxMode: thread.session.sandboxMode } | |
| 493 | : {}), | |
| 482 | 494 | activeTurnId: toTurnId(event.turnId) ?? null, |
| 483 | 495 | lastError: event.message, |
| 484 | 496 | updatedAt: now, |
packages/contracts/src/orchestration.ts+2
| ... | ... | @@ -142,6 +142,8 @@ export const OrchestrationSession = Schema.Struct({ |
| 142 | 142 | providerName: Schema.NullOr(TrimmedNonEmptyString), |
| 143 | 143 | providerSessionId: Schema.NullOr(ProviderSessionId), |
| 144 | 144 | providerThreadId: Schema.NullOr(ProviderThreadId), |
| 145 | approvalPolicy: Schema.optional(ProviderApprovalPolicy), | |
| 146 | sandboxMode: Schema.optional(ProviderSandboxMode), | |
| 145 | 147 | activeTurnId: Schema.NullOr(TurnId), |
| 146 | 148 | lastError: Schema.NullOr(TrimmedNonEmptyString), |
| 147 | 149 | updatedAt: IsoDateTime, |