| ... | @@ -0,0 +1,446 @@ |
| 1 | import { assert, it } from "@effect/vitest"; |
| 2 | import { |
| 3 | CommandId, |
| 4 | EventId, |
| 5 | MessageId, |
| 6 | NodeId, |
| 7 | ProjectId, |
| 8 | ProviderDriverKind, |
| 9 | ProviderInstanceId, |
| 10 | ProviderThreadId, |
| 11 | ProviderTurnId, |
| 12 | RunAttemptId, |
| 13 | RunId, |
| 14 | ThreadId, |
| 15 | TurnItemId, |
| 16 | type OrchestrationV2DomainEvent, |
| 17 | } from "@t3tools/contracts"; |
| 18 | import * as DateTime from "effect/DateTime"; |
| 19 | import * as Effect from "effect/Effect"; |
| 20 | import * as Fiber from "effect/Fiber"; |
| 21 | import * as Queue from "effect/Queue"; |
| 22 | import * as Stream from "effect/Stream"; |
| 23 | import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; |
| 24 | import * as EffectWorker from "./EffectWorker.ts"; |
| 25 | import * as EventSink from "./EventSink.ts"; |
| 26 | import * as Orchestrator from "./Orchestrator.ts"; |
| 27 | import type { |
| 28 | ProviderAdapterV2Event, |
| 29 | ProviderAdapterV2InterruptInput, |
| 30 | ProviderAdapterV2Shape, |
| 31 | ProviderAdapterV2TurnInput, |
| 32 | } from "./ProviderAdapter.ts"; |
| 33 | import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; |
| 34 | import { makeOrchestratorV2ReplayLayerWithRegistry } from "./testkit/ProviderReplayHarness.ts"; |
| 35 | import { checkpointWorkspace } from "./testkit/ReplayFixtureWorkspace.ts"; |
| 36 | |
| 37 | const driver = ProviderDriverKind.make("codex"); |
| 38 | const instanceId = ProviderInstanceId.make("codex"); |
| 39 | const modelSelection = { instanceId, model: "test-model" }; |
| 40 | |
| 41 | // Codex turns leave commands running, then the thread moves to another |
| 42 | // provider thread (a provider switch). Stop on the newer, settled run must |
| 43 | // reach both provider threads and end all of the Codex work. |
| 44 | it.effect("Stop reaches background work an earlier provider thread still runs", () => |
| 45 | Effect.scoped( |
| 46 | Effect.gen(function* () { |
| 47 | const cwd = yield* checkpointWorkspace("background-work-stop"); |
| 48 | const events = yield* Queue.unbounded<ProviderAdapterV2Event>(); |
| 49 | const started: ProviderAdapterV2TurnInput[] = []; |
| 50 | const interrupts: ProviderAdapterV2InterruptInput[] = []; |
| 51 | const adapter: ProviderAdapterV2Shape = { |
| 52 | instanceId, |
| 53 | driver, |
| 54 | getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), |
| 55 | planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), |
| 56 | openSession: (input) => |
| 57 | Effect.gen(function* () { |
| 58 | const now = yield* DateTime.now; |
| 59 | return { |
| 60 | instanceId, |
| 61 | driver, |
| 62 | providerSessionId: input.providerSessionId, |
| 63 | providerSession: { |
| 64 | id: input.providerSessionId, |
| 65 | driver, |
| 66 | providerInstanceId: instanceId, |
| 67 | status: "ready", |
| 68 | cwd, |
| 69 | model: modelSelection.model, |
| 70 | capabilities: CodexProviderCapabilitiesV2, |
| 71 | createdAt: now, |
| 72 | updatedAt: now, |
| 73 | lastError: null, |
| 74 | }, |
| 75 | events: Stream.fromQueue(events), |
| 76 | ensureThread: ({ threadId }) => |
| 77 | Effect.succeed({ |
| 78 | id: ProviderThreadId.make(`provider-thread:codex:${threadId}`), |
| 79 | driver, |
| 80 | providerInstanceId: instanceId, |
| 81 | providerSessionId: input.providerSessionId, |
| 82 | appThreadId: threadId, |
| 83 | ownerNodeId: null, |
| 84 | nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" }, |
| 85 | nativeConversationHeadRef: null, |
| 86 | status: "idle", |
| 87 | firstRunOrdinal: null, |
| 88 | lastRunOrdinal: null, |
| 89 | handoffIds: [], |
| 90 | forkedFrom: null, |
| 91 | createdAt: now, |
| 92 | updatedAt: now, |
| 93 | }), |
| 94 | resumeThread: ({ providerThread }) => Effect.succeed(providerThread), |
| 95 | startTurn: (turn) => |
| 96 | Effect.gen(function* () { |
| 97 | started.push(turn); |
| 98 | yield* Queue.offer(events, { |
| 99 | type: "provider_turn.updated", |
| 100 | driver, |
| 101 | providerTurn: { |
| 102 | id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`), |
| 103 | providerThreadId: turn.providerThread.id, |
| 104 | nodeId: turn.rootNodeId, |
| 105 | runAttemptId: turn.attemptId, |
| 106 | nativeTurnRef: { |
| 107 | driver, |
| 108 | nativeId: `native:${turn.attemptId}`, |
| 109 | strength: "strong", |
| 110 | }, |
| 111 | ordinal: turn.providerTurnOrdinal, |
| 112 | status: "running", |
| 113 | startedAt: now, |
| 114 | completedAt: null, |
| 115 | }, |
| 116 | }); |
| 117 | }), |
| 118 | steerTurn: () => Effect.die("unused"), |
| 119 | interruptTurn: (interrupt) => |
| 120 | Effect.sync(() => { |
| 121 | interrupts.push(interrupt); |
| 122 | }), |
| 123 | respondToRuntimeRequest: () => Effect.die("unused"), |
| 124 | readThreadSnapshot: () => Effect.die("unused"), |
| 125 | rollbackThread: () => Effect.die("unused"), |
| 126 | forkThread: () => Effect.die("unused"), |
| 127 | }; |
| 128 | }), |
| 129 | }; |
| 130 | yield* Effect.gen(function* () { |
| 131 | const orchestrator = yield* Orchestrator.OrchestratorV2; |
| 132 | const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; |
| 133 | const sink = yield* EventSink.EventSinkV2; |
| 134 | const threadId = ThreadId.make("thread:background-work-stop"); |
| 135 | const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) => |
| 136 | orchestrator.streamDomainEvents.pipe( |
| 137 | Stream.filter(predicate), |
| 138 | Stream.take(1), |
| 139 | Stream.runDrain, |
| 140 | Effect.forkScoped, |
| 141 | ); |
| 142 | yield* orchestrator.dispatch({ |
| 143 | type: "thread.create", |
| 144 | commandId: CommandId.make("create"), |
| 145 | threadId, |
| 146 | projectId: ProjectId.make("project:background-work-stop"), |
| 147 | title: "Background work stop", |
| 148 | modelSelection, |
| 149 | runtimeMode: "full-access", |
| 150 | interactionMode: "default", |
| 151 | branch: null, |
| 152 | worktreePath: cwd, |
| 153 | createdBy: "user", |
| 154 | creationSource: "web", |
| 155 | }); |
| 156 | const running = yield* watch( |
| 157 | (event) => event.type === "provider-turn.updated" && event.payload.status === "running", |
| 158 | ); |
| 159 | yield* orchestrator.dispatch({ |
| 160 | type: "message.dispatch", |
| 161 | commandId: CommandId.make("start-dev-server"), |
| 162 | threadId, |
| 163 | messageId: MessageId.make("message:start-dev-server"), |
| 164 | text: "Start the dev server", |
| 165 | attachments: [], |
| 166 | dispatchMode: { type: "start_immediately" }, |
| 167 | createdBy: "user", |
| 168 | creationSource: "web", |
| 169 | }); |
| 170 | yield* worker.drain(); |
| 171 | yield* Fiber.join(running); |
| 172 | const first = started[0]!; |
| 173 | const codexTurn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; |
| 174 | const devServerId = TurnItemId.make("turn-item:dev-server"); |
| 175 | const now = yield* DateTime.now; |
| 176 | yield* sink.write({ |
| 177 | events: [ |
| 178 | { |
| 179 | id: EventId.make("dev-server"), |
| 180 | type: "turn-item.updated", |
| 181 | threadId, |
| 182 | runId: first.runId, |
| 183 | occurredAt: now, |
| 184 | payload: { |
| 185 | id: devServerId, |
| 186 | threadId, |
| 187 | runId: first.runId, |
| 188 | nodeId: first.rootNodeId, |
| 189 | providerThreadId: codexTurn.providerThreadId, |
| 190 | providerTurnId: codexTurn.id, |
| 191 | nativeItemRef: null, |
| 192 | parentItemId: null, |
| 193 | ordinal: 100, |
| 194 | status: "running", |
| 195 | title: null, |
| 196 | startedAt: now, |
| 197 | completedAt: null, |
| 198 | updatedAt: now, |
| 199 | type: "command_execution", |
| 200 | input: "vp run dev --share", |
| 201 | }, |
| 202 | }, |
| 203 | ], |
| 204 | }); |
| 205 | const settled = yield* watch( |
| 206 | (event) => |
| 207 | event.type === "run.updated" && |
| 208 | event.payload.id === first.runId && |
| 209 | event.payload.status === "waiting", |
| 210 | ); |
| 211 | yield* Queue.offer(events, { |
| 212 | type: "provider_turn.updated", |
| 213 | driver, |
| 214 | providerTurn: { ...codexTurn, status: "completed", completedAt: now }, |
| 215 | }); |
| 216 | yield* Queue.offer(events, { |
| 217 | type: "turn.terminal", |
| 218 | driver, |
| 219 | providerThreadId: codexTurn.providerThreadId, |
| 220 | providerTurnId: codexTurn.id, |
| 221 | runOrdinal: first.runOrdinal, |
| 222 | status: "completed", |
| 223 | failure: null, |
| 224 | threadDisposition: "reusable", |
| 225 | }); |
| 226 | yield* Fiber.join(settled); |
| 227 | yield* worker.drain(); |
| 228 | |
| 229 | const codexProviderThread = (yield* orchestrator.getThreadProjection(threadId)) |
| 230 | .providerThreads[0]!; |
| 231 | // A settled run with its root node, attempt, provider turn and, |
| 232 | // optionally, a command or native subagent it left running. |
| 233 | const settledRun = (input: { |
| 234 | readonly ordinal: number; |
| 235 | readonly providerThreadId: ProviderThreadId; |
| 236 | readonly runningItem?: { readonly id: TurnItemId; readonly kind: "command" | "subagent" }; |
| 237 | }) => { |
| 238 | const runId = RunId.make(`run:${input.ordinal}`); |
| 239 | const attemptId = RunAttemptId.make(`attempt:${input.ordinal}`); |
| 240 | const nodeId = NodeId.make(`node:${input.ordinal}`); |
| 241 | const providerTurnId = ProviderTurnId.make(`provider-turn:${input.ordinal}`); |
| 242 | const events: Array<OrchestrationV2DomainEvent> = [ |
| 243 | { |
| 244 | id: EventId.make(`run:${input.ordinal}`), |
| 245 | type: "run.created", |
| 246 | threadId, |
| 247 | occurredAt: now, |
| 248 | payload: { |
| 249 | id: runId, |
| 250 | threadId, |
| 251 | ordinal: input.ordinal, |
| 252 | providerInstanceId: instanceId, |
| 253 | modelSelection, |
| 254 | providerThreadId: input.providerThreadId, |
| 255 | userMessageId: MessageId.make(`message:${input.ordinal}`), |
| 256 | rootNodeId: nodeId, |
| 257 | activeAttemptId: attemptId, |
| 258 | status: "completed", |
| 259 | requestedAt: now, |
| 260 | startedAt: now, |
| 261 | completedAt: now, |
| 262 | checkpointId: null, |
| 263 | contextHandoffId: null, |
| 264 | }, |
| 265 | }, |
| 266 | { |
| 267 | id: EventId.make(`node:${input.ordinal}`), |
| 268 | type: "node.updated", |
| 269 | threadId, |
| 270 | runId, |
| 271 | occurredAt: now, |
| 272 | payload: { |
| 273 | id: nodeId, |
| 274 | threadId, |
| 275 | runId, |
| 276 | parentNodeId: null, |
| 277 | rootNodeId: nodeId, |
| 278 | kind: "root_turn", |
| 279 | status: "completed", |
| 280 | countsForRun: true, |
| 281 | providerThreadId: input.providerThreadId, |
| 282 | providerTurnId, |
| 283 | nativeItemRef: null, |
| 284 | runtimeRequestId: null, |
| 285 | checkpointScopeId: null, |
| 286 | startedAt: now, |
| 287 | completedAt: now, |
| 288 | }, |
| 289 | }, |
| 290 | { |
| 291 | id: EventId.make(`attempt:${input.ordinal}`), |
| 292 | type: "run-attempt.created", |
| 293 | threadId, |
| 294 | occurredAt: now, |
| 295 | payload: { |
| 296 | id: attemptId, |
| 297 | runId, |
| 298 | attemptOrdinal: 1, |
| 299 | rootNodeId: nodeId, |
| 300 | providerInstanceId: instanceId, |
| 301 | providerThreadId: input.providerThreadId, |
| 302 | providerTurnId, |
| 303 | reason: "initial", |
| 304 | status: "completed", |
| 305 | startedAt: now, |
| 306 | completedAt: now, |
| 307 | }, |
| 308 | }, |
| 309 | { |
| 310 | id: EventId.make(`provider-turn:${input.ordinal}`), |
| 311 | type: "provider-turn.updated", |
| 312 | threadId, |
| 313 | occurredAt: now, |
| 314 | payload: { |
| 315 | id: providerTurnId, |
| 316 | providerThreadId: input.providerThreadId, |
| 317 | nodeId, |
| 318 | runAttemptId: attemptId, |
| 319 | nativeTurnRef: null, |
| 320 | ordinal: input.ordinal, |
| 321 | status: "completed", |
| 322 | startedAt: now, |
| 323 | completedAt: now, |
| 324 | }, |
| 325 | }, |
| 326 | ]; |
| 327 | const item = input.runningItem; |
| 328 | if (item !== undefined) { |
| 329 | const base = { |
| 330 | id: item.id, |
| 331 | threadId, |
| 332 | runId, |
| 333 | nodeId, |
| 334 | providerThreadId: input.providerThreadId, |
| 335 | providerTurnId, |
| 336 | nativeItemRef: null, |
| 337 | parentItemId: null, |
| 338 | ordinal: input.ordinal * 100, |
| 339 | status: "running", |
| 340 | title: null, |
| 341 | startedAt: now, |
| 342 | completedAt: null, |
| 343 | updatedAt: now, |
| 344 | } as const; |
| 345 | events.push({ |
| 346 | id: EventId.make(`item:${input.ordinal}`), |
| 347 | type: "turn-item.updated", |
| 348 | threadId, |
| 349 | runId, |
| 350 | occurredAt: now, |
| 351 | payload: |
| 352 | item.kind === "command" |
| 353 | ? { ...base, type: "command_execution", input: "vp run test --watch" } |
| 354 | : { |
| 355 | ...base, |
| 356 | // A native subagent item names its own provider thread |
| 357 | // but its parent's provider turn. |
| 358 | providerThreadId: ProviderThreadId.make("provider-thread:codex-subagent"), |
| 359 | type: "subagent", |
| 360 | subagentId: NodeId.make(`subagent:${input.ordinal}`), |
| 361 | origin: "provider_native", |
| 362 | driver, |
| 363 | providerInstanceId: instanceId, |
| 364 | childThreadId: null, |
| 365 | prompt: "Review the change", |
| 366 | result: null, |
| 367 | }, |
| 368 | }); |
| 369 | } |
| 370 | return { runId, providerTurnId, events }; |
| 371 | }; |
| 372 | |
| 373 | // Later Codex runs leave a second command and a native subagent. Then |
| 374 | // the thread moves on to another provider thread, which also has a |
| 375 | // live session. |
| 376 | const watcherId = TurnItemId.make("turn-item:watcher"); |
| 377 | const watcherRun = settledRun({ |
| 378 | ordinal: 2, |
| 379 | providerThreadId: codexProviderThread.id, |
| 380 | runningItem: { id: watcherId, kind: "command" }, |
| 381 | }); |
| 382 | const reviewerId = TurnItemId.make("turn-item:reviewer"); |
| 383 | const reviewerRun = settledRun({ |
| 384 | ordinal: 3, |
| 385 | providerThreadId: codexProviderThread.id, |
| 386 | runningItem: { id: reviewerId, kind: "subagent" }, |
| 387 | }); |
| 388 | const otherProviderThreadId = ProviderThreadId.make("provider-thread:other"); |
| 389 | const latestRun = settledRun({ ordinal: 4, providerThreadId: otherProviderThreadId }); |
| 390 | yield* sink.write({ |
| 391 | events: [ |
| 392 | ...watcherRun.events, |
| 393 | ...reviewerRun.events, |
| 394 | { |
| 395 | id: EventId.make("provider-thread:other"), |
| 396 | type: "provider-thread.updated", |
| 397 | threadId, |
| 398 | occurredAt: now, |
| 399 | payload: { |
| 400 | ...codexProviderThread, |
| 401 | id: otherProviderThreadId, |
| 402 | firstRunOrdinal: 4, |
| 403 | lastRunOrdinal: 4, |
| 404 | }, |
| 405 | }, |
| 406 | ...latestRun.events, |
| 407 | ], |
| 408 | }); |
| 409 | |
| 410 | yield* orchestrator.dispatch({ |
| 411 | type: "run.interrupt", |
| 412 | commandId: CommandId.make("stop-background-work"), |
| 413 | threadId, |
| 414 | runId: latestRun.runId, |
| 415 | }); |
| 416 | yield* worker.drain(); |
| 417 | |
| 418 | // Stop reaches both provider threads. The Codex one is interrupted at |
| 419 | // its latest pending work, the subagent's parent turn, so its settle |
| 420 | // covers all three Codex runs. |
| 421 | assert.sameDeepMembers( |
| 422 | interrupts.map((interrupt) => [interrupt.providerThread.id, interrupt.providerTurnId]), |
| 423 | [ |
| 424 | [otherProviderThreadId, latestRun.providerTurnId], |
| 425 | [codexProviderThread.id, reviewerRun.providerTurnId], |
| 426 | ], |
| 427 | ); |
| 428 | const after = yield* orchestrator.getThreadProjection(threadId); |
| 429 | assert.deepEqual( |
| 430 | [devServerId, watcherId, reviewerId].map( |
| 431 | (id) => after.turnItems.find((item) => item.id === id)?.status, |
| 432 | ), |
| 433 | ["interrupted", "interrupted", "interrupted"], |
| 434 | ); |
| 435 | }).pipe( |
| 436 | Effect.provide( |
| 437 | makeOrchestratorV2ReplayLayerWithRegistry( |
| 438 | { name: "background-work-stop" }, |
| 439 | ProviderAdapterRegistry.makeSingleLayer(adapter), |
| 440 | { runEffectWorker: false }, |
| 441 | ), |
| 442 | ), |
| 443 | ); |
| 444 | }), |
| 445 | ), |
| 446 | ); |
Theo Browne <me@t3.gg> 2026-10-03 16:20:17-07:00
GitHub <noreply@github.com> 2026-10-03 16:20:17-07:00