| ... | ... | @@ -10,6 +10,7 @@ import { |
| 10 | 10 | ProviderThreadId, |
| 11 | 11 | ProviderTurnId, |
| 12 | 12 | ThreadId, |
| 13 | type ModelSelection, |
| 13 | 14 | type OrchestrationV2DomainEvent, |
| 14 | 15 | } from "@t3tools/contracts"; |
| 15 | 16 | import * as DateTime from "effect/DateTime"; |
| ... | ... | @@ -402,183 +403,300 @@ for (const mailbox of [false, true]) { |
| 402 | 403 | } |
| 403 | 404 | } |
| 404 | 405 | |
| 405 | | // Claude steers live but cannot interrupt-and-restart. A composer whose model |
| 406 | | // options differ from the running run (an explicit `fastMode: false`) must still |
| 407 | | // steer, keeping the new selection for the next turn. |
| 408 | | it.effect("steers a changed turn-scoped selection into a provider that cannot restart", () => |
| 409 | | Effect.scoped( |
| 410 | | Effect.gen(function* () { |
| 411 | | const cwd = yield* checkpointWorkspace("steering-selection-change"); |
| 412 | | const events = yield* Queue.unbounded<ProviderAdapterV2Event>(); |
| 413 | | const started: ProviderAdapterV2TurnInput[] = []; |
| 414 | | const steered: string[] = []; |
| 415 | | const capabilities = { |
| 416 | | ...CodexProviderCapabilitiesV2, |
| 417 | | turns: { |
| 418 | | ...CodexProviderCapabilitiesV2.turns, |
| 419 | | supportsActiveSteering: true, |
| 420 | | supportsSteeringByInterruptRestart: false, |
| 421 | | }, |
| 422 | | }; |
| 423 | | const runSelection = { |
| 424 | | instanceId, |
| 425 | | model: "test-model", |
| 426 | | options: [{ id: "effort", value: "xhigh" }], |
| 427 | | }; |
| 428 | | const composerSelection = { |
| 429 | | ...runSelection, |
| 430 | | options: [...runSelection.options, { id: "fastMode", value: false }], |
| 431 | | }; |
| 432 | | const adapter: ProviderAdapterV2Shape = { |
| 433 | | instanceId, |
| 434 | | driver, |
| 435 | | getCapabilities: () => Effect.succeed(capabilities), |
| 436 | | planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), |
| 437 | | openSession: (input) => |
| 438 | | Effect.gen(function* () { |
| 439 | | const now = yield* DateTime.now; |
| 440 | | return { |
| 441 | | instanceId, |
| 406 | // Claude and Pi steer live but cannot interrupt-and-restart, so a changed |
| 407 | // selection waits for the next turn as the thread's saved selection. |
| 408 | const runSelection = { |
| 409 | instanceId, |
| 410 | model: "test-model", |
| 411 | options: [{ id: "effort", value: "xhigh" }], |
| 412 | }; |
| 413 | const composerSelection = { |
| 414 | ...runSelection, |
| 415 | options: [...runSelection.options, { id: "fastMode", value: false }], |
| 416 | }; |
| 417 | |
| 418 | const nextTurnSelectionHarness = Effect.fn("nextTurnSelectionHarness")(function* (name: string) { |
| 419 | const cwd = yield* checkpointWorkspace(name); |
| 420 | const events = yield* Queue.unbounded<ProviderAdapterV2Event>(); |
| 421 | const started: ProviderAdapterV2TurnInput[] = []; |
| 422 | const steered: string[] = []; |
| 423 | const capabilities = { |
| 424 | ...CodexProviderCapabilitiesV2, |
| 425 | turns: { |
| 426 | ...CodexProviderCapabilitiesV2.turns, |
| 427 | supportsActiveSteering: true, |
| 428 | supportsSteeringByInterruptRestart: false, |
| 429 | }, |
| 430 | }; |
| 431 | const adapter: ProviderAdapterV2Shape = { |
| 432 | instanceId, |
| 433 | driver, |
| 434 | getCapabilities: () => Effect.succeed(capabilities), |
| 435 | planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), |
| 436 | openSession: (input) => |
| 437 | Effect.gen(function* () { |
| 438 | const now = yield* DateTime.now; |
| 439 | return { |
| 440 | instanceId, |
| 441 | driver, |
| 442 | providerSessionId: input.providerSessionId, |
| 443 | providerSession: { |
| 444 | id: input.providerSessionId, |
| 445 | driver, |
| 446 | providerInstanceId: instanceId, |
| 447 | status: "ready", |
| 448 | cwd, |
| 449 | model: runSelection.model, |
| 450 | capabilities, |
| 451 | createdAt: now, |
| 452 | updatedAt: now, |
| 453 | lastError: null, |
| 454 | }, |
| 455 | events: Stream.fromQueue(events), |
| 456 | ensureThread: ({ threadId }) => |
| 457 | Effect.succeed({ |
| 458 | id: ProviderThreadId.make(`provider-thread:${threadId}`), |
| 442 | 459 | driver, |
| 460 | providerInstanceId: instanceId, |
| 443 | 461 | providerSessionId: input.providerSessionId, |
| 444 | | providerSession: { |
| 445 | | id: input.providerSessionId, |
| 462 | appThreadId: threadId, |
| 463 | ownerNodeId: null, |
| 464 | nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" }, |
| 465 | nativeConversationHeadRef: null, |
| 466 | status: "idle", |
| 467 | firstRunOrdinal: null, |
| 468 | lastRunOrdinal: null, |
| 469 | handoffIds: [], |
| 470 | forkedFrom: null, |
| 471 | createdAt: now, |
| 472 | updatedAt: now, |
| 473 | }), |
| 474 | resumeThread: ({ providerThread }) => Effect.succeed(providerThread), |
| 475 | startTurn: (turn) => |
| 476 | Effect.gen(function* () { |
| 477 | started.push(turn); |
| 478 | yield* Queue.offer(events, { |
| 479 | type: "provider_turn.updated", |
| 446 | 480 | driver, |
| 447 | | providerInstanceId: instanceId, |
| 448 | | status: "ready", |
| 449 | | cwd, |
| 450 | | model: runSelection.model, |
| 451 | | capabilities, |
| 452 | | createdAt: now, |
| 453 | | updatedAt: now, |
| 454 | | lastError: null, |
| 455 | | }, |
| 456 | | events: Stream.fromQueue(events), |
| 457 | | ensureThread: ({ threadId }) => |
| 458 | | Effect.succeed({ |
| 459 | | id: ProviderThreadId.make(`provider-thread:${threadId}`), |
| 460 | | driver, |
| 461 | | providerInstanceId: instanceId, |
| 462 | | providerSessionId: input.providerSessionId, |
| 463 | | appThreadId: threadId, |
| 464 | | ownerNodeId: null, |
| 465 | | nativeThreadRef: { driver, nativeId: "native-thread", strength: "strong" }, |
| 466 | | nativeConversationHeadRef: null, |
| 467 | | status: "idle", |
| 468 | | firstRunOrdinal: null, |
| 469 | | lastRunOrdinal: null, |
| 470 | | handoffIds: [], |
| 471 | | forkedFrom: null, |
| 472 | | createdAt: now, |
| 473 | | updatedAt: now, |
| 474 | | }), |
| 475 | | resumeThread: ({ providerThread }) => Effect.succeed(providerThread), |
| 476 | | startTurn: (turn) => |
| 477 | | Effect.gen(function* () { |
| 478 | | started.push(turn); |
| 479 | | yield* Queue.offer(events, { |
| 480 | | type: "provider_turn.updated", |
| 481 | providerTurn: { |
| 482 | id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`), |
| 483 | providerThreadId: turn.providerThread.id, |
| 484 | nodeId: turn.rootNodeId, |
| 485 | runAttemptId: turn.attemptId, |
| 486 | nativeTurnRef: { |
| 481 | 487 | driver, |
| 482 | | providerTurn: { |
| 483 | | id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`), |
| 484 | | providerThreadId: turn.providerThread.id, |
| 485 | | nodeId: turn.rootNodeId, |
| 486 | | runAttemptId: turn.attemptId, |
| 487 | | nativeTurnRef: { |
| 488 | | driver, |
| 489 | | nativeId: `native:${turn.attemptId}`, |
| 490 | | strength: "strong", |
| 491 | | }, |
| 492 | | ordinal: turn.providerTurnOrdinal, |
| 493 | | status: "running", |
| 494 | | startedAt: now, |
| 495 | | completedAt: null, |
| 496 | | }, |
| 497 | | }); |
| 498 | | }), |
| 499 | | steerTurn: (turn) => |
| 500 | | Effect.sync(() => { |
| 501 | | steered.push(turn.message.text); |
| 502 | | }), |
| 503 | | interruptTurn: () => Effect.void, |
| 504 | | respondToRuntimeRequest: () => Effect.void, |
| 505 | | readThreadSnapshot: () => Effect.die("unused"), |
| 506 | | rollbackThread: () => Effect.die("unused"), |
| 507 | | forkThread: () => Effect.die("unused"), |
| 508 | | }; |
| 509 | | }), |
| 510 | | }; |
| 488 | nativeId: `native:${turn.attemptId}`, |
| 489 | strength: "strong", |
| 490 | }, |
| 491 | ordinal: turn.providerTurnOrdinal, |
| 492 | status: "running", |
| 493 | startedAt: now, |
| 494 | completedAt: null, |
| 495 | }, |
| 496 | }); |
| 497 | }), |
| 498 | steerTurn: (turn) => |
| 499 | Effect.sync(() => { |
| 500 | steered.push(turn.message.text); |
| 501 | }), |
| 502 | interruptTurn: () => Effect.void, |
| 503 | respondToRuntimeRequest: () => Effect.void, |
| 504 | readThreadSnapshot: () => Effect.die("unused"), |
| 505 | rollbackThread: () => Effect.die("unused"), |
| 506 | forkThread: () => Effect.die("unused"), |
| 507 | }; |
| 508 | }), |
| 509 | }; |
| 510 | const layer = makeOrchestratorV2ReplayLayerWithRegistry( |
| 511 | { name }, |
| 512 | ProviderAdapterRegistry.makeSingleLayer(adapter), |
| 513 | { runEffectWorker: false }, |
| 514 | ); |
| 515 | // Creates the thread and starts its first turn on `runSelection`. |
| 516 | const startFirstTurn = Effect.gen(function* () { |
| 517 | const orchestrator = yield* Orchestrator.OrchestratorV2; |
| 518 | const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; |
| 519 | const threadId = ThreadId.make(`thread:${name}`); |
| 520 | yield* orchestrator.dispatch({ |
| 521 | type: "thread.create", |
| 522 | commandId: CommandId.make("create"), |
| 523 | threadId, |
| 524 | projectId: ProjectId.make(`project:${name}`), |
| 525 | title: "Steer with changed options", |
| 526 | modelSelection: runSelection, |
| 527 | runtimeMode: "full-access", |
| 528 | interactionMode: "default", |
| 529 | branch: null, |
| 530 | worktreePath: cwd, |
| 531 | createdBy: "user", |
| 532 | creationSource: "web", |
| 533 | }); |
| 534 | const running = yield* orchestrator.streamDomainEvents.pipe( |
| 535 | Stream.filter( |
| 536 | (event) => event.type === "provider-turn.updated" && event.payload.status === "running", |
| 537 | ), |
| 538 | Stream.take(1), |
| 539 | Stream.runDrain, |
| 540 | Effect.forkScoped, |
| 541 | ); |
| 542 | yield* orchestrator.dispatch({ |
| 543 | type: "message.dispatch", |
| 544 | commandId: CommandId.make("first"), |
| 545 | threadId, |
| 546 | messageId: MessageId.make("message:first"), |
| 547 | text: "first", |
| 548 | attachments: [], |
| 549 | modelSelection: runSelection, |
| 550 | dispatchMode: { type: "start_immediately" }, |
| 551 | createdBy: "user", |
| 552 | creationSource: "web", |
| 553 | }); |
| 554 | yield* worker.drain(); |
| 555 | yield* Fiber.join(running); |
| 556 | return threadId; |
| 557 | }); |
| 558 | return { events, started, steered, layer, startFirstTurn }; |
| 559 | }); |
| 560 | |
| 561 | it.effect("steers a changed turn-scoped selection into a provider that cannot restart", () => |
| 562 | Effect.scoped( |
| 563 | Effect.gen(function* () { |
| 564 | const { started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( |
| 565 | "steering-selection-change", |
| 566 | ); |
| 511 | 567 | yield* Effect.gen(function* () { |
| 512 | 568 | const orchestrator = yield* Orchestrator.OrchestratorV2; |
| 513 | 569 | const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; |
| 514 | | const threadId = ThreadId.make("thread:steering-selection-change"); |
| 570 | const threadId = yield* startFirstTurn; |
| 571 | const steer = (id: string, selection: ModelSelection) => |
| 572 | orchestrator |
| 573 | .dispatch({ |
| 574 | type: "message.dispatch", |
| 575 | commandId: CommandId.make(id), |
| 576 | threadId, |
| 577 | messageId: MessageId.make(`message:${id}`), |
| 578 | text: id, |
| 579 | attachments: [], |
| 580 | modelSelection: selection, |
| 581 | dispatchMode: { type: "steer_active", targetRunId: started[0]!.runId }, |
| 582 | createdBy: "user", |
| 583 | creationSource: "web", |
| 584 | }) |
| 585 | .pipe(Effect.andThen(worker.drain())); |
| 586 | |
| 587 | yield* steer("steer-changed", composerSelection); |
| 588 | const changed = yield* orchestrator.getThreadProjection(threadId); |
| 589 | assert.deepEqual(steered, ["steer-changed"]); |
| 590 | assert.equal(started.length, 1); |
| 591 | assert.lengthOf(changed.attempts, 1); |
| 592 | assert.equal(changed.runs[0]?.status, "running"); |
| 593 | assert.deepEqual(changed.runs[0]?.modelSelection, runSelection); |
| 594 | assert.deepEqual(changed.thread.modelSelection, composerSelection); |
| 595 | |
| 596 | // Choosing the running run's selection again replaces the saved choice. |
| 597 | yield* steer("steer-reverted", runSelection); |
| 598 | const reverted = yield* orchestrator.getThreadProjection(threadId); |
| 599 | assert.deepEqual(steered, ["steer-changed", "steer-reverted"]); |
| 600 | assert.lengthOf(reverted.attempts, 1); |
| 601 | assert.deepEqual(reverted.thread.modelSelection, runSelection); |
| 602 | |
| 603 | // The saved choice moves to another instance while the run keeps going. |
| 604 | // Steering with the run's selection brings the thread's instance back too. |
| 605 | const otherSelection = { |
| 606 | instanceId: ProviderInstanceId.make("codex-work"), |
| 607 | model: "other", |
| 608 | }; |
| 609 | const sink = yield* EventSink.EventSinkV2; |
| 610 | yield* sink.write({ |
| 611 | events: [ |
| 612 | { |
| 613 | id: EventId.make("switched-away"), |
| 614 | type: "thread.provider-switched", |
| 615 | threadId, |
| 616 | providerInstanceId: otherSelection.instanceId, |
| 617 | occurredAt: yield* DateTime.now, |
| 618 | payload: { |
| 619 | ...reverted.thread, |
| 620 | providerInstanceId: otherSelection.instanceId, |
| 621 | modelSelection: otherSelection, |
| 622 | }, |
| 623 | }, |
| 624 | ], |
| 625 | }); |
| 626 | yield* steer("steer-back", runSelection); |
| 627 | const back = yield* orchestrator.getThreadProjection(threadId); |
| 628 | assert.deepEqual(steered, ["steer-changed", "steer-reverted", "steer-back"]); |
| 629 | assert.equal(back.thread.providerInstanceId, instanceId); |
| 630 | assert.deepEqual(back.thread.modelSelection, runSelection); |
| 631 | }).pipe(Effect.provide(layer)); |
| 632 | }), |
| 633 | ), |
| 634 | ); |
| 635 | |
| 636 | it.effect("starts a steer that missed the turn on the saved next-turn selection", () => |
| 637 | Effect.scoped( |
| 638 | Effect.gen(function* () { |
| 639 | const { events, started, steered, layer, startFirstTurn } = yield* nextTurnSelectionHarness( |
| 640 | "steering-selection-follow-up", |
| 641 | ); |
| 642 | yield* Effect.gen(function* () { |
| 643 | const orchestrator = yield* Orchestrator.OrchestratorV2; |
| 644 | const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; |
| 645 | const threadId = yield* startFirstTurn; |
| 646 | const first = started[0]!; |
| 515 | 647 | yield* orchestrator.dispatch({ |
| 516 | | type: "thread.create", |
| 517 | | commandId: CommandId.make("create"), |
| 648 | type: "message.dispatch", |
| 649 | commandId: CommandId.make("steer"), |
| 518 | 650 | threadId, |
| 519 | | projectId: ProjectId.make("project:steering-selection-change"), |
| 520 | | title: "Steer with changed options", |
| 521 | | modelSelection: runSelection, |
| 522 | | runtimeMode: "full-access", |
| 523 | | interactionMode: "default", |
| 524 | | branch: null, |
| 525 | | worktreePath: cwd, |
| 651 | messageId: MessageId.make("message:steer"), |
| 652 | text: "late steer", |
| 653 | attachments: [], |
| 654 | modelSelection: composerSelection, |
| 655 | dispatchMode: { type: "steer_active", targetRunId: first.runId }, |
| 526 | 656 | createdBy: "user", |
| 527 | 657 | creationSource: "web", |
| 528 | 658 | }); |
| 529 | | const running = yield* orchestrator.streamDomainEvents.pipe( |
| 659 | // The turn ends before the worker delivers the steer, so the steer |
| 660 | // becomes a follow-up turn. |
| 661 | const settled = yield* orchestrator.streamDomainEvents.pipe( |
| 530 | 662 | Stream.filter( |
| 531 | | (event) => event.type === "provider-turn.updated" && event.payload.status === "running", |
| 663 | (event) => |
| 664 | event.type === "run.updated" && |
| 665 | event.payload.id === first.runId && |
| 666 | event.payload.status === "waiting", |
| 532 | 667 | ), |
| 533 | 668 | Stream.take(1), |
| 534 | 669 | Stream.runDrain, |
| 535 | 670 | Effect.forkScoped, |
| 536 | 671 | ); |
| 537 | | yield* orchestrator.dispatch({ |
| 538 | | type: "message.dispatch", |
| 539 | | commandId: CommandId.make("first"), |
| 540 | | threadId, |
| 541 | | messageId: MessageId.make("message:first"), |
| 542 | | text: "first", |
| 543 | | attachments: [], |
| 544 | | modelSelection: runSelection, |
| 545 | | dispatchMode: { type: "start_immediately" }, |
| 546 | | createdBy: "user", |
| 547 | | creationSource: "web", |
| 672 | const turn = (yield* orchestrator.getThreadProjection(threadId)).providerTurns[0]!; |
| 673 | yield* Queue.offer(events, { |
| 674 | type: "provider_turn.updated", |
| 675 | driver, |
| 676 | providerTurn: { ...turn, status: "completed", completedAt: yield* DateTime.now }, |
| 548 | 677 | }); |
| 549 | | yield* worker.drain(); |
| 550 | | yield* Fiber.join(running); |
| 551 | | |
| 552 | | yield* orchestrator.dispatch({ |
| 553 | | type: "message.dispatch", |
| 554 | | commandId: CommandId.make("steer"), |
| 555 | | threadId, |
| 556 | | messageId: MessageId.make("message:steer"), |
| 557 | | text: "use the File component", |
| 558 | | attachments: [], |
| 559 | | modelSelection: composerSelection, |
| 560 | | dispatchMode: { type: "steer_active", targetRunId: started[0]!.runId }, |
| 561 | | createdBy: "user", |
| 562 | | creationSource: "web", |
| 678 | yield* Queue.offer(events, { |
| 679 | type: "turn.terminal", |
| 680 | driver, |
| 681 | providerThreadId: turn.providerThreadId, |
| 682 | providerTurnId: turn.id, |
| 683 | runOrdinal: first.runOrdinal, |
| 684 | status: "completed", |
| 685 | failure: null, |
| 686 | threadDisposition: "reusable", |
| 563 | 687 | }); |
| 688 | yield* Fiber.join(settled); |
| 689 | yield* worker.drain(); |
| 690 | yield* orchestrator.resumeQueuedRuns; |
| 564 | 691 | yield* worker.drain(); |
| 565 | 692 | |
| 566 | 693 | const projection = yield* orchestrator.getThreadProjection(threadId); |
| 567 | | assert.deepEqual(steered, ["use the File component"]); |
| 568 | | assert.equal(started.length, 1); |
| 569 | | assert.lengthOf(projection.attempts, 1); |
| 570 | | assert.equal(projection.runs[0]?.status, "running"); |
| 571 | | assert.deepEqual(projection.runs[0]?.modelSelection, runSelection); |
| 694 | assert.deepEqual(steered, []); |
| 695 | assert.equal(started.length, 2); |
| 696 | assert.equal(started[1]?.message.messageId, MessageId.make("message:steer")); |
| 697 | assert.deepEqual(started[1]?.modelSelection, composerSelection); |
| 572 | 698 | assert.deepEqual(projection.thread.modelSelection, composerSelection); |
| 573 | | }).pipe( |
| 574 | | Effect.provide( |
| 575 | | makeOrchestratorV2ReplayLayerWithRegistry( |
| 576 | | { name: "steering-selection-change" }, |
| 577 | | ProviderAdapterRegistry.makeSingleLayer(adapter), |
| 578 | | { runEffectWorker: false }, |
| 579 | | ), |
| 580 | | ), |
| 581 | | ); |
| 699 | }).pipe(Effect.provide(layer)); |
| 582 | 700 | }), |
| 583 | 701 | ), |
| 584 | 702 | ); |
Julius Marminge <julius0216@outlook.com> 2026-10-02 15:06:18-07:00
GitHub <noreply@github.com> 2026-10-02 15:06:18-07:00