| 1 | import * as ScriptProgress from "@t3tools/provider-core/server/ScriptProgress"; |
| 2 | import * as NodeServices from "@effect/platform-node/NodeServices"; |
| 3 | import { describe, expect, it } from "@effect/vitest"; |
| 4 | import { AjvJsonSchemaValidator } from "@modelcontextprotocol/sdk/validation/ajv"; |
| 5 | import type { JsonSchemaType } from "@modelcontextprotocol/sdk/validation"; |
| 6 | import { |
| 7 | CommandId, |
| 8 | EnvironmentId, |
| 9 | EventId, |
| 10 | IsoDateTime, |
| 11 | isProviderNativeSubagentThread, |
| 12 | MessageId, |
| 13 | type ModelSelection, |
| 14 | type OrchestrationV2DelegatedCompletionDelivery, |
| 15 | type OrchestrationV2ProviderCapabilities, |
| 16 | type OrchestrationV2ProviderSession, |
| 17 | type OrchestrationV2ProviderThread, |
| 18 | type OrchestrationV2ThreadProjection, |
| 19 | OrchestratorMcpCreateThreadsResult, |
| 20 | OrchestratorMcpDelegateTaskResult, |
| 21 | OrchestratorMcpTaskCancelResult, |
| 22 | OrchestratorMcpThreadInterruptResult, |
| 23 | OrchestratorMcpThreadListResult, |
| 24 | OrchestratorMcpThreadReadResult, |
| 25 | OrchestratorMcpThreadSendResult, |
| 26 | OrchestratorMcpThreadWaitResult, |
| 27 | ProjectId, |
| 28 | ProviderDriverKind, |
| 29 | ProviderInstanceId, |
| 30 | type ProviderOptionDescriptor, |
| 31 | ProviderThreadId, |
| 32 | ProviderTurnId, |
| 33 | type ScheduledTask, |
| 34 | RuntimeRequestId, |
| 35 | ScheduledTaskId, |
| 36 | type ScheduledTaskUpsertInput, |
| 37 | type ServerProvider, |
| 38 | ThreadId, |
| 39 | ThreadMetadataMcpUpdateResult, |
| 40 | TurnItemId, |
| 41 | } from "@t3tools/contracts"; |
| 42 | import * as DateTime from "effect/DateTime"; |
| 43 | import * as Deferred from "effect/Deferred"; |
| 44 | import * as Effect from "effect/Effect"; |
| 45 | import * as FileSystem from "effect/FileSystem"; |
| 46 | import * as Layer from "effect/Layer"; |
| 47 | import * as Option from "effect/Option"; |
| 48 | import * as PubSub from "effect/PubSub"; |
| 49 | import * as Ref from "effect/Ref"; |
| 50 | import * as Schema from "effect/Schema"; |
| 51 | import * as Fiber from "effect/Fiber"; |
| 52 | import * as Stream from "effect/Stream"; |
| 53 | import { McpSchema, McpServer } from "effect/ai"; |
| 54 | |
| 55 | import { ClaudeProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/ClaudeAdapterV2.ts"; |
| 56 | import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexAdapterV2.ts"; |
| 57 | import { CodexOrchestratorReplayHarness } from "../orchestration-v2/Adapters/CodexAdapterV2.testkit.ts"; |
| 58 | import { threadShellFromProjection } from "../orchestration-v2/ProjectionStore.ts"; |
| 59 | import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts"; |
| 60 | import * as EventSink from "../orchestration-v2/EventSink.ts"; |
| 61 | import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; |
| 62 | import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts"; |
| 63 | import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; |
| 64 | import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; |
| 65 | import * as ServerSettings from "../serverSettings.ts"; |
| 66 | import * as SqlitePersistence from "../persistence/Sqlite.ts"; |
| 67 | import * as ThreadSearch from "../orchestration-v2/ThreadSearch.ts"; |
| 68 | import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts"; |
| 69 | import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; |
| 70 | import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; |
| 71 | import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace"; |
| 72 | import * as ProviderReplayHarness from "../orchestration-v2/testkit/ProviderReplayHarness.ts"; |
| 73 | import { |
| 74 | decodeProviderReplayNdjson, |
| 75 | materializeReplayTranscriptWorkspace, |
| 76 | } from "@t3tools/provider-testing/replayTranscript"; |
| 77 | import * as ProviderRegistryMock from "../provider/testUtils/providerRegistryMock.ts"; |
| 78 | import * as ProjectService from "../project/ProjectService.ts"; |
| 79 | import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; |
| 80 | import * as SecretRequests from "../secrets/SecretRequests.ts"; |
| 81 | import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; |
| 82 | import * as McpHttpServer from "./McpHttpServer.ts"; |
| 83 | import * as McpInvocationContext from "./McpInvocationContext.ts"; |
| 84 | import { delegatedTaskRun, hasPendingChildRuns } from "./OrchestratorMcpService.ts"; |
| 85 | import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; |
| 86 | |
| 87 | const layerOrchestratorToolkit = McpHttpServer.layerOrchestratorToolkit.pipe( |
| 88 | Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})), |
| 89 | Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})), |
| 90 | ); |
| 91 | |
| 92 | // Effect returns a declared tool failure as `isError` with its encoded payload |
| 93 | // as JSON text, never as `structuredContent`. |
| 94 | const declaredFailure = (result: McpSchema.CallToolResult) => { |
| 95 | const text = result.content[0]; |
| 96 | return result.isError === true && text?.type === "text" ? JSON.parse(text.text) : undefined; |
| 97 | }; |
| 98 | |
| 99 | const parentThreadId = ThreadId.make("thread:mcp-orchestrator-parent"); |
| 100 | const projectId = ProjectId.make("project:mcp-orchestrator"); |
| 101 | const codexInstanceId = ProviderInstanceId.make("codex"); |
| 102 | const claudeInstanceId = ProviderInstanceId.make("claudeAgent"); |
| 103 | const codexModel = "gpt-5.4"; |
| 104 | const claudeModel = "claude-sonnet-4-6"; |
| 105 | const parentPrompt = "Keep this parent turn active while orchestration tools are tested."; |
| 106 | const delegatedPrompt = "Inspect the delegated API boundary and return the result."; |
| 107 | const delegatedResult = "Delegated API boundary inspected."; |
| 108 | const cancellationPrompt = "Remain active until the parent cancels this delegated task."; |
| 109 | const createdThreadPrompt = "Complete the newly created ordinary thread."; |
| 110 | const queuedFollowupPrompt = "Complete the queued follow-up and return the final result."; |
| 111 | const queuedFollowupResult = "Queued delegated follow-up completed."; |
| 112 | |
| 113 | const decodeCreateThreadsResult = Schema.decodeUnknownEffect(OrchestratorMcpCreateThreadsResult); |
| 114 | const decodeDelegateTaskResult = Schema.decodeUnknownEffect(OrchestratorMcpDelegateTaskResult); |
| 115 | const decodeTaskCancelResult = Schema.decodeUnknownEffect(OrchestratorMcpTaskCancelResult); |
| 116 | const decodeThreadInterruptResult = Schema.decodeUnknownEffect( |
| 117 | OrchestratorMcpThreadInterruptResult, |
| 118 | ); |
| 119 | const decodeThreadListResult = Schema.decodeUnknownEffect(OrchestratorMcpThreadListResult); |
| 120 | const decodeThreadReadResult = Schema.decodeUnknownEffect(OrchestratorMcpThreadReadResult); |
| 121 | const decodeThreadSendResult = Schema.decodeUnknownEffect(OrchestratorMcpThreadSendResult); |
| 122 | const decodeThreadWaitResult = Schema.decodeUnknownEffect(OrchestratorMcpThreadWaitResult); |
| 123 | const decodeThreadUpdateResult = Schema.decodeUnknownEffect(ThreadMetadataMcpUpdateResult); |
| 124 | |
| 125 | const codexSelection = { |
| 126 | instanceId: codexInstanceId, |
| 127 | model: codexModel, |
| 128 | } satisfies ModelSelection; |
| 129 | |
| 130 | const claudeSelection = { |
| 131 | instanceId: claudeInstanceId, |
| 132 | model: claudeModel, |
| 133 | } satisfies ModelSelection; |
| 134 | |
| 135 | const delegatedTaskStatusTranscriptFile = new URL( |
| 136 | "../orchestration-v2/testkit/fixtures/delegated_task_status/codex_transcript.ndjson", |
| 137 | import.meta.url, |
| 138 | ); |
| 139 | |
| 140 | const readDelegatedTaskStatusTranscript = Effect.fn("readDelegatedTaskStatusTranscript")( |
| 141 | function* () { |
| 142 | const fs = yield* FileSystem.FileSystem; |
| 143 | const text = yield* fs.readFileString( |
| 144 | decodeURIComponent(delegatedTaskStatusTranscriptFile.pathname), |
| 145 | ); |
| 146 | return yield* decodeProviderReplayNdjson(text); |
| 147 | }, |
| 148 | Effect.provide(NodeServices.layer), |
| 149 | ); |
| 150 | |
| 151 | interface CapturedTurn { |
| 152 | readonly instanceId: ProviderInstanceId; |
| 153 | readonly threadId: ThreadId; |
| 154 | readonly text: string; |
| 155 | } |
| 156 | |
| 157 | function unsupported(driver: ProviderDriverKind, detail: string) { |
| 158 | return Effect.fail(new ProviderAdapter.ProviderAdapterProtocolError({ driver, detail })); |
| 159 | } |
| 160 | |
| 161 | function makeProviderSnapshot(input: { |
| 162 | readonly instanceId: ProviderInstanceId; |
| 163 | readonly driver: ProviderDriverKind; |
| 164 | readonly model: string; |
| 165 | readonly optionDescriptors?: ReadonlyArray<ProviderOptionDescriptor>; |
| 166 | }): ServerProvider { |
| 167 | return { |
| 168 | instanceId: input.instanceId, |
| 169 | driver: input.driver, |
| 170 | enabled: true, |
| 171 | installed: true, |
| 172 | version: "test", |
| 173 | status: "ready", |
| 174 | auth: { status: "authenticated" }, |
| 175 | checkedAt: "2026-06-17T00:00:00.000Z", |
| 176 | models: [ |
| 177 | { |
| 178 | slug: input.model, |
| 179 | name: input.model, |
| 180 | isCustom: false, |
| 181 | capabilities: |
| 182 | input.optionDescriptors === undefined |
| 183 | ? null |
| 184 | : { optionDescriptors: input.optionDescriptors }, |
| 185 | }, |
| 186 | ], |
| 187 | slashCommands: [], |
| 188 | skills: [], |
| 189 | }; |
| 190 | } |
| 191 | |
| 192 | function makeDeterministicAdapter(input: { |
| 193 | readonly instanceId: ProviderInstanceId; |
| 194 | readonly driver: ProviderDriverKind; |
| 195 | readonly capabilities: OrchestrationV2ProviderCapabilities; |
| 196 | readonly capturedTurns: Ref.Ref<ReadonlyArray<CapturedTurn>>; |
| 197 | readonly shouldComplete: (turn: ProviderAdapter.ProviderAdapterV2TurnInput) => boolean; |
| 198 | readonly terminalGate?: ( |
| 199 | turn: ProviderAdapter.ProviderAdapterV2TurnInput, |
| 200 | ) => Deferred.Deferred<void> | undefined; |
| 201 | readonly response: (turn: ProviderAdapter.ProviderAdapterV2TurnInput) => string; |
| 202 | }): ProviderAdapter.ProviderAdapterV2["Service"] { |
| 203 | return { |
| 204 | instanceId: input.instanceId, |
| 205 | driver: input.driver, |
| 206 | getCapabilities: () => Effect.succeed(input.capabilities), |
| 207 | planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), |
| 208 | openSession: (sessionInput) => |
| 209 | Effect.gen(function* () { |
| 210 | const events = yield* PubSub.unbounded<ProviderAdapter.ProviderAdapterV2Event>(); |
| 211 | const now = yield* DateTime.now; |
| 212 | const providerSession: OrchestrationV2ProviderSession = { |
| 213 | id: sessionInput.providerSessionId, |
| 214 | driver: input.driver, |
| 215 | providerInstanceId: input.instanceId, |
| 216 | status: "ready", |
| 217 | cwd: sessionInput.runtimePolicy.cwd ?? process.cwd(), |
| 218 | model: sessionInput.modelSelection.model, |
| 219 | capabilities: input.capabilities, |
| 220 | createdAt: now, |
| 221 | updatedAt: now, |
| 222 | lastError: null, |
| 223 | }; |
| 224 | |
| 225 | const publish = (providerEvents: ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>) => |
| 226 | Effect.forEach(providerEvents, (event) => PubSub.publish(events, event), { |
| 227 | discard: true, |
| 228 | }); |
| 229 | const runOrdinals = new Map<ProviderTurnId, number>(); |
| 230 | const turnInputs = new Map<ProviderTurnId, ProviderAdapter.ProviderAdapterV2TurnInput>(); |
| 231 | |
| 232 | return { |
| 233 | instanceId: input.instanceId, |
| 234 | driver: input.driver, |
| 235 | providerSessionId: sessionInput.providerSessionId, |
| 236 | providerSession, |
| 237 | events: Stream.fromPubSub(events), |
| 238 | ensureThread: (threadInput) => |
| 239 | Effect.gen(function* () { |
| 240 | const createdAt = yield* DateTime.now; |
| 241 | const nativeThreadId = `${input.driver}:${threadInput.threadId}`; |
| 242 | return { |
| 243 | id: ProviderThreadId.make(`provider-thread:${nativeThreadId}`), |
| 244 | driver: input.driver, |
| 245 | providerInstanceId: input.instanceId, |
| 246 | providerSessionId: sessionInput.providerSessionId, |
| 247 | appThreadId: threadInput.threadId, |
| 248 | ownerNodeId: null, |
| 249 | nativeThreadRef: { |
| 250 | driver: input.driver, |
| 251 | nativeId: nativeThreadId, |
| 252 | strength: "strong", |
| 253 | }, |
| 254 | nativeConversationHeadRef: null, |
| 255 | status: "idle", |
| 256 | firstRunOrdinal: null, |
| 257 | lastRunOrdinal: null, |
| 258 | handoffIds: [], |
| 259 | forkedFrom: null, |
| 260 | createdAt, |
| 261 | updatedAt: createdAt, |
| 262 | } satisfies OrchestrationV2ProviderThread; |
| 263 | }), |
| 264 | resumeThread: ({ providerThread }) => Effect.succeed(providerThread), |
| 265 | startTurn: (turnInput) => |
| 266 | Effect.gen(function* () { |
| 267 | yield* Ref.update(input.capturedTurns, (turns) => [ |
| 268 | ...turns, |
| 269 | { |
| 270 | instanceId: input.instanceId, |
| 271 | threadId: turnInput.threadId, |
| 272 | text: turnInput.message.text, |
| 273 | }, |
| 274 | ]); |
| 275 | const eventTime = yield* DateTime.now; |
| 276 | const providerTurnId = ProviderTurnId.make( |
| 277 | `provider-turn:${input.instanceId}:${turnInput.threadId}:${turnInput.runOrdinal}`, |
| 278 | ); |
| 279 | runOrdinals.set(providerTurnId, turnInput.runOrdinal); |
| 280 | turnInputs.set(providerTurnId, turnInput); |
| 281 | yield* publish([ |
| 282 | { |
| 283 | type: "provider_turn.updated", |
| 284 | driver: input.driver, |
| 285 | providerTurn: { |
| 286 | id: providerTurnId, |
| 287 | providerThreadId: turnInput.providerThread.id, |
| 288 | nodeId: turnInput.rootNodeId, |
| 289 | runAttemptId: turnInput.attemptId, |
| 290 | nativeTurnRef: { |
| 291 | driver: input.driver, |
| 292 | nativeId: `native-turn:${turnInput.threadId}:${turnInput.runOrdinal}`, |
| 293 | strength: "strong", |
| 294 | }, |
| 295 | ordinal: turnInput.providerTurnOrdinal, |
| 296 | status: "running", |
| 297 | startedAt: eventTime, |
| 298 | completedAt: null, |
| 299 | }, |
| 300 | }, |
| 301 | ]); |
| 302 | const terminalGate = input.terminalGate?.(turnInput); |
| 303 | if (terminalGate !== undefined) { |
| 304 | yield* Deferred.await(terminalGate); |
| 305 | } else if (!input.shouldComplete(turnInput)) { |
| 306 | return; |
| 307 | } |
| 308 | const response = input.response(turnInput); |
| 309 | yield* publish([ |
| 310 | { |
| 311 | type: "provider_turn.updated", |
| 312 | driver: input.driver, |
| 313 | providerTurn: { |
| 314 | id: providerTurnId, |
| 315 | providerThreadId: turnInput.providerThread.id, |
| 316 | nodeId: turnInput.rootNodeId, |
| 317 | runAttemptId: turnInput.attemptId, |
| 318 | nativeTurnRef: { |
| 319 | driver: input.driver, |
| 320 | nativeId: `native-turn:${turnInput.threadId}:${turnInput.runOrdinal}`, |
| 321 | strength: "strong", |
| 322 | }, |
| 323 | ordinal: turnInput.providerTurnOrdinal, |
| 324 | status: "completed", |
| 325 | startedAt: eventTime, |
| 326 | completedAt: eventTime, |
| 327 | }, |
| 328 | }, |
| 329 | { |
| 330 | type: "turn_item.updated", |
| 331 | driver: input.driver, |
| 332 | turnItem: { |
| 333 | id: TurnItemId.make( |
| 334 | `turn-item:${input.instanceId}:${turnInput.threadId}:${turnInput.runOrdinal}:assistant`, |
| 335 | ), |
| 336 | threadId: turnInput.threadId, |
| 337 | runId: turnInput.runId, |
| 338 | nodeId: turnInput.rootNodeId, |
| 339 | providerThreadId: turnInput.providerThread.id, |
| 340 | providerTurnId, |
| 341 | nativeItemRef: null, |
| 342 | parentItemId: null, |
| 343 | ordinal: turnInput.runOrdinal * 100 + 1, |
| 344 | status: "completed", |
| 345 | title: null, |
| 346 | startedAt: eventTime, |
| 347 | completedAt: eventTime, |
| 348 | updatedAt: eventTime, |
| 349 | type: "assistant_message", |
| 350 | messageId: MessageId.make( |
| 351 | `message:${input.instanceId}:${turnInput.threadId}:${turnInput.runOrdinal}:assistant`, |
| 352 | ), |
| 353 | text: response, |
| 354 | streaming: false, |
| 355 | }, |
| 356 | }, |
| 357 | { |
| 358 | type: "turn.terminal", |
| 359 | driver: input.driver, |
| 360 | providerThreadId: turnInput.providerThread.id, |
| 361 | providerTurnId, |
| 362 | runOrdinal: turnInput.runOrdinal, |
| 363 | status: "completed", |
| 364 | failure: null, |
| 365 | threadDisposition: "reusable", |
| 366 | }, |
| 367 | ]); |
| 368 | }), |
| 369 | steerTurn: () => Effect.void, |
| 370 | interruptTurn: ({ providerThread, providerTurnId }) => |
| 371 | Effect.gen(function* () { |
| 372 | const turnInput = turnInputs.get(providerTurnId); |
| 373 | const completedAt = yield* DateTime.now; |
| 374 | if (turnInput !== undefined) { |
| 375 | yield* publish([ |
| 376 | { |
| 377 | type: "provider_turn.updated", |
| 378 | driver: input.driver, |
| 379 | providerTurn: { |
| 380 | id: providerTurnId, |
| 381 | providerThreadId: providerThread.id, |
| 382 | nodeId: turnInput.rootNodeId, |
| 383 | runAttemptId: turnInput.attemptId, |
| 384 | nativeTurnRef: { |
| 385 | driver: input.driver, |
| 386 | nativeId: `native-turn:${turnInput.threadId}:${turnInput.runOrdinal}`, |
| 387 | strength: "strong", |
| 388 | }, |
| 389 | ordinal: turnInput.providerTurnOrdinal, |
| 390 | status: "interrupted", |
| 391 | startedAt: completedAt, |
| 392 | completedAt, |
| 393 | }, |
| 394 | }, |
| 395 | ]); |
| 396 | } |
| 397 | yield* publish([ |
| 398 | { |
| 399 | type: "turn.terminal", |
| 400 | driver: input.driver, |
| 401 | providerThreadId: providerThread.id, |
| 402 | providerTurnId, |
| 403 | runOrdinal: runOrdinals.get(providerTurnId) ?? 1, |
| 404 | status: "interrupted", |
| 405 | failure: null, |
| 406 | threadDisposition: "reusable", |
| 407 | }, |
| 408 | ]); |
| 409 | }), |
| 410 | respondToRuntimeRequest: () => Effect.void, |
| 411 | readThreadSnapshot: () => |
| 412 | unsupported(input.driver, "readThreadSnapshot is unused in this test"), |
| 413 | rollbackThread: () => unsupported(input.driver, "rollbackThread is unused in this test"), |
| 414 | forkThread: () => unsupported(input.driver, "forkThread is unused in this test"), |
| 415 | }; |
| 416 | }), |
| 417 | }; |
| 418 | } |
| 419 | |
| 420 | function waitForProjection( |
| 421 | orchestrator: Orchestrator.OrchestratorV2Shape, |
| 422 | threadId: ThreadId, |
| 423 | predicate: (projection: OrchestrationV2ThreadProjection) => boolean, |
| 424 | ) { |
| 425 | return Effect.gen(function* () { |
| 426 | for (let attempt = 0; attempt < 1_000; attempt += 1) { |
| 427 | const projection = yield* orchestrator.getThreadProjection(threadId); |
| 428 | if (predicate(projection)) { |
| 429 | return projection; |
| 430 | } |
| 431 | yield* Effect.sleep("5 millis"); |
| 432 | } |
| 433 | return yield* Effect.die( |
| 434 | new Error(`Timed out waiting for orchestration projection ${threadId}.`), |
| 435 | ); |
| 436 | }); |
| 437 | } |
| 438 | |
| 439 | const client = McpSchema.McpServerClient.of({ |
| 440 | clientId: 1, |
| 441 | protocolVersion: "2025-06-18", |
| 442 | clientCapabilities: {}, |
| 443 | clientInfo: { name: "orchestrator-mcp-test", version: "1.0.0" }, |
| 444 | initializePayload: { |
| 445 | protocolVersion: "2025-06-18", |
| 446 | capabilities: {}, |
| 447 | clientInfo: { name: "orchestrator-mcp-test", version: "1.0.0" }, |
| 448 | }, |
| 449 | getClient: Effect.die("unused"), |
| 450 | }); |
| 451 | |
| 452 | /** Build a persisted-looking ScheduledTask from an upsert input for the in-memory stub. */ |
| 453 | function scheduledTaskFromUpsert(input: ScheduledTaskUpsertInput): ScheduledTask { |
| 454 | const timestamp = IsoDateTime.make("2026-07-01T09:00:00.000Z"); |
| 455 | return { |
| 456 | id: input.id ?? ScheduledTaskId.make(`scheduled-task:${input.commandId ?? "stub"}`), |
| 457 | title: input.title, |
| 458 | prompt: input.prompt, |
| 459 | enabled: input.enabled, |
| 460 | schedule: |
| 461 | input.schedule.type === "webhook" |
| 462 | ? { |
| 463 | type: "webhook", |
| 464 | signature: |
| 465 | input.schedule.signature == null |
| 466 | ? null |
| 467 | : { |
| 468 | header: input.schedule.signature.header, |
| 469 | encoding: input.schedule.signature.encoding, |
| 470 | prefix: input.schedule.signature.prefix, |
| 471 | }, |
| 472 | } |
| 473 | : input.schedule, |
| 474 | projectId: input.projectId, |
| 475 | threadId: input.threadId ?? null, |
| 476 | workspaceStrategy: input.workspaceStrategy, |
| 477 | modelSelection: input.modelSelection, |
| 478 | runtimeMode: input.runtimeMode, |
| 479 | interactionMode: input.interactionMode, |
| 480 | createdBy: input.createdBy ?? "user", |
| 481 | creationSource: input.creationSource ?? "web", |
| 482 | createdAt: timestamp, |
| 483 | updatedAt: timestamp, |
| 484 | nextRunAt: null, |
| 485 | lastRunAt: null, |
| 486 | lastRunStatus: "never", |
| 487 | lastRunError: null, |
| 488 | runCount: 0, |
| 489 | }; |
| 490 | } |
| 491 | |
| 492 | /** In-memory server secret store for tests that exercise secret requests. */ |
| 493 | const layerMemorySecretStore = Layer.sync(ServerSecretStore.ServerSecretStore, () => { |
| 494 | const stored = new Map<string, Uint8Array>(); |
| 495 | return ServerSecretStore.ServerSecretStore.of({ |
| 496 | get: (name) => Effect.succeed(Option.fromNullishOr(stored.get(name))), |
| 497 | set: (name, value) => Effect.sync(() => void stored.set(name, value)), |
| 498 | create: (name, value) => Effect.sync(() => void stored.set(name, value)), |
| 499 | getOrCreateRandom: (name, bytes) => |
| 500 | Effect.sync(() => { |
| 501 | const existing = stored.get(name); |
| 502 | if (existing) return existing; |
| 503 | const value = new Uint8Array(bytes).fill(7); |
| 504 | stored.set(name, value); |
| 505 | return value; |
| 506 | }), |
| 507 | remove: (name) => Effect.sync(() => void stored.delete(name)), |
| 508 | }); |
| 509 | }); |
| 510 | |
| 511 | const layerUnusedScheduledTaskStub = Layer.succeed( |
| 512 | ScheduledTaskService.ScheduledTaskService, |
| 513 | ScheduledTaskService.ScheduledTaskService.of({ |
| 514 | list: () => Effect.succeed({ tasks: [] }), |
| 515 | subscribeList: () => Stream.succeed({ tasks: [] }), |
| 516 | upsert: () => Effect.die("ScheduledTaskService.upsert is unused in this test"), |
| 517 | setEnabled: () => Effect.die("ScheduledTaskService.setEnabled is unused in this test"), |
| 518 | delete: () => Effect.die("ScheduledTaskService.delete is unused in this test"), |
| 519 | runNow: () => Effect.die("ScheduledTaskService.runNow is unused in this test"), |
| 520 | rotateWebhookToken: () => Effect.die("unused in this test"), |
| 521 | listWebhookDeliveries: () => Effect.die("unused in this test"), |
| 522 | getWebhookDelivery: () => Effect.die("unused in this test"), |
| 523 | triggerWebhook: () => Effect.die("unused in this test"), |
| 524 | }), |
| 525 | ); |
| 526 | |
| 527 | describe("orchestrator MCP toolkit", () => { |
| 528 | it.live( |
| 529 | "delegates cross-provider tasks, polls and cancels children, and creates ordinary threads", |
| 530 | () => |
| 531 | Effect.scoped( |
| 532 | Effect.gen(function* () { |
| 533 | const cwd = yield* checkpointWorkspace("orchestrator-mcp-toolkit"); |
| 534 | const capturedTurns = yield* Ref.make<ReadonlyArray<CapturedTurn>>([]); |
| 535 | const parentTerminalGates = new Map<ThreadId, Deferred.Deferred<void>>(); |
| 536 | const deliveryTerminalGates = new Map<ThreadId, Deferred.Deferred<void>>(); |
| 537 | const layerRegistry = ProviderAdapterRegistry.layerFromAdapters([ |
| 538 | makeDeterministicAdapter({ |
| 539 | instanceId: codexInstanceId, |
| 540 | driver: ProviderDriverKind.make("codex"), |
| 541 | // Exercise queued completion ownership on a session without native steering. |
| 542 | // Native mailbox delivery and its completion races have dedicated integration tests. |
| 543 | capabilities: { |
| 544 | ...CodexProviderCapabilitiesV2, |
| 545 | turns: { ...CodexProviderCapabilitiesV2.turns, supportsActiveSteering: false }, |
| 546 | }, |
| 547 | capturedTurns, |
| 548 | shouldComplete: (turn) => |
| 549 | turn.threadId !== parentThreadId && turn.message.text !== cancellationPrompt, |
| 550 | terminalGate: (turn) => |
| 551 | turn.message.text.startsWith("Delegated task") || |
| 552 | turn.message.text.startsWith("Delegated tasks") |
| 553 | ? deliveryTerminalGates.get(turn.threadId) |
| 554 | : parentTerminalGates.get(turn.threadId), |
| 555 | response: (turn) => `Codex completed: ${turn.message.text}`, |
| 556 | }), |
| 557 | makeDeterministicAdapter({ |
| 558 | instanceId: claudeInstanceId, |
| 559 | driver: ProviderDriverKind.make("claudeAgent"), |
| 560 | capabilities: ClaudeProviderCapabilitiesV2, |
| 561 | capturedTurns, |
| 562 | shouldComplete: (turn) => turn.message.text !== cancellationPrompt, |
| 563 | terminalGate: (turn) => |
| 564 | turn.message.text.startsWith("Delegated task") || |
| 565 | turn.message.text.startsWith("Delegated tasks") |
| 566 | ? deliveryTerminalGates.get(turn.threadId) |
| 567 | : parentTerminalGates.get(turn.threadId), |
| 568 | response: (turn) => |
| 569 | turn.message.text === delegatedPrompt |
| 570 | ? delegatedResult |
| 571 | : `Claude completed: ${turn.message.text}`, |
| 572 | }), |
| 573 | ]); |
| 574 | // Captures parent-wake offers made when a delegated child |
| 575 | // terminalizes after the parent run settled. |
| 576 | const continuationOffers = yield* Ref.make< |
| 577 | ReadonlyArray<ProviderContinuationRequests.ProviderContinuationRequest> |
| 578 | >([]); |
| 579 | const layerContinuationProbe = Layer.succeed( |
| 580 | ProviderContinuationRequests.ProviderContinuationRequests, |
| 581 | { |
| 582 | offer: (request) => |
| 583 | Ref.update(continuationOffers, (existing) => [...existing, request]), |
| 584 | take: Effect.never, |
| 585 | }, |
| 586 | ); |
| 587 | // Offers land after the finalize projection writes, so poll briefly |
| 588 | // instead of asserting counts immediately. |
| 589 | const waitForContinuationOffers = (count: number) => |
| 590 | Effect.gen(function* () { |
| 591 | for (let attempt = 0; attempt < 1_000; attempt += 1) { |
| 592 | const current = yield* Ref.get(continuationOffers); |
| 593 | if (current.length >= count) { |
| 594 | return current; |
| 595 | } |
| 596 | yield* Effect.sleep("5 millis"); |
| 597 | } |
| 598 | return yield* Ref.get(continuationOffers); |
| 599 | }); |
| 600 | // Absence has no event to await, so sample repeatedly instead of |
| 601 | // trusting a single sleep to outlast a late offer. |
| 602 | const expectOffersToStay = (count: number) => |
| 603 | Effect.gen(function* () { |
| 604 | for (let sample = 0; sample < 4; sample += 1) { |
| 605 | yield* Effect.sleep("50 millis"); |
| 606 | expect(yield* Ref.get(continuationOffers)).toHaveLength(count); |
| 607 | } |
| 608 | }); |
| 609 | const layerOrchestrator = ProviderReplayHarness.layerWithRegistry( |
| 610 | { |
| 611 | name: "orchestrator-mcp-toolkit", |
| 612 | runtimePolicyOverride: { |
| 613 | cwd, |
| 614 | approvalPolicy: "never", |
| 615 | sandboxPolicy: { |
| 616 | type: "readOnly", |
| 617 | access: { type: "fullAccess" }, |
| 618 | networkAccess: false, |
| 619 | }, |
| 620 | }, |
| 621 | }, |
| 622 | layerRegistry, |
| 623 | ).pipe(Layer.provide(layerContinuationProbe)); |
| 624 | const layerOrchestration = Layer.merge( |
| 625 | layerOrchestrator, |
| 626 | ThreadManagementService.layer.pipe( |
| 627 | Layer.provide(layerOrchestrator), |
| 628 | Layer.provide(ServerSettings.layerTest()), |
| 629 | Layer.provide( |
| 630 | CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)), |
| 631 | ), |
| 632 | ), |
| 633 | ); |
| 634 | const layerProviderRegistry = ProviderRegistryMock.layer([ |
| 635 | makeProviderSnapshot({ |
| 636 | instanceId: codexInstanceId, |
| 637 | driver: ProviderDriverKind.make("codex"), |
| 638 | model: codexModel, |
| 639 | optionDescriptors: [ |
| 640 | { |
| 641 | id: "reasoning", |
| 642 | label: "Reasoning effort", |
| 643 | type: "select", |
| 644 | options: [ |
| 645 | { id: "low", label: "Low" }, |
| 646 | { id: "medium", label: "Medium" }, |
| 647 | { id: "high", label: "High" }, |
| 648 | ], |
| 649 | }, |
| 650 | ], |
| 651 | }), |
| 652 | makeProviderSnapshot({ |
| 653 | instanceId: claudeInstanceId, |
| 654 | driver: ProviderDriverKind.make("claudeAgent"), |
| 655 | model: claudeModel, |
| 656 | }), |
| 657 | makeProviderSnapshot({ |
| 658 | instanceId: ProviderInstanceId.make("opencode"), |
| 659 | driver: ProviderDriverKind.make("opencode"), |
| 660 | model: "opencode/test", |
| 661 | }), |
| 662 | ]); |
| 663 | // In-memory ScheduledTaskService stub so the schedule/list/update/ |
| 664 | // delete tools can be exercised without SQL/launch wiring. |
| 665 | const scheduledStore = yield* Ref.make<ReadonlyArray<ScheduledTask>>([]); |
| 666 | const layerScheduledTaskStub = Layer.succeed( |
| 667 | ScheduledTaskService.ScheduledTaskService, |
| 668 | ScheduledTaskService.ScheduledTaskService.of({ |
| 669 | list: () => Ref.get(scheduledStore).pipe(Effect.map((tasks) => ({ tasks }))), |
| 670 | subscribeList: () => Stream.empty, |
| 671 | upsert: (input) => |
| 672 | Effect.gen(function* () { |
| 673 | const task = scheduledTaskFromUpsert(input); |
| 674 | yield* Ref.update(scheduledStore, (all) => [ |
| 675 | ...all.filter((candidate) => candidate.id !== task.id), |
| 676 | task, |
| 677 | ]); |
| 678 | return { task }; |
| 679 | }), |
| 680 | setEnabled: () => |
| 681 | Effect.die("ScheduledTaskService.setEnabled is unused in this test"), |
| 682 | delete: (input) => |
| 683 | Ref.update(scheduledStore, (all) => |
| 684 | all.filter((candidate) => candidate.id !== input.id), |
| 685 | ).pipe(Effect.as({ id: input.id })), |
| 686 | runNow: () => Effect.die("ScheduledTaskService.runNow is unused in this test"), |
| 687 | rotateWebhookToken: () => Effect.die("unused in this test"), |
| 688 | listWebhookDeliveries: () => Effect.die("unused in this test"), |
| 689 | getWebhookDelivery: () => Effect.die("unused in this test"), |
| 690 | triggerWebhook: () => Effect.die("unused in this test"), |
| 691 | }), |
| 692 | ); |
| 693 | const layerTest = Layer.merge( |
| 694 | layerOrchestratorToolkit, |
| 695 | McpHttpServer.layerThreadToolkit, |
| 696 | ).pipe( |
| 697 | Layer.provideMerge(McpServer.McpServer.layer), |
| 698 | Layer.provideMerge(layerOrchestration), |
| 699 | Layer.provide(layerRegistry), |
| 700 | Layer.provide(layerProviderRegistry), |
| 701 | Layer.provide(ServerSettings.layerTest()), |
| 702 | Layer.provide(layerScheduledTaskStub), |
| 703 | Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})), |
| 704 | Layer.provide( |
| 705 | Layer.mock(ProjectService.ProjectService)({ |
| 706 | getById: (id) => |
| 707 | Effect.succeed( |
| 708 | id === projectId |
| 709 | ? Option.some({ id, defaultModelSelection: null } as never) |
| 710 | : Option.none(), |
| 711 | ), |
| 712 | }), |
| 713 | ), |
| 714 | Layer.provideMerge( |
| 715 | SecretRequests.layer.pipe( |
| 716 | Layer.provide(layerMemorySecretStore), |
| 717 | Layer.provide(layerOrchestration), |
| 718 | ), |
| 719 | ), |
| 720 | Layer.provide(NodeServices.layer), |
| 721 | ); |
| 722 | |
| 723 | yield* Effect.gen(function* () { |
| 724 | const orchestrator = yield* Orchestrator.OrchestratorV2; |
| 725 | const server = yield* McpServer.McpServer; |
| 726 | yield* orchestrator.dispatch({ |
| 727 | type: "thread.create", |
| 728 | createdBy: "user", |
| 729 | creationSource: "web", |
| 730 | commandId: CommandId.make("command:mcp-parent:create"), |
| 731 | threadId: parentThreadId, |
| 732 | projectId, |
| 733 | title: "MCP parent", |
| 734 | modelSelection: codexSelection, |
| 735 | runtimeMode: "full-access", |
| 736 | interactionMode: "default", |
| 737 | branch: null, |
| 738 | worktreePath: cwd, |
| 739 | }); |
| 740 | yield* orchestrator.dispatch({ |
| 741 | type: "message.dispatch", |
| 742 | createdBy: "user", |
| 743 | creationSource: "web", |
| 744 | commandId: CommandId.make("command:mcp-parent:start"), |
| 745 | threadId: parentThreadId, |
| 746 | messageId: MessageId.make("message:mcp-parent:start"), |
| 747 | text: parentPrompt, |
| 748 | attachments: [], |
| 749 | modelSelection: codexSelection, |
| 750 | dispatchMode: { type: "start_immediately" }, |
| 751 | }); |
| 752 | const parent = yield* waitForProjection( |
| 753 | orchestrator, |
| 754 | parentThreadId, |
| 755 | (projection) => |
| 756 | projection.runs.some((run) => |
| 757 | ["starting", "running", "waiting"].includes(run.status), |
| 758 | ) && projection.providerTurns.some((turn) => turn.status === "running"), |
| 759 | ); |
| 760 | const parentRun = parent.runs[0]; |
| 761 | expect(parentRun?.status).toBe("running"); |
| 762 | |
| 763 | const invocation: McpInvocationContext.McpInvocationScope = { |
| 764 | environmentId: EnvironmentId.make("environment:mcp-orchestrator"), |
| 765 | requestNamespace: "mcp-provider-session-parent", |
| 766 | thread: { |
| 767 | threadId: parentThreadId, |
| 768 | providerSessionId: "mcp-provider-session-parent", |
| 769 | providerInstanceId: codexInstanceId, |
| 770 | }, |
| 771 | client: undefined, |
| 772 | capabilities: new Set(["orchestration"]), |
| 773 | issuedAt: 1, |
| 774 | }; |
| 775 | const invokeAs = ( |
| 776 | callScope: McpInvocationContext.McpInvocationScope, |
| 777 | name: string, |
| 778 | args: Record<string, unknown>, |
| 779 | ) => |
| 780 | server |
| 781 | .callTool({ name, arguments: args }) |
| 782 | .pipe( |
| 783 | Effect.provideService(McpInvocationContext.McpInvocationContext, callScope), |
| 784 | Effect.provideService(McpSchema.McpServerClient, client), |
| 785 | ); |
| 786 | const invoke = (name: string, args: Record<string, unknown>) => |
| 787 | invokeAs(invocation, name, args); |
| 788 | |
| 789 | // Settling would stop the session, so the agent's own turn keeps running. |
| 790 | const deferredSettle = yield* invoke("t3_thread_organize", { action: "settle" }); |
| 791 | expect(deferredSettle.isError).toBe(false); |
| 792 | expect(deferredSettle.structuredContent).toEqual({ settlesWhenTurnEnds: true }); |
| 793 | const afterDeferredSettle = yield* orchestrator.getThreadProjection(parentThreadId); |
| 794 | expect(afterDeferredSettle.thread.settledOverride).not.toBe("settled"); |
| 795 | expect(afterDeferredSettle.runs.find((run) => run.id === parentRun?.id)?.status).toBe( |
| 796 | "running", |
| 797 | ); |
| 798 | |
| 799 | const pinned = yield* invoke("t3_thread_organize", { action: "pin" }); |
| 800 | expect(pinned.isError).toBe(false); |
| 801 | expect(pinned.structuredContent).toHaveProperty("sequence"); |
| 802 | expect((yield* orchestrator.getThreadShell(parentThreadId))?.pinnedAt).not.toBeNull(); |
| 803 | yield* invoke("t3_thread_organize", { action: "unpin" }); |
| 804 | expect((yield* orchestrator.getThreadShell(parentThreadId))?.pinnedAt).toBeNull(); |
| 805 | |
| 806 | if (parentRun === undefined || parentRun.rootNodeId === null) { |
| 807 | return yield* Effect.die(new Error("Parent run missing.")); |
| 808 | } |
| 809 | for (const name of ["t3_queue_edit", "t3_queue_cancel"]) { |
| 810 | const refusedQueueMutation = yield* invoke(name, { |
| 811 | queuedRunId: parentRun.id, |
| 812 | ...(name === "t3_queue_edit" ? { text: "Keep the active turn." } : {}), |
| 813 | }); |
| 814 | expect(refusedQueueMutation.isError).toBe(true); |
| 815 | expect(refusedQueueMutation.structuredContent).toBeUndefined(); |
| 816 | expect(declaredFailure(refusedQueueMutation)).toEqual({ |
| 817 | _tag: "OrchestratorMcpFailure", |
| 818 | code: "orchestration_error", |
| 819 | message: `Run ${parentRun.id} is not queued.`, |
| 820 | }); |
| 821 | } |
| 822 | |
| 823 | let parentRootNodeId = parentRun.rootNodeId; |
| 824 | const queueAutomaticCompletion = (suffix: string, taskText: string) => |
| 825 | Effect.gen(function* () { |
| 826 | const delegated = yield* orchestrator.dispatch({ |
| 827 | type: "delegated_task.request", |
| 828 | createdBy: "agent", |
| 829 | creationSource: "mcp", |
| 830 | commandId: CommandId.make(`command:mcp-parent:${suffix}:delegate`), |
| 831 | parentThreadId, |
| 832 | parentRunId: parentRun.id, |
| 833 | parentNodeId: parentRootNodeId, |
| 834 | task: taskText, |
| 835 | modelSelection: claudeSelection, |
| 836 | runtimeMode: "full-access", |
| 837 | interactionMode: "default", |
| 838 | completionWake: "always", |
| 839 | }); |
| 840 | const taskEvent = delegated.storedEvents.find( |
| 841 | (stored) => |
| 842 | stored.event.type === "subagent.updated" && |
| 843 | stored.event.payload.origin === "app_owned", |
| 844 | ); |
| 845 | if (taskEvent?.event.type !== "subagent.updated") { |
| 846 | return yield* Effect.die( |
| 847 | new Error(`Automatic completion task ${suffix} was not created.`), |
| 848 | ); |
| 849 | } |
| 850 | const task = taskEvent.event.payload; |
| 851 | const reserved = yield* waitForProjection( |
| 852 | orchestrator, |
| 853 | parentThreadId, |
| 854 | (projection) => { |
| 855 | const delivery = projection.runs.find((run) => run.id === parentRun.id) |
| 856 | ?.delegatedCompletion?.delivery; |
| 857 | return ( |
| 858 | projection.subagents.find((candidate) => candidate.id === task.id)?.status === |
| 859 | "completed" && |
| 860 | delivery !== undefined && |
| 861 | delivery !== null && |
| 862 | delivery.taskIds.includes(task.id) |
| 863 | ); |
| 864 | }, |
| 865 | ); |
| 866 | const delivery = reserved.runs.find((run) => run.id === parentRun.id) |
| 867 | ?.delegatedCompletion?.delivery; |
| 868 | if (delivery === undefined || delivery === null) { |
| 869 | return yield* Effect.die( |
| 870 | new Error(`Automatic completion delivery ${suffix} was not reserved.`), |
| 871 | ); |
| 872 | } |
| 873 | yield* orchestrator.dispatch({ |
| 874 | type: "message.dispatch", |
| 875 | createdBy: "agent", |
| 876 | creationSource: "server", |
| 877 | commandId: CommandId.make(`command:mcp-parent:${suffix}:dispatch`), |
| 878 | threadId: parentThreadId, |
| 879 | messageId: delivery.messageId, |
| 880 | text: "Delegated task reached a terminal state.", |
| 881 | attachments: [], |
| 882 | modelSelection: codexSelection, |
| 883 | dispatchMode: { type: "queue_after_active" }, |
| 884 | delegatedCompletion: { |
| 885 | parentRunId: parentRun.id, |
| 886 | generation: delivery.generation, |
| 887 | taskIds: delivery.taskIds, |
| 888 | }, |
| 889 | }); |
| 890 | const queued = yield* waitForProjection( |
| 891 | orchestrator, |
| 892 | parentThreadId, |
| 893 | (projection) => |
| 894 | projection.runs.some( |
| 895 | (run) => run.userMessageId === delivery.messageId && run.status === "queued", |
| 896 | ), |
| 897 | ); |
| 898 | const queuedRun = queued.runs.find( |
| 899 | (run) => run.userMessageId === delivery.messageId, |
| 900 | ); |
| 901 | if (queuedRun === undefined) { |
| 902 | return yield* Effect.die( |
| 903 | new Error(`Automatic completion delivery ${suffix} was not queued.`), |
| 904 | ); |
| 905 | } |
| 906 | return { task, delivery, queuedRun }; |
| 907 | }); |
| 908 | |
| 909 | // Several async terminals belonging to one parent run reserve one |
| 910 | // durable delivery. The continuation worker owns the later |
| 911 | // message.dispatch, so use the same metadata here while the test |
| 912 | // probe records the offers instead of starting a worker. |
| 913 | const coalescedTask = (suffix: string) => |
| 914 | orchestrator.dispatch({ |
| 915 | type: "delegated_task.request", |
| 916 | createdBy: "agent", |
| 917 | creationSource: "mcp", |
| 918 | commandId: CommandId.make(`command:mcp-parent:coalesced-${suffix}`), |
| 919 | parentThreadId, |
| 920 | parentRunId: parentRun.id, |
| 921 | parentNodeId: parentRootNodeId, |
| 922 | task: `Complete coalesced task ${suffix}.`, |
| 923 | modelSelection: claudeSelection, |
| 924 | runtimeMode: "full-access", |
| 925 | interactionMode: "default", |
| 926 | completionWake: "always", |
| 927 | }); |
| 928 | const firstCoalesced = yield* coalescedTask("one"); |
| 929 | const secondCoalesced = yield* coalescedTask("two"); |
| 930 | const coalescedTaskIds = [firstCoalesced, secondCoalesced].map((result) => { |
| 931 | const taskEvent = result.storedEvents.find( |
| 932 | (stored) => |
| 933 | stored.event.type === "subagent.updated" && |
| 934 | stored.event.payload.origin === "app_owned", |
| 935 | ); |
| 936 | if (taskEvent?.event.type !== "subagent.updated") { |
| 937 | throw new Error("Coalesced delegated task projection missing."); |
| 938 | } |
| 939 | return taskEvent.event.payload.id; |
| 940 | }); |
| 941 | const coalescedProjection = yield* waitForProjection( |
| 942 | orchestrator, |
| 943 | parentThreadId, |
| 944 | (projection) => |
| 945 | coalescedTaskIds.every( |
| 946 | (taskId) => |
| 947 | projection.subagents.find((task) => task.id === taskId)?.status === "completed", |
| 948 | ) && |
| 949 | projection.runs.find((run) => run.id === parentRun.id)?.delegatedCompletion !== |
| 950 | undefined, |
| 951 | ); |
| 952 | const coalescedCohort = coalescedProjection.runs.find( |
| 953 | (run) => run.id === parentRun.id, |
| 954 | )?.delegatedCompletion; |
| 955 | expect(coalescedCohort?.delivery?.taskIds).toEqual( |
| 956 | expect.arrayContaining(coalescedTaskIds), |
| 957 | ); |
| 958 | expect(coalescedCohort?.delivery?.taskIds).toHaveLength(2); |
| 959 | const coalescedOffers = yield* waitForContinuationOffers(1); |
| 960 | expect(coalescedOffers).toEqual( |
| 961 | expect.arrayContaining([ |
| 962 | expect.objectContaining({ |
| 963 | delegatedCompletion: expect.objectContaining({ |
| 964 | parentRunId: parentRun.id, |
| 965 | messageId: coalescedCohort?.delivery?.messageId, |
| 966 | }), |
| 967 | }), |
| 968 | ]), |
| 969 | ); |
| 970 | if (coalescedCohort?.delivery === undefined || coalescedCohort.delivery === null) { |
| 971 | return yield* Effect.die(new Error("Coalesced delivery reservation missing.")); |
| 972 | } |
| 973 | const coalescedDelivery = coalescedCohort.delivery; |
| 974 | yield* orchestrator.dispatch({ |
| 975 | type: "message.dispatch", |
| 976 | createdBy: "agent", |
| 977 | creationSource: "server", |
| 978 | commandId: CommandId.make("command:mcp-parent:dispatch-coalesced-delivery"), |
| 979 | threadId: parentThreadId, |
| 980 | messageId: coalescedDelivery.messageId, |
| 981 | text: "Delegated tasks reached terminal states.", |
| 982 | attachments: [], |
| 983 | modelSelection: codexSelection, |
| 984 | dispatchMode: { type: "queue_after_active" }, |
| 985 | delegatedCompletion: { |
| 986 | parentRunId: parentRun.id, |
| 987 | generation: coalescedDelivery.generation, |
| 988 | taskIds: coalescedDelivery.taskIds, |
| 989 | }, |
| 990 | }); |
| 991 | const queuedCoalesced = yield* waitForProjection( |
| 992 | orchestrator, |
| 993 | parentThreadId, |
| 994 | (projection) => |
| 995 | projection.runs.some( |
| 996 | (run) => |
| 997 | run.userMessageId === coalescedDelivery.messageId && run.status === "queued", |
| 998 | ), |
| 999 | ); |
| 1000 | const queuedCoalescedRun = queuedCoalesced.runs.find( |
| 1001 | (run) => run.userMessageId === coalescedDelivery.messageId, |
| 1002 | ); |
| 1003 | expect(queuedCoalescedRun?.status).toBe("queued"); |
| 1004 | if (queuedCoalescedRun === undefined) { |
| 1005 | return yield* Effect.die(new Error("Queued coalesced delivery missing.")); |
| 1006 | } |
| 1007 | |
| 1008 | // Server-owned delivery ownership is authoritative even if a |
| 1009 | // stale client tries to mutate the queue directly. |
| 1010 | const completionEditError = yield* orchestrator |
| 1011 | .dispatch({ |
| 1012 | type: "queued-run.edit", |
| 1013 | commandId: CommandId.make("command:mcp-parent:edit-coalesced-delivery"), |
| 1014 | threadId: parentThreadId, |
| 1015 | runId: queuedCoalescedRun.id, |
| 1016 | text: "Rewrite the automatic delivery.", |
| 1017 | }) |
| 1018 | .pipe(Effect.flip); |
| 1019 | expect(completionEditError._tag).toBe("OrchestratorCommandRejectedError"); |
| 1020 | const completionReorderError = yield* orchestrator |
| 1021 | .dispatch({ |
| 1022 | type: "queued-run.reorder", |
| 1023 | commandId: CommandId.make("command:mcp-parent:reorder-coalesced-delivery"), |
| 1024 | threadId: parentThreadId, |
| 1025 | runId: queuedCoalescedRun.id, |
| 1026 | beforeRunId: null, |
| 1027 | }) |
| 1028 | .pipe(Effect.flip); |
| 1029 | expect(completionReorderError._tag).toBe("OrchestratorDispatchError"); |
| 1030 | const completionSteerError = yield* orchestrator |
| 1031 | .dispatch({ |
| 1032 | type: "queued-message.promote-to-steer", |
| 1033 | commandId: CommandId.make("command:mcp-parent:steer-coalesced-delivery"), |
| 1034 | threadId: parentThreadId, |
| 1035 | queuedRunId: queuedCoalescedRun.id, |
| 1036 | targetRunId: parentRun.id, |
| 1037 | }) |
| 1038 | .pipe(Effect.flip); |
| 1039 | expect(completionSteerError._tag).toBe("OrchestratorDispatchError"); |
| 1040 | |
| 1041 | for (const taskId of coalescedTaskIds) { |
| 1042 | const statusCall = yield* invoke("task_status", { taskId }); |
| 1043 | expect(statusCall.isError).toBe(false); |
| 1044 | } |
| 1045 | const acknowledgedCoalesced = yield* waitForProjection( |
| 1046 | orchestrator, |
| 1047 | parentThreadId, |
| 1048 | (projection) => |
| 1049 | projection.runs.find((run) => run.id === queuedCoalescedRun.id)?.status === |
| 1050 | "cancelled" && |
| 1051 | coalescedTaskIds.every( |
| 1052 | (taskId) => |
| 1053 | projection.subagents.find((task) => task.id === taskId)?.completionDelivery |
| 1054 | ?.state === "acknowledged", |
| 1055 | ), |
| 1056 | ); |
| 1057 | expect( |
| 1058 | acknowledgedCoalesced.runs.find((run) => run.id === parentRun.id)?.delegatedCompletion |
| 1059 | ?.delivery, |
| 1060 | ).toBeNull(); |
| 1061 | yield* Ref.set(continuationOffers, []); |
| 1062 | |
| 1063 | // Waiting only observes the child run's status. A direct parent |
| 1064 | // read acknowledges delivery only after the terminal result is |
| 1065 | // returned untruncated, never from the child prompt or a partial |
| 1066 | // result page. |
| 1067 | const directRead = yield* queueAutomaticCompletion( |
| 1068 | "direct-child-read", |
| 1069 | "Complete before a parent reads this child result directly.", |
| 1070 | ); |
| 1071 | if (directRead.task.childThreadId === null) { |
| 1072 | return yield* Effect.die(new Error("Direct-read child thread missing.")); |
| 1073 | } |
| 1074 | const directChildThreadId = directRead.task.childThreadId; |
| 1075 | const directChildWaitCall = yield* invoke("t3_thread_wait", { |
| 1076 | threadId: directChildThreadId, |
| 1077 | timeoutMs: 10_000, |
| 1078 | }); |
| 1079 | const directChildWait = yield* decodeThreadWaitResult( |
| 1080 | directChildWaitCall.structuredContent, |
| 1081 | ).pipe(Effect.orDie); |
| 1082 | expect(directChildWait).toMatchObject({ |
| 1083 | threadId: directChildThreadId, |
| 1084 | status: "completed", |
| 1085 | timedOut: false, |
| 1086 | }); |
| 1087 | const pendingAfterWait = yield* orchestrator.getThreadProjection(parentThreadId); |
| 1088 | expect( |
| 1089 | pendingAfterWait.subagents.find((task) => task.id === directRead.task.id) |
| 1090 | ?.completionDelivery?.state, |
| 1091 | ).toBe("claimed"); |
| 1092 | expect( |
| 1093 | pendingAfterWait.runs.find((run) => run.id === directRead.queuedRun.id)?.status, |
| 1094 | ).toBe("queued"); |
| 1095 | |
| 1096 | const childPromptReadCall = yield* invoke("t3_thread_read", { |
| 1097 | threadId: directChildThreadId, |
| 1098 | limit: 1, |
| 1099 | }); |
| 1100 | const childPromptRead = yield* decodeThreadReadResult( |
| 1101 | childPromptReadCall.structuredContent, |
| 1102 | ).pipe(Effect.orDie); |
| 1103 | expect(childPromptRead.items.map((item) => item.type)).toEqual(["user_message"]); |
| 1104 | if (childPromptRead.nextPosition === null) { |
| 1105 | return yield* Effect.die(new Error("Direct-read child prompt position missing.")); |
| 1106 | } |
| 1107 | expect( |
| 1108 | (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( |
| 1109 | (task) => task.id === directRead.task.id, |
| 1110 | )?.completionDelivery?.state, |
| 1111 | ).toBe("claimed"); |
| 1112 | |
| 1113 | const truncatedResultReadCall = yield* invoke("t3_thread_read", { |
| 1114 | threadId: directChildThreadId, |
| 1115 | afterPosition: childPromptRead.nextPosition, |
| 1116 | limit: 1, |
| 1117 | maxCharsPerItem: 1, |
| 1118 | }); |
| 1119 | const truncatedResultRead = yield* decodeThreadReadResult( |
| 1120 | truncatedResultReadCall.structuredContent, |
| 1121 | ).pipe(Effect.orDie); |
| 1122 | expect(truncatedResultRead.items).toMatchObject([ |
| 1123 | { type: "assistant_message", textTruncated: true }, |
| 1124 | ]); |
| 1125 | expect( |
| 1126 | (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( |
| 1127 | (task) => task.id === directRead.task.id, |
| 1128 | )?.completionDelivery?.state, |
| 1129 | ).toBe("claimed"); |
| 1130 | |
| 1131 | const firstChunk = truncatedResultRead.items[0]!; |
| 1132 | expect(firstChunk.nextTextOffset).toBe(1); |
| 1133 | const remainderCall = yield* invoke("t3_thread_read", { |
| 1134 | threadId: directChildThreadId, |
| 1135 | itemId: firstChunk.itemId, |
| 1136 | textOffset: firstChunk.nextTextOffset, |
| 1137 | maxCharsPerItem: 50_000, |
| 1138 | }); |
| 1139 | const remainder = yield* decodeThreadReadResult(remainderCall.structuredContent).pipe( |
| 1140 | Effect.orDie, |
| 1141 | ); |
| 1142 | expect(remainder.items[0]?.nextTextOffset).toBeNull(); |
| 1143 | expect(firstChunk.text + (remainder.items[0]?.text ?? "")).toBe( |
| 1144 | "Claude completed: Complete before a parent reads this child result directly.", |
| 1145 | ); |
| 1146 | // Reading a suffix alone cannot acknowledge a whole child result. |
| 1147 | expect( |
| 1148 | (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( |
| 1149 | (task) => task.id === directRead.task.id, |
| 1150 | )?.completionDelivery?.state, |
| 1151 | ).toBe("claimed"); |
| 1152 | |
| 1153 | const terminalResultReadCall = yield* invoke("t3_thread_read", { |
| 1154 | threadId: directChildThreadId, |
| 1155 | afterPosition: childPromptRead.nextPosition, |
| 1156 | limit: 1, |
| 1157 | }); |
| 1158 | const terminalResultRead = yield* decodeThreadReadResult( |
| 1159 | terminalResultReadCall.structuredContent, |
| 1160 | ).pipe(Effect.orDie); |
| 1161 | expect(terminalResultRead.items).toMatchObject([ |
| 1162 | { |
| 1163 | type: "assistant_message", |
| 1164 | text: "Claude completed: Complete before a parent reads this child result directly.", |
| 1165 | textTruncated: false, |
| 1166 | }, |
| 1167 | ]); |
| 1168 | const longText = "界🧪\n".repeat(20_000) + "FINAL_CONSTRAINT"; |
| 1169 | const childProjection = yield* orchestrator.getThreadProjection(directChildThreadId); |
| 1170 | const sourceItem = childProjection.turnItems.find( |
| 1171 | (item) => item.type === "user_message", |
| 1172 | )!; |
| 1173 | const oversizedItem = { |
| 1174 | ...sourceItem, |
| 1175 | id: TurnItemId.make("item:oversized-retrieval"), |
| 1176 | type: "assistant_message" as const, |
| 1177 | text: longText, |
| 1178 | streaming: false, |
| 1179 | ordinal: 999, |
| 1180 | }; |
| 1181 | yield* (yield* EventSink.EventSinkV2).write({ |
| 1182 | events: [ |
| 1183 | { |
| 1184 | id: EventId.make("event:oversized-retrieval"), |
| 1185 | type: "turn-item.updated", |
| 1186 | threadId: directChildThreadId, |
| 1187 | occurredAt: yield* DateTime.now, |
| 1188 | payload: oversizedItem, |
| 1189 | }, |
| 1190 | ], |
| 1191 | }); |
| 1192 | let recovered = ""; |
| 1193 | let textOffset: number | null = 0; |
| 1194 | while (textOffset !== null) { |
| 1195 | const pageCall = yield* invoke("t3_thread_read", { |
| 1196 | threadId: directChildThreadId, |
| 1197 | itemId: oversizedItem.id, |
| 1198 | textOffset, |
| 1199 | maxCharsPerItem: 50_000, |
| 1200 | }); |
| 1201 | const page: OrchestratorMcpThreadReadResult = yield* decodeThreadReadResult( |
| 1202 | pageCall.structuredContent, |
| 1203 | ).pipe(Effect.orDie); |
| 1204 | expect(page.items).toHaveLength(1); |
| 1205 | recovered += page.items[0]!.text; |
| 1206 | textOffset = page.items[0]!.nextTextOffset ?? null; |
| 1207 | } |
| 1208 | expect(recovered).toBe(longText); |
| 1209 | |
| 1210 | const acknowledgedByDirectRead = yield* waitForProjection( |
| 1211 | orchestrator, |
| 1212 | parentThreadId, |
| 1213 | (projection) => |
| 1214 | projection.runs.find((run) => run.id === directRead.queuedRun.id)?.status === |
| 1215 | "cancelled" && |
| 1216 | projection.subagents.find((task) => task.id === directRead.task.id) |
| 1217 | ?.completionDelivery?.state === "acknowledged", |
| 1218 | ); |
| 1219 | expect( |
| 1220 | acknowledgedByDirectRead.subagents.find((task) => task.id === directRead.task.id) |
| 1221 | ?.completionDelivery, |
| 1222 | ).toMatchObject({ state: "acknowledged", observedByRunId: parentRun.id }); |
| 1223 | yield* Ref.set(continuationOffers, []); |
| 1224 | |
| 1225 | // New user work retains its own queue entry while an explicit |
| 1226 | // observation disposes the stale automatic one. |
| 1227 | const queueRace = yield* queueAutomaticCompletion( |
| 1228 | "queue-race", |
| 1229 | "Complete before a user queues follow-up work.", |
| 1230 | ); |
| 1231 | const queuedUserMessageId = MessageId.make("message:mcp-parent:queue-race:user"); |
| 1232 | yield* orchestrator.dispatch({ |
| 1233 | type: "message.dispatch", |
| 1234 | createdBy: "user", |
| 1235 | creationSource: "web", |
| 1236 | commandId: CommandId.make("command:mcp-parent:queue-race:user"), |
| 1237 | threadId: parentThreadId, |
| 1238 | messageId: queuedUserMessageId, |
| 1239 | text: "🙂".repeat(16001), |
| 1240 | attachments: [], |
| 1241 | modelSelection: codexSelection, |
| 1242 | dispatchMode: { type: "queue_after_active" }, |
| 1243 | }); |
| 1244 | const queueRaceQueued = yield* waitForProjection( |
| 1245 | orchestrator, |
| 1246 | parentThreadId, |
| 1247 | (projection) => |
| 1248 | projection.runs.some((run) => run.id === queueRace.queuedRun.id) && |
| 1249 | projection.runs.some( |
| 1250 | (run) => run.userMessageId === queuedUserMessageId && run.status === "queued", |
| 1251 | ), |
| 1252 | ); |
| 1253 | const queuedUserRun = queueRaceQueued.runs.find( |
| 1254 | (run) => run.userMessageId === queuedUserMessageId, |
| 1255 | ); |
| 1256 | if (queuedUserRun === undefined) { |
| 1257 | return yield* Effect.die(new Error("Queued user follow-up missing.")); |
| 1258 | } |
| 1259 | const queueFirstPage = yield* invoke("t3_queue_list", { limit: 1 }); |
| 1260 | expect(queueFirstPage.isError).toBe(false); |
| 1261 | expect(queueFirstPage.structuredContent).toMatchObject({ |
| 1262 | items: [{ queuedRunId: queueRace.queuedRun.id }], |
| 1263 | nextCursor: 1, |
| 1264 | }); |
| 1265 | const queueDefinition = server.tools.find(({ tool }) => tool.name === "t3_queue_list"); |
| 1266 | const validateQueue = new AjvJsonSchemaValidator().getValidator( |
| 1267 | queueDefinition!.tool.outputSchema! as JsonSchemaType, |
| 1268 | ); |
| 1269 | expect(validateQueue(queueFirstPage.structuredContent).valid).toBe(true); |
| 1270 | const missingThreadId = ThreadId.make("00000000-0000-4000-8000-000000000000"); |
| 1271 | const missingThreadQueue = yield* invoke("t3_queue_list", { |
| 1272 | threadId: missingThreadId, |
| 1273 | limit: 1, |
| 1274 | }); |
| 1275 | expect(missingThreadQueue.isError).toBe(true); |
| 1276 | expect(declaredFailure(missingThreadQueue)).toMatchObject({ |
| 1277 | _tag: "OrchestratorMcpFailure", |
| 1278 | code: "thread_not_found", |
| 1279 | message: "The thread was not found.", |
| 1280 | }); |
| 1281 | expect(missingThreadQueue.structuredContent).toBeUndefined(); |
| 1282 | expect(validateQueue({ items: "invalid", nextCursor: null }).valid).toBe(false); |
| 1283 | const missingThreadRead = yield* invoke("t3_thread_read", { |
| 1284 | threadId: missingThreadId, |
| 1285 | }); |
| 1286 | expect(missingThreadRead.isError).toBe(true); |
| 1287 | expect(declaredFailure(missingThreadRead)).toMatchObject({ |
| 1288 | _tag: "OrchestratorMcpFailure", |
| 1289 | }); |
| 1290 | const queueSecondPage = yield* invoke("t3_queue_list", { cursor: 1, limit: 1 }); |
| 1291 | expect(queueSecondPage.structuredContent).toEqual({ |
| 1292 | items: [{ queuedRunId: queuedUserRun.id, text: "🙂".repeat(1000), truncated: true }], |
| 1293 | nextCursor: null, |
| 1294 | }); |
| 1295 | const queueRead = yield* invoke("t3_queue_read", { queuedRunId: queuedUserRun.id }); |
| 1296 | expect(queueRead.structuredContent).toEqual({ |
| 1297 | queuedRunId: queuedUserRun.id, |
| 1298 | text: "🙂".repeat(16000), |
| 1299 | truncated: true, |
| 1300 | }); |
| 1301 | const missingQueueRead = yield* invoke("t3_queue_read", { queuedRunId: parentRun.id }); |
| 1302 | expect(declaredFailure(missingQueueRead)).toMatchObject({ code: "invalid_request" }); |
| 1303 | const queueRaceStatus = yield* invoke("task_status", { taskId: queueRace.task.id }); |
| 1304 | expect(queueRaceStatus.isError).toBe(false); |
| 1305 | yield* waitForProjection( |
| 1306 | orchestrator, |
| 1307 | parentThreadId, |
| 1308 | (projection) => |
| 1309 | projection.runs.find((run) => run.id === queueRace.queuedRun.id)?.status === |
| 1310 | "cancelled" && |
| 1311 | projection.runs.find((run) => run.id === queuedUserRun.id)?.status === "queued" && |
| 1312 | projection.subagents.find((task) => task.id === queueRace.task.id) |
| 1313 | ?.completionDelivery?.state === "acknowledged", |
| 1314 | ); |
| 1315 | yield* orchestrator.dispatch({ |
| 1316 | type: "queued-run.cancel", |
| 1317 | commandId: CommandId.make("command:mcp-parent:queue-race:cleanup"), |
| 1318 | threadId: parentThreadId, |
| 1319 | runId: queuedUserRun.id, |
| 1320 | }); |
| 1321 | |
| 1322 | // A user Steer keeps the parent run active. Once the |
| 1323 | // parent observes the child result, its queued delivery is stale. |
| 1324 | const steerRace = yield* queueAutomaticCompletion( |
| 1325 | "steer-race", |
| 1326 | "Complete before a user steers the parent.", |
| 1327 | ); |
| 1328 | const steerMessageId = MessageId.make("message:mcp-parent:steer-race:user"); |
| 1329 | yield* orchestrator.dispatch({ |
| 1330 | type: "message.dispatch", |
| 1331 | createdBy: "user", |
| 1332 | creationSource: "web", |
| 1333 | commandId: CommandId.make("command:mcp-parent:steer-race:user"), |
| 1334 | threadId: parentThreadId, |
| 1335 | messageId: steerMessageId, |
| 1336 | text: "Steer the active parent after receiving the result.", |
| 1337 | attachments: [], |
| 1338 | modelSelection: codexSelection, |
| 1339 | dispatchMode: { type: "steer_active", targetRunId: parentRun.id }, |
| 1340 | }); |
| 1341 | const steerRaceStatus = yield* invoke("task_status", { taskId: steerRace.task.id }); |
| 1342 | expect(steerRaceStatus.isError).toBe(false); |
| 1343 | const acknowledgedSteerRace = yield* waitForProjection( |
| 1344 | orchestrator, |
| 1345 | parentThreadId, |
| 1346 | (projection) => |
| 1347 | projection.messages.some((message) => message.id === steerMessageId) && |
| 1348 | projection.runs.find((run) => run.id === parentRun.id)?.status === "running" && |
| 1349 | projection.runs.find((run) => run.id === steerRace.queuedRun.id)?.status === |
| 1350 | "cancelled" && |
| 1351 | projection.subagents.find((task) => task.id === steerRace.task.id) |
| 1352 | ?.completionDelivery?.state === "acknowledged", |
| 1353 | ); |
| 1354 | expect( |
| 1355 | acknowledgedSteerRace.runs.find((run) => run.id === parentRun.id)?.delegatedCompletion |
| 1356 | ?.delivery, |
| 1357 | ).toBeNull(); |
| 1358 | |
| 1359 | // Restart creates a new attempt for the same parent cohort. The |
| 1360 | // old terminal must not turn that continuation into a Stop |
| 1361 | // barrier, and an acknowledged result cancels the stale delivery. |
| 1362 | const restartRace = yield* queueAutomaticCompletion( |
| 1363 | "restart-race", |
| 1364 | "Complete before a user restarts the parent.", |
| 1365 | ); |
| 1366 | const attemptBeforeRestart = (yield* orchestrator.getThreadProjection( |
| 1367 | parentThreadId, |
| 1368 | )).runs.find((run) => run.id === parentRun.id)?.activeAttemptId; |
| 1369 | const restartMessageId = MessageId.make("message:mcp-parent:restart-race:user"); |
| 1370 | yield* orchestrator.dispatch({ |
| 1371 | type: "message.dispatch", |
| 1372 | createdBy: "user", |
| 1373 | creationSource: "web", |
| 1374 | commandId: CommandId.make("command:mcp-parent:restart-race:user"), |
| 1375 | threadId: parentThreadId, |
| 1376 | messageId: restartMessageId, |
| 1377 | text: "Restart the active parent after receiving the result.", |
| 1378 | attachments: [], |
| 1379 | modelSelection: codexSelection, |
| 1380 | dispatchMode: { type: "restart_active", targetRunId: parentRun.id }, |
| 1381 | }); |
| 1382 | const restartedParent = yield* waitForProjection( |
| 1383 | orchestrator, |
| 1384 | parentThreadId, |
| 1385 | (projection) => { |
| 1386 | const run = projection.runs.find((candidate) => candidate.id === parentRun.id); |
| 1387 | return ( |
| 1388 | run?.status === "running" && |
| 1389 | run.activeAttemptId !== attemptBeforeRestart && |
| 1390 | run.rootNodeId !== null |
| 1391 | ); |
| 1392 | }, |
| 1393 | ); |
| 1394 | const currentParentRun = restartedParent.runs.find((run) => run.id === parentRun.id); |
| 1395 | if (currentParentRun?.rootNodeId === null || currentParentRun === undefined) { |
| 1396 | return yield* Effect.die(new Error("Restarted parent run missing.")); |
| 1397 | } |
| 1398 | parentRootNodeId = currentParentRun.rootNodeId; |
| 1399 | expect(currentParentRun.delegatedCompletion?.disposition).toBe("open"); |
| 1400 | const restartRaceStatus = yield* invoke("task_status", { |
| 1401 | taskId: restartRace.task.id, |
| 1402 | }); |
| 1403 | expect(restartRaceStatus.isError).toBe(false); |
| 1404 | yield* waitForProjection( |
| 1405 | orchestrator, |
| 1406 | parentThreadId, |
| 1407 | (projection) => |
| 1408 | projection.messages.some((message) => message.id === restartMessageId) && |
| 1409 | projection.runs.find((run) => run.id === restartRace.queuedRun.id)?.status === |
| 1410 | "cancelled" && |
| 1411 | projection.subagents.find((task) => task.id === restartRace.task.id) |
| 1412 | ?.completionDelivery?.state === "acknowledged" && |
| 1413 | projection.runs.find((run) => run.id === parentRun.id)?.delegatedCompletion |
| 1414 | ?.disposition === "open", |
| 1415 | ); |
| 1416 | yield* Ref.set(continuationOffers, []); |
| 1417 | |
| 1418 | const capabilitiesTool = server.tools.find( |
| 1419 | ({ tool }) => tool.name === "orchestrator_capabilities", |
| 1420 | ); |
| 1421 | expect(capabilitiesTool?.tool.annotations?.readOnlyHint).toBe(true); |
| 1422 | expect(capabilitiesTool?.tool.annotations?.idempotentHint).toBe(true); |
| 1423 | const delegateTool = server.tools.find(({ tool }) => tool.name === "delegate_task"); |
| 1424 | expect(delegateTool?.tool.annotations?.destructiveHint).toBe(true); |
| 1425 | expect(delegateTool?.tool.annotations?.openWorldHint).toBe(true); |
| 1426 | const taskStatusTool = server.tools.find(({ tool }) => tool.name === "task_status"); |
| 1427 | expect(taskStatusTool?.tool.annotations?.readOnlyHint).toBe(false); |
| 1428 | expect(taskStatusTool?.tool.annotations?.idempotentHint).toBe(true); |
| 1429 | const createThreadsTool = server.tools.find( |
| 1430 | ({ tool }) => tool.name === "create_threads", |
| 1431 | ); |
| 1432 | expect(createThreadsTool?.tool.annotations?.destructiveHint).toBe(true); |
| 1433 | const threadListTool = server.tools.find(({ tool }) => tool.name === "t3_thread_list"); |
| 1434 | expect(threadListTool?.tool.annotations?.readOnlyHint).toBe(true); |
| 1435 | expect(threadListTool?.tool.annotations?.idempotentHint).toBe(true); |
| 1436 | const threadReadTool = server.tools.find(({ tool }) => tool.name === "t3_thread_read"); |
| 1437 | expect(threadReadTool?.tool.annotations?.readOnlyHint).toBe(false); |
| 1438 | expect(threadReadTool?.tool.inputSchema).toMatchObject({ additionalProperties: false }); |
| 1439 | const unknownPagingParamError = yield* invoke("t3_thread_read", { |
| 1440 | threadId: parentThreadId, |
| 1441 | cursor: 90, |
| 1442 | }).pipe(Effect.flip); |
| 1443 | expect(unknownPagingParamError.message).toContain("Expected no excess property"); |
| 1444 | const threadUpdateTool = server.tools.find( |
| 1445 | ({ tool }) => tool.name === "t3_thread_update", |
| 1446 | ); |
| 1447 | expect(threadUpdateTool?.tool.annotations?.destructiveHint).toBe(true); |
| 1448 | expect(threadUpdateTool?.tool.annotations?.idempotentHint).toBe(false); |
| 1449 | expect(threadUpdateTool?.tool.inputSchema).toMatchObject({ |
| 1450 | type: "object", |
| 1451 | properties: { |
| 1452 | action: expect.any(Object), |
| 1453 | title: expect.any(Object), |
| 1454 | pullRequest: expect.any(Object), |
| 1455 | }, |
| 1456 | }); |
| 1457 | const deniedThreadUpdate = yield* invokeAs( |
| 1458 | { ...invocation, capabilities: new Set() }, |
| 1459 | "t3_thread_update", |
| 1460 | { action: "rename", title: "Denied title" }, |
| 1461 | ); |
| 1462 | expect(declaredFailure(deniedThreadUpdate)).toMatchObject({ |
| 1463 | _tag: "OrchestratorMcpFailure", |
| 1464 | code: "capability_denied", |
| 1465 | }); |
| 1466 | const threadSendTool = server.tools.find(({ tool }) => tool.name === "t3_thread_send"); |
| 1467 | expect(threadSendTool?.tool.annotations?.destructiveHint).toBe(true); |
| 1468 | const threadWaitTool = server.tools.find(({ tool }) => tool.name === "t3_thread_wait"); |
| 1469 | expect(threadWaitTool?.tool.annotations?.readOnlyHint).toBe(true); |
| 1470 | const threadInterruptTool = server.tools.find( |
| 1471 | ({ tool }) => tool.name === "t3_thread_interrupt", |
| 1472 | ); |
| 1473 | expect(threadInterruptTool?.tool.annotations?.destructiveHint).toBe(true); |
| 1474 | |
| 1475 | const capabilities = yield* invoke("orchestrator_capabilities", {}); |
| 1476 | expect(capabilities.isError).toBe(false); |
| 1477 | expect(capabilities.structuredContent).toMatchObject({ |
| 1478 | inheritedProviderInstanceId: codexInstanceId, |
| 1479 | inheritedModel: codexModel, |
| 1480 | features: { |
| 1481 | appOwnedSubagents: true, |
| 1482 | asyncPolling: true, |
| 1483 | cancellation: true, |
| 1484 | batchThreadCreation: true, |
| 1485 | threadManagement: true, |
| 1486 | incrementalThreadRead: true, |
| 1487 | }, |
| 1488 | providers: expect.arrayContaining([ |
| 1489 | expect.objectContaining({ |
| 1490 | providerInstanceId: claudeInstanceId, |
| 1491 | canRunCrossProviderChildTask: true, |
| 1492 | }), |
| 1493 | // No opencode adapter is registered in this harness, so the |
| 1494 | // capability view must not claim delegation can target it. |
| 1495 | expect.objectContaining({ |
| 1496 | providerInstanceId: "opencode", |
| 1497 | canRunChildTask: false, |
| 1498 | }), |
| 1499 | // Models advertise their option descriptors so agents can |
| 1500 | // discover valid target.options ids and values. |
| 1501 | expect.objectContaining({ |
| 1502 | providerInstanceId: codexInstanceId, |
| 1503 | models: [ |
| 1504 | expect.objectContaining({ |
| 1505 | id: codexModel, |
| 1506 | options: [expect.objectContaining({ id: "reasoning", type: "select" })], |
| 1507 | }), |
| 1508 | ], |
| 1509 | }), |
| 1510 | ]), |
| 1511 | }); |
| 1512 | |
| 1513 | const scheduleTool = server.tools.find(({ tool }) => tool.name === "schedule_task"); |
| 1514 | expect(scheduleTool?.tool.annotations?.destructiveHint).toBe(true); |
| 1515 | const scheduleCall = yield* invoke("schedule_task", { |
| 1516 | prompt: "wake up in this thread and say hello", |
| 1517 | schedule: { type: "interval", everyMs: 60_000 }, |
| 1518 | clientRequestId: "schedule-hello-1", |
| 1519 | }); |
| 1520 | expect(scheduleCall.isError).toBe(false); |
| 1521 | expect(scheduleCall.structuredContent).toMatchObject({ |
| 1522 | // Defaults to binding the calling thread. |
| 1523 | boundThreadId: parentThreadId, |
| 1524 | projectId, |
| 1525 | enabled: true, |
| 1526 | schedule: { type: "interval", everyMs: 60_000 }, |
| 1527 | // Title is derived from the prompt when omitted. |
| 1528 | title: "wake up in this thread and say hello", |
| 1529 | }); |
| 1530 | const scheduledTaskId = (scheduleCall.structuredContent as { scheduledTaskId: string }) |
| 1531 | .scheduledTaskId; |
| 1532 | const storedAfterCreate = yield* Ref.get(scheduledStore); |
| 1533 | expect(storedAfterCreate).toHaveLength(1); |
| 1534 | expect(storedAfterCreate[0]).toMatchObject({ |
| 1535 | threadId: parentThreadId, |
| 1536 | projectId, |
| 1537 | createdBy: "agent", |
| 1538 | creationSource: "mcp", |
| 1539 | // Inherits the parent thread's model selection. |
| 1540 | modelSelection: codexSelection, |
| 1541 | }); |
| 1542 | |
| 1543 | // list_scheduled_tasks returns the task scoped to this project. |
| 1544 | const scheduledListCall = yield* invoke("list_scheduled_tasks", {}); |
| 1545 | expect(scheduledListCall.isError).toBe(false); |
| 1546 | expect(scheduledListCall.structuredContent).toMatchObject({ |
| 1547 | tasks: [{ scheduledTaskId, boundThreadId: parentThreadId }], |
| 1548 | }); |
| 1549 | |
| 1550 | // update_scheduled_task pauses without deleting. |
| 1551 | const scheduledUpdateCall = yield* invoke("update_scheduled_task", { |
| 1552 | scheduledTaskId, |
| 1553 | enabled: false, |
| 1554 | }); |
| 1555 | expect(scheduledUpdateCall.isError).toBe(false); |
| 1556 | expect(scheduledUpdateCall.structuredContent).toMatchObject({ |
| 1557 | scheduledTaskId, |
| 1558 | enabled: false, |
| 1559 | }); |
| 1560 | |
| 1561 | // delete_scheduled_task removes it entirely. |
| 1562 | const scheduledDeleteCall = yield* invoke("delete_scheduled_task", { scheduledTaskId }); |
| 1563 | expect(scheduledDeleteCall.isError).toBe(false); |
| 1564 | expect(scheduledDeleteCall.structuredContent).toMatchObject({ |
| 1565 | scheduledTaskId, |
| 1566 | deleted: true, |
| 1567 | }); |
| 1568 | expect(yield* Ref.get(scheduledStore)).toHaveLength(0); |
| 1569 | |
| 1570 | // OpenCode 1.15 has emitted this exact nested-object-as-JSON-string |
| 1571 | // shape. Decode it at the MCP boundary rather than failing a task |
| 1572 | // the model otherwise specified correctly. |
| 1573 | const serializedScheduleCall = yield* invoke("schedule_task", { |
| 1574 | prompt: "check for new pull requests", |
| 1575 | schedule: '{"type":"interval","everyMs":3600000}', |
| 1576 | clientRequestId: "schedule-opencode-compat-1", |
| 1577 | }); |
| 1578 | expect(serializedScheduleCall.isError).toBe(false); |
| 1579 | expect(serializedScheduleCall.structuredContent).toMatchObject({ |
| 1580 | schedule: { type: "interval", everyMs: 3_600_000 }, |
| 1581 | boundThreadId: parentThreadId, |
| 1582 | }); |
| 1583 | const serializedScheduledTaskId = ( |
| 1584 | serializedScheduleCall.structuredContent as { scheduledTaskId: string } |
| 1585 | ).scheduledTaskId; |
| 1586 | yield* invoke("delete_scheduled_task", { |
| 1587 | scheduledTaskId: serializedScheduledTaskId, |
| 1588 | }); |
| 1589 | expect(yield* Ref.get(scheduledStore)).toHaveLength(0); |
| 1590 | |
| 1591 | // The agent asks for a secret; the tool waits for the user and |
| 1592 | // returns a one-use ref, never the value, which a signed webhook |
| 1593 | // task then consumes. |
| 1594 | const secretRequests = yield* SecretRequests.SecretRequests; |
| 1595 | const secretFiber = yield* invoke("request_secret", { |
| 1596 | label: "GitHub webhook secret", |
| 1597 | reason: "Signs release webhooks. Enter the same value in GitHub's webhook settings.", |
| 1598 | placeholder: "Paste the webhook secret", |
| 1599 | clientRequestId: "release-webhook-secret", |
| 1600 | }).pipe(Effect.forkChild); |
| 1601 | // Polled without the helper's short budget: under load the tool's |
| 1602 | // own reads come first. |
| 1603 | const asked = yield* Effect.gen(function* () { |
| 1604 | while (true) { |
| 1605 | const projection = yield* orchestrator.getThreadProjection(parentThreadId); |
| 1606 | if ( |
| 1607 | projection.turnItems.some( |
| 1608 | (item) => item.type === "secret_request" && item.secretStatus === "pending", |
| 1609 | ) |
| 1610 | ) { |
| 1611 | return projection; |
| 1612 | } |
| 1613 | yield* Effect.sleep("5 millis"); |
| 1614 | } |
| 1615 | }); |
| 1616 | const card = asked.turnItems.find((item) => item.type === "secret_request"); |
| 1617 | if (card?.type !== "secret_request") { |
| 1618 | return yield* Effect.die(new Error("Secret request card missing.")); |
| 1619 | } |
| 1620 | expect(card).toMatchObject({ |
| 1621 | label: "GitHub webhook secret", |
| 1622 | placeholder: "Paste the webhook secret", |
| 1623 | }); |
| 1624 | // Asking again with the same id leaves the open card exactly as it was. |
| 1625 | yield* orchestrator.dispatch({ |
| 1626 | type: "secret_request.record", |
| 1627 | commandId: CommandId.make("command:test:secret-request-replay"), |
| 1628 | threadId: parentThreadId, |
| 1629 | runId: card.runId!, |
| 1630 | nodeId: card.nodeId!, |
| 1631 | turnItemId: card.id, |
| 1632 | label: "Something else", |
| 1633 | reason: "A different reason.", |
| 1634 | secretStatus: "pending", |
| 1635 | }); |
| 1636 | expect( |
| 1637 | (yield* orchestrator.getThreadProjection(parentThreadId)).turnItems.find( |
| 1638 | (item) => item.id === card.id, |
| 1639 | ), |
| 1640 | ).toMatchObject({ label: "GitHub webhook secret", runId: card.runId }); |
| 1641 | // The agent is blocked on the user, so the thread asks for input |
| 1642 | // like a question does, in both shell paths. |
| 1643 | expect( |
| 1644 | (yield* orchestrator.getThreadShell(parentThreadId))?.pendingRuntimeRequest, |
| 1645 | ).toMatchObject({ kind: "user_input" }); |
| 1646 | expect(threadShellFromProjection(asked).pendingRuntimeRequest).toMatchObject({ |
| 1647 | kind: "user_input", |
| 1648 | }); |
| 1649 | // What the card's Save sends (secrets.answerRequest). |
| 1650 | yield* secretRequests.answer({ |
| 1651 | threadId: parentThreadId, |
| 1652 | turnItemId: card.id, |
| 1653 | answer: { type: "save", secret: "github-webhook-secret" }, |
| 1654 | }); |
| 1655 | const secretCall = yield* Fiber.join(secretFiber); |
| 1656 | expect(secretCall.isError).toBe(false); |
| 1657 | const secretResult = secretCall.structuredContent as { |
| 1658 | status: string; |
| 1659 | secretRef?: string; |
| 1660 | }; |
| 1661 | expect(secretResult.status).toBe("saved"); |
| 1662 | expect( |
| 1663 | (yield* orchestrator.getThreadShell(parentThreadId))?.pendingRuntimeRequest ?? null, |
| 1664 | ).toBeNull(); |
| 1665 | expect(secretResult.secretRef).toMatch(/^secret-ref:[0-9a-f]{32}$/); |
| 1666 | // The value appears nowhere in what the agent received. |
| 1667 | const received = [ |
| 1668 | ...secretCall.content.map((part) => ("text" in part ? part.text : "")), |
| 1669 | ...Object.values(secretResult), |
| 1670 | ]; |
| 1671 | expect(received.some((value) => value.includes("github-webhook-secret"))).toBe(false); |
| 1672 | |
| 1673 | // A retry that lost the first result gets the same answer, with no |
| 1674 | // second card for the user. |
| 1675 | const retried = yield* invoke("request_secret", { |
| 1676 | label: "GitHub webhook secret", |
| 1677 | reason: "Signs release webhooks. Enter the same value in GitHub's webhook settings.", |
| 1678 | clientRequestId: "release-webhook-secret", |
| 1679 | }); |
| 1680 | expect(retried.structuredContent).toEqual(secretResult); |
| 1681 | expect( |
| 1682 | (yield* orchestrator.getThreadProjection(parentThreadId)).turnItems.filter( |
| 1683 | (item) => item.type === "secret_request", |
| 1684 | ), |
| 1685 | ).toHaveLength(1); |
| 1686 | |
| 1687 | // The ref is the secret for exactly one consumer. |
| 1688 | expect( |
| 1689 | yield* secretRequests.consume({ |
| 1690 | ref: secretResult.secretRef as never, |
| 1691 | projectId, |
| 1692 | }), |
| 1693 | ).toBe("github-webhook-secret"); |
| 1694 | const reused = yield* secretRequests |
| 1695 | .consume({ ref: secretResult.secretRef as never, projectId }) |
| 1696 | .pipe(Effect.flip); |
| 1697 | expect(reused.message).toContain("already used"); |
| 1698 | |
| 1699 | const delegatedCall = yield* invoke("delegate_task", { |
| 1700 | task: delegatedPrompt, |
| 1701 | target: { |
| 1702 | providerInstanceId: claudeInstanceId, |
| 1703 | model: claudeModel, |
| 1704 | }, |
| 1705 | mode: "wait", |
| 1706 | timeoutMs: 10_000, |
| 1707 | clientRequestId: "delegate-claude-1", |
| 1708 | }); |
| 1709 | expect(delegatedCall.isError).toBe(false); |
| 1710 | const delegated = yield* decodeDelegateTaskResult(delegatedCall.structuredContent).pipe( |
| 1711 | Effect.orDie, |
| 1712 | ); |
| 1713 | const delegatedSource = yield* orchestrator.getThreadProjection( |
| 1714 | delegated.childThreadId, |
| 1715 | ); |
| 1716 | expect(delegatedSource.messages[0]).toMatchObject({ |
| 1717 | senderThreadId: parentThreadId, |
| 1718 | }); |
| 1719 | expect( |
| 1720 | delegatedSource.turnItems.find((item) => item.type === "user_message"), |
| 1721 | ).toMatchObject({ |
| 1722 | senderThreadId: parentThreadId, |
| 1723 | }); |
| 1724 | expect(delegated.status).toBe("completed"); |
| 1725 | expect(delegated.summary).toBe(delegatedResult); |
| 1726 | expect(delegated.providerInstanceId).toBe(claudeInstanceId); |
| 1727 | |
| 1728 | const completedParent = yield* waitForProjection( |
| 1729 | orchestrator, |
| 1730 | parentThreadId, |
| 1731 | (projection) => |
| 1732 | projection.subagents.some( |
| 1733 | (task) => |
| 1734 | task.id === delegated.taskId && |
| 1735 | task.status === "completed" && |
| 1736 | task.result === delegatedResult, |
| 1737 | ) && |
| 1738 | projection.contextTransfers.some( |
| 1739 | (transfer) => |
| 1740 | transfer.type === "subagent_result" && |
| 1741 | transfer.sourceThreadId === delegated.childThreadId, |
| 1742 | ), |
| 1743 | ); |
| 1744 | const completedTask = completedParent.subagents.find( |
| 1745 | (task) => task.id === delegated.taskId, |
| 1746 | ); |
| 1747 | expect(completedTask).toMatchObject({ |
| 1748 | origin: "app_owned", |
| 1749 | createdBy: "agent", |
| 1750 | childThreadId: delegated.childThreadId, |
| 1751 | status: "completed", |
| 1752 | result: delegatedResult, |
| 1753 | }); |
| 1754 | const child = yield* orchestrator.getThreadProjection(delegated.childThreadId); |
| 1755 | expect(child.thread.lineage).toEqual({ |
| 1756 | parentThreadId, |
| 1757 | relationshipToParent: "subagent", |
| 1758 | rootThreadId: parentThreadId, |
| 1759 | }); |
| 1760 | expect(child.thread).toMatchObject({ |
| 1761 | createdBy: "agent", |
| 1762 | creationSource: "mcp", |
| 1763 | }); |
| 1764 | expect(child.thread.modelSelection).toEqual(claudeSelection); |
| 1765 | expect( |
| 1766 | child.messages |
| 1767 | .filter((message) => message.role === "user") |
| 1768 | .map((message) => message.text), |
| 1769 | ).toEqual([delegatedPrompt]); |
| 1770 | expect( |
| 1771 | child.contextTransfers.some( |
| 1772 | (transfer) => |
| 1773 | transfer.type === "subagent_spawn" && transfer.sourceThreadId === parentThreadId, |
| 1774 | ), |
| 1775 | ).toBe(true); |
| 1776 | const capturedAfterDelegate = yield* Ref.get(capturedTurns); |
| 1777 | expect( |
| 1778 | capturedAfterDelegate.filter((turn) => turn.threadId === delegated.childThreadId), |
| 1779 | ).toEqual([ |
| 1780 | { |
| 1781 | instanceId: claudeInstanceId, |
| 1782 | threadId: delegated.childThreadId, |
| 1783 | text: delegatedPrompt, |
| 1784 | }, |
| 1785 | ]); |
| 1786 | expect( |
| 1787 | capturedAfterDelegate.some( |
| 1788 | (turn) => |
| 1789 | turn.threadId === delegated.childThreadId && turn.text.includes(parentPrompt), |
| 1790 | ), |
| 1791 | ).toBe(false); |
| 1792 | |
| 1793 | const delegatedStatusCall = yield* invoke("task_status", { |
| 1794 | taskId: delegated.taskId, |
| 1795 | }); |
| 1796 | const delegatedStatus = yield* decodeDelegateTaskResult( |
| 1797 | delegatedStatusCall.structuredContent, |
| 1798 | ).pipe(Effect.orDie); |
| 1799 | expect(delegatedStatus).toMatchObject({ |
| 1800 | childRunId: delegated.childRunId, |
| 1801 | status: "completed", |
| 1802 | hasPendingChildRuns: false, |
| 1803 | latestTerminalRunId: delegated.childRunId, |
| 1804 | latestTerminalStatus: "completed", |
| 1805 | latestTerminalSummary: delegatedResult, |
| 1806 | }); |
| 1807 | expect(delegatedStatus.resultContextTransferId).not.toBeNull(); |
| 1808 | expect(delegatedStatus.latestTerminalResultContextTransferId).not.toBeNull(); |
| 1809 | |
| 1810 | // Wait-mode completion already returned its result while the parent was live. |
| 1811 | yield* expectOffersToStay(0); |
| 1812 | |
| 1813 | const childFollowupCall = yield* invoke("t3_thread_send", { |
| 1814 | threadId: delegated.childThreadId, |
| 1815 | message: "Confirm the delegated API boundary remains inspected.", |
| 1816 | clientRequestId: "delegated-child-followup-1", |
| 1817 | }); |
| 1818 | const childFollowup = yield* decodeThreadSendResult( |
| 1819 | childFollowupCall.structuredContent, |
| 1820 | ).pipe(Effect.orDie); |
| 1821 | const childFollowupWaitCall = yield* invoke("t3_thread_wait", { |
| 1822 | threadId: delegated.childThreadId, |
| 1823 | runId: childFollowup.runId, |
| 1824 | timeoutMs: 10_000, |
| 1825 | }); |
| 1826 | const childFollowupWait = yield* decodeThreadWaitResult( |
| 1827 | childFollowupWaitCall.structuredContent, |
| 1828 | ).pipe(Effect.orDie); |
| 1829 | expect(childFollowupWait).toMatchObject({ |
| 1830 | runId: childFollowup.runId, |
| 1831 | status: "completed", |
| 1832 | timedOut: false, |
| 1833 | }); |
| 1834 | const delegatedStatusAfterFollowupCall = yield* invoke("task_status", { |
| 1835 | taskId: delegated.taskId, |
| 1836 | }); |
| 1837 | const delegatedStatusAfterFollowup = yield* decodeDelegateTaskResult( |
| 1838 | delegatedStatusAfterFollowupCall.structuredContent, |
| 1839 | ).pipe(Effect.orDie); |
| 1840 | expect(delegatedStatusAfterFollowup).toMatchObject({ |
| 1841 | childRunId: delegated.childRunId, |
| 1842 | status: "completed", |
| 1843 | summary: delegatedResult, |
| 1844 | hasPendingChildRuns: false, |
| 1845 | latestTerminalRunId: childFollowup.runId, |
| 1846 | latestTerminalStatus: "completed", |
| 1847 | }); |
| 1848 | expect(delegatedStatusAfterFollowup.latestTerminalSummary).not.toBeNull(); |
| 1849 | |
| 1850 | const activeChildFollowupCall = yield* invoke("t3_thread_send", { |
| 1851 | threadId: delegated.childThreadId, |
| 1852 | message: cancellationPrompt, |
| 1853 | clientRequestId: "delegated-child-active-followup-1", |
| 1854 | }); |
| 1855 | const activeChildFollowup = yield* decodeThreadSendResult( |
| 1856 | activeChildFollowupCall.structuredContent, |
| 1857 | ).pipe(Effect.orDie); |
| 1858 | yield* waitForProjection(orchestrator, delegated.childThreadId, (projection) => |
| 1859 | projection.runs.some( |
| 1860 | (run) => run.id === activeChildFollowup.runId && run.status === "running", |
| 1861 | ), |
| 1862 | ); |
| 1863 | const delegatedStatusDuringFollowupCall = yield* invoke("task_status", { |
| 1864 | taskId: delegated.taskId, |
| 1865 | }); |
| 1866 | const delegatedStatusDuringFollowup = yield* decodeDelegateTaskResult( |
| 1867 | delegatedStatusDuringFollowupCall.structuredContent, |
| 1868 | ).pipe(Effect.orDie); |
| 1869 | expect(delegatedStatusDuringFollowup).toMatchObject({ |
| 1870 | childRunId: delegated.childRunId, |
| 1871 | status: "completed", |
| 1872 | summary: delegatedResult, |
| 1873 | hasPendingChildRuns: true, |
| 1874 | latestTerminalRunId: childFollowup.runId, |
| 1875 | latestTerminalStatus: "completed", |
| 1876 | }); |
| 1877 | const activeChildProjection = yield* orchestrator.getThreadProjection( |
| 1878 | delegated.childThreadId, |
| 1879 | ); |
| 1880 | const legacyChildProjection = { |
| 1881 | ...activeChildProjection, |
| 1882 | contextTransfers: activeChildProjection.contextTransfers.filter( |
| 1883 | (transfer) => transfer.type !== "subagent_spawn", |
| 1884 | ), |
| 1885 | }; |
| 1886 | const legacyDelegatedRun = delegatedTaskRun(legacyChildProjection, completedTask!); |
| 1887 | expect(legacyDelegatedRun?.id).toBe(delegated.childRunId); |
| 1888 | expect(hasPendingChildRuns(legacyChildProjection, legacyDelegatedRun)).toBe(true); |
| 1889 | expect( |
| 1890 | hasPendingChildRuns( |
| 1891 | { |
| 1892 | ...legacyChildProjection, |
| 1893 | runs: legacyChildProjection.runs.map((run) => |
| 1894 | run.id === activeChildFollowup.runId ? { ...run, status: "queued" } : run, |
| 1895 | ), |
| 1896 | }, |
| 1897 | legacyDelegatedRun, |
| 1898 | ), |
| 1899 | ).toBe(true); |
| 1900 | // A queue Stop held waits for the user; it is not pending child work. |
| 1901 | expect( |
| 1902 | hasPendingChildRuns( |
| 1903 | { |
| 1904 | ...legacyChildProjection, |
| 1905 | runs: legacyChildProjection.runs.map((run) => |
| 1906 | run.id === activeChildFollowup.runId |
| 1907 | ? { ...run, status: "queued", queueHeld: true } |
| 1908 | : run, |
| 1909 | ), |
| 1910 | }, |
| 1911 | legacyDelegatedRun, |
| 1912 | ), |
| 1913 | ).toBe(false); |
| 1914 | const completedTaskCancelCall = yield* invoke("task_cancel", { |
| 1915 | taskId: delegated.taskId, |
| 1916 | reason: "Stop the child's later work too.", |
| 1917 | clientRequestId: "cancel-completed-delegated-task-1", |
| 1918 | }); |
| 1919 | const completedTaskCancel = yield* decodeTaskCancelResult( |
| 1920 | completedTaskCancelCall.structuredContent, |
| 1921 | ).pipe(Effect.orDie); |
| 1922 | expect(completedTaskCancel).toEqual({ |
| 1923 | taskId: delegated.taskId, |
| 1924 | status: "completed", |
| 1925 | }); |
| 1926 | expect( |
| 1927 | (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( |
| 1928 | (task) => task.id === delegated.taskId, |
| 1929 | ), |
| 1930 | ).toMatchObject({ |
| 1931 | result: delegatedResult, |
| 1932 | completionDelivery: { state: "disposed" }, |
| 1933 | }); |
| 1934 | // Cancelling a finished task still stops the child thread's later work. |
| 1935 | yield* waitForProjection(orchestrator, delegated.childThreadId, (projection) => |
| 1936 | projection.runs.some( |
| 1937 | (run) => run.id === activeChildFollowup.runId && run.status === "interrupted", |
| 1938 | ), |
| 1939 | ); |
| 1940 | const delegatedStatusAfterCancelCall = yield* invoke("task_status", { |
| 1941 | taskId: delegated.taskId, |
| 1942 | }); |
| 1943 | const delegatedStatusAfterCancel = yield* decodeDelegateTaskResult( |
| 1944 | delegatedStatusAfterCancelCall.structuredContent, |
| 1945 | ).pipe(Effect.orDie); |
| 1946 | expect(delegatedStatusAfterCancel).toMatchObject({ |
| 1947 | childRunId: delegated.childRunId, |
| 1948 | status: "completed", |
| 1949 | summary: delegatedResult, |
| 1950 | hasPendingChildRuns: false, |
| 1951 | latestTerminalRunId: activeChildFollowup.runId, |
| 1952 | latestTerminalStatus: "interrupted", |
| 1953 | }); |
| 1954 | |
| 1955 | // Later requested work has its own delivery, including an interrupted follow-up. |
| 1956 | expect(childFollowup.taskId).toBeDefined(); |
| 1957 | expect(activeChildFollowup.taskId).toBeDefined(); |
| 1958 | yield* waitForContinuationOffers(2); |
| 1959 | yield* expectOffersToStay(2); |
| 1960 | yield* invoke("task_status", { taskId: childFollowup.taskId }); |
| 1961 | yield* invoke("task_status", { taskId: activeChildFollowup.taskId }); |
| 1962 | yield* Ref.set(continuationOffers, []); |
| 1963 | |
| 1964 | const repeatedDelegatedCall = yield* invoke("delegate_task", { |
| 1965 | task: delegatedPrompt, |
| 1966 | target: { |
| 1967 | providerInstanceId: claudeInstanceId, |
| 1968 | model: claudeModel, |
| 1969 | }, |
| 1970 | mode: "async", |
| 1971 | clientRequestId: "delegate-claude-1", |
| 1972 | }); |
| 1973 | const repeatedDelegated = yield* decodeDelegateTaskResult( |
| 1974 | repeatedDelegatedCall.structuredContent, |
| 1975 | ).pipe(Effect.orDie); |
| 1976 | expect(repeatedDelegated.taskId).toBe(delegated.taskId); |
| 1977 | expect( |
| 1978 | (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.filter( |
| 1979 | (task) => task.id === delegated.taskId, |
| 1980 | ), |
| 1981 | ).toHaveLength(1); |
| 1982 | |
| 1983 | // Options outside the model's advertised descriptors are rejected |
| 1984 | // before any child thread is created. |
| 1985 | const rejectedOptionsCall = yield* invoke("delegate_task", { |
| 1986 | task: cancellationPrompt, |
| 1987 | target: { |
| 1988 | providerInstanceId: codexInstanceId, |
| 1989 | model: codexModel, |
| 1990 | options: { reasoning: "extreme" }, |
| 1991 | }, |
| 1992 | mode: "async", |
| 1993 | clientRequestId: "delegate-rejected-options-1", |
| 1994 | }); |
| 1995 | expect(declaredFailure(rejectedOptionsCall)).toMatchObject({ |
| 1996 | _tag: "OrchestratorMcpFailure", |
| 1997 | code: "invalid_request", |
| 1998 | message: expect.stringContaining("rejected options"), |
| 1999 | }); |
| 2000 | |
| 2001 | // Duplicate option ids fail fast: downstream consumers disagree on |
| 2002 | // whether the first or last value of a duplicated id wins. |
| 2003 | const duplicateOptionsCall = yield* invoke("delegate_task", { |
| 2004 | task: cancellationPrompt, |
| 2005 | target: { |
| 2006 | providerInstanceId: codexInstanceId, |
| 2007 | model: codexModel, |
| 2008 | options: [ |
| 2009 | { id: "reasoning", value: "low" }, |
| 2010 | { id: "reasoning", value: "high" }, |
| 2011 | ], |
| 2012 | }, |
| 2013 | mode: "async", |
| 2014 | clientRequestId: "delegate-duplicate-options-1", |
| 2015 | }); |
| 2016 | expect(declaredFailure(duplicateOptionsCall)).toMatchObject({ |
| 2017 | _tag: "OrchestratorMcpFailure", |
| 2018 | code: "invalid_request", |
| 2019 | message: expect.stringContaining("more than once"), |
| 2020 | }); |
| 2021 | |
| 2022 | const cancellableCall = yield* invoke("delegate_task", { |
| 2023 | task: cancellationPrompt, |
| 2024 | target: { |
| 2025 | providerInstanceId: codexInstanceId, |
| 2026 | model: codexModel, |
| 2027 | // Shorthand record form; decodes to the canonical array. |
| 2028 | options: { reasoning: "low" }, |
| 2029 | }, |
| 2030 | mode: "async", |
| 2031 | clientRequestId: "delegate-cancel-1", |
| 2032 | }); |
| 2033 | const cancellable = yield* decodeDelegateTaskResult( |
| 2034 | cancellableCall.structuredContent, |
| 2035 | ).pipe(Effect.orDie); |
| 2036 | expect(cancellable.status).toBe("running"); |
| 2037 | // Original delegated run alone is not "later" work. |
| 2038 | expect(cancellable.hasPendingChildRuns).toBe(false); |
| 2039 | yield* waitForProjection(orchestrator, cancellable.childThreadId, (projection) => |
| 2040 | projection.providerTurns.some((turn) => turn.status === "running"), |
| 2041 | ); |
| 2042 | const statusWhileOriginalRunningCall = yield* invoke("task_status", { |
| 2043 | taskId: cancellable.taskId, |
| 2044 | }); |
| 2045 | const statusWhileOriginalRunning = yield* decodeDelegateTaskResult( |
| 2046 | statusWhileOriginalRunningCall.structuredContent, |
| 2047 | ).pipe(Effect.orDie); |
| 2048 | expect(statusWhileOriginalRunning).toMatchObject({ |
| 2049 | childRunId: cancellable.childRunId, |
| 2050 | status: "running", |
| 2051 | hasPendingChildRuns: false, |
| 2052 | latestTerminalRunId: null, |
| 2053 | latestTerminalStatus: null, |
| 2054 | }); |
| 2055 | // The requested model options reach the child thread's selection. |
| 2056 | const optionedChild = yield* orchestrator.getThreadProjection( |
| 2057 | cancellable.childThreadId, |
| 2058 | ); |
| 2059 | expect(optionedChild.thread.modelSelection).toEqual({ |
| 2060 | instanceId: codexInstanceId, |
| 2061 | model: codexModel, |
| 2062 | options: [{ id: "reasoning", value: "low" }], |
| 2063 | }); |
| 2064 | const cancelCall = yield* invoke("task_cancel", { |
| 2065 | taskId: cancellable.taskId, |
| 2066 | reason: "Parent no longer needs this work.", |
| 2067 | clientRequestId: "cancel-1", |
| 2068 | }); |
| 2069 | const cancelResult = yield* decodeTaskCancelResult(cancelCall.structuredContent).pipe( |
| 2070 | Effect.orDie, |
| 2071 | ); |
| 2072 | expect(cancelResult.status).toBe("cancel_requested"); |
| 2073 | yield* waitForProjection(orchestrator, cancellable.childThreadId, (projection) => |
| 2074 | projection.runs.some((run) => run.status === "interrupted"), |
| 2075 | ); |
| 2076 | const cancelledStatusCall = yield* invoke("task_status", { |
| 2077 | taskId: cancellable.taskId, |
| 2078 | }); |
| 2079 | const cancelledStatus = yield* decodeDelegateTaskResult( |
| 2080 | cancelledStatusCall.structuredContent, |
| 2081 | ).pipe(Effect.orDie); |
| 2082 | expect(cancelledStatus.status).toBe("interrupted"); |
| 2083 | |
| 2084 | // Explicit task_cancel disposes automatic delivery after the child |
| 2085 | // interrupt succeeds. The interrupted result remains readable, |
| 2086 | // but it cannot create a parent continuation. |
| 2087 | expect( |
| 2088 | (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( |
| 2089 | (task) => task.id === cancellable.taskId, |
| 2090 | )?.completionDelivery, |
| 2091 | ).toMatchObject({ state: "disposed" }); |
| 2092 | yield* expectOffersToStay(0); |
| 2093 | |
| 2094 | const createInput = { |
| 2095 | clientRequestId: "create-thread-batch-1", |
| 2096 | threads: [ |
| 2097 | { |
| 2098 | title: "Inherited empty thread", |
| 2099 | }, |
| 2100 | { |
| 2101 | title: "Claude ordinary thread", |
| 2102 | prompt: createdThreadPrompt, |
| 2103 | target: { |
| 2104 | driverKind: "claudeAgent", |
| 2105 | }, |
| 2106 | }, |
| 2107 | ], |
| 2108 | }; |
| 2109 | const createCall = yield* invoke("create_threads", createInput); |
| 2110 | expect(createCall.isError).toBe(false); |
| 2111 | const created = yield* decodeCreateThreadsResult(createCall.structuredContent).pipe( |
| 2112 | Effect.orDie, |
| 2113 | ); |
| 2114 | expect(created.threads).toHaveLength(2); |
| 2115 | const emptyThread = created.threads[0]!; |
| 2116 | const promptedThread = created.threads[1]!; |
| 2117 | expect(emptyThread).toMatchObject({ |
| 2118 | status: "idle", |
| 2119 | createdBy: "agent", |
| 2120 | creationSource: "mcp", |
| 2121 | providerInstanceId: codexInstanceId, |
| 2122 | model: codexModel, |
| 2123 | }); |
| 2124 | expect(promptedThread).toMatchObject({ |
| 2125 | createdBy: "agent", |
| 2126 | creationSource: "mcp", |
| 2127 | providerInstanceId: claudeInstanceId, |
| 2128 | model: claudeModel, |
| 2129 | }); |
| 2130 | const createdSource = yield* orchestrator.getThreadProjection(promptedThread.threadId); |
| 2131 | expect(createdSource.messages[0]).toMatchObject({ |
| 2132 | senderThreadId: parentThreadId, |
| 2133 | }); |
| 2134 | expect( |
| 2135 | createdSource.turnItems.find((item) => item.type === "user_message"), |
| 2136 | ).toMatchObject({ |
| 2137 | senderThreadId: parentThreadId, |
| 2138 | }); |
| 2139 | const emptyProjection = yield* orchestrator.getThreadProjection(emptyThread.threadId); |
| 2140 | expect(emptyProjection.thread.lineage).toEqual({ |
| 2141 | parentThreadId: null, |
| 2142 | relationshipToParent: null, |
| 2143 | rootThreadId: emptyThread.threadId, |
| 2144 | }); |
| 2145 | expect(emptyProjection.thread).toMatchObject({ |
| 2146 | createdBy: "agent", |
| 2147 | creationSource: "mcp", |
| 2148 | }); |
| 2149 | expect(emptyProjection.thread.forkedFrom).toBeNull(); |
| 2150 | expect(emptyProjection.runs).toEqual([]); |
| 2151 | |
| 2152 | const defaultRenameCall = yield* invoke("t3_thread_update", { |
| 2153 | action: "rename", |
| 2154 | title: "MCP parent metadata", |
| 2155 | clientRequestId: "metadata-default-thread-1", |
| 2156 | }); |
| 2157 | const defaultRenamed = yield* decodeThreadUpdateResult( |
| 2158 | defaultRenameCall.structuredContent, |
| 2159 | ).pipe(Effect.orDie); |
| 2160 | expect(defaultRenamed).toMatchObject({ |
| 2161 | threadId: parentThreadId, |
| 2162 | action: "rename", |
| 2163 | title: "MCP parent metadata", |
| 2164 | }); |
| 2165 | |
| 2166 | const renameCall = yield* invoke("t3_thread_update", { |
| 2167 | threadId: emptyThread.threadId, |
| 2168 | action: "rename", |
| 2169 | title: "Metadata-managed thread", |
| 2170 | clientRequestId: "metadata-rename-1", |
| 2171 | }); |
| 2172 | const renamed = yield* decodeThreadUpdateResult(renameCall.structuredContent).pipe( |
| 2173 | Effect.orDie, |
| 2174 | ); |
| 2175 | expect(renamed).toMatchObject({ |
| 2176 | threadId: emptyThread.threadId, |
| 2177 | action: "rename", |
| 2178 | title: "Metadata-managed thread", |
| 2179 | linkedPullRequest: null, |
| 2180 | }); |
| 2181 | const repeatedRenameCall = yield* invoke("t3_thread_update", { |
| 2182 | threadId: emptyThread.threadId, |
| 2183 | action: "rename", |
| 2184 | title: "Metadata-managed thread", |
| 2185 | clientRequestId: "metadata-rename-1", |
| 2186 | }); |
| 2187 | const repeatedRename = yield* decodeThreadUpdateResult( |
| 2188 | repeatedRenameCall.structuredContent, |
| 2189 | ).pipe(Effect.orDie); |
| 2190 | expect(repeatedRename.commandId).toBe(renamed.commandId); |
| 2191 | expect(repeatedRename.sequence).toBe(renamed.sequence); |
| 2192 | |
| 2193 | const linkedCall = yield* invoke("t3_thread_update", { |
| 2194 | threadId: emptyThread.threadId, |
| 2195 | action: "link_pull_request", |
| 2196 | pullRequest: { |
| 2197 | repository: "pingdotgg/t3code", |
| 2198 | number: 8689, |
| 2199 | url: "https://github.com/pingdotgg/t3code/pull/8689", |
| 2200 | }, |
| 2201 | clientRequestId: "metadata-link-1", |
| 2202 | }); |
| 2203 | const linked = yield* decodeThreadUpdateResult(linkedCall.structuredContent).pipe( |
| 2204 | Effect.orDie, |
| 2205 | ); |
| 2206 | expect(linked.linkedPullRequest).toEqual({ |
| 2207 | projectId, |
| 2208 | repository: "pingdotgg/t3code", |
| 2209 | number: 8689, |
| 2210 | url: "https://github.com/pingdotgg/t3code/pull/8689", |
| 2211 | }); |
| 2212 | const metadataReadCall = yield* invoke("t3_thread_read", { |
| 2213 | threadId: emptyThread.threadId, |
| 2214 | }); |
| 2215 | const metadataRead = yield* decodeThreadReadResult( |
| 2216 | metadataReadCall.structuredContent, |
| 2217 | ).pipe(Effect.orDie); |
| 2218 | expect(metadataRead.thread).toMatchObject({ |
| 2219 | title: "Metadata-managed thread", |
| 2220 | linkedPullRequest: linked.linkedPullRequest, |
| 2221 | }); |
| 2222 | const metadataListCall = yield* invoke("t3_thread_list", { limit: 100 }); |
| 2223 | const metadataList = yield* decodeThreadListResult( |
| 2224 | metadataListCall.structuredContent, |
| 2225 | ).pipe(Effect.orDie); |
| 2226 | expect( |
| 2227 | metadataList.threads.find((thread) => thread.threadId === emptyThread.threadId), |
| 2228 | ).toMatchObject({ |
| 2229 | title: "Metadata-managed thread", |
| 2230 | linkedPullRequest: linked.linkedPullRequest, |
| 2231 | settled: false, |
| 2232 | settledAt: null, |
| 2233 | }); |
| 2234 | expect(metadataRead.thread).toMatchObject({ settled: false, settledAt: null }); |
| 2235 | |
| 2236 | yield* orchestrator.dispatch({ |
| 2237 | type: "thread.settle", |
| 2238 | commandId: CommandId.make("command:mcp-empty:settle"), |
| 2239 | threadId: emptyThread.threadId, |
| 2240 | settledAt: DateTime.makeUnsafe("2026-01-01T00:00:00.000Z"), |
| 2241 | }); |
| 2242 | const settledReadCall = yield* invoke("t3_thread_read", { |
| 2243 | threadId: emptyThread.threadId, |
| 2244 | }); |
| 2245 | const settledRead = yield* decodeThreadReadResult( |
| 2246 | settledReadCall.structuredContent, |
| 2247 | ).pipe(Effect.orDie); |
| 2248 | expect(settledRead.thread).toMatchObject({ |
| 2249 | settled: true, |
| 2250 | settledAt: "2026-01-01T00:00:00.000Z", |
| 2251 | }); |
| 2252 | const settledListCall = yield* invoke("t3_thread_list", { settled: true, limit: 100 }); |
| 2253 | const settledList = yield* decodeThreadListResult( |
| 2254 | settledListCall.structuredContent, |
| 2255 | ).pipe(Effect.orDie); |
| 2256 | expect(settledList.threads.map((thread) => thread.threadId)).toEqual([ |
| 2257 | emptyThread.threadId, |
| 2258 | ]); |
| 2259 | expect(settledList.threads[0]).toMatchObject({ |
| 2260 | settled: true, |
| 2261 | settledAt: "2026-01-01T00:00:00.000Z", |
| 2262 | }); |
| 2263 | const activeListCall = yield* invoke("t3_thread_list", { settled: false, limit: 100 }); |
| 2264 | const activeList = yield* decodeThreadListResult(activeListCall.structuredContent).pipe( |
| 2265 | Effect.orDie, |
| 2266 | ); |
| 2267 | expect( |
| 2268 | activeList.threads.some((thread) => thread.threadId === emptyThread.threadId), |
| 2269 | ).toBe(false); |
| 2270 | yield* orchestrator.dispatch({ |
| 2271 | type: "thread.unsettle", |
| 2272 | commandId: CommandId.make("command:mcp-empty:unsettle"), |
| 2273 | threadId: emptyThread.threadId, |
| 2274 | reason: "user", |
| 2275 | }); |
| 2276 | |
| 2277 | expect(metadataRead.thread).toMatchObject({ snoozed: false, snoozedUntil: null }); |
| 2278 | yield* orchestrator.dispatch({ |
| 2279 | type: "thread.snooze", |
| 2280 | commandId: CommandId.make("command:mcp-empty:snooze"), |
| 2281 | threadId: emptyThread.threadId, |
| 2282 | snoozedUntil: "2099-01-01T00:00:00.000Z", |
| 2283 | }); |
| 2284 | const snoozedListCall = yield* invoke("t3_thread_list", { snoozed: true, limit: 100 }); |
| 2285 | const snoozedList = yield* decodeThreadListResult( |
| 2286 | snoozedListCall.structuredContent, |
| 2287 | ).pipe(Effect.orDie); |
| 2288 | expect(snoozedList.threads.map((thread) => thread.threadId)).toEqual([ |
| 2289 | emptyThread.threadId, |
| 2290 | ]); |
| 2291 | expect(snoozedList.threads[0]).toMatchObject({ |
| 2292 | snoozed: true, |
| 2293 | snoozedUntil: "2099-01-01T00:00:00.000Z", |
| 2294 | }); |
| 2295 | const snoozedReadCall = yield* invoke("t3_thread_read", { |
| 2296 | threadId: emptyThread.threadId, |
| 2297 | }); |
| 2298 | const snoozedRead = yield* decodeThreadReadResult( |
| 2299 | snoozedReadCall.structuredContent, |
| 2300 | ).pipe(Effect.orDie); |
| 2301 | expect(snoozedRead.thread).toMatchObject({ |
| 2302 | snoozed: true, |
| 2303 | snoozedUntil: "2099-01-01T00:00:00.000Z", |
| 2304 | }); |
| 2305 | yield* orchestrator.dispatch({ |
| 2306 | type: "thread.unsnooze", |
| 2307 | commandId: CommandId.make("command:mcp-empty:unsnooze"), |
| 2308 | threadId: emptyThread.threadId, |
| 2309 | reason: "user", |
| 2310 | }); |
| 2311 | |
| 2312 | const unlinkedCall = yield* invoke("t3_thread_update", { |
| 2313 | threadId: emptyThread.threadId, |
| 2314 | action: "unlink_pull_request", |
| 2315 | clientRequestId: "metadata-unlink-1", |
| 2316 | }); |
| 2317 | const unlinked = yield* decodeThreadUpdateResult(unlinkedCall.structuredContent).pipe( |
| 2318 | Effect.orDie, |
| 2319 | ); |
| 2320 | expect(unlinked.linkedPullRequest).toBeNull(); |
| 2321 | |
| 2322 | const regenerateCall = yield* invoke("t3_thread_update", { |
| 2323 | threadId: emptyThread.threadId, |
| 2324 | action: "regenerate_title", |
| 2325 | clientRequestId: "metadata-regenerate-title-1", |
| 2326 | }); |
| 2327 | const regenerating = yield* decodeThreadUpdateResult( |
| 2328 | regenerateCall.structuredContent, |
| 2329 | ).pipe(Effect.orDie); |
| 2330 | expect(regenerating).toMatchObject({ |
| 2331 | threadId: emptyThread.threadId, |
| 2332 | action: "regenerate_title", |
| 2333 | titleRegeneration: { requestId: regenerating.commandId }, |
| 2334 | }); |
| 2335 | const promptedProjection = yield* waitForProjection( |
| 2336 | orchestrator, |
| 2337 | promptedThread.threadId, |
| 2338 | (projection) => projection.runs.some((run) => run.status === "completed"), |
| 2339 | ); |
| 2340 | expect(promptedProjection.thread.lineage.parentThreadId).toBeNull(); |
| 2341 | expect( |
| 2342 | promptedProjection.messages |
| 2343 | .filter((message) => message.role === "user") |
| 2344 | .map((message) => message.text), |
| 2345 | ).toEqual([createdThreadPrompt]); |
| 2346 | const createdThreadItems = (yield* orchestrator.getThreadProjection( |
| 2347 | parentThreadId, |
| 2348 | )).visibleTurnItems |
| 2349 | .map((row) => row.item) |
| 2350 | .filter((item) => item.type === "thread_created"); |
| 2351 | expect( |
| 2352 | createdThreadItems.map((item) => ({ |
| 2353 | targetThreadId: item.targetThreadId, |
| 2354 | targetRunId: item.targetRunId, |
| 2355 | title: item.title, |
| 2356 | providerInstanceId: item.targetProviderInstanceId, |
| 2357 | model: item.targetModel, |
| 2358 | })), |
| 2359 | ).toEqual([ |
| 2360 | { |
| 2361 | targetThreadId: emptyThread.threadId, |
| 2362 | targetRunId: null, |
| 2363 | title: emptyThread.title, |
| 2364 | providerInstanceId: codexInstanceId, |
| 2365 | model: codexModel, |
| 2366 | }, |
| 2367 | { |
| 2368 | targetThreadId: promptedThread.threadId, |
| 2369 | targetRunId: promptedThread.runId, |
| 2370 | title: promptedThread.title, |
| 2371 | providerInstanceId: claudeInstanceId, |
| 2372 | model: claudeModel, |
| 2373 | }, |
| 2374 | ]); |
| 2375 | |
| 2376 | const repeatedCreateCall = yield* invoke("create_threads", createInput); |
| 2377 | const repeatedCreated = yield* decodeCreateThreadsResult( |
| 2378 | repeatedCreateCall.structuredContent, |
| 2379 | ).pipe(Effect.orDie); |
| 2380 | expect(repeatedCreated.threads.map((thread) => thread.threadId)).toEqual( |
| 2381 | created.threads.map((thread) => thread.threadId), |
| 2382 | ); |
| 2383 | expect( |
| 2384 | (yield* orchestrator.getThreadProjection(parentThreadId)).visibleTurnItems.filter( |
| 2385 | (row) => { |
| 2386 | if (row.item.type !== "thread_created") return false; |
| 2387 | const targetThreadId = row.item.targetThreadId; |
| 2388 | return created.threads.some((thread) => thread.threadId === targetThreadId); |
| 2389 | }, |
| 2390 | ), |
| 2391 | ).toHaveLength(2); |
| 2392 | |
| 2393 | const promptedReadCall = yield* invoke("t3_thread_read", { |
| 2394 | threadId: promptedThread.threadId, |
| 2395 | limit: 1, |
| 2396 | }); |
| 2397 | const promptedRead = yield* decodeThreadReadResult( |
| 2398 | promptedReadCall.structuredContent, |
| 2399 | ).pipe(Effect.orDie); |
| 2400 | expect(promptedRead.thread.status).toBe("completed"); |
| 2401 | expect(promptedRead.thread).toMatchObject({ |
| 2402 | createdBy: "agent", |
| 2403 | creationSource: "mcp", |
| 2404 | }); |
| 2405 | expect(promptedRead.items.map((item) => item.type)).toEqual(["user_message"]); |
| 2406 | expect(promptedRead.items[0]).toMatchObject({ |
| 2407 | createdBy: "agent", |
| 2408 | creationSource: "mcp", |
| 2409 | }); |
| 2410 | expect(promptedRead.hasMore).toBe(true); |
| 2411 | const promptedReadNextCall = yield* invoke("t3_thread_read", { |
| 2412 | threadId: promptedThread.threadId, |
| 2413 | afterPosition: promptedRead.nextPosition, |
| 2414 | limit: 1, |
| 2415 | }); |
| 2416 | const promptedReadNext = yield* decodeThreadReadResult( |
| 2417 | promptedReadNextCall.structuredContent, |
| 2418 | ).pipe(Effect.orDie); |
| 2419 | expect(promptedReadNext.items.map((item) => item.type)).toEqual(["assistant_message"]); |
| 2420 | expect(promptedReadNext.items[0]?.text).toBe( |
| 2421 | `Claude completed: ${createdThreadPrompt}`, |
| 2422 | ); |
| 2423 | |
| 2424 | const forkedThreadId = ThreadId.make("thread:mcp-orchestrator-inherited-read"); |
| 2425 | yield* orchestrator.dispatch({ |
| 2426 | type: "thread.fork", |
| 2427 | createdBy: "user", |
| 2428 | creationSource: "web", |
| 2429 | commandId: CommandId.make("command:mcp-orchestrator-inherited-read"), |
| 2430 | sourceThreadId: promptedThread.threadId, |
| 2431 | targetThreadId: forkedThreadId, |
| 2432 | sourcePoint: { type: "latest_stable" }, |
| 2433 | title: "Inherited read thread", |
| 2434 | }); |
| 2435 | const forkedProjection = yield* orchestrator.getThreadProjection(forkedThreadId); |
| 2436 | expect(forkedProjection.messages).toEqual([]); |
| 2437 | expect( |
| 2438 | forkedProjection.visibleTurnItems.some( |
| 2439 | (row) => |
| 2440 | row.sourceThreadId === promptedThread.threadId && |
| 2441 | row.item.type === "user_message", |
| 2442 | ), |
| 2443 | ).toBe(true); |
| 2444 | |
| 2445 | const forkedReadCall = yield* invoke("t3_thread_read", { |
| 2446 | threadId: forkedThreadId, |
| 2447 | }); |
| 2448 | const forkedRead = yield* decodeThreadReadResult(forkedReadCall.structuredContent).pipe( |
| 2449 | Effect.orDie, |
| 2450 | ); |
| 2451 | expect( |
| 2452 | forkedRead.items.find((item) => item.text === createdThreadPrompt), |
| 2453 | ).toMatchObject({ |
| 2454 | sourceThreadId: promptedThread.threadId, |
| 2455 | createdBy: "agent", |
| 2456 | creationSource: "mcp", |
| 2457 | }); |
| 2458 | |
| 2459 | const ordinaryLoopPrompt = "Run an ordinary thread loop iteration."; |
| 2460 | const sendCall = yield* invoke("t3_thread_send", { |
| 2461 | threadId: emptyThread.threadId, |
| 2462 | message: ordinaryLoopPrompt, |
| 2463 | clientRequestId: "ordinary-loop-send-1", |
| 2464 | }); |
| 2465 | const sent = yield* decodeThreadSendResult(sendCall.structuredContent).pipe( |
| 2466 | Effect.orDie, |
| 2467 | ); |
| 2468 | const sentSource = yield* orchestrator.getThreadProjection(emptyThread.threadId); |
| 2469 | expect( |
| 2470 | sentSource.messages.find((message) => message.id === sent.messageId), |
| 2471 | ).toMatchObject({ |
| 2472 | senderThreadId: parentThreadId, |
| 2473 | }); |
| 2474 | expect( |
| 2475 | sentSource.turnItems.find( |
| 2476 | (item) => item.type === "user_message" && item.messageId === sent.messageId, |
| 2477 | ), |
| 2478 | ).toMatchObject({ |
| 2479 | senderThreadId: parentThreadId, |
| 2480 | }); |
| 2481 | expect(sent.delivery).toBe("started"); |
| 2482 | const waitCall = yield* invoke("t3_thread_wait", { |
| 2483 | threadId: emptyThread.threadId, |
| 2484 | runId: sent.runId, |
| 2485 | timeoutMs: 10_000, |
| 2486 | }); |
| 2487 | const waited = yield* decodeThreadWaitResult(waitCall.structuredContent).pipe( |
| 2488 | Effect.orDie, |
| 2489 | ); |
| 2490 | expect(waited).toMatchObject({ |
| 2491 | runId: sent.runId, |
| 2492 | status: "completed", |
| 2493 | timedOut: false, |
| 2494 | }); |
| 2495 | const repeatedSendCall = yield* invoke("t3_thread_send", { |
| 2496 | threadId: emptyThread.threadId, |
| 2497 | message: ordinaryLoopPrompt, |
| 2498 | clientRequestId: "ordinary-loop-send-1", |
| 2499 | }); |
| 2500 | const repeatedSend = yield* decodeThreadSendResult( |
| 2501 | repeatedSendCall.structuredContent, |
| 2502 | ).pipe(Effect.orDie); |
| 2503 | expect(repeatedSend.runId).toBe(sent.runId); |
| 2504 | expect( |
| 2505 | (yield* orchestrator.getThreadProjection(emptyThread.threadId)).runs, |
| 2506 | ).toHaveLength(1); |
| 2507 | |
| 2508 | const activeThreadCall = yield* invoke("create_threads", { |
| 2509 | threads: [{ prompt: cancellationPrompt, title: "Managed active thread" }], |
| 2510 | clientRequestId: "managed-active-thread-1", |
| 2511 | }); |
| 2512 | const activeThread = (yield* decodeCreateThreadsResult( |
| 2513 | activeThreadCall.structuredContent, |
| 2514 | ).pipe(Effect.orDie)).threads[0]!; |
| 2515 | const activeThreadItem = (yield* orchestrator.getThreadProjection( |
| 2516 | parentThreadId, |
| 2517 | )).visibleTurnItems |
| 2518 | .map((row) => row.item) |
| 2519 | .find( |
| 2520 | (item) => |
| 2521 | item.type === "thread_created" && item.targetThreadId === activeThread.threadId, |
| 2522 | ); |
| 2523 | expect(activeThreadItem).toMatchObject({ |
| 2524 | type: "thread_created", |
| 2525 | title: "Managed active thread", |
| 2526 | targetThreadId: activeThread.threadId, |
| 2527 | targetRunId: activeThread.runId, |
| 2528 | targetProviderInstanceId: codexInstanceId, |
| 2529 | targetModel: codexModel, |
| 2530 | }); |
| 2531 | const activeProjection = yield* waitForProjection( |
| 2532 | orchestrator, |
| 2533 | activeThread.threadId, |
| 2534 | (projection) => |
| 2535 | projection.runs.some((run) => run.status === "running") && |
| 2536 | projection.providerTurns.some((turn) => turn.status === "running"), |
| 2537 | ); |
| 2538 | const activeRun = activeProjection.runs[0]!; |
| 2539 | const activeTimeoutCall = yield* invoke("t3_thread_wait", { |
| 2540 | threadId: activeThread.threadId, |
| 2541 | runId: activeRun.id, |
| 2542 | timeoutMs: 1, |
| 2543 | }); |
| 2544 | const activeTimeout = yield* decodeThreadWaitResult( |
| 2545 | activeTimeoutCall.structuredContent, |
| 2546 | ).pipe(Effect.orDie); |
| 2547 | expect(activeTimeout).toMatchObject({ |
| 2548 | runId: activeRun.id, |
| 2549 | status: "running", |
| 2550 | timedOut: true, |
| 2551 | }); |
| 2552 | const steerCall = yield* invoke("t3_thread_send", { |
| 2553 | threadId: activeThread.threadId, |
| 2554 | message: "Include the latest parent guidance before finishing.", |
| 2555 | mode: "steer", |
| 2556 | clientRequestId: "managed-active-steer-1", |
| 2557 | }); |
| 2558 | const steered = yield* decodeThreadSendResult(steerCall.structuredContent).pipe( |
| 2559 | Effect.orDie, |
| 2560 | ); |
| 2561 | expect(steered).toMatchObject({ |
| 2562 | runId: activeRun.id, |
| 2563 | delivery: "steered", |
| 2564 | }); |
| 2565 | const steeredSource = yield* orchestrator.getThreadProjection(activeThread.threadId); |
| 2566 | expect( |
| 2567 | steeredSource.messages.find((message) => message.id === steered.messageId), |
| 2568 | ).toMatchObject({ |
| 2569 | senderThreadId: parentThreadId, |
| 2570 | }); |
| 2571 | expect( |
| 2572 | steeredSource.turnItems.find( |
| 2573 | (item) => item.type === "user_message" && item.messageId === steered.messageId, |
| 2574 | ), |
| 2575 | ).toMatchObject({ |
| 2576 | senderThreadId: parentThreadId, |
| 2577 | }); |
| 2578 | const interruptCall = yield* invoke("t3_thread_interrupt", { |
| 2579 | threadId: activeThread.threadId, |
| 2580 | reason: "The orchestration loop has enough evidence.", |
| 2581 | clientRequestId: "managed-active-interrupt-1", |
| 2582 | }); |
| 2583 | const interrupted = yield* decodeThreadInterruptResult( |
| 2584 | interruptCall.structuredContent, |
| 2585 | ).pipe(Effect.orDie); |
| 2586 | expect(interrupted).toMatchObject({ |
| 2587 | runId: activeRun.id, |
| 2588 | status: "interrupt_requested", |
| 2589 | }); |
| 2590 | const interruptedWaitCall = yield* invoke("t3_thread_wait", { |
| 2591 | threadId: activeThread.threadId, |
| 2592 | runId: activeRun.id, |
| 2593 | timeoutMs: 10_000, |
| 2594 | }); |
| 2595 | const interruptedWait = yield* decodeThreadWaitResult( |
| 2596 | interruptedWaitCall.structuredContent, |
| 2597 | ).pipe(Effect.orDie); |
| 2598 | expect(interruptedWait.status).toBe("interrupted"); |
| 2599 | const repeatedInterruptCall = yield* invoke("t3_thread_interrupt", { |
| 2600 | threadId: activeThread.threadId, |
| 2601 | runId: activeRun.id, |
| 2602 | }); |
| 2603 | const repeatedInterrupt = yield* decodeThreadInterruptResult( |
| 2604 | repeatedInterruptCall.structuredContent, |
| 2605 | ).pipe(Effect.orDie); |
| 2606 | expect(repeatedInterrupt.status).toBe("interrupted"); |
| 2607 | |
| 2608 | const foreignThreadId = ThreadId.make("thread:mcp-foreign-project"); |
| 2609 | yield* orchestrator.dispatch({ |
| 2610 | type: "thread.create", |
| 2611 | createdBy: "user", |
| 2612 | creationSource: "web", |
| 2613 | commandId: CommandId.make("command:mcp-foreign-project:create"), |
| 2614 | threadId: foreignThreadId, |
| 2615 | projectId: ProjectId.make("project:mcp-foreign"), |
| 2616 | title: "Foreign project thread", |
| 2617 | modelSelection: codexSelection, |
| 2618 | runtimeMode: "full-access", |
| 2619 | interactionMode: "default", |
| 2620 | branch: null, |
| 2621 | worktreePath: cwd, |
| 2622 | }); |
| 2623 | // Targets reach the whole environment; the caller's modes still cap writes. |
| 2624 | const foreignOrganizeCall = yield* invoke("t3_thread_organize", { |
| 2625 | threadId: foreignThreadId, |
| 2626 | action: "pin", |
| 2627 | }); |
| 2628 | expect(foreignOrganizeCall.isError).toBe(false); |
| 2629 | expect((yield* orchestrator.getThreadShell(foreignThreadId))?.pinnedAt).not.toBeNull(); |
| 2630 | |
| 2631 | const foreignReadCall = yield* invoke("t3_thread_read", { |
| 2632 | threadId: foreignThreadId, |
| 2633 | }); |
| 2634 | expect(foreignReadCall.structuredContent).toMatchObject({ |
| 2635 | thread: { threadId: foreignThreadId, projectId: "project:mcp-foreign" }, |
| 2636 | }); |
| 2637 | const foreignUpdateCall = yield* invoke("t3_thread_update", { |
| 2638 | threadId: foreignThreadId, |
| 2639 | action: "rename", |
| 2640 | title: "Renamed from another project", |
| 2641 | }); |
| 2642 | expect(foreignUpdateCall.structuredContent).toMatchObject({ |
| 2643 | threadId: foreignThreadId, |
| 2644 | title: "Renamed from another project", |
| 2645 | }); |
| 2646 | const foreignListCall = yield* invoke("t3_thread_list", { |
| 2647 | projectId: "project:mcp-foreign", |
| 2648 | }); |
| 2649 | const foreignListed = yield* decodeThreadListResult( |
| 2650 | foreignListCall.structuredContent, |
| 2651 | ).pipe(Effect.orDie); |
| 2652 | expect(foreignListed.threads.map((thread) => thread.threadId)).toEqual([ |
| 2653 | foreignThreadId, |
| 2654 | ]); |
| 2655 | const listCall = yield* invoke("t3_thread_list", { |
| 2656 | includeSubagents: false, |
| 2657 | limit: 100, |
| 2658 | }); |
| 2659 | const listed = yield* decodeThreadListResult(listCall.structuredContent).pipe( |
| 2660 | Effect.orDie, |
| 2661 | ); |
| 2662 | expect(listed.projectId).toBe(projectId); |
| 2663 | expect(listed.threads.map((thread) => thread.threadId)).toEqual( |
| 2664 | expect.arrayContaining([ |
| 2665 | parentThreadId, |
| 2666 | emptyThread.threadId, |
| 2667 | promptedThread.threadId, |
| 2668 | activeThread.threadId, |
| 2669 | ]), |
| 2670 | ); |
| 2671 | expect( |
| 2672 | listed.threads.find((thread) => thread.threadId === emptyThread.threadId), |
| 2673 | ).toMatchObject({ |
| 2674 | createdBy: "agent", |
| 2675 | creationSource: "mcp", |
| 2676 | }); |
| 2677 | expect(listed.threads.some((thread) => thread.threadId === foreignThreadId)).toBe( |
| 2678 | false, |
| 2679 | ); |
| 2680 | expect( |
| 2681 | listed.threads.some((thread) => thread.relationshipToParent === "subagent"), |
| 2682 | ).toBe(false); |
| 2683 | |
| 2684 | // A wait-mode delegation whose blocking wait times out no longer |
| 2685 | // owns delivery, so delegate_task upgrades the task to "always". |
| 2686 | // Its terminal then wakes the parent even mid-turn. |
| 2687 | const upgradedCall = yield* invoke("delegate_task", { |
| 2688 | task: cancellationPrompt, |
| 2689 | target: { |
| 2690 | providerInstanceId: codexInstanceId, |
| 2691 | model: codexModel, |
| 2692 | }, |
| 2693 | mode: "wait", |
| 2694 | timeoutMs: 1, |
| 2695 | clientRequestId: "delegate-wait-upgrade-1", |
| 2696 | }); |
| 2697 | const upgradedDelegated = yield* decodeDelegateTaskResult( |
| 2698 | upgradedCall.structuredContent, |
| 2699 | ).pipe(Effect.orDie); |
| 2700 | expect(upgradedDelegated.status).toBe("running"); |
| 2701 | yield* waitForProjection(orchestrator, upgradedDelegated.childThreadId, (projection) => |
| 2702 | projection.providerTurns.some((turn) => turn.status === "running"), |
| 2703 | ); |
| 2704 | expect( |
| 2705 | (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( |
| 2706 | (task) => task.id === upgradedDelegated.taskId, |
| 2707 | )?.completionWake, |
| 2708 | ).toBe("always"); |
| 2709 | const upgradeCancelCall = yield* invoke("task_cancel", { |
| 2710 | taskId: upgradedDelegated.taskId, |
| 2711 | reason: "Terminalize while the parent run is live.", |
| 2712 | clientRequestId: "cancel-wait-upgrade-1", |
| 2713 | }); |
| 2714 | expect(upgradeCancelCall.isError).toBe(false); |
| 2715 | yield* waitForProjection(orchestrator, parentThreadId, (projection) => |
| 2716 | projection.subagents.some( |
| 2717 | (task) => task.id === upgradedDelegated.taskId && task.status === "interrupted", |
| 2718 | ), |
| 2719 | ); |
| 2720 | expect( |
| 2721 | (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( |
| 2722 | (task) => task.id === upgradedDelegated.taskId, |
| 2723 | )?.completionDelivery, |
| 2724 | ).toMatchObject({ state: "disposed" }); |
| 2725 | yield* expectOffersToStay(0); |
| 2726 | |
| 2727 | // A live question ends the wait and leaves completion delivery armed. |
| 2728 | const questionWait = yield* invoke("delegate_task", { |
| 2729 | task: cancellationPrompt, |
| 2730 | title: "Wait question", |
| 2731 | target: { providerInstanceId: codexInstanceId, model: codexModel }, |
| 2732 | mode: "wait", |
| 2733 | timeoutMs: 30_000, |
| 2734 | clientRequestId: "delegate-wait-question-1", |
| 2735 | }).pipe(Effect.forkChild); |
| 2736 | const question = (yield* waitForProjection(orchestrator, parentThreadId, (p) => |
| 2737 | p.subagents.some( |
| 2738 | (task) => task.title === "Wait question" && task.childThreadId !== null, |
| 2739 | ), |
| 2740 | )).subagents.find((task) => task.title === "Wait question")!; |
| 2741 | const asking = yield* waitForProjection(orchestrator, question.childThreadId!, (p) => |
| 2742 | p.providerTurns.some((turn) => turn.status === "running"), |
| 2743 | ); |
| 2744 | const askedAt = yield* DateTime.now; |
| 2745 | yield* (yield* EventSink.EventSinkV2).write({ |
| 2746 | events: [ |
| 2747 | { |
| 2748 | id: EventId.make("event:wait-question:live"), |
| 2749 | type: "runtime-request.updated", |
| 2750 | threadId: question.childThreadId!, |
| 2751 | occurredAt: askedAt, |
| 2752 | payload: { |
| 2753 | id: RuntimeRequestId.make("request:wait-question"), |
| 2754 | nodeId: question.id, |
| 2755 | providerTurnId: null, |
| 2756 | nativeRequestRef: null, |
| 2757 | kind: "user_input", |
| 2758 | status: "pending", |
| 2759 | responseCapability: { |
| 2760 | type: "live", |
| 2761 | providerSessionId: asking.providerSessions[0]!.id, |
| 2762 | }, |
| 2763 | createdAt: askedAt, |
| 2764 | resolvedAt: null, |
| 2765 | }, |
| 2766 | }, |
| 2767 | ], |
| 2768 | }); |
| 2769 | expect((yield* Fiber.join(questionWait)).structuredContent).toMatchObject({ |
| 2770 | status: "running", |
| 2771 | hasPendingQuestion: true, |
| 2772 | waitTimedOut: false, |
| 2773 | }); |
| 2774 | expect( |
| 2775 | (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( |
| 2776 | (task) => task.id === question.id, |
| 2777 | )?.completionWake, |
| 2778 | ).toBe("always"); |
| 2779 | yield* invoke("task_cancel", { |
| 2780 | taskId: question.id, |
| 2781 | clientRequestId: "cancel-wait-question-1", |
| 2782 | }); |
| 2783 | yield* waitForProjection(orchestrator, parentThreadId, (p) => |
| 2784 | p.subagents.some((task) => task.id === question.id && task.status === "interrupted"), |
| 2785 | ); |
| 2786 | |
| 2787 | // The MCP tool cannot force the reverse interleaving (child |
| 2788 | // terminal before the upgrade lands), so dispatch the command |
| 2789 | // directly. The first wait-mode delegation completed while the |
| 2790 | // parent run was live, so finalize skipped its offer under |
| 2791 | // settled_only; the upgrade must accept, persist the policy, and |
| 2792 | // deliver the wake finalize declined. |
| 2793 | const terminalUpgrade = yield* orchestrator.dispatch({ |
| 2794 | type: "delegated_task.wake-policy", |
| 2795 | commandId: CommandId.make("command:mcp-parent:wake-policy-terminal"), |
| 2796 | parentThreadId, |
| 2797 | taskId: delegated.taskId, |
| 2798 | completionWake: "always", |
| 2799 | }); |
| 2800 | expect( |
| 2801 | terminalUpgrade.storedEvents.some( |
| 2802 | (stored) => |
| 2803 | stored.event.type === "subagent.updated" && |
| 2804 | stored.event.payload.id === delegated.taskId && |
| 2805 | stored.event.payload.completionWake === "always", |
| 2806 | ), |
| 2807 | ).toBe(true); |
| 2808 | expect( |
| 2809 | (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( |
| 2810 | (task) => task.id === delegated.taskId, |
| 2811 | )?.completionWake, |
| 2812 | ).toBe("always"); |
| 2813 | // task_status above acknowledged this terminal result, so making |
| 2814 | // its policy eager later cannot re-arm a stale parent wake. |
| 2815 | yield* expectOffersToStay(0); |
| 2816 | |
| 2817 | // Legacy records omit completionWake and stay settled_only. The |
| 2818 | // MCP service always sets the field now, so dispatch the request |
| 2819 | // directly to cover the legacy shape. |
| 2820 | const legacyDispatch = yield* orchestrator.dispatch({ |
| 2821 | type: "delegated_task.request", |
| 2822 | createdBy: "agent", |
| 2823 | creationSource: "mcp", |
| 2824 | commandId: CommandId.make("command:mcp-parent:delegate-legacy"), |
| 2825 | parentThreadId, |
| 2826 | parentRunId: parentRun.id, |
| 2827 | parentNodeId: parentRootNodeId, |
| 2828 | task: cancellationPrompt, |
| 2829 | modelSelection: codexSelection, |
| 2830 | runtimeMode: "full-access", |
| 2831 | interactionMode: "default", |
| 2832 | }); |
| 2833 | const legacyTaskEvent = legacyDispatch.storedEvents.find( |
| 2834 | (stored) => |
| 2835 | stored.event.type === "subagent.updated" && |
| 2836 | stored.event.payload.origin === "app_owned", |
| 2837 | ); |
| 2838 | if (legacyTaskEvent?.event.type !== "subagent.updated") { |
| 2839 | return yield* Effect.die(new Error("Legacy delegated task projection missing.")); |
| 2840 | } |
| 2841 | const legacyTask = legacyTaskEvent.event.payload; |
| 2842 | expect(legacyTask.completionWake).toBeUndefined(); |
| 2843 | if (legacyTask.childThreadId === null) { |
| 2844 | return yield* Effect.die(new Error("Legacy delegated task child thread missing.")); |
| 2845 | } |
| 2846 | const legacyChildThreadId = legacyTask.childThreadId; |
| 2847 | yield* waitForProjection(orchestrator, legacyChildThreadId, (projection) => |
| 2848 | projection.providerTurns.some((turn) => turn.status === "running"), |
| 2849 | ); |
| 2850 | // Nothing terminalized here, so the offer count must hold. |
| 2851 | yield* expectOffersToStay(0); |
| 2852 | |
| 2853 | // Stop cancels a queued server-owned delivery for this cohort and |
| 2854 | // records a barrier before any still-running child reaches a |
| 2855 | // terminal state. |
| 2856 | const stopQueuedDispatch = yield* orchestrator.dispatch({ |
| 2857 | type: "delegated_task.request", |
| 2858 | createdBy: "agent", |
| 2859 | creationSource: "mcp", |
| 2860 | commandId: CommandId.make("command:mcp-parent:stop-queued-delivery"), |
| 2861 | parentThreadId, |
| 2862 | parentRunId: parentRun.id, |
| 2863 | parentNodeId: parentRootNodeId, |
| 2864 | task: "Complete the stop barrier delivery task.", |
| 2865 | modelSelection: claudeSelection, |
| 2866 | runtimeMode: "full-access", |
| 2867 | interactionMode: "default", |
| 2868 | completionWake: "always", |
| 2869 | }); |
| 2870 | const stopQueuedTaskEvent = stopQueuedDispatch.storedEvents.find( |
| 2871 | (stored) => |
| 2872 | stored.event.type === "subagent.updated" && |
| 2873 | stored.event.payload.origin === "app_owned", |
| 2874 | ); |
| 2875 | if (stopQueuedTaskEvent?.event.type !== "subagent.updated") { |
| 2876 | return yield* Effect.die( |
| 2877 | new Error("Stop barrier delegated task projection missing."), |
| 2878 | ); |
| 2879 | } |
| 2880 | const stopQueuedTaskId = stopQueuedTaskEvent.event.payload.id; |
| 2881 | const stopQueuedProjection = yield* waitForProjection( |
| 2882 | orchestrator, |
| 2883 | parentThreadId, |
| 2884 | (projection) => |
| 2885 | projection.subagents.find((task) => task.id === stopQueuedTaskId)?.status === |
| 2886 | "completed" && |
| 2887 | projection.runs.find((run) => run.id === parentRun.id)?.delegatedCompletion |
| 2888 | ?.delivery !== null && |
| 2889 | projection.runs.find((run) => run.id === parentRun.id)?.delegatedCompletion |
| 2890 | ?.delivery !== undefined, |
| 2891 | ); |
| 2892 | const stopQueuedDelivery = stopQueuedProjection.runs.find( |
| 2893 | (run) => run.id === parentRun.id, |
| 2894 | )?.delegatedCompletion?.delivery; |
| 2895 | if (stopQueuedDelivery === undefined || stopQueuedDelivery === null) { |
| 2896 | return yield* Effect.die(new Error("Stop barrier delivery reservation missing.")); |
| 2897 | } |
| 2898 | yield* orchestrator.dispatch({ |
| 2899 | type: "message.dispatch", |
| 2900 | createdBy: "agent", |
| 2901 | creationSource: "server", |
| 2902 | commandId: CommandId.make("command:mcp-parent:dispatch-stop-queued-delivery"), |
| 2903 | threadId: parentThreadId, |
| 2904 | messageId: stopQueuedDelivery.messageId, |
| 2905 | text: "Delegated task reached a terminal state.", |
| 2906 | attachments: [], |
| 2907 | modelSelection: codexSelection, |
| 2908 | dispatchMode: { type: "queue_after_active" }, |
| 2909 | delegatedCompletion: { |
| 2910 | parentRunId: parentRun.id, |
| 2911 | generation: stopQueuedDelivery.generation, |
| 2912 | taskIds: stopQueuedDelivery.taskIds, |
| 2913 | }, |
| 2914 | }); |
| 2915 | const queuedStopDelivery = yield* waitForProjection( |
| 2916 | orchestrator, |
| 2917 | parentThreadId, |
| 2918 | (projection) => |
| 2919 | projection.runs.some( |
| 2920 | (run) => |
| 2921 | run.userMessageId === stopQueuedDelivery.messageId && run.status === "queued", |
| 2922 | ), |
| 2923 | ); |
| 2924 | const queuedStopDeliveryRun = queuedStopDelivery.runs.find( |
| 2925 | (run) => run.userMessageId === stopQueuedDelivery.messageId, |
| 2926 | ); |
| 2927 | if (queuedStopDeliveryRun === undefined) { |
| 2928 | return yield* Effect.die(new Error("Queued stop barrier delivery missing.")); |
| 2929 | } |
| 2930 | yield* Ref.set(continuationOffers, []); |
| 2931 | |
| 2932 | const parentStop = yield* orchestrator.dispatch({ |
| 2933 | type: "run.interrupt", |
| 2934 | commandId: CommandId.make("command:mcp-parent:interrupt-wake"), |
| 2935 | threadId: parentThreadId, |
| 2936 | runId: parentRun.id, |
| 2937 | reason: "Settle the parent before the legacy child terminalizes.", |
| 2938 | }); |
| 2939 | yield* waitForProjection(orchestrator, parentThreadId, (projection) => |
| 2940 | projection.runs.every( |
| 2941 | (run) => |
| 2942 | run.status !== "preparing" && |
| 2943 | run.status !== "starting" && |
| 2944 | run.status !== "running", |
| 2945 | ), |
| 2946 | ); |
| 2947 | const stoppedParent = yield* orchestrator.getThreadProjection(parentThreadId); |
| 2948 | expect(stoppedParent.runs.some((run) => run.id === parentRun.id)).toBe(true); |
| 2949 | expect( |
| 2950 | parentStop.storedEvents.some( |
| 2951 | (stored) => |
| 2952 | stored.event.type === "run.updated" && |
| 2953 | stored.event.payload.id === parentRun.id && |
| 2954 | stored.event.payload.delegatedCompletion?.disposition === "stopped", |
| 2955 | ), |
| 2956 | ).toBe(true); |
| 2957 | expect( |
| 2958 | stoppedParent.runs.find((run) => run.id === parentRun.id)?.delegatedCompletion, |
| 2959 | ).toMatchObject({ disposition: "stopped", delivery: null }); |
| 2960 | expect( |
| 2961 | stoppedParent.runs.find((run) => run.id === queuedStopDeliveryRun.id)?.status, |
| 2962 | ).toBe("cancelled"); |
| 2963 | expect( |
| 2964 | stoppedParent.subagents.find((task) => task.id === legacyTask.id)?.completionDelivery, |
| 2965 | ).toMatchObject({ state: "disposed" }); |
| 2966 | const legacyChildRun = (yield* orchestrator.getThreadProjection( |
| 2967 | legacyChildThreadId, |
| 2968 | )).runs.find((run) => run.status === "running"); |
| 2969 | if (legacyChildRun === undefined) { |
| 2970 | return yield* Effect.die( |
| 2971 | new Error("Still-running child was missing after parent Stop."), |
| 2972 | ); |
| 2973 | } |
| 2974 | yield* orchestrator.dispatch({ |
| 2975 | type: "run.interrupt", |
| 2976 | commandId: CommandId.make("command:mcp-parent:interrupt-legacy-after-parent-stop"), |
| 2977 | threadId: legacyChildThreadId, |
| 2978 | runId: legacyChildRun.id, |
| 2979 | reason: "Terminalize after the parent Stop barrier.", |
| 2980 | }); |
| 2981 | yield* waitForProjection(orchestrator, legacyChildThreadId, (projection) => |
| 2982 | projection.runs.some((run) => run.status === "interrupted"), |
| 2983 | ); |
| 2984 | expect( |
| 2985 | (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( |
| 2986 | (task) => task.id === legacyTask.id, |
| 2987 | )?.completionDelivery, |
| 2988 | ).toMatchObject({ state: "disposed" }); |
| 2989 | yield* expectOffersToStay(0); |
| 2990 | |
| 2991 | // Same command against a terminal task whose finalize already |
| 2992 | // offered (the parent was settled then, and still is): the policy |
| 2993 | // must persist without a second offer, or the parent wakes twice. |
| 2994 | const settledUpgrade = yield* orchestrator.dispatch({ |
| 2995 | type: "delegated_task.wake-policy", |
| 2996 | commandId: CommandId.make("command:mcp-parent:wake-policy-settled"), |
| 2997 | parentThreadId, |
| 2998 | taskId: legacyTask.id, |
| 2999 | completionWake: "always", |
| 3000 | }); |
| 3001 | expect( |
| 3002 | settledUpgrade.storedEvents.some( |
| 3003 | (stored) => |
| 3004 | stored.event.type === "subagent.updated" && |
| 3005 | stored.event.payload.id === legacyTask.id && |
| 3006 | stored.event.payload.completionWake === "always", |
| 3007 | ), |
| 3008 | ).toBe(true); |
| 3009 | expect( |
| 3010 | (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( |
| 3011 | (task) => task.id === legacyTask.id, |
| 3012 | )?.completionWake, |
| 3013 | ).toBe("always"); |
| 3014 | yield* expectOffersToStay(0); |
| 3015 | |
| 3016 | // A child that terminalizes after the coalesced delivery started |
| 3017 | // is held for one successor. It must not fan out into a second |
| 3018 | // concurrent delivery for the same parent run. |
| 3019 | const lateParentThreadId = ThreadId.make("thread:mcp-late-completion-parent"); |
| 3020 | const lateParentGate = yield* Deferred.make<void>(); |
| 3021 | const lateDeliveryGate = yield* Deferred.make<void>(); |
| 3022 | parentTerminalGates.set(lateParentThreadId, lateParentGate); |
| 3023 | deliveryTerminalGates.set(lateParentThreadId, lateDeliveryGate); |
| 3024 | yield* Ref.set(continuationOffers, []); |
| 3025 | yield* orchestrator.dispatch({ |
| 3026 | type: "thread.create", |
| 3027 | createdBy: "user", |
| 3028 | creationSource: "web", |
| 3029 | commandId: CommandId.make("command:mcp-late-parent:create"), |
| 3030 | threadId: lateParentThreadId, |
| 3031 | projectId, |
| 3032 | title: "Late completion parent", |
| 3033 | modelSelection: codexSelection, |
| 3034 | runtimeMode: "full-access", |
| 3035 | interactionMode: "default", |
| 3036 | branch: null, |
| 3037 | worktreePath: cwd, |
| 3038 | }); |
| 3039 | yield* orchestrator.dispatch({ |
| 3040 | type: "message.dispatch", |
| 3041 | createdBy: "user", |
| 3042 | creationSource: "web", |
| 3043 | commandId: CommandId.make("command:mcp-late-parent:start"), |
| 3044 | threadId: lateParentThreadId, |
| 3045 | messageId: MessageId.make("message:mcp-late-parent:start"), |
| 3046 | text: "Hold this parent until its completion delivery is queued.", |
| 3047 | attachments: [], |
| 3048 | modelSelection: codexSelection, |
| 3049 | dispatchMode: { type: "start_immediately" }, |
| 3050 | }); |
| 3051 | const lateParentProjection = yield* waitForProjection( |
| 3052 | orchestrator, |
| 3053 | lateParentThreadId, |
| 3054 | (projection) => |
| 3055 | projection.runs.some((run) => run.status === "running") && |
| 3056 | projection.providerTurns.some((turn) => turn.status === "running"), |
| 3057 | ); |
| 3058 | const lateParentRun = lateParentProjection.runs.find((run) => run.status === "running"); |
| 3059 | if (lateParentRun?.rootNodeId === null || lateParentRun === undefined) { |
| 3060 | return yield* Effect.die(new Error("Late completion parent run missing.")); |
| 3061 | } |
| 3062 | const earlyLateDelivery = yield* orchestrator.dispatch({ |
| 3063 | type: "delegated_task.request", |
| 3064 | createdBy: "agent", |
| 3065 | creationSource: "mcp", |
| 3066 | commandId: CommandId.make("command:mcp-late-parent:early-task"), |
| 3067 | parentThreadId: lateParentThreadId, |
| 3068 | parentRunId: lateParentRun.id, |
| 3069 | parentNodeId: lateParentRun.rootNodeId, |
| 3070 | task: "Complete before the first delivery starts.", |
| 3071 | modelSelection: claudeSelection, |
| 3072 | runtimeMode: "full-access", |
| 3073 | interactionMode: "default", |
| 3074 | completionWake: "always", |
| 3075 | }); |
| 3076 | const lateChild = yield* orchestrator.dispatch({ |
| 3077 | type: "delegated_task.request", |
| 3078 | createdBy: "agent", |
| 3079 | creationSource: "mcp", |
| 3080 | commandId: CommandId.make("command:mcp-late-parent:late-task"), |
| 3081 | parentThreadId: lateParentThreadId, |
| 3082 | parentRunId: lateParentRun.id, |
| 3083 | parentNodeId: lateParentRun.rootNodeId, |
| 3084 | task: cancellationPrompt, |
| 3085 | modelSelection: codexSelection, |
| 3086 | runtimeMode: "full-access", |
| 3087 | interactionMode: "default", |
| 3088 | completionWake: "always", |
| 3089 | }); |
| 3090 | const taskFromDispatch = (result: typeof earlyLateDelivery) => { |
| 3091 | const taskEvent = result.storedEvents.find( |
| 3092 | (stored) => |
| 3093 | stored.event.type === "subagent.updated" && |
| 3094 | stored.event.payload.origin === "app_owned", |
| 3095 | ); |
| 3096 | if (taskEvent?.event.type !== "subagent.updated") { |
| 3097 | throw new Error("Late completion delegated task projection missing."); |
| 3098 | } |
| 3099 | return taskEvent.event.payload; |
| 3100 | }; |
| 3101 | const earlyLateTask = taskFromDispatch(earlyLateDelivery); |
| 3102 | const lateTask = taskFromDispatch(lateChild); |
| 3103 | const thirdLateChild = yield* orchestrator.dispatch({ |
| 3104 | type: "delegated_task.request", |
| 3105 | createdBy: "agent", |
| 3106 | creationSource: "mcp", |
| 3107 | commandId: CommandId.make("command:mcp-late-parent:third-late-task"), |
| 3108 | parentThreadId: lateParentThreadId, |
| 3109 | parentRunId: lateParentRun.id, |
| 3110 | parentNodeId: lateParentRun.rootNodeId, |
| 3111 | task: cancellationPrompt, |
| 3112 | modelSelection: codexSelection, |
| 3113 | runtimeMode: "full-access", |
| 3114 | interactionMode: "default", |
| 3115 | completionWake: "always", |
| 3116 | }); |
| 3117 | const thirdLateTask = taskFromDispatch(thirdLateChild); |
| 3118 | const fourthLateTask = taskFromDispatch( |
| 3119 | yield* orchestrator.dispatch({ |
| 3120 | type: "delegated_task.request", |
| 3121 | createdBy: "agent", |
| 3122 | creationSource: "mcp", |
| 3123 | commandId: CommandId.make("command:mcp-late-parent:fourth-late-task"), |
| 3124 | parentThreadId: lateParentThreadId, |
| 3125 | parentRunId: lateParentRun.id, |
| 3126 | parentNodeId: lateParentRun.rootNodeId, |
| 3127 | task: cancellationPrompt, |
| 3128 | modelSelection: codexSelection, |
| 3129 | runtimeMode: "full-access", |
| 3130 | interactionMode: "default", |
| 3131 | completionWake: "always", |
| 3132 | }), |
| 3133 | ); |
| 3134 | if ( |
| 3135 | lateTask.childThreadId === null || |
| 3136 | thirdLateTask.childThreadId === null || |
| 3137 | fourthLateTask.childThreadId === null |
| 3138 | ) { |
| 3139 | return yield* Effect.die(new Error("Late completion child thread missing.")); |
| 3140 | } |
| 3141 | const beforeFirstDelivery = yield* waitForProjection( |
| 3142 | orchestrator, |
| 3143 | lateParentThreadId, |
| 3144 | (projection) => |
| 3145 | projection.subagents.find((task) => task.id === earlyLateTask.id)?.status === |
| 3146 | "completed" && |
| 3147 | projection.subagents.find((task) => task.id === lateTask.id)?.status === |
| 3148 | "running" && |
| 3149 | projection.runs.find((run) => run.id === lateParentRun.id)?.delegatedCompletion |
| 3150 | ?.delivery !== null && |
| 3151 | projection.runs.find((run) => run.id === lateParentRun.id)?.delegatedCompletion |
| 3152 | ?.delivery !== undefined, |
| 3153 | ); |
| 3154 | const firstLateDelivery = beforeFirstDelivery.runs.find( |
| 3155 | (run) => run.id === lateParentRun.id, |
| 3156 | )?.delegatedCompletion?.delivery; |
| 3157 | if (firstLateDelivery === undefined || firstLateDelivery === null) { |
| 3158 | return yield* Effect.die(new Error("First late completion delivery missing.")); |
| 3159 | } |
| 3160 | yield* orchestrator.dispatch({ |
| 3161 | type: "message.dispatch", |
| 3162 | createdBy: "agent", |
| 3163 | creationSource: "server", |
| 3164 | commandId: CommandId.make("command:mcp-late-parent:dispatch-first-delivery"), |
| 3165 | threadId: lateParentThreadId, |
| 3166 | messageId: firstLateDelivery.messageId, |
| 3167 | text: "Delegated task reached a terminal state.", |
| 3168 | attachments: [], |
| 3169 | modelSelection: codexSelection, |
| 3170 | dispatchMode: { type: "queue_after_active" }, |
| 3171 | delegatedCompletion: { |
| 3172 | parentRunId: lateParentRun.id, |
| 3173 | generation: firstLateDelivery.generation, |
| 3174 | taskIds: firstLateDelivery.taskIds, |
| 3175 | }, |
| 3176 | }); |
| 3177 | yield* Deferred.succeed(lateParentGate, undefined); |
| 3178 | const activeFirstDelivery = yield* waitForProjection( |
| 3179 | orchestrator, |
| 3180 | lateParentThreadId, |
| 3181 | (projection) => |
| 3182 | projection.runs.some( |
| 3183 | (run) => |
| 3184 | run.userMessageId === firstLateDelivery.messageId && run.status === "running", |
| 3185 | ), |
| 3186 | ); |
| 3187 | const activeFirstDeliveryRun = activeFirstDelivery.runs.find( |
| 3188 | (run) => run.userMessageId === firstLateDelivery.messageId, |
| 3189 | ); |
| 3190 | if (activeFirstDeliveryRun === undefined) { |
| 3191 | return yield* Effect.die(new Error("First late completion delivery did not start.")); |
| 3192 | } |
| 3193 | const lateChildProjection = yield* orchestrator.getThreadProjection( |
| 3194 | lateTask.childThreadId, |
| 3195 | ); |
| 3196 | const lateChildRun = lateChildProjection.runs[0]; |
| 3197 | if (lateChildRun === undefined) { |
| 3198 | return yield* Effect.die(new Error("Late completion child run missing.")); |
| 3199 | } |
| 3200 | yield* orchestrator.dispatch({ |
| 3201 | type: "run.interrupt", |
| 3202 | commandId: CommandId.make("command:mcp-late-parent:interrupt-late-child"), |
| 3203 | threadId: lateTask.childThreadId, |
| 3204 | runId: lateChildRun.id, |
| 3205 | reason: "Terminalize after the first completion delivery started.", |
| 3206 | }); |
| 3207 | yield* waitForProjection( |
| 3208 | orchestrator, |
| 3209 | lateParentThreadId, |
| 3210 | (projection) => |
| 3211 | projection.subagents.find((task) => task.id === lateTask.id)?.completionDelivery |
| 3212 | ?.state === "pending", |
| 3213 | ); |
| 3214 | yield* Deferred.succeed(lateDeliveryGate, undefined); |
| 3215 | const successorReserved = yield* waitForProjection( |
| 3216 | orchestrator, |
| 3217 | lateParentThreadId, |
| 3218 | (projection) => { |
| 3219 | const delivery = projection.runs.find((run) => run.id === lateParentRun.id) |
| 3220 | ?.delegatedCompletion?.delivery; |
| 3221 | return ( |
| 3222 | delivery !== undefined && |
| 3223 | delivery !== null && |
| 3224 | delivery.generation === firstLateDelivery.generation + 1 && |
| 3225 | delivery.taskIds.length === 1 && |
| 3226 | delivery.taskIds[0] === lateTask.id |
| 3227 | ); |
| 3228 | }, |
| 3229 | ); |
| 3230 | const successorDelivery = successorReserved.runs.find( |
| 3231 | (run) => run.id === lateParentRun.id, |
| 3232 | )?.delegatedCompletion?.delivery; |
| 3233 | expect(successorDelivery).toMatchObject({ |
| 3234 | generation: firstLateDelivery.generation + 1, |
| 3235 | taskIds: [lateTask.id], |
| 3236 | }); |
| 3237 | const lateOffers = yield* waitForContinuationOffers(2); |
| 3238 | expect(lateOffers).toHaveLength(2); |
| 3239 | expect( |
| 3240 | successorReserved.runs.filter((run) => { |
| 3241 | const message = successorReserved.messages.find( |
| 3242 | (candidate) => candidate.id === run.userMessageId, |
| 3243 | ); |
| 3244 | return message?.delegatedCompletion?.parentRunId === lateParentRun.id; |
| 3245 | }), |
| 3246 | ).toHaveLength(1); |
| 3247 | if (successorDelivery === undefined || successorDelivery === null) { |
| 3248 | return yield* Effect.die(new Error("Late completion successor delivery missing.")); |
| 3249 | } |
| 3250 | |
| 3251 | // Results that land while the successor runs wait for it to settle, |
| 3252 | // then go out together in one more delivery. Nothing is left |
| 3253 | // pending once the parent has been told about every child. |
| 3254 | const successorGate = yield* Deferred.make<void>(); |
| 3255 | deliveryTerminalGates.set(lateParentThreadId, successorGate); |
| 3256 | yield* orchestrator.dispatch({ |
| 3257 | type: "message.dispatch", |
| 3258 | createdBy: "agent", |
| 3259 | creationSource: "server", |
| 3260 | commandId: CommandId.make("command:mcp-late-parent:dispatch-successor-delivery"), |
| 3261 | threadId: lateParentThreadId, |
| 3262 | messageId: successorDelivery.messageId, |
| 3263 | text: "Delegated task reached a terminal state.", |
| 3264 | attachments: [], |
| 3265 | modelSelection: codexSelection, |
| 3266 | dispatchMode: { type: "queue_after_active" }, |
| 3267 | delegatedCompletion: { |
| 3268 | parentRunId: lateParentRun.id, |
| 3269 | generation: successorDelivery.generation, |
| 3270 | taskIds: successorDelivery.taskIds, |
| 3271 | }, |
| 3272 | }); |
| 3273 | const activeSuccessor = yield* waitForProjection( |
| 3274 | orchestrator, |
| 3275 | lateParentThreadId, |
| 3276 | (projection) => |
| 3277 | projection.runs.some( |
| 3278 | (run) => |
| 3279 | run.userMessageId === successorDelivery.messageId && run.status === "running", |
| 3280 | ), |
| 3281 | ); |
| 3282 | const activeSuccessorRun = activeSuccessor.runs.find( |
| 3283 | (run) => run.userMessageId === successorDelivery.messageId, |
| 3284 | ); |
| 3285 | if (activeSuccessorRun === undefined) { |
| 3286 | return yield* Effect.die(new Error("Late completion successor did not start.")); |
| 3287 | } |
| 3288 | expect(activeSuccessor.turnItems).toEqual( |
| 3289 | expect.arrayContaining([ |
| 3290 | expect.objectContaining({ |
| 3291 | type: "notification", |
| 3292 | runId: activeSuccessorRun.id, |
| 3293 | source: { |
| 3294 | kind: "delegated_task", |
| 3295 | taskIds: successorDelivery.taskIds, |
| 3296 | childThreadId: lateTask.childThreadId, |
| 3297 | }, |
| 3298 | outcome: "cancelled", |
| 3299 | summary: `Delegated task "${cancellationPrompt}" stopped`, |
| 3300 | }), |
| 3301 | ]), |
| 3302 | ); |
| 3303 | expect( |
| 3304 | activeSuccessor.turnItems.some( |
| 3305 | (item) => |
| 3306 | item.type === "user_message" && item.messageId === successorDelivery.messageId, |
| 3307 | ), |
| 3308 | ).toBe(false); |
| 3309 | for (const [label, childThreadId] of [ |
| 3310 | ["third", thirdLateTask.childThreadId], |
| 3311 | ["fourth", fourthLateTask.childThreadId], |
| 3312 | ] as const) { |
| 3313 | const childProjection = yield* waitForProjection( |
| 3314 | orchestrator, |
| 3315 | childThreadId, |
| 3316 | (projection) => |
| 3317 | projection.runs.some((run) => run.status === "running") && |
| 3318 | projection.providerTurns.some((turn) => turn.status === "running"), |
| 3319 | ); |
| 3320 | const childRun = childProjection.runs.find((run) => run.status === "running"); |
| 3321 | if (childRun === undefined) { |
| 3322 | return yield* Effect.die(new Error(`${label} late completion child run missing.`)); |
| 3323 | } |
| 3324 | yield* orchestrator.dispatch({ |
| 3325 | type: "run.interrupt", |
| 3326 | commandId: CommandId.make(`command:mcp-late-parent:interrupt-${label}-late-child`), |
| 3327 | threadId: childThreadId, |
| 3328 | runId: childRun.id, |
| 3329 | reason: "Terminalize while the successor delivery is running.", |
| 3330 | }); |
| 3331 | } |
| 3332 | const pendingDuringSuccessor = yield* waitForProjection( |
| 3333 | orchestrator, |
| 3334 | lateParentThreadId, |
| 3335 | (projection) => |
| 3336 | [thirdLateTask.id, fourthLateTask.id].every( |
| 3337 | (taskId) => |
| 3338 | projection.subagents.find((task) => task.id === taskId)?.completionDelivery |
| 3339 | ?.state === "pending", |
| 3340 | ), |
| 3341 | ); |
| 3342 | // The running successor still owns the cohort's only reservation. |
| 3343 | expect( |
| 3344 | pendingDuringSuccessor.runs.find((run) => run.id === lateParentRun.id) |
| 3345 | ?.delegatedCompletion?.delivery, |
| 3346 | ).toMatchObject({ |
| 3347 | generation: successorDelivery.generation, |
| 3348 | taskIds: successorDelivery.taskIds, |
| 3349 | }); |
| 3350 | yield* expectOffersToStay(2); |
| 3351 | yield* Deferred.succeed(successorGate, undefined); |
| 3352 | const batchedReserved = yield* waitForProjection( |
| 3353 | orchestrator, |
| 3354 | lateParentThreadId, |
| 3355 | (projection) => { |
| 3356 | const delivery = projection.runs.find((run) => run.id === lateParentRun.id) |
| 3357 | ?.delegatedCompletion?.delivery; |
| 3358 | return ( |
| 3359 | delivery !== undefined && |
| 3360 | delivery !== null && |
| 3361 | delivery.generation === successorDelivery.generation + 1 && |
| 3362 | projection.runs.find((run) => run.id === activeSuccessorRun.id)?.status === |
| 3363 | "completed" |
| 3364 | ); |
| 3365 | }, |
| 3366 | ); |
| 3367 | const batchedDelivery = batchedReserved.runs.find((run) => run.id === lateParentRun.id) |
| 3368 | ?.delegatedCompletion?.delivery; |
| 3369 | if (batchedDelivery === undefined || batchedDelivery === null) { |
| 3370 | return yield* Effect.die(new Error("Batched late completion delivery missing.")); |
| 3371 | } |
| 3372 | expect([...batchedDelivery.taskIds].toSorted()).toEqual( |
| 3373 | [thirdLateTask.id, fourthLateTask.id].toSorted(), |
| 3374 | ); |
| 3375 | expect( |
| 3376 | [lateTask.id, thirdLateTask.id, fourthLateTask.id].map( |
| 3377 | (taskId) => |
| 3378 | batchedReserved.subagents.find((task) => task.id === taskId)?.completionDelivery |
| 3379 | ?.state, |
| 3380 | ), |
| 3381 | ).toEqual(["delivered", "claimed", "claimed"]); |
| 3382 | // One offer per delivery: first, successor, and this batch. |
| 3383 | yield* waitForContinuationOffers(3); |
| 3384 | yield* expectOffersToStay(3); |
| 3385 | expect( |
| 3386 | batchedReserved.runs.filter((run) => { |
| 3387 | const message = batchedReserved.messages.find( |
| 3388 | (candidate) => candidate.id === run.userMessageId, |
| 3389 | ); |
| 3390 | return ( |
| 3391 | message?.delegatedCompletion?.parentRunId === lateParentRun.id && |
| 3392 | run.status !== "completed" |
| 3393 | ); |
| 3394 | }), |
| 3395 | ).toHaveLength(0); |
| 3396 | yield* orchestrator.dispatch({ |
| 3397 | type: "message.dispatch", |
| 3398 | createdBy: "agent", |
| 3399 | creationSource: "server", |
| 3400 | commandId: CommandId.make("command:mcp-late-parent:dispatch-batched-delivery"), |
| 3401 | threadId: lateParentThreadId, |
| 3402 | messageId: batchedDelivery.messageId, |
| 3403 | text: "Delegated tasks reached terminal states.", |
| 3404 | attachments: [], |
| 3405 | modelSelection: codexSelection, |
| 3406 | dispatchMode: { type: "queue_after_active" }, |
| 3407 | delegatedCompletion: { |
| 3408 | parentRunId: lateParentRun.id, |
| 3409 | generation: batchedDelivery.generation, |
| 3410 | taskIds: batchedDelivery.taskIds, |
| 3411 | }, |
| 3412 | }); |
| 3413 | const drainedCohort = yield* waitForProjection( |
| 3414 | orchestrator, |
| 3415 | lateParentThreadId, |
| 3416 | (projection) => |
| 3417 | projection.runs.find((run) => run.id === lateParentRun.id)?.delegatedCompletion |
| 3418 | ?.delivery === null && |
| 3419 | projection.runs.some( |
| 3420 | (run) => |
| 3421 | run.userMessageId === batchedDelivery.messageId && run.status === "completed", |
| 3422 | ), |
| 3423 | ); |
| 3424 | expect( |
| 3425 | [earlyLateTask.id, lateTask.id, thirdLateTask.id, fourthLateTask.id].map( |
| 3426 | (taskId) => |
| 3427 | drainedCohort.subagents.find((task) => task.id === taskId)?.completionDelivery |
| 3428 | ?.state, |
| 3429 | ), |
| 3430 | ).toEqual(["delivered", "delivered", "delivered", "delivered"]); |
| 3431 | yield* expectOffersToStay(3); |
| 3432 | |
| 3433 | // Queue Remove is a durable disposal action, not a local queue |
| 3434 | // edit. Start a fresh parent-run cohort so removing this delivery |
| 3435 | // cannot interfere with the bounded late-delivery assertions. |
| 3436 | const removeParentGate = yield* Deferred.make<void>(); |
| 3437 | parentTerminalGates.set(lateParentThreadId, removeParentGate); |
| 3438 | const removeParentMessageId = MessageId.make("message:mcp-late-parent:remove-start"); |
| 3439 | yield* orchestrator.dispatch({ |
| 3440 | type: "message.dispatch", |
| 3441 | createdBy: "user", |
| 3442 | creationSource: "web", |
| 3443 | commandId: CommandId.make("command:mcp-late-parent:remove-start"), |
| 3444 | threadId: lateParentThreadId, |
| 3445 | messageId: removeParentMessageId, |
| 3446 | text: "Keep the parent active while removing automatic delivery.", |
| 3447 | attachments: [], |
| 3448 | modelSelection: codexSelection, |
| 3449 | dispatchMode: { type: "start_immediately" }, |
| 3450 | }); |
| 3451 | const removeParentProjection = yield* waitForProjection( |
| 3452 | orchestrator, |
| 3453 | lateParentThreadId, |
| 3454 | (projection) => |
| 3455 | projection.runs.some( |
| 3456 | (run) => run.userMessageId === removeParentMessageId && run.status === "running", |
| 3457 | ), |
| 3458 | ); |
| 3459 | const removeParentRun = removeParentProjection.runs.find( |
| 3460 | (run) => run.userMessageId === removeParentMessageId, |
| 3461 | ); |
| 3462 | if (removeParentRun?.rootNodeId === null || removeParentRun === undefined) { |
| 3463 | return yield* Effect.die(new Error("Queue Remove parent run missing.")); |
| 3464 | } |
| 3465 | const removeDelegation = yield* orchestrator.dispatch({ |
| 3466 | type: "delegated_task.request", |
| 3467 | createdBy: "agent", |
| 3468 | creationSource: "mcp", |
| 3469 | commandId: CommandId.make("command:mcp-late-parent:remove-delegate"), |
| 3470 | parentThreadId: lateParentThreadId, |
| 3471 | parentRunId: removeParentRun.id, |
| 3472 | parentNodeId: removeParentRun.rootNodeId, |
| 3473 | task: "Complete before Queue Remove disposes this automatic delivery.", |
| 3474 | modelSelection: claudeSelection, |
| 3475 | runtimeMode: "full-access", |
| 3476 | interactionMode: "default", |
| 3477 | completionWake: "always", |
| 3478 | }); |
| 3479 | const removeTaskEvent = removeDelegation.storedEvents.find( |
| 3480 | (stored) => |
| 3481 | stored.event.type === "subagent.updated" && |
| 3482 | stored.event.payload.origin === "app_owned", |
| 3483 | ); |
| 3484 | if (removeTaskEvent?.event.type !== "subagent.updated") { |
| 3485 | return yield* Effect.die( |
| 3486 | new Error("Queue Remove delegated task projection missing."), |
| 3487 | ); |
| 3488 | } |
| 3489 | const removeTask = removeTaskEvent.event.payload; |
| 3490 | const removeReserved = yield* waitForProjection( |
| 3491 | orchestrator, |
| 3492 | lateParentThreadId, |
| 3493 | (projection) => |
| 3494 | projection.subagents.find((task) => task.id === removeTask.id)?.status === |
| 3495 | "completed" && |
| 3496 | projection.runs.find((run) => run.id === removeParentRun.id)?.delegatedCompletion |
| 3497 | ?.delivery !== null && |
| 3498 | projection.runs.find((run) => run.id === removeParentRun.id)?.delegatedCompletion |
| 3499 | ?.delivery !== undefined, |
| 3500 | ); |
| 3501 | const removeDelivery = removeReserved.runs.find((run) => run.id === removeParentRun.id) |
| 3502 | ?.delegatedCompletion?.delivery; |
| 3503 | if (removeDelivery === undefined || removeDelivery === null) { |
| 3504 | return yield* Effect.die(new Error("Queue Remove delivery reservation missing.")); |
| 3505 | } |
| 3506 | yield* orchestrator.dispatch({ |
| 3507 | type: "message.dispatch", |
| 3508 | createdBy: "agent", |
| 3509 | creationSource: "server", |
| 3510 | commandId: CommandId.make("command:mcp-late-parent:remove-dispatch"), |
| 3511 | threadId: lateParentThreadId, |
| 3512 | messageId: removeDelivery.messageId, |
| 3513 | text: "Delegated task reached a terminal state.", |
| 3514 | attachments: [], |
| 3515 | modelSelection: codexSelection, |
| 3516 | dispatchMode: { type: "queue_after_active" }, |
| 3517 | delegatedCompletion: { |
| 3518 | parentRunId: removeParentRun.id, |
| 3519 | generation: removeDelivery.generation, |
| 3520 | taskIds: removeDelivery.taskIds, |
| 3521 | }, |
| 3522 | }); |
| 3523 | const removeQueued = yield* waitForProjection( |
| 3524 | orchestrator, |
| 3525 | lateParentThreadId, |
| 3526 | (projection) => |
| 3527 | projection.runs.some( |
| 3528 | (run) => |
| 3529 | run.userMessageId === removeDelivery.messageId && run.status === "queued", |
| 3530 | ), |
| 3531 | ); |
| 3532 | const removeQueuedRun = removeQueued.runs.find( |
| 3533 | (run) => run.userMessageId === removeDelivery.messageId, |
| 3534 | ); |
| 3535 | if (removeQueuedRun === undefined) { |
| 3536 | return yield* Effect.die(new Error("Queue Remove delivery did not queue.")); |
| 3537 | } |
| 3538 | yield* Ref.set(continuationOffers, []); |
| 3539 | yield* orchestrator.dispatch({ |
| 3540 | type: "queued-run.cancel", |
| 3541 | commandId: CommandId.make("command:mcp-late-parent:remove-queued-delivery"), |
| 3542 | threadId: lateParentThreadId, |
| 3543 | runId: removeQueuedRun.id, |
| 3544 | }); |
| 3545 | const removedDelivery = yield* waitForProjection( |
| 3546 | orchestrator, |
| 3547 | lateParentThreadId, |
| 3548 | (projection) => |
| 3549 | projection.runs.find((run) => run.id === removeParentRun.id)?.delegatedCompletion |
| 3550 | ?.disposition === "disposed" && |
| 3551 | projection.runs.find((run) => run.id === removeQueuedRun.id)?.status === |
| 3552 | "cancelled" && |
| 3553 | projection.subagents.find((task) => task.id === removeTask.id)?.completionDelivery |
| 3554 | ?.state === "disposed", |
| 3555 | ); |
| 3556 | expect( |
| 3557 | removedDelivery.runs.find((run) => run.id === removeParentRun.id) |
| 3558 | ?.delegatedCompletion, |
| 3559 | ).toMatchObject({ disposition: "disposed", delivery: null }); |
| 3560 | expect( |
| 3561 | removedDelivery.subagents.find((task) => task.id === removeTask.id), |
| 3562 | ).toMatchObject({ result: expect.any(String), status: "completed" }); |
| 3563 | yield* expectOffersToStay(0); |
| 3564 | |
| 3565 | // A wide fan-out keeps waking its parent until every result is |
| 3566 | // delivered. Two children finish before the first delivery starts, |
| 3567 | // five while it runs, three before the successor starts, and two |
| 3568 | // while the successor runs. Three batched deliveries cover all |
| 3569 | // twelve, and the cohort never has two deliveries outstanding. |
| 3570 | const fanoutParentThreadId = ThreadId.make("thread:mcp-fanout-parent"); |
| 3571 | const fanoutParentGate = yield* Deferred.make<void>(); |
| 3572 | const firstFanoutDeliveryGate = yield* Deferred.make<void>(); |
| 3573 | const secondFanoutDeliveryGate = yield* Deferred.make<void>(); |
| 3574 | parentTerminalGates.set(fanoutParentThreadId, fanoutParentGate); |
| 3575 | deliveryTerminalGates.set(fanoutParentThreadId, firstFanoutDeliveryGate); |
| 3576 | yield* Ref.set(continuationOffers, []); |
| 3577 | yield* orchestrator.dispatch({ |
| 3578 | type: "thread.create", |
| 3579 | createdBy: "user", |
| 3580 | creationSource: "web", |
| 3581 | commandId: CommandId.make("command:mcp-fanout-parent:create"), |
| 3582 | threadId: fanoutParentThreadId, |
| 3583 | projectId, |
| 3584 | title: "Fan-out parent", |
| 3585 | modelSelection: codexSelection, |
| 3586 | runtimeMode: "full-access", |
| 3587 | interactionMode: "default", |
| 3588 | branch: null, |
| 3589 | worktreePath: cwd, |
| 3590 | }); |
| 3591 | yield* orchestrator.dispatch({ |
| 3592 | type: "message.dispatch", |
| 3593 | createdBy: "user", |
| 3594 | creationSource: "web", |
| 3595 | commandId: CommandId.make("command:mcp-fanout-parent:start"), |
| 3596 | threadId: fanoutParentThreadId, |
| 3597 | messageId: MessageId.make("message:mcp-fanout-parent:start"), |
| 3598 | text: "Fan out twelve review agents.", |
| 3599 | attachments: [], |
| 3600 | modelSelection: codexSelection, |
| 3601 | dispatchMode: { type: "start_immediately" }, |
| 3602 | }); |
| 3603 | const fanoutStarted = yield* waitForProjection( |
| 3604 | orchestrator, |
| 3605 | fanoutParentThreadId, |
| 3606 | (projection) => |
| 3607 | projection.runs.some((run) => run.status === "running") && |
| 3608 | projection.providerTurns.some((turn) => turn.status === "running"), |
| 3609 | ); |
| 3610 | const fanoutRun = fanoutStarted.runs.find((run) => run.status === "running"); |
| 3611 | if (fanoutRun === undefined || fanoutRun.rootNodeId === null) { |
| 3612 | return yield* Effect.die(new Error("Fan-out parent run missing.")); |
| 3613 | } |
| 3614 | const fanoutRootNodeId = fanoutRun.rootNodeId; |
| 3615 | const fanoutTasks = yield* Effect.forEach( |
| 3616 | Array.from({ length: 12 }, (_, index) => index), |
| 3617 | (index) => |
| 3618 | orchestrator |
| 3619 | .dispatch({ |
| 3620 | type: "delegated_task.request", |
| 3621 | createdBy: "agent", |
| 3622 | creationSource: "mcp", |
| 3623 | commandId: CommandId.make(`command:mcp-fanout-parent:task-${index}`), |
| 3624 | parentThreadId: fanoutParentThreadId, |
| 3625 | parentRunId: fanoutRun.id, |
| 3626 | parentNodeId: fanoutRootNodeId, |
| 3627 | task: cancellationPrompt, |
| 3628 | modelSelection: codexSelection, |
| 3629 | runtimeMode: "full-access", |
| 3630 | interactionMode: "default", |
| 3631 | completionWake: "always", |
| 3632 | }) |
| 3633 | .pipe(Effect.map(taskFromDispatch)), |
| 3634 | ); |
| 3635 | type FanoutTask = (typeof fanoutTasks)[number]; |
| 3636 | const finishFanoutChildren = (tasks: ReadonlyArray<FanoutTask>) => |
| 3637 | Effect.forEach( |
| 3638 | tasks, |
| 3639 | (task) => |
| 3640 | Effect.gen(function* () { |
| 3641 | const childThreadId = task.childThreadId; |
| 3642 | if (childThreadId === null) { |
| 3643 | return yield* Effect.die(new Error("Fan-out child thread missing.")); |
| 3644 | } |
| 3645 | const child = yield* waitForProjection( |
| 3646 | orchestrator, |
| 3647 | childThreadId, |
| 3648 | (projection) => |
| 3649 | projection.runs.some((run) => run.status === "running") && |
| 3650 | projection.providerTurns.some((turn) => turn.status === "running"), |
| 3651 | ); |
| 3652 | const childRun = child.runs.find((run) => run.status === "running"); |
| 3653 | if (childRun === undefined) { |
| 3654 | return yield* Effect.die(new Error("Fan-out child run missing.")); |
| 3655 | } |
| 3656 | yield* orchestrator.dispatch({ |
| 3657 | type: "run.interrupt", |
| 3658 | commandId: CommandId.make(`command:mcp-fanout-parent:finish-${task.id}`), |
| 3659 | threadId: childThreadId, |
| 3660 | runId: childRun.id, |
| 3661 | reason: "Finish this fan-out child.", |
| 3662 | }); |
| 3663 | }), |
| 3664 | { discard: true }, |
| 3665 | ); |
| 3666 | const fanoutDelivery = (projection: OrchestrationV2ThreadProjection) => |
| 3667 | projection.runs.find((run) => run.id === fanoutRun.id)?.delegatedCompletion |
| 3668 | ?.delivery ?? null; |
| 3669 | const fanoutDeliveryRuns = (projection: OrchestrationV2ThreadProjection) => |
| 3670 | projection.runs.filter( |
| 3671 | (run) => |
| 3672 | projection.messages.find((message) => message.id === run.userMessageId) |
| 3673 | ?.delegatedCompletion?.parentRunId === fanoutRun.id, |
| 3674 | ); |
| 3675 | // The cohort holds one reservation, and at most one of its |
| 3676 | // delivery runs is queued or running at a time. |
| 3677 | const expectAtMostOneOutstandingDelivery = ( |
| 3678 | projection: OrchestrationV2ThreadProjection, |
| 3679 | ) => |
| 3680 | expect( |
| 3681 | fanoutDeliveryRuns(projection).filter( |
| 3682 | (run) => |
| 3683 | run.status !== "completed" && |
| 3684 | run.status !== "failed" && |
| 3685 | run.status !== "cancelled" && |
| 3686 | run.status !== "interrupted", |
| 3687 | ).length, |
| 3688 | ).toBeLessThanOrEqual(1); |
| 3689 | const deliveryStates = ( |
| 3690 | projection: OrchestrationV2ThreadProjection, |
| 3691 | tasks: ReadonlyArray<FanoutTask>, |
| 3692 | ) => |
| 3693 | tasks.map( |
| 3694 | (task) => |
| 3695 | projection.subagents.find((candidate) => candidate.id === task.id) |
| 3696 | ?.completionDelivery?.state, |
| 3697 | ); |
| 3698 | const sortedIds = (ids: ReadonlyArray<string>) => [...ids].toSorted(); |
| 3699 | const idsOf = (tasks: ReadonlyArray<FanoutTask>) => |
| 3700 | sortedIds(tasks.map((task) => task.id)); |
| 3701 | const dispatchFanoutDelivery = ( |
| 3702 | label: string, |
| 3703 | delivery: OrchestrationV2DelegatedCompletionDelivery, |
| 3704 | ) => |
| 3705 | orchestrator.dispatch({ |
| 3706 | type: "message.dispatch", |
| 3707 | createdBy: "agent", |
| 3708 | creationSource: "server", |
| 3709 | commandId: CommandId.make(`command:mcp-fanout-parent:dispatch-${label}`), |
| 3710 | threadId: fanoutParentThreadId, |
| 3711 | messageId: delivery.messageId, |
| 3712 | text: "Delegated tasks reached terminal states.", |
| 3713 | attachments: [], |
| 3714 | modelSelection: codexSelection, |
| 3715 | dispatchMode: { type: "queue_after_active" }, |
| 3716 | delegatedCompletion: { |
| 3717 | parentRunId: fanoutRun.id, |
| 3718 | generation: delivery.generation, |
| 3719 | taskIds: delivery.taskIds, |
| 3720 | }, |
| 3721 | }); |
| 3722 | const deliveryRunStatus = ( |
| 3723 | projection: OrchestrationV2ThreadProjection, |
| 3724 | delivery: OrchestrationV2DelegatedCompletionDelivery, |
| 3725 | ) => projection.runs.find((run) => run.userMessageId === delivery.messageId)?.status; |
| 3726 | |
| 3727 | const beforeFirstStarts = fanoutTasks.slice(0, 2); |
| 3728 | const duringFirst = fanoutTasks.slice(2, 7); |
| 3729 | const beforeSecondStarts = fanoutTasks.slice(7, 10); |
| 3730 | const duringSecond = fanoutTasks.slice(10, 12); |
| 3731 | |
| 3732 | yield* finishFanoutChildren(beforeFirstStarts); |
| 3733 | const firstReserved = yield* waitForProjection( |
| 3734 | orchestrator, |
| 3735 | fanoutParentThreadId, |
| 3736 | (projection) => |
| 3737 | deliveryStates(projection, beforeFirstStarts).every( |
| 3738 | (state) => state === "claimed", |
| 3739 | ) && fanoutDelivery(projection)?.taskIds.length === 2, |
| 3740 | ); |
| 3741 | const firstFanoutDelivery = fanoutDelivery(firstReserved); |
| 3742 | if (firstFanoutDelivery === null) { |
| 3743 | return yield* Effect.die(new Error("First fan-out delivery missing.")); |
| 3744 | } |
| 3745 | expect(sortedIds(firstFanoutDelivery.taskIds)).toEqual(idsOf(beforeFirstStarts)); |
| 3746 | yield* dispatchFanoutDelivery("first", firstFanoutDelivery); |
| 3747 | yield* Deferred.succeed(fanoutParentGate, undefined); |
| 3748 | yield* waitForProjection( |
| 3749 | orchestrator, |
| 3750 | fanoutParentThreadId, |
| 3751 | (projection) => deliveryRunStatus(projection, firstFanoutDelivery) === "running", |
| 3752 | ); |
| 3753 | |
| 3754 | yield* finishFanoutChildren(duringFirst); |
| 3755 | const pendingDuringFirst = yield* waitForProjection( |
| 3756 | orchestrator, |
| 3757 | fanoutParentThreadId, |
| 3758 | (projection) => |
| 3759 | deliveryStates(projection, duringFirst).every((state) => state === "pending"), |
| 3760 | ); |
| 3761 | expect(fanoutDelivery(pendingDuringFirst)).toEqual(firstFanoutDelivery); |
| 3762 | expectAtMostOneOutstandingDelivery(pendingDuringFirst); |
| 3763 | |
| 3764 | deliveryTerminalGates.set(fanoutParentThreadId, secondFanoutDeliveryGate); |
| 3765 | yield* Deferred.succeed(firstFanoutDeliveryGate, undefined); |
| 3766 | const secondReserved = yield* waitForProjection( |
| 3767 | orchestrator, |
| 3768 | fanoutParentThreadId, |
| 3769 | (projection) => |
| 3770 | deliveryRunStatus(projection, firstFanoutDelivery) === "completed" && |
| 3771 | fanoutDelivery(projection)?.generation === firstFanoutDelivery.generation + 1, |
| 3772 | ); |
| 3773 | expect(sortedIds(fanoutDelivery(secondReserved)?.taskIds ?? [])).toEqual( |
| 3774 | idsOf(duringFirst), |
| 3775 | ); |
| 3776 | expect(deliveryStates(secondReserved, beforeFirstStarts)).toEqual([ |
| 3777 | "delivered", |
| 3778 | "delivered", |
| 3779 | ]); |
| 3780 | |
| 3781 | // Results that land before the successor starts join it. |
| 3782 | yield* finishFanoutChildren(beforeSecondStarts); |
| 3783 | const secondJoined = yield* waitForProjection( |
| 3784 | orchestrator, |
| 3785 | fanoutParentThreadId, |
| 3786 | (projection) => |
| 3787 | deliveryStates(projection, beforeSecondStarts).every( |
| 3788 | (state) => state === "claimed", |
| 3789 | ) && fanoutDelivery(projection)?.taskIds.length === 8, |
| 3790 | ); |
| 3791 | const secondFanoutDelivery = fanoutDelivery(secondJoined); |
| 3792 | if (secondFanoutDelivery === null) { |
| 3793 | return yield* Effect.die(new Error("Second fan-out delivery missing.")); |
| 3794 | } |
| 3795 | expect(sortedIds(secondFanoutDelivery.taskIds)).toEqual( |
| 3796 | idsOf([...duringFirst, ...beforeSecondStarts]), |
| 3797 | ); |
| 3798 | expectAtMostOneOutstandingDelivery(secondJoined); |
| 3799 | yield* dispatchFanoutDelivery("second", secondFanoutDelivery); |
| 3800 | yield* waitForProjection( |
| 3801 | orchestrator, |
| 3802 | fanoutParentThreadId, |
| 3803 | (projection) => deliveryRunStatus(projection, secondFanoutDelivery) === "running", |
| 3804 | ); |
| 3805 | |
| 3806 | yield* finishFanoutChildren(duringSecond); |
| 3807 | const pendingDuringSecond = yield* waitForProjection( |
| 3808 | orchestrator, |
| 3809 | fanoutParentThreadId, |
| 3810 | (projection) => |
| 3811 | deliveryStates(projection, duringSecond).every((state) => state === "pending"), |
| 3812 | ); |
| 3813 | expectAtMostOneOutstandingDelivery(pendingDuringSecond); |
| 3814 | |
| 3815 | // Later deliveries complete as soon as they start. |
| 3816 | deliveryTerminalGates.delete(fanoutParentThreadId); |
| 3817 | yield* Deferred.succeed(secondFanoutDeliveryGate, undefined); |
| 3818 | const thirdReserved = yield* waitForProjection( |
| 3819 | orchestrator, |
| 3820 | fanoutParentThreadId, |
| 3821 | (projection) => |
| 3822 | deliveryRunStatus(projection, secondFanoutDelivery) === "completed" && |
| 3823 | fanoutDelivery(projection)?.generation === secondFanoutDelivery.generation + 1, |
| 3824 | ); |
| 3825 | const thirdFanoutDelivery = fanoutDelivery(thirdReserved); |
| 3826 | if (thirdFanoutDelivery === null) { |
| 3827 | return yield* Effect.die(new Error("Third fan-out delivery missing.")); |
| 3828 | } |
| 3829 | expect(sortedIds(thirdFanoutDelivery.taskIds)).toEqual(idsOf(duringSecond)); |
| 3830 | yield* dispatchFanoutDelivery("third", thirdFanoutDelivery); |
| 3831 | const fanoutDrained = yield* waitForProjection( |
| 3832 | orchestrator, |
| 3833 | fanoutParentThreadId, |
| 3834 | (projection) => |
| 3835 | deliveryRunStatus(projection, thirdFanoutDelivery) === "completed" && |
| 3836 | fanoutDelivery(projection) === null, |
| 3837 | ); |
| 3838 | expect(deliveryStates(fanoutDrained, fanoutTasks)).toEqual( |
| 3839 | fanoutTasks.map(() => "delivered"), |
| 3840 | ); |
| 3841 | expect( |
| 3842 | fanoutTasks.map( |
| 3843 | (task) => |
| 3844 | fanoutDrained.subagents.find((candidate) => candidate.id === task.id)?.status, |
| 3845 | ), |
| 3846 | ).toEqual(fanoutTasks.map(() => "interrupted")); |
| 3847 | const drainedDeliveryRuns = fanoutDeliveryRuns(fanoutDrained); |
| 3848 | expect(drainedDeliveryRuns.map((run) => run.status)).toEqual([ |
| 3849 | "completed", |
| 3850 | "completed", |
| 3851 | "completed", |
| 3852 | ]); |
| 3853 | const deliveredTaskIds = drainedDeliveryRuns.flatMap( |
| 3854 | (run) => |
| 3855 | fanoutDrained.messages.find((message) => message.id === run.userMessageId) |
| 3856 | ?.delegatedCompletion?.taskIds ?? [], |
| 3857 | ); |
| 3858 | expect(sortedIds(deliveredTaskIds)).toEqual(idsOf(fanoutTasks)); |
| 3859 | const offeredDeliveries = new Set( |
| 3860 | (yield* Ref.get(continuationOffers)).map( |
| 3861 | (offer) => offer.delegatedCompletion?.messageId, |
| 3862 | ), |
| 3863 | ); |
| 3864 | expect(offeredDeliveries).toEqual( |
| 3865 | new Set([ |
| 3866 | firstFanoutDelivery.messageId, |
| 3867 | secondFanoutDelivery.messageId, |
| 3868 | thirdFanoutDelivery.messageId, |
| 3869 | ]), |
| 3870 | ); |
| 3871 | }).pipe(Effect.provide(layerTest)); |
| 3872 | }), |
| 3873 | ), |
| 3874 | ); |
| 3875 | |
| 3876 | it.live("reports running and queued child follow-ups from a Codex replay transcript", () => |
| 3877 | Effect.scoped( |
| 3878 | Effect.gen(function* () { |
| 3879 | const cwd = yield* checkpointWorkspace("delegated-task-status"); |
| 3880 | const rawTranscript = yield* readDelegatedTaskStatusTranscript(); |
| 3881 | const transcript = yield* CodexOrchestratorReplayHarness.decodeTranscript( |
| 3882 | materializeReplayTranscriptWorkspace(rawTranscript, cwd), |
| 3883 | ); |
| 3884 | const layerOrchestrator = ProviderReplayHarness.layerProviderReplay( |
| 3885 | { |
| 3886 | name: "delegated-task-status/codex", |
| 3887 | transcript, |
| 3888 | commands: [], |
| 3889 | runtimePolicyOverride: { cwd }, |
| 3890 | }, |
| 3891 | CodexOrchestratorReplayHarness, |
| 3892 | ); |
| 3893 | const layerOrchestration = Layer.merge( |
| 3894 | layerOrchestrator, |
| 3895 | ThreadManagementService.layer.pipe( |
| 3896 | Layer.provide(layerOrchestrator), |
| 3897 | Layer.provide(ServerSettings.layerTest()), |
| 3898 | Layer.provide( |
| 3899 | CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)), |
| 3900 | ), |
| 3901 | ), |
| 3902 | ); |
| 3903 | const layerProviderRegistry = ProviderRegistryMock.layer([ |
| 3904 | makeProviderSnapshot({ |
| 3905 | instanceId: codexInstanceId, |
| 3906 | driver: ProviderDriverKind.make("codex"), |
| 3907 | model: codexModel, |
| 3908 | }), |
| 3909 | ]); |
| 3910 | const layerTest = layerOrchestratorToolkit.pipe( |
| 3911 | Layer.provideMerge(McpServer.McpServer.layer), |
| 3912 | Layer.provideMerge(layerOrchestration), |
| 3913 | Layer.provide( |
| 3914 | CodexOrchestratorReplayHarness.makeProviderAdapterRegistryLayer(transcript).pipe( |
| 3915 | Layer.provide( |
| 3916 | Layer.merge(McpProviderSessions.layer, ScriptProgress.layerUnavailable), |
| 3917 | ), |
| 3918 | ), |
| 3919 | ), |
| 3920 | Layer.provide(layerProviderRegistry), |
| 3921 | Layer.provide(ServerSettings.layerTest()), |
| 3922 | Layer.provide(layerUnusedScheduledTaskStub), |
| 3923 | Layer.provide(Layer.mock(ProjectService.ProjectService)({})), |
| 3924 | Layer.provideMerge( |
| 3925 | SecretRequests.layer.pipe( |
| 3926 | Layer.provide(layerMemorySecretStore), |
| 3927 | Layer.provide(layerOrchestration), |
| 3928 | ), |
| 3929 | ), |
| 3930 | Layer.provide(NodeServices.layer), |
| 3931 | ); |
| 3932 | |
| 3933 | yield* Effect.gen(function* () { |
| 3934 | const orchestrator = yield* Orchestrator.OrchestratorV2; |
| 3935 | const server = yield* McpServer.McpServer; |
| 3936 | const parentCreate = yield* orchestrator.dispatch({ |
| 3937 | type: "thread.create", |
| 3938 | createdBy: "user", |
| 3939 | creationSource: "web", |
| 3940 | commandId: CommandId.make("command:mcp-replay-parent:create"), |
| 3941 | threadId: parentThreadId, |
| 3942 | projectId, |
| 3943 | title: "MCP replay parent", |
| 3944 | modelSelection: codexSelection, |
| 3945 | runtimeMode: "full-access", |
| 3946 | interactionMode: "default", |
| 3947 | branch: null, |
| 3948 | worktreePath: cwd, |
| 3949 | }); |
| 3950 | yield* orchestrator.dispatch({ |
| 3951 | type: "message.dispatch", |
| 3952 | createdBy: "user", |
| 3953 | creationSource: "web", |
| 3954 | commandId: CommandId.make("command:mcp-replay-parent:start"), |
| 3955 | threadId: parentThreadId, |
| 3956 | messageId: MessageId.make("message:mcp-replay-parent:start"), |
| 3957 | text: parentPrompt, |
| 3958 | attachments: [], |
| 3959 | modelSelection: codexSelection, |
| 3960 | dispatchMode: { type: "start_immediately" }, |
| 3961 | }); |
| 3962 | yield* orchestrator |
| 3963 | .streamStoredEventsFrom({ |
| 3964 | threadId: parentThreadId, |
| 3965 | afterSequence: parentCreate.sequence, |
| 3966 | }) |
| 3967 | .pipe( |
| 3968 | Stream.filter( |
| 3969 | (stored) => |
| 3970 | stored.event.type === "provider-turn.updated" && |
| 3971 | stored.event.payload.status === "running", |
| 3972 | ), |
| 3973 | Stream.runHead, |
| 3974 | Effect.flatMap( |
| 3975 | Option.match({ |
| 3976 | onNone: () => Effect.die("Parent provider turn did not start."), |
| 3977 | onSome: () => Effect.void, |
| 3978 | }), |
| 3979 | ), |
| 3980 | ); |
| 3981 | |
| 3982 | const invocation: McpInvocationContext.McpInvocationScope = { |
| 3983 | environmentId: EnvironmentId.make("environment:mcp-replay"), |
| 3984 | requestNamespace: "mcp-provider-session-replay-parent", |
| 3985 | thread: { |
| 3986 | threadId: parentThreadId, |
| 3987 | providerSessionId: "mcp-provider-session-replay-parent", |
| 3988 | providerInstanceId: codexInstanceId, |
| 3989 | }, |
| 3990 | client: undefined, |
| 3991 | capabilities: new Set(["orchestration"]), |
| 3992 | issuedAt: 1, |
| 3993 | }; |
| 3994 | const invoke = (name: string, args: Record<string, unknown>) => |
| 3995 | server |
| 3996 | .callTool({ name, arguments: args }) |
| 3997 | .pipe( |
| 3998 | Effect.provideService(McpInvocationContext.McpInvocationContext, invocation), |
| 3999 | Effect.provideService(McpSchema.McpServerClient, client), |
| 4000 | ); |
| 4001 | |
| 4002 | const delegationStartSequence = |
| 4003 | yield* orchestrator.getThreadEventSequence(parentThreadId); |
| 4004 | const delegatedCall = yield* invoke("delegate_task", { |
| 4005 | task: delegatedPrompt, |
| 4006 | target: { |
| 4007 | providerInstanceId: codexInstanceId, |
| 4008 | model: codexModel, |
| 4009 | }, |
| 4010 | mode: "async", |
| 4011 | clientRequestId: "delegate-codex-replay-1", |
| 4012 | }); |
| 4013 | const delegatedStart = yield* decodeDelegateTaskResult( |
| 4014 | delegatedCall.structuredContent, |
| 4015 | ).pipe(Effect.orDie); |
| 4016 | expect(delegatedStart.childRunId).not.toBeNull(); |
| 4017 | yield* orchestrator |
| 4018 | .streamStoredEventsFrom({ |
| 4019 | threadId: parentThreadId, |
| 4020 | afterSequence: delegationStartSequence, |
| 4021 | }) |
| 4022 | .pipe( |
| 4023 | Stream.filter( |
| 4024 | (stored) => |
| 4025 | stored.event.type === "context-transfer.created" && |
| 4026 | stored.event.payload.type === "subagent_result" && |
| 4027 | stored.event.payload.sourceThreadId === delegatedStart.childThreadId && |
| 4028 | stored.event.payload.sourcePoint.runId === delegatedStart.childRunId, |
| 4029 | ), |
| 4030 | Stream.runHead, |
| 4031 | Effect.flatMap( |
| 4032 | Option.match({ |
| 4033 | onNone: () => Effect.die("Delegated result transfer was not created."), |
| 4034 | onSome: () => Effect.void, |
| 4035 | }), |
| 4036 | ), |
| 4037 | ); |
| 4038 | const delegatedStatusCall = yield* invoke("task_status", { |
| 4039 | taskId: delegatedStart.taskId, |
| 4040 | }); |
| 4041 | const delegated = yield* decodeDelegateTaskResult( |
| 4042 | delegatedStatusCall.structuredContent, |
| 4043 | ).pipe(Effect.orDie); |
| 4044 | expect(delegated).toMatchObject({ |
| 4045 | status: "completed", |
| 4046 | summary: delegatedResult, |
| 4047 | providerInstanceId: codexInstanceId, |
| 4048 | hasPendingChildRuns: false, |
| 4049 | latestTerminalRunId: delegated.childRunId, |
| 4050 | latestTerminalStatus: "completed", |
| 4051 | latestTerminalSummary: delegatedResult, |
| 4052 | }); |
| 4053 | expect(delegated.resultContextTransferId).not.toBeNull(); |
| 4054 | expect(delegated.latestTerminalResultContextTransferId).toBe( |
| 4055 | delegated.resultContextTransferId, |
| 4056 | ); |
| 4057 | |
| 4058 | // Delegated children are subagent threads too, but T3 owns them, so |
| 4059 | // they keep taking follow-ups (provider-native children do not). |
| 4060 | const delegatedChild = yield* orchestrator.getThreadProjection(delegated.childThreadId); |
| 4061 | expect(delegatedChild.thread.lineage.relationshipToParent).toBe("subagent"); |
| 4062 | expect(isProviderNativeSubagentThread(delegatedChild.thread)).toBe(false); |
| 4063 | const followupStartSequence = yield* orchestrator.getThreadEventSequence( |
| 4064 | delegated.childThreadId, |
| 4065 | ); |
| 4066 | const runningFollowupCall = yield* invoke("t3_thread_send", { |
| 4067 | threadId: delegated.childThreadId, |
| 4068 | message: cancellationPrompt, |
| 4069 | clientRequestId: "delegated-child-replay-running-1", |
| 4070 | }); |
| 4071 | const runningFollowup = yield* decodeThreadSendResult( |
| 4072 | runningFollowupCall.structuredContent, |
| 4073 | ).pipe(Effect.orDie); |
| 4074 | yield* orchestrator |
| 4075 | .streamStoredEventsFrom({ |
| 4076 | threadId: delegated.childThreadId, |
| 4077 | afterSequence: followupStartSequence, |
| 4078 | }) |
| 4079 | .pipe( |
| 4080 | Stream.filter( |
| 4081 | (stored) => |
| 4082 | stored.event.type === "provider-turn.updated" && |
| 4083 | stored.event.payload.status === "running", |
| 4084 | ), |
| 4085 | Stream.runHead, |
| 4086 | Effect.flatMap( |
| 4087 | Option.match({ |
| 4088 | onNone: () => Effect.die("Follow-up provider turn did not start."), |
| 4089 | onSome: () => Effect.void, |
| 4090 | }), |
| 4091 | ), |
| 4092 | ); |
| 4093 | |
| 4094 | const queuedFollowupCall = yield* invoke("t3_thread_send", { |
| 4095 | threadId: delegated.childThreadId, |
| 4096 | message: queuedFollowupPrompt, |
| 4097 | mode: "queue", |
| 4098 | clientRequestId: "delegated-child-replay-queued-1", |
| 4099 | }); |
| 4100 | const queuedFollowup = yield* decodeThreadSendResult( |
| 4101 | queuedFollowupCall.structuredContent, |
| 4102 | ).pipe(Effect.orDie); |
| 4103 | expect(queuedFollowup).toMatchObject({ |
| 4104 | status: "queued", |
| 4105 | delivery: "queued", |
| 4106 | }); |
| 4107 | |
| 4108 | const pendingProjection = yield* orchestrator.getThreadProjection( |
| 4109 | delegated.childThreadId, |
| 4110 | ); |
| 4111 | const followupTasks = (yield* orchestrator.getThreadProjection( |
| 4112 | parentThreadId, |
| 4113 | )).subagents.filter( |
| 4114 | (task) => |
| 4115 | task.childThreadId === delegated.childThreadId && task.id !== delegated.taskId, |
| 4116 | ); |
| 4117 | expect(followupTasks).toHaveLength(2); |
| 4118 | expect(followupTasks.every((task) => task.result === null)).toBe(true); |
| 4119 | expect(followupTasks.map((task) => task.id)).toEqual([ |
| 4120 | runningFollowup.taskId, |
| 4121 | queuedFollowup.taskId, |
| 4122 | ]); |
| 4123 | expect( |
| 4124 | pendingProjection.runs.find((run) => run.id === delegated.childRunId)?.status, |
| 4125 | ).toBe("completed"); |
| 4126 | expect( |
| 4127 | pendingProjection.runs.find((run) => run.id === runningFollowup.runId)?.status, |
| 4128 | ).toBe("running"); |
| 4129 | expect( |
| 4130 | pendingProjection.runs.find((run) => run.id === queuedFollowup.runId)?.status, |
| 4131 | ).toBe("queued"); |
| 4132 | expect( |
| 4133 | (yield* orchestrator.getThreadProjection(parentThreadId)).runs.some( |
| 4134 | (run) => run.status === "running", |
| 4135 | ), |
| 4136 | ).toBe(true); |
| 4137 | |
| 4138 | const pendingStatusCall = yield* invoke("task_status", { |
| 4139 | taskId: delegated.taskId, |
| 4140 | }); |
| 4141 | const pendingStatus = yield* decodeDelegateTaskResult( |
| 4142 | pendingStatusCall.structuredContent, |
| 4143 | ).pipe(Effect.orDie); |
| 4144 | expect(pendingStatus).toMatchObject({ |
| 4145 | childRunId: delegated.childRunId, |
| 4146 | status: "completed", |
| 4147 | summary: delegatedResult, |
| 4148 | resultContextTransferId: delegated.resultContextTransferId, |
| 4149 | hasPendingChildRuns: true, |
| 4150 | latestTerminalRunId: delegated.childRunId, |
| 4151 | latestTerminalStatus: "completed", |
| 4152 | latestTerminalSummary: delegatedResult, |
| 4153 | latestTerminalResultContextTransferId: delegated.resultContextTransferId, |
| 4154 | }); |
| 4155 | |
| 4156 | const finalSequence = yield* orchestrator.getThreadEventSequence(delegated.childThreadId); |
| 4157 | const interruptCall = yield* invoke("t3_thread_interrupt", { |
| 4158 | threadId: delegated.childThreadId, |
| 4159 | runId: runningFollowup.runId, |
| 4160 | reason: "Allow the queued replay follow-up to run.", |
| 4161 | clientRequestId: "interrupt-delegated-child-replay-1", |
| 4162 | }); |
| 4163 | const interrupt = yield* decodeThreadInterruptResult( |
| 4164 | interruptCall.structuredContent, |
| 4165 | ).pipe(Effect.orDie); |
| 4166 | expect(interrupt).toMatchObject({ |
| 4167 | runId: runningFollowup.runId, |
| 4168 | status: "interrupt_requested", |
| 4169 | }); |
| 4170 | yield* orchestrator |
| 4171 | .streamStoredEventsFrom({ |
| 4172 | threadId: delegated.childThreadId, |
| 4173 | afterSequence: finalSequence, |
| 4174 | }) |
| 4175 | .pipe( |
| 4176 | Stream.filter( |
| 4177 | (stored) => |
| 4178 | stored.event.type === "run.updated" && |
| 4179 | stored.event.payload.id === queuedFollowup.runId && |
| 4180 | stored.event.payload.status === "completed", |
| 4181 | ), |
| 4182 | Stream.runHead, |
| 4183 | Effect.flatMap( |
| 4184 | Option.match({ |
| 4185 | onNone: () => Effect.die("Queued follow-up did not complete."), |
| 4186 | onSome: () => Effect.void, |
| 4187 | }), |
| 4188 | ), |
| 4189 | ); |
| 4190 | |
| 4191 | const finalProjection = yield* orchestrator.getThreadProjection(delegated.childThreadId); |
| 4192 | expect(finalProjection.runs.find((run) => run.id === runningFollowup.runId)?.status).toBe( |
| 4193 | "interrupted", |
| 4194 | ); |
| 4195 | expect(finalProjection.runs.find((run) => run.id === queuedFollowup.runId)?.status).toBe( |
| 4196 | "completed", |
| 4197 | ); |
| 4198 | yield* orchestrator |
| 4199 | .streamStoredEventsFrom({ |
| 4200 | threadId: parentThreadId, |
| 4201 | afterSequence: finalSequence, |
| 4202 | }) |
| 4203 | .pipe( |
| 4204 | Stream.filter( |
| 4205 | (stored) => |
| 4206 | stored.event.type === "subagent.updated" && |
| 4207 | stored.event.payload.id === followupTasks[1]!.id && |
| 4208 | stored.event.payload.status === "completed", |
| 4209 | ), |
| 4210 | Stream.runHead, |
| 4211 | Effect.flatMap( |
| 4212 | Option.match({ |
| 4213 | onNone: () => Effect.die("Follow-up task did not settle."), |
| 4214 | onSome: () => Effect.void, |
| 4215 | }), |
| 4216 | ), |
| 4217 | ); |
| 4218 | const finishedParent = yield* orchestrator.getThreadProjection(parentThreadId); |
| 4219 | expect( |
| 4220 | finishedParent.subagents.find((task) => task.id === followupTasks[0]!.id)?.status, |
| 4221 | ).toBe("interrupted"); |
| 4222 | expect( |
| 4223 | finishedParent.subagents.find((task) => task.id === followupTasks[1]!.id), |
| 4224 | ).toMatchObject({ |
| 4225 | status: "completed", |
| 4226 | result: queuedFollowupResult, |
| 4227 | }); |
| 4228 | expect( |
| 4229 | finishedParent.contextTransfers.filter( |
| 4230 | (transfer) => |
| 4231 | transfer.type === "subagent_result" && |
| 4232 | transfer.sourceThreadId === delegated.childThreadId, |
| 4233 | ), |
| 4234 | ).toHaveLength(3); |
| 4235 | expect( |
| 4236 | finishedParent.subagents.find((task) => task.id === followupTasks[1]!.id) |
| 4237 | ?.completionDelivery?.state, |
| 4238 | ).toBe("claimed"); |
| 4239 | const finalStatusCall = yield* invoke("task_status", { |
| 4240 | taskId: delegated.taskId, |
| 4241 | }); |
| 4242 | const finalStatus = yield* decodeDelegateTaskResult( |
| 4243 | finalStatusCall.structuredContent, |
| 4244 | ).pipe(Effect.orDie); |
| 4245 | expect(finalStatus).toMatchObject({ |
| 4246 | childRunId: delegated.childRunId, |
| 4247 | status: "completed", |
| 4248 | summary: delegatedResult, |
| 4249 | resultContextTransferId: delegated.resultContextTransferId, |
| 4250 | hasPendingChildRuns: false, |
| 4251 | latestTerminalRunId: queuedFollowup.runId, |
| 4252 | latestTerminalStatus: "completed", |
| 4253 | latestTerminalSummary: queuedFollowupResult, |
| 4254 | }); |
| 4255 | expect(finalStatus.latestTerminalResultContextTransferId).not.toBeNull(); |
| 4256 | const read = yield* invoke("t3_thread_read", { |
| 4257 | threadId: delegated.childThreadId, |
| 4258 | limit: 100, |
| 4259 | }); |
| 4260 | expect(read.isError).not.toBe(true); |
| 4261 | const afterRead = yield* orchestrator.getThreadProjection(parentThreadId); |
| 4262 | expect( |
| 4263 | afterRead.subagents.find((task) => task.id === queuedFollowup.taskId) |
| 4264 | ?.completionDelivery?.state, |
| 4265 | ).toBe("acknowledged"); |
| 4266 | }).pipe(Effect.provide(layerTest)); |
| 4267 | }), |
| 4268 | ), |
| 4269 | ); |
| 4270 | }); |