| 1 | import * as Scheduler from "../scheduling/Scheduler.ts"; |
| 2 | import * as WorktreeSetupTracker from "../project/WorktreeSetupTracker.ts"; |
| 3 | import * as ProjectCloneTracker from "../project/ProjectCloneTracker.ts"; |
| 4 | import * as TerminalManager from "../terminal/Manager.ts"; |
| 5 | import * as NodeServices from "@effect/platform-node/NodeServices"; |
| 6 | import * as FileSystem from "effect/FileSystem"; |
| 7 | import * as ServerConfig from "../config.ts"; |
| 8 | import { createPendingAttachmentId, resolveAttachmentPath } from "../attachmentStore.ts"; |
| 9 | import * as ThreadMessageIntake from "./ThreadMessageIntake.ts"; |
| 10 | import { assert, it, vi } from "@effect/vitest"; |
| 11 | import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; |
| 12 | import { |
| 13 | ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE, |
| 14 | ChatAttachmentId, |
| 15 | ComposerContextId, |
| 16 | type ChatAttachment, |
| 17 | CommandId, |
| 18 | EnvironmentId, |
| 19 | EventId, |
| 20 | DEFAULT_SERVER_SETTINGS, |
| 21 | GitCommandError, |
| 22 | MessageId, |
| 23 | NodeId, |
| 24 | ProjectId, |
| 25 | ProviderDriverKind, |
| 26 | ProviderInstanceId, |
| 27 | OrchestrationV2ThreadProjectionJson, |
| 28 | ScheduledTaskId, |
| 29 | type ServerProvider, |
| 30 | ThreadId, |
| 31 | TurnItemId, |
| 32 | } from "@t3tools/contracts"; |
| 33 | import * as Cause from "effect/Cause"; |
| 34 | import * as Deferred from "effect/Deferred"; |
| 35 | import * as DateTime from "effect/DateTime"; |
| 36 | import * as Duration from "effect/Duration"; |
| 37 | import * as Effect from "effect/Effect"; |
| 38 | import * as Exit from "effect/Exit"; |
| 39 | import * as Layer from "effect/Layer"; |
| 40 | import * as Option from "effect/Option"; |
| 41 | import * as Ref from "effect/Ref"; |
| 42 | import * as Stream from "effect/Stream"; |
| 43 | import * as Schema from "effect/Schema"; |
| 44 | import * as TestClock from "effect/testing/TestClock"; |
| 45 | |
| 46 | import * as GitWorkflow from "../git/GitWorkflowService.ts"; |
| 47 | import * as SqlitePersistence from "../persistence/Sqlite.ts"; |
| 48 | import * as ProjectStore from "./ProjectStore.ts"; |
| 49 | import * as ProjectService from "../project/ProjectService.ts"; |
| 50 | import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts"; |
| 51 | import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts"; |
| 52 | import * as ProviderRegistryMock from "../provider/testUtils/providerRegistryMock.ts"; |
| 53 | import * as ServerSettings from "../serverSettings.ts"; |
| 54 | import * as ScheduledTasks from "../scheduledTasks/ScheduledTaskService.ts"; |
| 55 | import * as SecretRequests from "../secrets/SecretRequests.ts"; |
| 56 | import * as TextGeneration from "../textGeneration/TextGeneration.ts"; |
| 57 | import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; |
| 58 | import * as CommandReceiptStore from "./CommandReceiptStore.ts"; |
| 59 | import * as EffectOutbox from "./EffectOutbox.ts"; |
| 60 | import * as EventSink from "./EventSink.ts"; |
| 61 | import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; |
| 62 | import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; |
| 63 | import * as ThreadLaunch from "./ThreadLaunchService.ts"; |
| 64 | import * as OrchestratorMcpService from "../mcp/OrchestratorMcpService.ts"; |
| 65 | import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; |
| 66 | import * as ThreadManagement from "./ThreadManagementService.ts"; |
| 67 | import * as ThreadTitleRegeneration from "./ThreadTitleRegenerationService.ts"; |
| 68 | import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; |
| 69 | import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; |
| 70 | |
| 71 | const projectId = ProjectId.make("project:launch-test"); |
| 72 | const otherProjectId = ProjectId.make("project:launch-other"); |
| 73 | const encodeThreadProjection = Schema.encodeEffect(OrchestrationV2ThreadProjectionJson); |
| 74 | const modelSelection = { |
| 75 | instanceId: ProviderInstanceId.make("codex"), |
| 76 | model: "gpt-5.1-codex", |
| 77 | } as const; |
| 78 | const project = { |
| 79 | id: projectId, |
| 80 | title: "Project", |
| 81 | workspaceRoot: "/repo", |
| 82 | repositoryIdentity: null, |
| 83 | faviconPath: null, |
| 84 | defaultModelSelection: modelSelection, |
| 85 | defaultThreadEnvMode: null, |
| 86 | scripts: [], |
| 87 | createdAt: "2026-06-20T00:00:00.000Z", |
| 88 | updatedAt: "2026-06-20T00:00:00.000Z", |
| 89 | deletedAt: null, |
| 90 | } as const; |
| 91 | |
| 92 | const otherProject = { |
| 93 | ...project, |
| 94 | id: otherProjectId, |
| 95 | title: "Other", |
| 96 | } as const; |
| 97 | |
| 98 | const adapter = { |
| 99 | instanceId: modelSelection.instanceId, |
| 100 | driver: ProviderDriverKind.make("codex"), |
| 101 | getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), |
| 102 | planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }), |
| 103 | openSession: () => Effect.die("provider execution is disabled in launch tests"), |
| 104 | } as ProviderAdapter.ProviderAdapterV2["Service"]; |
| 105 | |
| 106 | interface HarnessOptions { |
| 107 | readonly closeTerminal?: TerminalManager.TerminalManager["Service"]["close"]; |
| 108 | readonly adapters?: ReadonlyArray<ProviderAdapter.ProviderAdapterV2["Service"]>; |
| 109 | readonly managedFolders?: Layer.Layer<ManagedProjectFolders.ManagedProjectFolders>; |
| 110 | readonly createWorktree?: GitWorkflow.GitWorkflowService["Service"]["createWorktree"]; |
| 111 | readonly fetchRemote?: GitWorkflow.GitWorkflowService["Service"]["fetchRemote"]; |
| 112 | readonly hasCommit?: GitWorkflow.GitWorkflowService["Service"]["hasCommit"]; |
| 113 | readonly renameBranch?: GitWorkflow.GitWorkflowService["Service"]["renameBranch"]; |
| 114 | readonly runSetup?: ProjectSetupScriptRunner.ProjectSetupScriptRunner["Service"]["runForThread"]; |
| 115 | readonly generateTitle?: TextGeneration.TextGeneration["Service"]["generateThreadTitle"]; |
| 116 | readonly generateBranchName?: TextGeneration.TextGeneration["Service"]["generateBranchName"]; |
| 117 | readonly serverSettings?: Parameters<typeof ServerSettings.layerTest>[0]; |
| 118 | readonly providers?: ReadonlyArray<ServerProvider>; |
| 119 | readonly git?: Partial<GitWorkflow.GitWorkflowService["Service"]>; |
| 120 | } |
| 121 | |
| 122 | function makeHarness(options: HarnessOptions = {}) { |
| 123 | const layerDatabase = SqlitePersistence.layerMemory; |
| 124 | const layerRegistry = ProviderAdapterRegistry.layerFromAdapters(options.adapters ?? [adapter]); |
| 125 | const layerOrchestrator = ProviderReplayHarness.layerWithRegistry( |
| 126 | { name: "thread-launch" }, |
| 127 | layerRegistry, |
| 128 | { databaseLayer: layerDatabase, runEffectWorker: false }, |
| 129 | ); |
| 130 | const layerServerSettings = ServerSettings.layerTest(options.serverSettings); |
| 131 | const layerReceipts = CommandReceiptStore.layer.pipe(Layer.provide(layerDatabase)); |
| 132 | const layerThreadManagement = ThreadManagement.layer.pipe( |
| 133 | Layer.provide(layerOrchestrator), |
| 134 | Layer.provide(layerServerSettings), |
| 135 | Layer.provide(layerReceipts), |
| 136 | ); |
| 137 | const layerOutbox = EffectOutbox.layer.pipe(Layer.provide(layerDatabase)); |
| 138 | const createWorktree = vi.fn( |
| 139 | options.createWorktree ?? |
| 140 | ((input) => |
| 141 | Effect.succeed({ |
| 142 | worktree: { path: "/repo-worktrees/feature", refName: input.newRefName, headSha: "abc" }, |
| 143 | } as never)), |
| 144 | ); |
| 145 | const renameBranch = vi.fn( |
| 146 | options.renameBranch ?? ((input) => Effect.succeed({ branch: input.newBranch })), |
| 147 | ); |
| 148 | const removeWorktree = vi.fn( |
| 149 | (_input: Parameters<GitWorkflow.GitWorkflowService["Service"]["removeWorktree"]>[0]) => |
| 150 | Effect.void, |
| 151 | ); |
| 152 | const runSetup = vi.fn( |
| 153 | options.runSetup ?? (() => Effect.succeed({ status: "no-script" as const })), |
| 154 | ); |
| 155 | const closeTerminal = vi.fn(options.closeTerminal ?? (() => Effect.void)); |
| 156 | const generateBranchName = vi.fn( |
| 157 | options.generateBranchName ?? (() => Effect.succeed({ branch: "generated-branch" })), |
| 158 | ); |
| 159 | const generateThreadTitle = vi.fn( |
| 160 | options.generateTitle ?? (() => Effect.succeed({ title: "Generated title" })), |
| 161 | ); |
| 162 | const layerExternalServices = Layer.mergeAll( |
| 163 | WorktreeSetupTracker.layer, |
| 164 | Layer.mock(ProjectCloneTracker.ProjectCloneTracker)({ get: () => Effect.succeed(null) }), |
| 165 | Layer.mock(TerminalManager.TerminalManager)({ close: closeTerminal }), |
| 166 | Layer.succeed(ProjectService.ProjectService, { |
| 167 | create: () => Effect.die("unused"), |
| 168 | bootstrap: () => Effect.die("unused"), |
| 169 | update: () => Effect.die("unused"), |
| 170 | delete: () => Effect.die("unused"), |
| 171 | getById: (id) => |
| 172 | Effect.succeed( |
| 173 | id === projectId |
| 174 | ? Option.some(project) |
| 175 | : id === otherProjectId |
| 176 | ? Option.some(otherProject) |
| 177 | : Option.none(), |
| 178 | ), |
| 179 | getByWorkspaceRoot: () => Effect.succeed(Option.some(project)), |
| 180 | snapshot: Effect.die("unused"), |
| 181 | getShell: () => Effect.die("unused"), |
| 182 | listShells: () => Effect.die("unused"), |
| 183 | }), |
| 184 | Layer.mock(GitWorkflow.GitWorkflowService)({ |
| 185 | createWorktree, |
| 186 | renameBranch, |
| 187 | fetchRemote: options.fetchRemote ?? (() => Effect.void), |
| 188 | hasCommit: options.hasCommit ?? (() => Effect.succeed(false)), |
| 189 | remoteExists: () => Effect.succeed(true), |
| 190 | remoteBranchExists: () => Effect.succeed(true), |
| 191 | removeWorktree, |
| 192 | resolveRemoteTrackingCommit: () => |
| 193 | Effect.succeed({ commitSha: "remote-main-sha", remoteRefName: "origin/main" }), |
| 194 | ...options.git, |
| 195 | }), |
| 196 | Layer.succeed(ProjectSetupScriptRunner.ProjectSetupScriptRunner, { |
| 197 | runForThread: runSetup, |
| 198 | }), |
| 199 | Layer.mock(TextGeneration.TextGeneration)({ |
| 200 | generateThreadTitle, |
| 201 | generateBranchName, |
| 202 | }), |
| 203 | layerServerSettings, |
| 204 | ProviderRegistryMock.layer(options.providers), |
| 205 | options.managedFolders ?? |
| 206 | Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ |
| 207 | namedProjectsRoot: "/projects", |
| 208 | folderForThread: () => Effect.succeed(Option.none()), |
| 209 | }), |
| 210 | ); |
| 211 | const layerLaunch = ThreadLaunch.layer.pipe( |
| 212 | Layer.provide( |
| 213 | Layer.mergeAll( |
| 214 | layerExternalServices, |
| 215 | layerThreadManagement, |
| 216 | layerReceipts, |
| 217 | IdAllocator.layer, |
| 218 | ), |
| 219 | ), |
| 220 | ); |
| 221 | const layerProjectedProjects = Layer.mock(ProjectStore.ProjectStoreV2)({ |
| 222 | get: (requestedProjectId) => |
| 223 | Effect.succeed( |
| 224 | requestedProjectId === projectId |
| 225 | ? Option.some({ |
| 226 | projectId, |
| 227 | title: project.title, |
| 228 | workspaceRoot: project.workspaceRoot, |
| 229 | defaultModelSelection: project.defaultModelSelection, |
| 230 | defaultThreadEnvMode: null, |
| 231 | autoPull: false, |
| 232 | faviconPath: null, |
| 233 | projectIcon: null, |
| 234 | scripts: project.scripts, |
| 235 | createdAt: project.createdAt, |
| 236 | updatedAt: project.updatedAt, |
| 237 | deletedAt: project.deletedAt, |
| 238 | }) |
| 239 | : Option.none(), |
| 240 | ), |
| 241 | }); |
| 242 | const layerTitleRegeneration = ThreadTitleRegeneration.layer.pipe( |
| 243 | Layer.provide( |
| 244 | Layer.mergeAll(layerThreadManagement, layerProjectedProjects, layerExternalServices), |
| 245 | ), |
| 246 | ); |
| 247 | return { |
| 248 | layer: Layer.mergeAll( |
| 249 | layerLaunch, |
| 250 | layerRegistry, |
| 251 | layerThreadManagement, |
| 252 | layerTitleRegeneration, |
| 253 | layerOutbox, |
| 254 | layerDatabase, |
| 255 | layerOrchestrator, |
| 256 | layerExternalServices, |
| 257 | ), |
| 258 | createWorktree, |
| 259 | removeWorktree, |
| 260 | renameBranch, |
| 261 | generateBranchName, |
| 262 | generateThreadTitle, |
| 263 | runSetup, |
| 264 | closeTerminal, |
| 265 | }; |
| 266 | } |
| 267 | |
| 268 | function launchInput(input: { |
| 269 | readonly command: string; |
| 270 | readonly thread: string; |
| 271 | readonly message?: string; |
| 272 | readonly workspace?: ThreadLaunch.ThreadLaunchWorkspaceStrategy; |
| 273 | }) { |
| 274 | return { |
| 275 | commandId: CommandId.make(input.command), |
| 276 | threadId: ThreadId.make(input.thread), |
| 277 | projectId, |
| 278 | title: "New thread", |
| 279 | modelSelection, |
| 280 | runtimeMode: "full-access" as const, |
| 281 | interactionMode: "default" as const, |
| 282 | workspaceStrategy: input.workspace ?? { type: "root" as const }, |
| 283 | ...(input.message === undefined |
| 284 | ? {} |
| 285 | : { |
| 286 | initialMessage: { |
| 287 | messageId: MessageId.make(`${input.message}:id`), |
| 288 | text: input.message, |
| 289 | attachments: [], |
| 290 | }, |
| 291 | }), |
| 292 | createdBy: "user" as const, |
| 293 | creationSource: "web" as const, |
| 294 | }; |
| 295 | } |
| 296 | |
| 297 | function waitUntil<E, R>(predicate: () => Effect.Effect<boolean, E, R>): Effect.Effect<void, E, R> { |
| 298 | return Effect.gen(function* () { |
| 299 | for (let attempt = 0; attempt < 200; attempt += 1) { |
| 300 | if (yield* predicate()) return; |
| 301 | yield* Effect.promise( |
| 302 | () => |
| 303 | new Promise<void>((resolve) => { |
| 304 | setImmediate(resolve); |
| 305 | }), |
| 306 | ); |
| 307 | } |
| 308 | assert.fail("Condition was not reached before timeout."); |
| 309 | }); |
| 310 | } |
| 311 | |
| 312 | it.effect.each( |
| 313 | (["new", "existing"] as const).flatMap((target) => |
| 314 | (["user", "agent"] as const).map((createdBy) => ({ target, createdBy })), |
| 315 | ), |
| 316 | )( |
| 317 | "attributes $createdBy-configured automations in $target threads without changing their prompt", |
| 318 | ({ target, createdBy }) => { |
| 319 | const harness = makeHarness(); |
| 320 | const layerScheduledTasks = ScheduledTasks.layer.pipe( |
| 321 | Layer.provide( |
| 322 | Layer.mergeAll( |
| 323 | harness.layer, |
| 324 | NodeCrypto.layer, |
| 325 | Scheduler.layer, |
| 326 | Layer.mock(SecretRequests.SecretRequests)({}), |
| 327 | ), |
| 328 | ), |
| 329 | ); |
| 330 | return Effect.gen(function* () { |
| 331 | const tasks = yield* ScheduledTasks.ScheduledTaskService; |
| 332 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 333 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 334 | const existing = |
| 335 | target === "existing" |
| 336 | ? yield* launches.launch( |
| 337 | launchInput({ command: "command:existing", thread: "thread:existing" }), |
| 338 | ) |
| 339 | : null; |
| 340 | const { task } = yield* tasks.upsert({ |
| 341 | id: ScheduledTaskId.make("scheduled-task:attribution"), |
| 342 | title: "Daily audit", |
| 343 | prompt: "Audit performance and crashes.", |
| 344 | enabled: false, |
| 345 | schedule: { type: "interval", everyMs: 60_000 }, |
| 346 | projectId, |
| 347 | threadId: existing?.threadId ?? null, |
| 348 | workspaceStrategy: { type: "root" }, |
| 349 | modelSelection, |
| 350 | runtimeMode: "full-access", |
| 351 | interactionMode: "default", |
| 352 | createdBy, |
| 353 | creationSource: createdBy === "agent" ? "mcp" : "web", |
| 354 | }); |
| 355 | const result = yield* tasks.runNow({ id: task.id }); |
| 356 | assert.equal(result.task.lastRunStatus, "succeeded"); |
| 357 | const projectThreads = yield* threads.listProjectThreads({ |
| 358 | projectId, |
| 359 | includeSubagents: false, |
| 360 | }); |
| 361 | const thread = |
| 362 | projectThreads.find((candidate) => candidate.id === existing?.threadId) ?? |
| 363 | projectThreads[0]; |
| 364 | assert.isDefined(thread); |
| 365 | const projection = yield* threads.getThreadProjection(thread!.id); |
| 366 | // Encoding the persisted projection exercises both message and turn-item wire schemas. |
| 367 | const wire = yield* encodeThreadProjection(projection); |
| 368 | assert.equal(wire.messages[0]?.text, task.prompt); |
| 369 | assert.equal(wire.messages[0]?.scheduledTaskId, task.id); |
| 370 | assert.equal(wire.messages[0]?.createdBy, createdBy); |
| 371 | const turnItem = wire.turnItems.find( |
| 372 | (item): item is Extract<typeof item, { type: "user_message" }> => |
| 373 | item.type === "user_message", |
| 374 | ); |
| 375 | assert.equal(turnItem?.text, task.prompt); |
| 376 | assert.equal(turnItem?.scheduledTaskId, task.id); |
| 377 | }).pipe(Effect.provide(Layer.mergeAll(harness.layer, layerScheduledTasks))); |
| 378 | }, |
| 379 | ); |
| 380 | |
| 381 | it.effect("retains automation and sender attribution while a message waits in the queue", () => { |
| 382 | const harness = makeHarness({ runSetup: () => Effect.never }); |
| 383 | return Effect.gen(function* () { |
| 384 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 385 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 386 | const launched = yield* launches.launch( |
| 387 | launchInput({ |
| 388 | command: "command:automation:queue", |
| 389 | thread: "thread:automation:queue", |
| 390 | message: "First message", |
| 391 | }), |
| 392 | ); |
| 393 | const scheduledTaskId = ScheduledTaskId.make("scheduled-task:queued"); |
| 394 | const senderThreadId = ThreadId.make("thread:agent-sender"); |
| 395 | const queued = yield* threads.sendToThread({ |
| 396 | projectId, |
| 397 | commandId: CommandId.make("command:automation:queued"), |
| 398 | threadId: launched.threadId, |
| 399 | messageId: MessageId.make("message:automation:queued"), |
| 400 | scheduledTaskId, |
| 401 | senderThreadId, |
| 402 | text: "Run the audit", |
| 403 | attachments: [], |
| 404 | mode: "queue", |
| 405 | createdBy: "agent", |
| 406 | creationSource: "mcp", |
| 407 | }); |
| 408 | assert.equal(queued.delivery, "queued"); |
| 409 | const projection = yield* threads.getThreadProjection(launched.threadId); |
| 410 | const message = projection.messages.find((item) => item.id === queued.message.id); |
| 411 | assert.equal(message?.scheduledTaskId, scheduledTaskId); |
| 412 | assert.equal(message?.senderThreadId, senderThreadId); |
| 413 | assert.equal(message?.text, "Run the audit"); |
| 414 | }).pipe(Effect.provide(harness.layer)); |
| 415 | }); |
| 416 | |
| 417 | it.effect("returns a visible preparing message while provisioning is still blocked", () => |
| 418 | Effect.gen(function* () { |
| 419 | const worktreeEntered = yield* Deferred.make<void>(); |
| 420 | const allowWorktree = yield* Deferred.make<void>(); |
| 421 | const setupEntered = yield* Deferred.make<void>(); |
| 422 | const allowSetup = yield* Deferred.make<void>(); |
| 423 | const harness = makeHarness({ |
| 424 | createWorktree: () => |
| 425 | Deferred.succeed(worktreeEntered, undefined).pipe( |
| 426 | Effect.andThen(Deferred.await(allowWorktree)), |
| 427 | Effect.as({ |
| 428 | worktree: { path: "/repo-worktrees/feature", refName: "feature", headSha: "abc" }, |
| 429 | } as never), |
| 430 | ), |
| 431 | runSetup: () => |
| 432 | Deferred.succeed(setupEntered, undefined).pipe( |
| 433 | Effect.andThen(Deferred.await(allowSetup)), |
| 434 | Effect.as({ status: "no-script" as const }), |
| 435 | ), |
| 436 | }); |
| 437 | yield* Effect.gen(function* () { |
| 438 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 439 | const outbox = yield* EffectOutbox.EffectOutboxV2; |
| 440 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 441 | const input = launchInput({ |
| 442 | command: "command:launch:blocked", |
| 443 | thread: "thread:launch:blocked", |
| 444 | message: "Build the feature", |
| 445 | workspace: { type: "worktree", baseRef: "main" }, |
| 446 | }); |
| 447 | const launched = yield* launches.launch(input); |
| 448 | assert.equal(launched.projection.messages[0]?.text, "Build the feature"); |
| 449 | assert.equal(launched.projection.runs[0]?.status, "preparing"); |
| 450 | assert.equal( |
| 451 | launched.projection.turnItems.find((item) => item.type === "command_execution")?.status, |
| 452 | "running", |
| 453 | ); |
| 454 | yield* Deferred.await(worktreeEntered); |
| 455 | let current = yield* threads.getThreadProjection(launched.threadId); |
| 456 | assert.equal( |
| 457 | current.turnItems.find((item) => item.type === "command_execution")?.title, |
| 458 | "Preparing worktree", |
| 459 | ); |
| 460 | yield* Deferred.succeed(allowWorktree, undefined); |
| 461 | const entered = yield* Deferred.await(setupEntered).pipe( |
| 462 | Effect.timeoutOption(Duration.seconds(2)), |
| 463 | ); |
| 464 | if (Option.isNone(entered)) { |
| 465 | current = yield* threads.getThreadProjection(launched.threadId); |
| 466 | assert.fail( |
| 467 | `Setup was not reached; run=${current.runs[0]?.status ?? "missing"}, worklog=${current.turnItems.find((item) => item.type === "command_execution")?.title ?? "missing"}.`, |
| 468 | ); |
| 469 | } |
| 470 | current = yield* threads.getThreadProjection(launched.threadId); |
| 471 | assert.equal( |
| 472 | current.turnItems.find((item) => item.type === "command_execution")?.title, |
| 473 | "Starting setup script", |
| 474 | ); |
| 475 | const prematureEffects = yield* outbox.listByCommandId( |
| 476 | CommandId.make("command:launch:blocked:initial-message"), |
| 477 | ); |
| 478 | assert.isEmpty(prematureEffects); |
| 479 | yield* Deferred.succeed(allowSetup, undefined); |
| 480 | }).pipe(Effect.provide(harness.layer)); |
| 481 | }), |
| 482 | ); |
| 483 | |
| 484 | it.effect("provisions independent launches concurrently instead of behind a global semaphore", () => |
| 485 | Effect.gen(function* () { |
| 486 | const setupCount = yield* Ref.make(0); |
| 487 | const bothEntered = yield* Deferred.make<void>(); |
| 488 | const allowSetup = yield* Deferred.make<void>(); |
| 489 | const harness = makeHarness({ |
| 490 | runSetup: () => |
| 491 | Ref.updateAndGet(setupCount, (count) => count + 1).pipe( |
| 492 | Effect.tap((count) => |
| 493 | count === 2 ? Deferred.succeed(bothEntered, undefined) : Effect.void, |
| 494 | ), |
| 495 | Effect.andThen(Deferred.await(allowSetup)), |
| 496 | Effect.as({ status: "no-script" as const }), |
| 497 | ), |
| 498 | }); |
| 499 | yield* Effect.gen(function* () { |
| 500 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 501 | const results = yield* Effect.all( |
| 502 | [ |
| 503 | launches.launch( |
| 504 | launchInput({ |
| 505 | command: "command:launch:concurrent-a", |
| 506 | thread: "thread:launch:concurrent-a", |
| 507 | message: "First", |
| 508 | }), |
| 509 | ), |
| 510 | launches.launch( |
| 511 | launchInput({ |
| 512 | command: "command:launch:concurrent-b", |
| 513 | thread: "thread:launch:concurrent-b", |
| 514 | message: "Second", |
| 515 | }), |
| 516 | ), |
| 517 | ], |
| 518 | { concurrency: "unbounded" }, |
| 519 | ); |
| 520 | assert.deepEqual( |
| 521 | results.map((result) => result.projection.runs[0]?.status), |
| 522 | ["preparing", "preparing"], |
| 523 | ); |
| 524 | yield* Deferred.await(bothEntered); |
| 525 | assert.equal(yield* Ref.get(setupCount), 2); |
| 526 | yield* Deferred.succeed(allowSetup, undefined); |
| 527 | }).pipe(Effect.provide(harness.layer)); |
| 528 | }), |
| 529 | ); |
| 530 | |
| 531 | it.effect("enqueues provider work only after setup has been initiated", () => |
| 532 | Effect.gen(function* () { |
| 533 | const setupEntered = yield* Deferred.make<void>(); |
| 534 | const allowSetup = yield* Deferred.make<void>(); |
| 535 | const harness = makeHarness({ |
| 536 | runSetup: () => |
| 537 | Deferred.succeed(setupEntered, undefined).pipe( |
| 538 | Effect.andThen(Deferred.await(allowSetup)), |
| 539 | Effect.as({ |
| 540 | status: "started" as const, |
| 541 | async: false, |
| 542 | scriptId: "setup", |
| 543 | scriptName: "Setup", |
| 544 | scriptCommand: "vp install", |
| 545 | terminalId: "setup", |
| 546 | cwd: "/repo", |
| 547 | }), |
| 548 | ), |
| 549 | }); |
| 550 | yield* Effect.gen(function* () { |
| 551 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 552 | const outbox = yield* EffectOutbox.EffectOutboxV2; |
| 553 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 554 | const input = launchInput({ |
| 555 | command: "command:launch:release", |
| 556 | thread: "thread:launch:release", |
| 557 | message: "Start after setup", |
| 558 | workspace: { type: "worktree", baseRef: "main" }, |
| 559 | }); |
| 560 | const launched = yield* launches.launch(input); |
| 561 | yield* Deferred.await(setupEntered); |
| 562 | assert.isEmpty( |
| 563 | yield* outbox.listByCommandId(CommandId.make("command:launch:release:release")), |
| 564 | ); |
| 565 | yield* Deferred.succeed(allowSetup, undefined); |
| 566 | yield* waitUntil(() => |
| 567 | outbox |
| 568 | .listByCommandId(CommandId.make("command:launch:release:release")) |
| 569 | .pipe(Effect.map((effects) => effects.length === 1)), |
| 570 | ); |
| 571 | const projection = yield* threads.getThreadProjection(launched.threadId); |
| 572 | assert.equal(projection.runs[0]?.status, "starting"); |
| 573 | assert.equal(projection.checkpointScopes[0]?.cwd, "/repo-worktrees/feature"); |
| 574 | assert.equal( |
| 575 | projection.turnItems.find((item) => item.type === "command_execution")?.status, |
| 576 | "completed", |
| 577 | ); |
| 578 | }).pipe(Effect.provide(harness.layer)); |
| 579 | }), |
| 580 | ); |
| 581 | |
| 582 | it.effect( |
| 583 | "queues follow-up messages behind preparation and checkpoints them in the final workspace", |
| 584 | () => |
| 585 | Effect.gen(function* () { |
| 586 | const setupEntered = yield* Deferred.make<void>(); |
| 587 | const failSetup = yield* Deferred.make<void>(); |
| 588 | const harness = makeHarness({ |
| 589 | runSetup: () => |
| 590 | Deferred.succeed(setupEntered, undefined).pipe( |
| 591 | Effect.andThen(Deferred.await(failSetup)), |
| 592 | Effect.andThen(Effect.fail(new Error("setup failed") as never)), |
| 593 | ), |
| 594 | }); |
| 595 | yield* Effect.gen(function* () { |
| 596 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 597 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 598 | const launched = yield* launches.launch( |
| 599 | launchInput({ |
| 600 | command: "command:launch:queued-during-preparation", |
| 601 | thread: "thread:launch:queued-during-preparation", |
| 602 | message: "Prepare the workspace", |
| 603 | workspace: { type: "worktree", baseRef: "main" }, |
| 604 | }), |
| 605 | ); |
| 606 | yield* Deferred.await(setupEntered); |
| 607 | |
| 608 | const followUp = yield* threads.sendToThread({ |
| 609 | projectId, |
| 610 | commandId: CommandId.make("command:launch:queued-follow-up"), |
| 611 | threadId: launched.threadId, |
| 612 | messageId: MessageId.make("message:launch:queued-follow-up"), |
| 613 | text: "Run after preparation", |
| 614 | attachments: [], |
| 615 | mode: "auto", |
| 616 | createdBy: "user", |
| 617 | creationSource: "web", |
| 618 | }); |
| 619 | assert.equal(followUp.delivery, "queued"); |
| 620 | assert.equal(followUp.run.status, "queued"); |
| 621 | assert.equal( |
| 622 | (yield* threads.getThreadRecords(launched.threadId, ["nodes"])).nodes.find( |
| 623 | (node) => node.runId === followUp.run.id && node.kind === "root_turn", |
| 624 | )?.checkpointScopeId, |
| 625 | null, |
| 626 | ); |
| 627 | |
| 628 | yield* Deferred.succeed(failSetup, undefined); |
| 629 | yield* waitUntil(() => |
| 630 | threads |
| 631 | .getThreadProjection(launched.threadId) |
| 632 | .pipe( |
| 633 | Effect.map( |
| 634 | (projection) => |
| 635 | projection.runs.find((run) => run.id === followUp.run.id)?.status === "starting", |
| 636 | ), |
| 637 | ), |
| 638 | ); |
| 639 | |
| 640 | const projection = yield* threads.getThreadProjection(launched.threadId); |
| 641 | const rootNode = projection.nodes.find( |
| 642 | (node) => node.runId === followUp.run.id && node.kind === "root_turn", |
| 643 | ); |
| 644 | assert.isNotNull(rootNode?.checkpointScopeId); |
| 645 | assert.equal( |
| 646 | projection.checkpointScopes.find((scope) => scope.id === rootNode?.checkpointScopeId) |
| 647 | ?.cwd, |
| 648 | "/repo-worktrees/feature", |
| 649 | ); |
| 650 | }).pipe(Effect.provide(harness.layer)); |
| 651 | }), |
| 652 | ); |
| 653 | |
| 654 | it.effect.each([" /COMPACT ", "/logout"])( |
| 655 | "uses the first conversation message for a title after %s", |
| 656 | (nativeCommand) => |
| 657 | Effect.gen(function* () { |
| 658 | const harness = makeHarness(); |
| 659 | yield* Effect.gen(function* () { |
| 660 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 661 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 662 | const outbox = yield* EffectOutbox.EffectOutboxV2; |
| 663 | const launched = yield* launches.launch({ |
| 664 | ...launchInput({ |
| 665 | command: "compact-title", |
| 666 | thread: "compact-title-thread", |
| 667 | message: nativeCommand, |
| 668 | }), |
| 669 | generateTitle: true, |
| 670 | }); |
| 671 | assert.isUndefined( |
| 672 | (yield* threads.getThreadProjection(launched.threadId)).thread.titleRegeneration, |
| 673 | ); |
| 674 | assert.isFalse( |
| 675 | (yield* outbox.listByCommandId(CommandId.make("compact-title:initial-message"))).some( |
| 676 | (effect) => effect.request.type === "thread-title.generate", |
| 677 | ), |
| 678 | ); |
| 679 | const commandId = CommandId.make("compact-title-conversation"); |
| 680 | const messageId = MessageId.make("compact-title-conversation-message"); |
| 681 | yield* threads.dispatch({ |
| 682 | type: "message.dispatch", |
| 683 | commandId, |
| 684 | threadId: launched.threadId, |
| 685 | messageId, |
| 686 | createdBy: "user", |
| 687 | creationSource: "web", |
| 688 | text: "Fix the failing parser", |
| 689 | attachments: [], |
| 690 | dispatchMode: { type: "defer_start" }, |
| 691 | }); |
| 692 | assert.equal( |
| 693 | (yield* threads.getThreadProjection(launched.threadId)).thread.titleRegeneration |
| 694 | ?.requestId, |
| 695 | commandId, |
| 696 | ); |
| 697 | assert.deepEqual( |
| 698 | (yield* outbox.listByCommandId(commandId)) |
| 699 | .filter((effect) => effect.request.type === "thread-title.generate") |
| 700 | .map((effect) => effect.request), |
| 701 | [{ type: "thread-title.generate", kind: { type: "initial", messageId } }], |
| 702 | ); |
| 703 | }).pipe(Effect.provide(harness.layer)); |
| 704 | }), |
| 705 | ); |
| 706 | |
| 707 | it.effect("keeps native maintenance commands out of steering and restart messages", () => |
| 708 | Effect.gen(function* () { |
| 709 | const harness = makeHarness(); |
| 710 | yield* Effect.gen(function* () { |
| 711 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 712 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 713 | const outbox = yield* EffectOutbox.EffectOutboxV2; |
| 714 | for (const scenario of [ |
| 715 | { |
| 716 | name: "compact-steer", |
| 717 | first: "Fix the parser", |
| 718 | next: " /COMPACT ", |
| 719 | mode: "steer_active", |
| 720 | }, |
| 721 | { |
| 722 | name: "compact-restart", |
| 723 | first: "Fix the parser", |
| 724 | next: "/compact", |
| 725 | mode: "restart_active", |
| 726 | }, |
| 727 | { |
| 728 | name: "logout-steer", |
| 729 | first: "Fix the parser", |
| 730 | next: "/logout", |
| 731 | mode: "steer_active", |
| 732 | }, |
| 733 | { |
| 734 | name: "logout-restart", |
| 735 | first: "Fix the parser", |
| 736 | next: "/logout", |
| 737 | mode: "restart_active", |
| 738 | }, |
| 739 | { |
| 740 | name: "steer-logout", |
| 741 | first: "/logout", |
| 742 | next: "Continue with the parser", |
| 743 | mode: "steer_active", |
| 744 | }, |
| 745 | ] as const) { |
| 746 | const launched = yield* launches.launch( |
| 747 | launchInput({ |
| 748 | command: `${scenario.name}:launch`, |
| 749 | thread: scenario.name, |
| 750 | message: scenario.first, |
| 751 | }), |
| 752 | ); |
| 753 | const before = yield* threads.getThreadProjection(launched.threadId); |
| 754 | const targetRun = before.runs[0]; |
| 755 | if (targetRun === undefined) return yield* Effect.die("Launch must create a run"); |
| 756 | const commandId = CommandId.make(`${scenario.name}:message`); |
| 757 | const failure = yield* threads |
| 758 | .dispatch({ |
| 759 | type: "message.dispatch", |
| 760 | commandId, |
| 761 | threadId: launched.threadId, |
| 762 | messageId: MessageId.make(`${scenario.name}:message`), |
| 763 | createdBy: "user", |
| 764 | creationSource: "web", |
| 765 | text: scenario.next, |
| 766 | attachments: [], |
| 767 | dispatchMode: { type: scenario.mode, targetRunId: targetRun.id }, |
| 768 | }) |
| 769 | .pipe(Effect.flip); |
| 770 | assert.include( |
| 771 | String(failure.cause).toLowerCase(), |
| 772 | scenario.name.includes("logout") ? "sign" : "context compaction", |
| 773 | ); |
| 774 | const after = yield* threads.getThreadProjection(launched.threadId); |
| 775 | assert.deepEqual(after.messages, before.messages); |
| 776 | assert.deepEqual(after.thread.titleRegeneration, before.thread.titleRegeneration); |
| 777 | assert.deepEqual(yield* outbox.listByCommandId(commandId), []); |
| 778 | } |
| 779 | }).pipe(Effect.provide(harness.layer)); |
| 780 | }), |
| 781 | ); |
| 782 | |
| 783 | it.effect.each([ |
| 784 | ...(["manual", "automatic", "pending"] as const).flatMap((compaction) => |
| 785 | (["steer", "restart", "auto", "targeted-steer", "targeted-restart"] as const).map((intent) => ({ |
| 786 | compaction, |
| 787 | intent, |
| 788 | })), |
| 789 | ), |
| 790 | { compaction: "completed", intent: "steer" }, |
| 791 | { compaction: "child", intent: "steer" }, |
| 792 | ] as const)( |
| 793 | "resolves $intent during $compaction compaction without losing the submission", |
| 794 | ({ compaction, intent }) => |
| 795 | Effect.gen(function* () { |
| 796 | const harness = makeHarness(); |
| 797 | yield* Effect.gen(function* () { |
| 798 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 799 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 800 | const sink = yield* EventSink.EventSinkV2; |
| 801 | const outbox = yield* EffectOutbox.EffectOutboxV2; |
| 802 | const launched = yield* launches.launch( |
| 803 | launchInput({ |
| 804 | command: "compaction-queue:launch", |
| 805 | thread: "compaction-queue", |
| 806 | message: compaction === "manual" ? " /COMPACT " : "Fix the parser", |
| 807 | }), |
| 808 | ); |
| 809 | const before = yield* threads.getThreadProjection(launched.threadId); |
| 810 | const active = before.runs[0]!; |
| 811 | const now = yield* DateTime.now; |
| 812 | yield* sink.write({ |
| 813 | events: [ |
| 814 | { |
| 815 | id: EventId.make("compaction-queue:running"), |
| 816 | type: "run.updated", |
| 817 | threadId: launched.threadId, |
| 818 | occurredAt: now, |
| 819 | payload: { ...active, status: "running", startedAt: now }, |
| 820 | }, |
| 821 | ...(compaction === "manual" |
| 822 | ? [] |
| 823 | : [ |
| 824 | { |
| 825 | id: EventId.make("compaction-queue:compacting"), |
| 826 | type: "turn-item.updated" as const, |
| 827 | threadId: launched.threadId, |
| 828 | occurredAt: now, |
| 829 | payload: { |
| 830 | id: TurnItemId.make("compaction-queue:item"), |
| 831 | type: "compaction" as const, |
| 832 | threadId: launched.threadId, |
| 833 | runId: active.id, |
| 834 | nodeId: |
| 835 | compaction === "child" |
| 836 | ? NodeId.make("compaction-child") |
| 837 | : active.rootNodeId, |
| 838 | providerThreadId: active.providerThreadId, |
| 839 | providerTurnId: null, |
| 840 | nativeItemRef: null, |
| 841 | parentItemId: null, |
| 842 | ordinal: 2, |
| 843 | status: |
| 844 | compaction === "completed" |
| 845 | ? ("completed" as const) |
| 846 | : compaction === "pending" |
| 847 | ? ("pending" as const) |
| 848 | : ("running" as const), |
| 849 | title: null, |
| 850 | driver: ProviderDriverKind.make("codex"), |
| 851 | startedAt: now, |
| 852 | completedAt: compaction === "completed" ? now : null, |
| 853 | updatedAt: now, |
| 854 | }, |
| 855 | }, |
| 856 | ]), |
| 857 | ], |
| 858 | }); |
| 859 | const attachments: ChatAttachment[] = [ |
| 860 | { |
| 861 | type: "image", |
| 862 | id: ChatAttachmentId.make("compaction-queue-image"), |
| 863 | name: "parser.png", |
| 864 | mimeType: "image/png", |
| 865 | sizeBytes: 42, |
| 866 | }, |
| 867 | ]; |
| 868 | const command = { |
| 869 | type: "message.dispatch" as const, |
| 870 | commandId: CommandId.make("compaction-queue:follow-up"), |
| 871 | threadId: launched.threadId, |
| 872 | messageId: MessageId.make("compaction-queue:follow-up"), |
| 873 | text: "Continue with the parser [image](t3-context://v1/file/parser)", |
| 874 | context: { |
| 875 | version: 1 as const, |
| 876 | records: [ |
| 877 | { |
| 878 | version: 1 as const, |
| 879 | kind: "file" as const, |
| 880 | contextId: ComposerContextId.make("parser"), |
| 881 | label: "Parser screenshot", |
| 882 | attachmentId: attachments[0]!.id, |
| 883 | name: "parser.png", |
| 884 | mimeType: "image/png", |
| 885 | sizeBytes: 42, |
| 886 | }, |
| 887 | ], |
| 888 | }, |
| 889 | attachments, |
| 890 | modelSelection: { ...modelSelection, model: "gpt-5.2-codex" }, |
| 891 | createdBy: "user" as const, |
| 892 | creationSource: "web" as const, |
| 893 | dispatchMode: { |
| 894 | type: |
| 895 | intent === "targeted-restart" |
| 896 | ? ("restart_active" as const) |
| 897 | : ("steer_active" as const), |
| 898 | targetRunId: active.id, |
| 899 | }, |
| 900 | ...(intent === "targeted-steer" || intent === "targeted-restart" |
| 901 | ? {} |
| 902 | : { deliveryIntent: intent }), |
| 903 | }; |
| 904 | if (compaction === "completed" || compaction === "child") { |
| 905 | const failure = yield* threads.dispatch(command).pipe(Effect.flip); |
| 906 | assert.include(String(failure.cause), "No running provider turn found"); |
| 907 | assert.deepEqual( |
| 908 | (yield* threads.getThreadProjection(launched.threadId)).messages, |
| 909 | before.messages, |
| 910 | ); |
| 911 | return; |
| 912 | } |
| 913 | const accepted = yield* threads.dispatch(command); |
| 914 | const retried = yield* threads.dispatch(command); |
| 915 | assert.equal(retried.sequence, accepted.sequence); |
| 916 | assert.deepEqual(retried.storedEvents, accepted.storedEvents); |
| 917 | |
| 918 | const queued = yield* threads.getThreadProjection(launched.threadId); |
| 919 | assert.equal(queued.runs.length, 2); |
| 920 | assert.equal(queued.runs[0]?.status, "running"); |
| 921 | const queuedRun = queued.runs[1]!; |
| 922 | assert.equal(queuedRun.status, "queued"); |
| 923 | assert.deepEqual(queuedRun.modelSelection, command.modelSelection); |
| 924 | const message = queued.messages.find((candidate) => candidate.id === command.messageId)!; |
| 925 | assert.equal(message.text, command.text); |
| 926 | assert.deepEqual(message.attachments, command.attachments); |
| 927 | assert.deepEqual(message.context, command.context); |
| 928 | assert.equal(message.createdBy, command.createdBy); |
| 929 | assert.equal(message.creationSource, command.creationSource); |
| 930 | assert.equal(message.runId, queuedRun.id); |
| 931 | assert.isFalse( |
| 932 | (yield* outbox.listByCommandId(command.commandId)).some((effect) => |
| 933 | effect.request.type.startsWith("provider-turn."), |
| 934 | ), |
| 935 | ); |
| 936 | |
| 937 | const promote = { |
| 938 | type: "queued-message.promote-to-steer" as const, |
| 939 | commandId: CommandId.make("compaction-queue:promote"), |
| 940 | threadId: launched.threadId, |
| 941 | queuedRunId: queuedRun.id, |
| 942 | targetRunId: active.id, |
| 943 | }; |
| 944 | assert.deepEqual((yield* threads.dispatch(promote)).storedEvents, []); |
| 945 | assert.deepEqual((yield* threads.dispatch(promote)).storedEvents, []); |
| 946 | const afterPromotion = yield* threads.getThreadProjection(launched.threadId); |
| 947 | assert.deepEqual(afterPromotion.runs, queued.runs); |
| 948 | assert.deepEqual(afterPromotion.messages, queued.messages); |
| 949 | |
| 950 | yield* sink.write({ |
| 951 | events: [ |
| 952 | { |
| 953 | id: EventId.make("compaction-queue:completed"), |
| 954 | type: "run.updated", |
| 955 | threadId: launched.threadId, |
| 956 | occurredAt: now, |
| 957 | payload: { ...active, status: "completed", completedAt: now }, |
| 958 | }, |
| 959 | ], |
| 960 | }); |
| 961 | yield* threads.dispatch({ |
| 962 | type: "queue.resume", |
| 963 | commandId: CommandId.make("compaction-queue:resume"), |
| 964 | threadId: launched.threadId, |
| 965 | }); |
| 966 | const resumed = yield* threads.getThreadProjection(launched.threadId); |
| 967 | assert.equal(resumed.runs.find((run) => run.id === queuedRun.id)?.status, "starting"); |
| 968 | assert.equal( |
| 969 | resumed.messages.filter((candidate) => candidate.id === command.messageId).length, |
| 970 | 1, |
| 971 | ); |
| 972 | }).pipe(Effect.provide(harness.layer)); |
| 973 | }), |
| 974 | ); |
| 975 | |
| 976 | it.effect("arms durable title generation after accepting the first message", () => |
| 977 | Effect.gen(function* () { |
| 978 | const harness = makeHarness({ |
| 979 | generateTitle: (input) => |
| 980 | Effect.succeed({ |
| 981 | title: input.previousTitle === undefined ? "Generated title" : "Regenerated title", |
| 982 | }), |
| 983 | }); |
| 984 | yield* Effect.gen(function* () { |
| 985 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 986 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 987 | const outbox = yield* EffectOutbox.EffectOutboxV2; |
| 988 | const titleRegeneration = yield* ThreadTitleRegeneration.ThreadTitleRegenerationService; |
| 989 | const input = { |
| 990 | ...launchInput({ |
| 991 | command: "command:launch:title-generation", |
| 992 | thread: "thread:launch:title-generation", |
| 993 | message: "Generate my title", |
| 994 | }), |
| 995 | title: "Generate my title", |
| 996 | generateTitle: true, |
| 997 | }; |
| 998 | const launched = yield* launches.launch(input); |
| 999 | const generationCommandId = CommandId.make("command:launch:title-generation:initial-message"); |
| 1000 | |
| 1001 | const projection = yield* threads.getThreadProjection(launched.threadId); |
| 1002 | assert.equal(projection.thread.title, "Generate my title"); |
| 1003 | assert.equal(projection.thread.titleRegeneration?.requestId, generationCommandId); |
| 1004 | assert.deepEqual( |
| 1005 | (yield* outbox.listByCommandId(generationCommandId)).map((effect) => effect.request), |
| 1006 | [ |
| 1007 | { |
| 1008 | type: "thread-title.generate", |
| 1009 | kind: { type: "initial", messageId: MessageId.make("Generate my title:id") }, |
| 1010 | }, |
| 1011 | ], |
| 1012 | ); |
| 1013 | yield* titleRegeneration.execute({ |
| 1014 | threadId: launched.threadId, |
| 1015 | requestId: generationCommandId, |
| 1016 | kind: { type: "initial", messageId: MessageId.make("Generate my title:id") }, |
| 1017 | }); |
| 1018 | const generated = yield* threads.getThreadProjection(launched.threadId); |
| 1019 | assert.equal(generated.thread.title, "Generated title"); |
| 1020 | assert.deepEqual( |
| 1021 | harness.generateThreadTitle.mock.calls[0]?.[0]?.modelSelection, |
| 1022 | DEFAULT_SERVER_SETTINGS.textGenerationModelSelection, |
| 1023 | ); |
| 1024 | |
| 1025 | const manualRequestId = CommandId.make("command:title-generation:manual"); |
| 1026 | yield* threads.dispatch({ |
| 1027 | type: "thread.metadata.update", |
| 1028 | commandId: manualRequestId, |
| 1029 | threadId: launched.threadId, |
| 1030 | regenerateTitle: true, |
| 1031 | }); |
| 1032 | assert.deepEqual( |
| 1033 | (yield* outbox.listByCommandId(manualRequestId)).map((effect) => effect.request), |
| 1034 | [{ type: "thread-title.generate", kind: { type: "regenerate" } }], |
| 1035 | ); |
| 1036 | yield* titleRegeneration.execute({ |
| 1037 | threadId: launched.threadId, |
| 1038 | requestId: manualRequestId, |
| 1039 | kind: { type: "regenerate" }, |
| 1040 | }); |
| 1041 | const regenerated = yield* threads.getThreadProjection(launched.threadId); |
| 1042 | assert.equal(regenerated.thread.title, "Regenerated title"); |
| 1043 | assert.equal( |
| 1044 | harness.generateThreadTitle.mock.calls[1]?.[0]?.previousTitle, |
| 1045 | "Generated title", |
| 1046 | ); |
| 1047 | |
| 1048 | yield* threads.dispatch({ |
| 1049 | type: "thread.metadata.update", |
| 1050 | commandId: CommandId.make("command:title-generation:user-rename"), |
| 1051 | threadId: launched.threadId, |
| 1052 | title: "Keep my title", |
| 1053 | }); |
| 1054 | const renamed = yield* threads.getThreadProjection(launched.threadId); |
| 1055 | yield* TestClock.adjust(Duration.seconds(1)); |
| 1056 | yield* threads.dispatch({ |
| 1057 | type: "thread.title.regeneration.complete", |
| 1058 | commandId: CommandId.make("command:title-generation:stale-completion"), |
| 1059 | threadId: launched.threadId, |
| 1060 | requestId: generationCommandId, |
| 1061 | title: "Stale generated title", |
| 1062 | }); |
| 1063 | const afterStaleCompletion = yield* threads.getThreadProjection(launched.threadId); |
| 1064 | assert.equal(afterStaleCompletion.thread.title, "Keep my title"); |
| 1065 | assert.equal( |
| 1066 | DateTime.toEpochMillis(afterStaleCompletion.thread.updatedAt), |
| 1067 | DateTime.toEpochMillis(renamed.thread.updatedAt), |
| 1068 | ); |
| 1069 | }).pipe(Effect.provide(harness.layer)); |
| 1070 | }), |
| 1071 | ); |
| 1072 | |
| 1073 | it.effect("does not update a reused thread title when the initial message is rejected", () => |
| 1074 | Effect.gen(function* () { |
| 1075 | const harness = makeHarness(); |
| 1076 | yield* Effect.gen(function* () { |
| 1077 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1078 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 1079 | const outbox = yield* EffectOutbox.EffectOutboxV2; |
| 1080 | const threadId = ThreadId.make("thread:launch:reused-title-failure"); |
| 1081 | yield* threads.dispatch({ |
| 1082 | type: "thread.create", |
| 1083 | commandId: CommandId.make("command:launch:reused-title-failure:create"), |
| 1084 | threadId, |
| 1085 | projectId, |
| 1086 | title: "Original title", |
| 1087 | modelSelection, |
| 1088 | runtimeMode: "full-access", |
| 1089 | interactionMode: "default", |
| 1090 | branch: null, |
| 1091 | worktreePath: null, |
| 1092 | createdBy: "user", |
| 1093 | creationSource: "web", |
| 1094 | }); |
| 1095 | |
| 1096 | const commandId = CommandId.make("command:launch:reused-title-failure"); |
| 1097 | const failed = yield* launches |
| 1098 | .launch({ |
| 1099 | ...launchInput({ |
| 1100 | command: commandId, |
| 1101 | thread: threadId, |
| 1102 | message: "Generate a provisional title", |
| 1103 | }), |
| 1104 | reuseExistingThread: true, |
| 1105 | title: "Generate a provisional title", |
| 1106 | generateTitle: true, |
| 1107 | modelSelection: { |
| 1108 | instanceId: ProviderInstanceId.make("missing-provider"), |
| 1109 | model: "missing-model", |
| 1110 | }, |
| 1111 | }) |
| 1112 | .pipe(Effect.exit); |
| 1113 | |
| 1114 | assert.isTrue(Exit.isFailure(failed)); |
| 1115 | const projection = yield* threads.getThreadProjection(threadId); |
| 1116 | assert.equal(projection.thread.title, "Original title"); |
| 1117 | assert.isUndefined(projection.thread.titleRegeneration); |
| 1118 | assert.isEmpty(projection.messages); |
| 1119 | assert.isEmpty(yield* outbox.listByCommandId(CommandId.make(`${commandId}:initial-message`))); |
| 1120 | }).pipe(Effect.provide(harness.layer)); |
| 1121 | }), |
| 1122 | ); |
| 1123 | |
| 1124 | it.effect("generates an initial title for an attachment-only message", () => |
| 1125 | Effect.gen(function* () { |
| 1126 | const harness = makeHarness(); |
| 1127 | yield* Effect.gen(function* () { |
| 1128 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1129 | const titleRegeneration = yield* ThreadTitleRegeneration.ThreadTitleRegenerationService; |
| 1130 | const messageId = MessageId.make("message:image-only"); |
| 1131 | const input = { |
| 1132 | ...launchInput({ |
| 1133 | command: "command:launch:image-only", |
| 1134 | thread: "thread:launch:image-only", |
| 1135 | }), |
| 1136 | title: "Image: screenshot.png", |
| 1137 | generateTitle: true, |
| 1138 | initialMessage: { |
| 1139 | messageId, |
| 1140 | text: "", |
| 1141 | attachments: [ |
| 1142 | { |
| 1143 | type: "image" as const, |
| 1144 | id: "attachment-image-only", |
| 1145 | name: "screenshot.png", |
| 1146 | mimeType: "image/png", |
| 1147 | sizeBytes: 128, |
| 1148 | }, |
| 1149 | ], |
| 1150 | }, |
| 1151 | }; |
| 1152 | |
| 1153 | const launched = yield* launches.launch(input); |
| 1154 | yield* titleRegeneration.execute({ |
| 1155 | threadId: launched.threadId, |
| 1156 | requestId: CommandId.make("command:launch:image-only:initial-message"), |
| 1157 | kind: { type: "initial", messageId }, |
| 1158 | }); |
| 1159 | |
| 1160 | assert.equal(harness.generateThreadTitle.mock.calls[0]?.[0]?.message, ""); |
| 1161 | assert.equal( |
| 1162 | harness.generateThreadTitle.mock.calls[0]?.[0]?.attachments?.[0]?.name, |
| 1163 | "screenshot.png", |
| 1164 | ); |
| 1165 | }).pipe(Effect.provide(harness.layer)); |
| 1166 | }), |
| 1167 | ); |
| 1168 | |
| 1169 | it.effect("uses the available source control writer for generated worktree branches", () => |
| 1170 | Effect.gen(function* () { |
| 1171 | const writerInstanceId = ProviderInstanceId.make("source-control-writer"); |
| 1172 | const writerModelSelection = { |
| 1173 | instanceId: writerInstanceId, |
| 1174 | model: "branch-writer-model", |
| 1175 | } as const; |
| 1176 | const harness = makeHarness({ |
| 1177 | serverSettings: { |
| 1178 | providerInstances: { |
| 1179 | [writerInstanceId]: { |
| 1180 | driver: ProviderDriverKind.make("codex"), |
| 1181 | config: {}, |
| 1182 | }, |
| 1183 | }, |
| 1184 | sourceControlWriterModelSelection: writerModelSelection, |
| 1185 | }, |
| 1186 | providers: [ |
| 1187 | { |
| 1188 | instanceId: writerInstanceId, |
| 1189 | driver: ProviderDriverKind.make("codex"), |
| 1190 | enabled: true, |
| 1191 | installed: true, |
| 1192 | version: null, |
| 1193 | status: "ready", |
| 1194 | auth: { status: "authenticated" }, |
| 1195 | checkedAt: "2026-07-28T00:00:00.000Z", |
| 1196 | availability: "available", |
| 1197 | models: [], |
| 1198 | slashCommands: [], |
| 1199 | skills: [], |
| 1200 | }, |
| 1201 | ], |
| 1202 | }); |
| 1203 | |
| 1204 | yield* Effect.gen(function* () { |
| 1205 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1206 | yield* launches.launch( |
| 1207 | launchInput({ |
| 1208 | command: "command:launch:source-control-writer", |
| 1209 | thread: "thread:launch:source-control-writer", |
| 1210 | message: "Generate a branch with the configured writer", |
| 1211 | workspace: { type: "worktree", baseRef: "main" }, |
| 1212 | }), |
| 1213 | ); |
| 1214 | yield* waitUntil(() => Effect.sync(() => harness.generateBranchName.mock.calls.length === 1)); |
| 1215 | assert.deepEqual( |
| 1216 | harness.generateBranchName.mock.calls[0]?.[0]?.modelSelection, |
| 1217 | writerModelSelection, |
| 1218 | ); |
| 1219 | }).pipe(Effect.provide(harness.layer)); |
| 1220 | }), |
| 1221 | ); |
| 1222 | |
| 1223 | it.effect("falls back when the source control writer is unavailable", () => |
| 1224 | Effect.gen(function* () { |
| 1225 | const writerInstanceId = ProviderInstanceId.make("missing-source-control-writer"); |
| 1226 | const harness = makeHarness({ |
| 1227 | serverSettings: { |
| 1228 | providerInstances: { |
| 1229 | [writerInstanceId]: { |
| 1230 | driver: ProviderDriverKind.make("missing-driver"), |
| 1231 | config: {}, |
| 1232 | }, |
| 1233 | }, |
| 1234 | sourceControlWriterModelSelection: { |
| 1235 | instanceId: writerInstanceId, |
| 1236 | model: "missing-branch-writer-model", |
| 1237 | }, |
| 1238 | }, |
| 1239 | providers: [], |
| 1240 | }); |
| 1241 | |
| 1242 | yield* Effect.gen(function* () { |
| 1243 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1244 | yield* launches.launch( |
| 1245 | launchInput({ |
| 1246 | command: "command:launch:source-control-writer-fallback", |
| 1247 | thread: "thread:launch:source-control-writer-fallback", |
| 1248 | message: "Generate a branch with the available writer", |
| 1249 | workspace: { type: "worktree", baseRef: "main" }, |
| 1250 | }), |
| 1251 | ); |
| 1252 | yield* waitUntil(() => Effect.sync(() => harness.generateBranchName.mock.calls.length === 1)); |
| 1253 | assert.deepEqual( |
| 1254 | harness.generateBranchName.mock.calls[0]?.[0]?.modelSelection, |
| 1255 | DEFAULT_SERVER_SETTINGS.textGenerationModelSelection, |
| 1256 | ); |
| 1257 | }).pipe(Effect.provide(harness.layer)); |
| 1258 | }), |
| 1259 | ); |
| 1260 | |
| 1261 | it.effect("runs a Scratch thread launched at the root in its own folder", () => |
| 1262 | Effect.gen(function* () { |
| 1263 | // Only `projectId` stands in for the Scratch project here. |
| 1264 | const claimed: Array<{ readonly threadId: ThreadId; readonly text: string }> = []; |
| 1265 | const harness = makeHarness({ |
| 1266 | managedFolders: Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ |
| 1267 | namedProjectsRoot: "/projects", |
| 1268 | folderForThread: (input) => |
| 1269 | Effect.sync(() => { |
| 1270 | if (input.projectId !== projectId) return Option.none(); |
| 1271 | claimed.push({ threadId: input.threadId, text: input.text }); |
| 1272 | return Option.some(`/scratch/folder-${claimed.length}`); |
| 1273 | }), |
| 1274 | }), |
| 1275 | }); |
| 1276 | yield* Effect.gen(function* () { |
| 1277 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1278 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 1279 | const input = launchInput({ |
| 1280 | command: "command:launch:scratch", |
| 1281 | thread: "thread:launch:scratch", |
| 1282 | message: "Convert these PNGs", |
| 1283 | }); |
| 1284 | const launched = yield* launches.launch(input); |
| 1285 | assert.deepEqual(claimed, [{ threadId: launched.threadId, text: "Convert these PNGs" }]); |
| 1286 | assert.equal(launched.projection.thread.worktreePath, "/scratch/folder-1"); |
| 1287 | yield* waitUntil(() => Effect.sync(() => harness.runSetup.mock.calls.length === 1)); |
| 1288 | assert.equal(harness.runSetup.mock.calls[0]?.[0]?.worktreePath, "/scratch/folder-1"); |
| 1289 | assert.equal(harness.createWorktree.mock.calls.length, 0); |
| 1290 | |
| 1291 | // A retry replays the first attempt and claims no second folder. |
| 1292 | const retried = yield* launches.launch(input); |
| 1293 | assert.isTrue(retried.resumed); |
| 1294 | assert.lengthOf(claimed, 1); |
| 1295 | assert.equal( |
| 1296 | (yield* threads.getThreadProjection(launched.threadId)).thread.worktreePath, |
| 1297 | "/scratch/folder-1", |
| 1298 | ); |
| 1299 | |
| 1300 | const other = yield* launches.launch({ |
| 1301 | ...launchInput({ |
| 1302 | command: "command:launch:scratch-other", |
| 1303 | thread: "thread:launch:scratch-other", |
| 1304 | message: "Elsewhere", |
| 1305 | }), |
| 1306 | projectId: otherProjectId, |
| 1307 | }); |
| 1308 | assert.lengthOf(claimed, 1); |
| 1309 | assert.isNull(other.projection.thread.worktreePath); |
| 1310 | }).pipe(Effect.provide(harness.layer)); |
| 1311 | }), |
| 1312 | ); |
| 1313 | |
| 1314 | it.effect("names the worktree itself when the client provides no branch", () => |
| 1315 | Effect.gen(function* () { |
| 1316 | const harness = makeHarness(); |
| 1317 | yield* Effect.gen(function* () { |
| 1318 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1319 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 1320 | const launched = yield* launches.launch( |
| 1321 | launchInput({ |
| 1322 | command: "command:launch:server-named-branch", |
| 1323 | thread: "thread:launch:server-named-branch", |
| 1324 | message: "Build the feature", |
| 1325 | workspace: { type: "worktree", baseRef: "main" }, |
| 1326 | }), |
| 1327 | ); |
| 1328 | yield* waitUntil(() => Effect.sync(() => harness.createWorktree.mock.calls.length === 1)); |
| 1329 | assert.match( |
| 1330 | harness.createWorktree.mock.calls[0]?.[0]?.newRefName ?? "", |
| 1331 | /^t3\/[0-9a-f]{8}$/u, |
| 1332 | ); |
| 1333 | yield* waitUntil(() => |
| 1334 | threads |
| 1335 | .getThreadProjection(launched.threadId) |
| 1336 | .pipe(Effect.map((projection) => projection.thread.branch === "generated-branch")), |
| 1337 | ); |
| 1338 | }).pipe(Effect.provide(harness.layer)); |
| 1339 | }), |
| 1340 | ); |
| 1341 | |
| 1342 | it.effect("renames a temporary t3/<hash> branch off the provisioning critical path", () => |
| 1343 | Effect.gen(function* () { |
| 1344 | const branchNameStarted = yield* Deferred.make<void>(); |
| 1345 | const allowBranchName = yield* Deferred.make<void>(); |
| 1346 | const harness = makeHarness({ |
| 1347 | createWorktree: (input) => |
| 1348 | Effect.succeed({ |
| 1349 | worktree: { path: "/repo-worktrees/temp", refName: input.newRefName, headSha: "abc" }, |
| 1350 | } as never), |
| 1351 | generateBranchName: () => |
| 1352 | Deferred.succeed(branchNameStarted, undefined).pipe( |
| 1353 | Effect.andThen(Deferred.await(allowBranchName)), |
| 1354 | Effect.as({ branch: "generated-branch" }), |
| 1355 | ), |
| 1356 | }); |
| 1357 | yield* Effect.gen(function* () { |
| 1358 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1359 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 1360 | const launched = yield* launches.launch( |
| 1361 | launchInput({ |
| 1362 | command: "command:launch:temp-branch", |
| 1363 | thread: "thread:launch:temp-branch", |
| 1364 | message: "Build the feature", |
| 1365 | workspace: { type: "worktree", baseRef: "main", branch: "t3/abcd1234" }, |
| 1366 | }), |
| 1367 | ); |
| 1368 | yield* Deferred.await(branchNameStarted); |
| 1369 | assert.equal(harness.createWorktree.mock.calls[0]?.[0]?.newRefName, "t3/abcd1234"); |
| 1370 | yield* waitUntil(() => |
| 1371 | threads |
| 1372 | .getThreadProjection(launched.threadId) |
| 1373 | .pipe(Effect.map((projection) => projection.runs[0]?.status === "starting")), |
| 1374 | ); |
| 1375 | assert.equal( |
| 1376 | (yield* threads.getThreadProjection(launched.threadId)).thread.branch, |
| 1377 | "t3/abcd1234", |
| 1378 | ); |
| 1379 | yield* Deferred.succeed(allowBranchName, undefined); |
| 1380 | yield* waitUntil(() => |
| 1381 | threads |
| 1382 | .getThreadProjection(launched.threadId) |
| 1383 | .pipe(Effect.map((projection) => projection.thread.branch === "generated-branch")), |
| 1384 | ); |
| 1385 | assert.deepEqual(harness.renameBranch.mock.calls[0]?.[0], { |
| 1386 | cwd: "/repo-worktrees/temp", |
| 1387 | oldBranch: "t3/abcd1234", |
| 1388 | newBranch: "generated-branch", |
| 1389 | }); |
| 1390 | }).pipe(Effect.provide(harness.layer)); |
| 1391 | }), |
| 1392 | ); |
| 1393 | |
| 1394 | it.effect("provisions under t3-<hash> when a plain t3 branch blocks t3/*", () => |
| 1395 | Effect.gen(function* () { |
| 1396 | const harness = makeHarness({ |
| 1397 | hasCommit: (input) => Effect.succeed(input.refName === "refs/heads/t3"), |
| 1398 | createWorktree: (input) => |
| 1399 | Effect.succeed({ |
| 1400 | worktree: { path: "/repo-worktrees/temp", refName: input.newRefName, headSha: "abc" }, |
| 1401 | } as never), |
| 1402 | }); |
| 1403 | yield* Effect.gen(function* () { |
| 1404 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1405 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 1406 | const launched = yield* launches.launch( |
| 1407 | launchInput({ |
| 1408 | command: "command:launch:blocked-namespace", |
| 1409 | thread: "thread:launch:blocked-namespace", |
| 1410 | message: "Build the feature", |
| 1411 | workspace: { type: "worktree", baseRef: "main", branch: "t3/abcd1234" }, |
| 1412 | }), |
| 1413 | ); |
| 1414 | yield* waitUntil(() => |
| 1415 | threads |
| 1416 | .getThreadProjection(launched.threadId) |
| 1417 | .pipe(Effect.map((projection) => projection.thread.branch === "generated-branch")), |
| 1418 | ); |
| 1419 | assert.equal(harness.createWorktree.mock.calls[0]?.[0]?.newRefName, "t3-abcd1234"); |
| 1420 | assert.equal(harness.renameBranch.mock.calls[0]?.[0]?.oldBranch, "t3-abcd1234"); |
| 1421 | }).pipe(Effect.provide(harness.layer)); |
| 1422 | }), |
| 1423 | ); |
| 1424 | |
| 1425 | it.effect("keeps an explicit branch name instead of generating one", () => |
| 1426 | Effect.gen(function* () { |
| 1427 | const harness = makeHarness(); |
| 1428 | yield* Effect.gen(function* () { |
| 1429 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1430 | yield* launches.launch( |
| 1431 | launchInput({ |
| 1432 | command: "command:launch:explicit-branch", |
| 1433 | thread: "thread:launch:explicit-branch", |
| 1434 | message: "Build the feature", |
| 1435 | workspace: { type: "worktree", baseRef: "main", branch: "my-feature" }, |
| 1436 | }), |
| 1437 | ); |
| 1438 | yield* waitUntil(() => Effect.sync(() => harness.createWorktree.mock.calls.length === 1)); |
| 1439 | assert.equal(harness.generateBranchName.mock.calls.length, 0); |
| 1440 | assert.equal(harness.createWorktree.mock.calls[0]?.[0]?.newRefName, "my-feature"); |
| 1441 | }).pipe(Effect.provide(harness.layer)); |
| 1442 | }), |
| 1443 | ); |
| 1444 | |
| 1445 | it.effect("keeps the temporary branch when branch generation fails", () => |
| 1446 | Effect.gen(function* () { |
| 1447 | const harness = makeHarness({ |
| 1448 | createWorktree: (input) => |
| 1449 | Effect.succeed({ |
| 1450 | worktree: { path: "/repo-worktrees/temp", refName: input.newRefName, headSha: "abc" }, |
| 1451 | } as never), |
| 1452 | generateBranchName: () => Effect.die("branch generation is down"), |
| 1453 | }); |
| 1454 | yield* Effect.gen(function* () { |
| 1455 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1456 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 1457 | const launched = yield* launches.launch( |
| 1458 | launchInput({ |
| 1459 | command: "command:launch:branch-fallback", |
| 1460 | thread: "thread:launch:branch-fallback", |
| 1461 | message: "Build the feature", |
| 1462 | workspace: { type: "worktree", baseRef: "main", branch: "t3/abcd1234" }, |
| 1463 | }), |
| 1464 | ); |
| 1465 | yield* waitUntil(() => Effect.sync(() => harness.generateBranchName.mock.calls.length === 1)); |
| 1466 | assert.equal(harness.createWorktree.mock.calls[0]?.[0]?.newRefName, "t3/abcd1234"); |
| 1467 | yield* waitUntil(() => |
| 1468 | threads |
| 1469 | .getThreadProjection(launched.threadId) |
| 1470 | .pipe(Effect.map((projection) => projection.runs[0]?.status === "starting")), |
| 1471 | ); |
| 1472 | assert.equal(harness.renameBranch.mock.calls.length, 0); |
| 1473 | assert.equal( |
| 1474 | (yield* threads.getThreadProjection(launched.threadId)).thread.branch, |
| 1475 | "t3/abcd1234", |
| 1476 | ); |
| 1477 | }).pipe(Effect.provide(harness.layer)); |
| 1478 | }), |
| 1479 | ); |
| 1480 | |
| 1481 | it.effect("renames a temporary branch on an existing worktree to a generated name", () => |
| 1482 | Effect.gen(function* () { |
| 1483 | const harness = makeHarness(); |
| 1484 | yield* Effect.gen(function* () { |
| 1485 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1486 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 1487 | const launched = yield* launches.launch( |
| 1488 | launchInput({ |
| 1489 | command: "command:launch:existing-worktree-rename", |
| 1490 | thread: "thread:launch:existing-worktree-rename", |
| 1491 | message: "Build the feature", |
| 1492 | workspace: { |
| 1493 | type: "existing_worktree", |
| 1494 | worktreePath: "/repo-worktrees/t3-abcd1234", |
| 1495 | branch: "t3/abcd1234", |
| 1496 | }, |
| 1497 | }), |
| 1498 | ); |
| 1499 | yield* waitUntil(() => |
| 1500 | threads |
| 1501 | .getThreadProjection(launched.threadId) |
| 1502 | .pipe(Effect.map((projection) => projection.thread.branch === "generated-branch")), |
| 1503 | ); |
| 1504 | assert.deepEqual(harness.renameBranch.mock.calls[0]?.[0], { |
| 1505 | cwd: "/repo-worktrees/t3-abcd1234", |
| 1506 | oldBranch: "t3/abcd1234", |
| 1507 | newBranch: "generated-branch", |
| 1508 | }); |
| 1509 | }).pipe(Effect.provide(harness.layer)); |
| 1510 | }), |
| 1511 | ); |
| 1512 | |
| 1513 | it.effect("shows the fetch diagnosis when preparing a worktree from origin fails", () => { |
| 1514 | const detail = |
| 1515 | "Git could not authenticate with the remote. Check Git credentials or SSH access on the server, then retry."; |
| 1516 | const harness = makeHarness({ |
| 1517 | fetchRemote: () => |
| 1518 | Effect.fail( |
| 1519 | new GitCommandError({ |
| 1520 | operation: "GitVcsDriver.fetchRemote", |
| 1521 | command: "git", |
| 1522 | cwd: project.workspaceRoot, |
| 1523 | detail, |
| 1524 | exitCode: 128, |
| 1525 | }), |
| 1526 | ), |
| 1527 | }); |
| 1528 | return Effect.gen(function* () { |
| 1529 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1530 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 1531 | const launched = yield* launches.launch( |
| 1532 | launchInput({ |
| 1533 | command: "command:launch:fetch-failure", |
| 1534 | thread: "thread:launch:fetch-failure", |
| 1535 | message: "Start from origin", |
| 1536 | workspace: { type: "worktree", baseRef: "main", startFromOrigin: true }, |
| 1537 | }), |
| 1538 | ); |
| 1539 | yield* threads.streamStoredEventsFrom({ threadId: launched.threadId }).pipe( |
| 1540 | Stream.filter( |
| 1541 | (stored) => stored.event.type === "run.updated" && stored.event.payload.status === "failed", |
| 1542 | ), |
| 1543 | Stream.runHead, |
| 1544 | ); |
| 1545 | const projection = yield* threads.getThreadProjection(launched.threadId); |
| 1546 | assert.equal(projection.messages[0]?.text, "Start from origin"); |
| 1547 | assert.equal(projection.runs[0]?.status, "failed"); |
| 1548 | assert.equal(projection.thread.worktreePath, null); |
| 1549 | assert.equal( |
| 1550 | projection.turnItems.find((item) => item.type === "command_execution")?.status, |
| 1551 | "failed", |
| 1552 | ); |
| 1553 | assert.include( |
| 1554 | projection.turnItems.find((item) => item.type === "error")?.failure.message ?? "", |
| 1555 | detail, |
| 1556 | ); |
| 1557 | assert.equal(harness.createWorktree.mock.calls.length, 0); |
| 1558 | assert.equal(harness.runSetup.mock.calls.length, 0); |
| 1559 | }).pipe(Effect.provide(harness.layer)); |
| 1560 | }); |
| 1561 | |
| 1562 | it.effect("retries a failed workspace preparation on the same run", () => { |
| 1563 | let fetchFailures = 1; |
| 1564 | const harness = makeHarness({ |
| 1565 | fetchRemote: () => |
| 1566 | fetchFailures-- > 0 |
| 1567 | ? Effect.fail( |
| 1568 | new GitCommandError({ |
| 1569 | operation: "GitVcsDriver.fetchRemote", |
| 1570 | command: "git", |
| 1571 | cwd: project.workspaceRoot, |
| 1572 | detail: "Git could not update a local reference.", |
| 1573 | exitCode: 1, |
| 1574 | }), |
| 1575 | ) |
| 1576 | : Effect.void, |
| 1577 | }); |
| 1578 | return Effect.gen(function* () { |
| 1579 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1580 | const outbox = yield* EffectOutbox.EffectOutboxV2; |
| 1581 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 1582 | const launched = yield* launches.launch( |
| 1583 | launchInput({ |
| 1584 | command: "command:launch:retry", |
| 1585 | thread: "thread:launch:retry", |
| 1586 | message: "Retry me", |
| 1587 | workspace: { type: "worktree", baseRef: "main", startFromOrigin: true }, |
| 1588 | }), |
| 1589 | ); |
| 1590 | yield* waitUntil(() => |
| 1591 | threads |
| 1592 | .getThreadProjection(launched.threadId) |
| 1593 | .pipe(Effect.map((projection) => projection.runs[0]?.status === "failed")), |
| 1594 | ); |
| 1595 | const failed = yield* threads.getThreadProjection(launched.threadId); |
| 1596 | const runId = failed.runs[0]!.id; |
| 1597 | assert.equal( |
| 1598 | failed.turnItems.find((item) => item.type === "error")?.failure.code, |
| 1599 | ORCHESTRATION_V2_WORKSPACE_PREPARATION_FAILURE_CODE, |
| 1600 | ); |
| 1601 | |
| 1602 | const retry = { |
| 1603 | commandId: CommandId.make("command:launch:retry:1"), |
| 1604 | threadId: launched.threadId, |
| 1605 | runId, |
| 1606 | }; |
| 1607 | yield* launches.retryPreparation(retry); |
| 1608 | yield* waitUntil(() => |
| 1609 | outbox |
| 1610 | .listByCommandId(CommandId.make("command:launch:retry:1:release")) |
| 1611 | .pipe(Effect.map((effects) => effects.length === 1)), |
| 1612 | ); |
| 1613 | const retried = yield* threads.getThreadProjection(launched.threadId); |
| 1614 | assert.equal(retried.runs.length, 1); |
| 1615 | assert.equal(retried.runs[0]?.status, "starting"); |
| 1616 | assert.equal(retried.thread.worktreePath, "/repo-worktrees/feature"); |
| 1617 | assert.equal(retried.turnItems.find((item) => item.type === "error")?.status, "cancelled"); |
| 1618 | assert.equal( |
| 1619 | retried.turnItems.find((item) => item.type === "command_execution")?.status, |
| 1620 | "completed", |
| 1621 | ); |
| 1622 | assert.equal(harness.createWorktree.mock.calls.length, 1); |
| 1623 | |
| 1624 | // The run left preparation, so a second retry has nothing to do. |
| 1625 | const rejected = yield* launches |
| 1626 | .retryPreparation({ ...retry, commandId: CommandId.make("command:launch:retry:2") }) |
| 1627 | .pipe(Effect.flip); |
| 1628 | assert.equal(rejected._tag, "OrchestratorDispatchError"); |
| 1629 | }).pipe(Effect.provide(harness.layer)); |
| 1630 | }); |
| 1631 | |
| 1632 | it.effect.each(["fails", "is interrupted"] as const)( |
| 1633 | "holds a message queued during setup when the workspace preparation %s", |
| 1634 | (ending) => |
| 1635 | Effect.gen(function* () { |
| 1636 | const fetchEntered = yield* Deferred.make<void>(); |
| 1637 | const allowFetch = yield* Deferred.make<void>(); |
| 1638 | let fetchFailures = 1; |
| 1639 | const harness = makeHarness({ |
| 1640 | fetchRemote: () => |
| 1641 | fetchFailures-- <= 0 |
| 1642 | ? Effect.void |
| 1643 | : Deferred.succeed(fetchEntered, undefined).pipe( |
| 1644 | Effect.andThen(Deferred.await(allowFetch)), |
| 1645 | Effect.andThen( |
| 1646 | Effect.fail( |
| 1647 | new GitCommandError({ |
| 1648 | operation: "GitVcsDriver.fetchRemote", |
| 1649 | command: "git", |
| 1650 | cwd: project.workspaceRoot, |
| 1651 | detail: "Git could not reach the remote.", |
| 1652 | exitCode: 128, |
| 1653 | }), |
| 1654 | ), |
| 1655 | ), |
| 1656 | ), |
| 1657 | }); |
| 1658 | yield* Effect.gen(function* () { |
| 1659 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1660 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 1661 | const launched = yield* launches.launch( |
| 1662 | launchInput({ |
| 1663 | command: `command:launch:queued-setup-${ending}`, |
| 1664 | thread: `thread:launch:queued-setup-${ending}`, |
| 1665 | message: "First message", |
| 1666 | workspace: { type: "worktree", baseRef: "main", startFromOrigin: true }, |
| 1667 | }), |
| 1668 | ); |
| 1669 | yield* Deferred.await(fetchEntered); |
| 1670 | const queued = yield* threads.sendToThread({ |
| 1671 | projectId, |
| 1672 | commandId: CommandId.make(`command:launch:queued-setup-${ending}:follow-up`), |
| 1673 | threadId: launched.threadId, |
| 1674 | messageId: MessageId.make(`message:launch:queued-setup-${ending}:follow-up`), |
| 1675 | text: "Sent during setup", |
| 1676 | attachments: [], |
| 1677 | mode: "queue", |
| 1678 | createdBy: "user", |
| 1679 | creationSource: "web", |
| 1680 | }); |
| 1681 | assert.equal(queued.delivery, "queued"); |
| 1682 | if (ending === "fails") { |
| 1683 | yield* Deferred.succeed(allowFetch, undefined); |
| 1684 | } else { |
| 1685 | // An agent stop does not ask to hold the queue. |
| 1686 | yield* threads.dispatch({ |
| 1687 | type: "run.interrupt", |
| 1688 | commandId: CommandId.make("command:launch:queued-setup-interrupt"), |
| 1689 | threadId: launched.threadId, |
| 1690 | runId: launched.projection.runs[0]!.id, |
| 1691 | }); |
| 1692 | } |
| 1693 | |
| 1694 | // Without its worktree the follow-up would start in the project checkout. |
| 1695 | yield* threads.streamStoredEventsFrom({ threadId: launched.threadId }).pipe( |
| 1696 | Stream.filter( |
| 1697 | (stored) => |
| 1698 | stored.event.type === "run.updated" && stored.event.payload.queueHeld === true, |
| 1699 | ), |
| 1700 | Stream.runHead, |
| 1701 | ); |
| 1702 | const projection = yield* threads.getThreadProjection(launched.threadId); |
| 1703 | assert.equal(projection.runs.at(-1)?.status, "queued"); |
| 1704 | assert.equal(projection.thread.worktreePath, null); |
| 1705 | |
| 1706 | if (ending === "fails") { |
| 1707 | // A successful retry lets the held message follow the first turn again. |
| 1708 | yield* launches.retryPreparation({ |
| 1709 | commandId: CommandId.make("command:launch:queued-setup-retry"), |
| 1710 | threadId: launched.threadId, |
| 1711 | runId: launched.projection.runs[0]!.id, |
| 1712 | }); |
| 1713 | const retried = yield* threads.getThreadProjection(launched.threadId); |
| 1714 | assert.equal(retried.runs.at(-1)?.queueHeld, false); |
| 1715 | } |
| 1716 | }).pipe(Effect.provide(harness.layer)); |
| 1717 | }), |
| 1718 | ); |
| 1719 | |
| 1720 | it.effect.each(["failed", "interrupted"] as const)( |
| 1721 | "holds a message queued during setup that arrives after the setup %s", |
| 1722 | (ending) => { |
| 1723 | let fetchEntered: Deferred.Deferred<void> | null = null; |
| 1724 | const harness = makeHarness({ |
| 1725 | fetchRemote: () => |
| 1726 | ending === "interrupted" |
| 1727 | ? Deferred.succeed(fetchEntered!, undefined).pipe(Effect.andThen(Effect.never)) |
| 1728 | : Effect.fail( |
| 1729 | new GitCommandError({ |
| 1730 | operation: "GitVcsDriver.fetchRemote", |
| 1731 | command: "git", |
| 1732 | cwd: project.workspaceRoot, |
| 1733 | detail: "Git could not reach the remote.", |
| 1734 | exitCode: 128, |
| 1735 | }), |
| 1736 | ), |
| 1737 | }); |
| 1738 | return Effect.gen(function* () { |
| 1739 | fetchEntered = yield* Deferred.make<void>(); |
| 1740 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1741 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 1742 | const launched = yield* launches.launch( |
| 1743 | launchInput({ |
| 1744 | command: `command:launch:late-queued-follow-up-${ending}`, |
| 1745 | thread: `thread:launch:late-queued-follow-up-${ending}`, |
| 1746 | message: "First message", |
| 1747 | workspace: { type: "worktree", baseRef: "main", startFromOrigin: true }, |
| 1748 | }), |
| 1749 | ); |
| 1750 | if (ending === "interrupted") { |
| 1751 | yield* Deferred.await(fetchEntered); |
| 1752 | // An agent stop does not ask to hold the queue. |
| 1753 | yield* threads.dispatch({ |
| 1754 | type: "run.interrupt", |
| 1755 | commandId: CommandId.make("command:launch:late-queued-follow-up:interrupt"), |
| 1756 | threadId: launched.threadId, |
| 1757 | runId: launched.projection.runs[0]!.id, |
| 1758 | }); |
| 1759 | } |
| 1760 | yield* threads.streamStoredEventsFrom({ threadId: launched.threadId }).pipe( |
| 1761 | Stream.filter( |
| 1762 | (stored) => stored.event.type === "run.updated" && stored.event.payload.status === ending, |
| 1763 | ), |
| 1764 | Stream.runHead, |
| 1765 | ); |
| 1766 | |
| 1767 | // Queued while setup ran, delivered only after it failed: it must not start |
| 1768 | // in the project checkout. |
| 1769 | const queued = yield* threads.sendToThread({ |
| 1770 | projectId, |
| 1771 | commandId: CommandId.make(`command:launch:late-queued-follow-up-${ending}:send`), |
| 1772 | threadId: launched.threadId, |
| 1773 | messageId: MessageId.make(`message:launch:late-queued-follow-up-${ending}:send`), |
| 1774 | text: "Sent during setup", |
| 1775 | attachments: [], |
| 1776 | mode: "queue", |
| 1777 | createdBy: "user", |
| 1778 | creationSource: "web", |
| 1779 | }); |
| 1780 | assert.equal(queued.delivery, "queued"); |
| 1781 | const projection = yield* threads.getThreadProjection(launched.threadId); |
| 1782 | assert.equal(projection.runs.at(-1)?.status, "queued"); |
| 1783 | assert.equal(projection.runs.at(-1)?.queueHeld, true); |
| 1784 | }).pipe(Effect.provide(harness.layer)); |
| 1785 | }, |
| 1786 | ); |
| 1787 | |
| 1788 | it.effect("a retry reuses a recorded worktree without undoing its branch rename", () => { |
| 1789 | let setupFailures = 1; |
| 1790 | const harness = makeHarness({ |
| 1791 | runSetup: () => |
| 1792 | setupFailures-- > 0 |
| 1793 | ? Effect.fail(new Error("setup failed") as never) |
| 1794 | : Effect.succeed({ status: "no-script" as const }), |
| 1795 | }); |
| 1796 | return Effect.gen(function* () { |
| 1797 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1798 | const outbox = yield* EffectOutbox.EffectOutboxV2; |
| 1799 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 1800 | const tracker = yield* WorktreeSetupTracker.WorktreeSetupTracker; |
| 1801 | const launched = yield* launches.launch( |
| 1802 | launchInput({ |
| 1803 | command: "command:launch:reuse", |
| 1804 | thread: "thread:launch:reuse", |
| 1805 | message: "Reuse the worktree", |
| 1806 | workspace: { type: "worktree", baseRef: "main" }, |
| 1807 | }), |
| 1808 | ); |
| 1809 | yield* waitUntil(() => |
| 1810 | threads |
| 1811 | .getThreadProjection(launched.threadId) |
| 1812 | .pipe( |
| 1813 | Effect.map( |
| 1814 | (projection) => |
| 1815 | projection.runs[0]?.status === "failed" && |
| 1816 | projection.thread.branch === "generated-branch", |
| 1817 | ), |
| 1818 | ), |
| 1819 | ); |
| 1820 | const failed = yield* threads.getThreadProjection(launched.threadId); |
| 1821 | assert.equal(failed.thread.worktreePath, "/repo-worktrees/feature"); |
| 1822 | |
| 1823 | yield* launches.retryPreparation({ |
| 1824 | commandId: CommandId.make("command:launch:reuse:retry"), |
| 1825 | threadId: launched.threadId, |
| 1826 | runId: failed.runs[0]!.id, |
| 1827 | }); |
| 1828 | yield* waitUntil(() => |
| 1829 | outbox |
| 1830 | .listByCommandId(CommandId.make("command:launch:reuse:retry:release")) |
| 1831 | .pipe(Effect.map((effects) => effects.length === 1)), |
| 1832 | ); |
| 1833 | const retried = yield* threads.getThreadProjection(launched.threadId); |
| 1834 | assert.equal(retried.runs[0]?.status, "starting"); |
| 1835 | // The retry neither checks out again nor puts back the temporary branch. |
| 1836 | assert.equal(harness.createWorktree.mock.calls.length, 1); |
| 1837 | assert.equal(harness.renameBranch.mock.calls.length, 1); |
| 1838 | assert.equal(retried.thread.branch, "generated-branch"); |
| 1839 | assert.equal(retried.thread.worktreePath, "/repo-worktrees/feature"); |
| 1840 | // Clients see the retry's setup, not the failed one it replaced. |
| 1841 | const snapshot = yield* tracker.get(launched.threadId); |
| 1842 | assert.equal(snapshot?.phase, "done"); |
| 1843 | assert.deepEqual( |
| 1844 | snapshot?.stages.map((stage) => stage.id), |
| 1845 | ["setup-script", "agent"], |
| 1846 | ); |
| 1847 | }).pipe(Effect.provide(harness.layer)); |
| 1848 | }); |
| 1849 | |
| 1850 | it.effect("removes a worktree that failed before the thread recorded it", () => { |
| 1851 | const harness = makeHarness({ |
| 1852 | // A checkout that dies after claiming its directory. |
| 1853 | createWorktree: (_input, options) => |
| 1854 | (options?.progress?.onWorktreeClaimed?.("/repo-worktrees/partial") ?? Effect.void).pipe( |
| 1855 | Effect.andThen(Effect.fail(new Error("checkout failed") as never)), |
| 1856 | ), |
| 1857 | }); |
| 1858 | return Effect.gen(function* () { |
| 1859 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1860 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 1861 | const launched = yield* launches.launch( |
| 1862 | launchInput({ |
| 1863 | command: "command:launch:partial-worktree", |
| 1864 | thread: "thread:launch:partial-worktree", |
| 1865 | message: "Partial checkout", |
| 1866 | workspace: { type: "worktree", baseRef: "main" }, |
| 1867 | }), |
| 1868 | ); |
| 1869 | yield* waitUntil(() => |
| 1870 | threads |
| 1871 | .getThreadProjection(launched.threadId) |
| 1872 | .pipe(Effect.map((projection) => projection.runs[0]?.status === "failed")), |
| 1873 | ); |
| 1874 | const projection = yield* threads.getThreadProjection(launched.threadId); |
| 1875 | // Unrecorded, so a retry would create a second checkout beside it. |
| 1876 | assert.equal(projection.thread.worktreePath, null); |
| 1877 | assert.deepEqual( |
| 1878 | harness.removeWorktree.mock.calls.map(([input]) => input.path), |
| 1879 | ["/repo-worktrees/partial"], |
| 1880 | ); |
| 1881 | }).pipe(Effect.provide(harness.layer)); |
| 1882 | }); |
| 1883 | |
| 1884 | it.effect.each(["worktree", "setup"] as const)( |
| 1885 | "%s failure keeps the thread and message visible and emits failure items", |
| 1886 | (failurePoint) => |
| 1887 | Effect.gen(function* () { |
| 1888 | const failure = new Error(`${failurePoint} failed`); |
| 1889 | const harness = makeHarness( |
| 1890 | failurePoint === "worktree" |
| 1891 | ? { createWorktree: () => Effect.fail(failure as never) } |
| 1892 | : { runSetup: () => Effect.fail(failure as never) }, |
| 1893 | ); |
| 1894 | yield* Effect.gen(function* () { |
| 1895 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1896 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 1897 | const input = launchInput({ |
| 1898 | command: `command:launch:${failurePoint}-failure`, |
| 1899 | thread: `thread:launch:${failurePoint}-failure`, |
| 1900 | message: `Fail during ${failurePoint}`, |
| 1901 | workspace: { type: "worktree", baseRef: "main" }, |
| 1902 | }); |
| 1903 | const launched = yield* launches.launch(input); |
| 1904 | yield* waitUntil(() => |
| 1905 | threads |
| 1906 | .getThreadProjection(launched.threadId) |
| 1907 | .pipe(Effect.map((projection) => projection.runs[0]?.status === "failed")), |
| 1908 | ); |
| 1909 | const projection = yield* threads.getThreadProjection(launched.threadId); |
| 1910 | assert.equal(projection.messages[0]?.text, `Fail during ${failurePoint}`); |
| 1911 | assert.equal(projection.runs[0]?.status, "failed"); |
| 1912 | assert.equal( |
| 1913 | projection.turnItems.find((item) => item.type === "command_execution")?.status, |
| 1914 | "failed", |
| 1915 | ); |
| 1916 | assert.match( |
| 1917 | projection.turnItems.find((item) => item.type === "error")?.failure.message ?? "", |
| 1918 | new RegExp(`${failurePoint} failed`, "u"), |
| 1919 | ); |
| 1920 | }).pipe(Effect.provide(harness.layer)); |
| 1921 | }), |
| 1922 | ); |
| 1923 | |
| 1924 | it.effect("replays a server-allocated launch", () => |
| 1925 | Effect.gen(function* () { |
| 1926 | const setupEntered = yield* Deferred.make<void>(); |
| 1927 | const allowSetup = yield* Deferred.make<void>(); |
| 1928 | const harness = makeHarness({ |
| 1929 | runSetup: () => |
| 1930 | Deferred.succeed(setupEntered, undefined).pipe( |
| 1931 | Effect.andThen(Deferred.await(allowSetup)), |
| 1932 | Effect.as({ status: "no-script" as const }), |
| 1933 | ), |
| 1934 | }); |
| 1935 | yield* Effect.gen(function* () { |
| 1936 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1937 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 1938 | const { threadId: _unusedThreadId, ...rest } = launchInput({ |
| 1939 | command: "command:launch:allocated-retry", |
| 1940 | thread: "unused", |
| 1941 | message: "Only once", |
| 1942 | }); |
| 1943 | const first = yield* launches.launch(rest); |
| 1944 | yield* Deferred.await(setupEntered); |
| 1945 | const retry = yield* launches.launch(rest); |
| 1946 | assert.equal(first.threadId, retry.threadId); |
| 1947 | assert.isFalse(first.resumed); |
| 1948 | assert.isTrue(retry.resumed); |
| 1949 | assert.equal(harness.runSetup.mock.calls.length, 1); |
| 1950 | assert.equal(retry.projection.messages.length, 1); |
| 1951 | assert.equal(retry.projection.runs.length, 1); |
| 1952 | assert.equal(retry.projection.messages[0]?.id, first.projection.messages[0]?.id); |
| 1953 | assert.equal(retry.projection.runs[0]?.id, first.projection.runs[0]?.id); |
| 1954 | yield* Deferred.succeed(allowSetup, undefined); |
| 1955 | yield* threads.streamStoredEventsFrom({ threadId: first.threadId }).pipe( |
| 1956 | Stream.filter( |
| 1957 | (stored) => |
| 1958 | stored.commandId === CommandId.make(`${rest.commandId}:release`) && |
| 1959 | stored.event.type === "run.updated", |
| 1960 | ), |
| 1961 | Stream.runHead, |
| 1962 | ); |
| 1963 | const settled = yield* launches.launch(rest); |
| 1964 | assert.equal(settled.threadId, first.threadId); |
| 1965 | assert.isTrue(settled.resumed); |
| 1966 | assert.equal(settled.projection.messages[0]?.id, first.projection.messages[0]?.id); |
| 1967 | assert.equal(settled.projection.runs[0]?.id, first.projection.runs[0]?.id); |
| 1968 | assert.equal(harness.runSetup.mock.calls.length, 1); |
| 1969 | }).pipe(Effect.provide(harness.layer)); |
| 1970 | }), |
| 1971 | ); |
| 1972 | |
| 1973 | it.effect("refuses a launch when the project disables the model's provider instance", () => { |
| 1974 | const harness = makeHarness({ |
| 1975 | serverSettings: { |
| 1976 | projectSettingsOverrides: { |
| 1977 | [projectId]: { providerInstances: { [modelSelection.instanceId]: { enabled: false } } }, |
| 1978 | }, |
| 1979 | }, |
| 1980 | }); |
| 1981 | return Effect.gen(function* () { |
| 1982 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 1983 | const failed = yield* launches |
| 1984 | .launch( |
| 1985 | launchInput({ |
| 1986 | command: "command:launch:project-disabled-instance", |
| 1987 | thread: "thread:launch:project-disabled-instance", |
| 1988 | message: "Hello", |
| 1989 | }), |
| 1990 | ) |
| 1991 | .pipe(Effect.flip); |
| 1992 | if (failed._tag !== "ThreadManagementProviderInstanceDisabledError") { |
| 1993 | assert.fail(`expected ThreadManagementProviderInstanceDisabledError, got ${failed._tag}`); |
| 1994 | } |
| 1995 | assert.equal(failed.projectId, projectId); |
| 1996 | assert.equal(failed.instanceId, modelSelection.instanceId); |
| 1997 | }).pipe(Effect.provide(harness.layer)); |
| 1998 | }); |
| 1999 | |
| 2000 | it.effect("replays an accepted launch after the project disables the instance", () => { |
| 2001 | const harness = makeHarness(); |
| 2002 | return Effect.gen(function* () { |
| 2003 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2004 | const settings = yield* ServerSettings.ServerSettingsService; |
| 2005 | const input = launchInput({ |
| 2006 | command: "command:launch:retry-after-disable", |
| 2007 | thread: "thread:launch:retry-after-disable", |
| 2008 | message: "Hello", |
| 2009 | }); |
| 2010 | const first = yield* launches.launch(input); |
| 2011 | |
| 2012 | yield* settings.updateSettings({ |
| 2013 | projectSettingsOverrides: { |
| 2014 | [projectId]: { providerInstances: { [modelSelection.instanceId]: { enabled: false } } }, |
| 2015 | }, |
| 2016 | }); |
| 2017 | |
| 2018 | const retry = yield* launches.launch(input); |
| 2019 | assert.equal(retry.threadId, first.threadId); |
| 2020 | assert.isTrue(retry.resumed); |
| 2021 | |
| 2022 | const fresh = yield* launches |
| 2023 | .launch( |
| 2024 | launchInput({ |
| 2025 | command: "command:launch:fresh-after-disable", |
| 2026 | thread: "thread:launch:fresh-after-disable", |
| 2027 | message: "Hello", |
| 2028 | }), |
| 2029 | ) |
| 2030 | .pipe(Effect.flip); |
| 2031 | if (fresh._tag !== "ThreadManagementProviderInstanceDisabledError") { |
| 2032 | assert.fail(`expected ThreadManagementProviderInstanceDisabledError, got ${fresh._tag}`); |
| 2033 | } |
| 2034 | }).pipe(Effect.provide(harness.layer)); |
| 2035 | }); |
| 2036 | |
| 2037 | it.effect("rejects a server-allocated launch replay with a mismatching thread id", () => { |
| 2038 | const harness = makeHarness(); |
| 2039 | return Effect.gen(function* () { |
| 2040 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2041 | const { threadId: _unusedThreadId, ...rest } = launchInput({ |
| 2042 | command: "command:launch:allocated-mismatch", |
| 2043 | thread: "unused", |
| 2044 | message: "Mismatch", |
| 2045 | }); |
| 2046 | const first = yield* launches.launch(rest); |
| 2047 | const failed = yield* launches |
| 2048 | .launch({ |
| 2049 | ...rest, |
| 2050 | threadId: ThreadId.make("thread:launch:allocated-mismatch"), |
| 2051 | }) |
| 2052 | .pipe(Effect.flip); |
| 2053 | assert.notEqual(first.threadId, ThreadId.make("thread:launch:allocated-mismatch")); |
| 2054 | if (failed._tag !== "ThreadLaunchError") { |
| 2055 | assert.fail(`expected ThreadLaunchError, got ${failed._tag}`); |
| 2056 | } |
| 2057 | assert.equal(failed.operation, "create-thread"); |
| 2058 | assert.include(String(failed.cause), "cannot be replayed"); |
| 2059 | }).pipe(Effect.provide(harness.layer)); |
| 2060 | }); |
| 2061 | |
| 2062 | it.effect("rejects a server-allocated launch receipt from another project", () => { |
| 2063 | const harness = makeHarness(); |
| 2064 | return Effect.gen(function* () { |
| 2065 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2066 | const { threadId: _unusedThreadId, ...rest } = launchInput({ |
| 2067 | command: "command:launch:allocated-wrong-project", |
| 2068 | thread: "unused", |
| 2069 | message: "Wrong project", |
| 2070 | }); |
| 2071 | const first = yield* launches.launch(rest); |
| 2072 | const failed = yield* launches |
| 2073 | .launch({ |
| 2074 | ...rest, |
| 2075 | projectId: otherProjectId, |
| 2076 | }) |
| 2077 | .pipe(Effect.flip); |
| 2078 | if (failed._tag !== "ThreadLaunchError") { |
| 2079 | assert.fail(`expected ThreadLaunchError, got ${failed._tag}`); |
| 2080 | } |
| 2081 | assert.equal(failed.operation, "resolve-project"); |
| 2082 | assert.equal(failed.threadId, first.threadId); |
| 2083 | assert.equal(failed.cause, "Project identity changed."); |
| 2084 | }).pipe(Effect.provide(harness.layer)); |
| 2085 | }); |
| 2086 | |
| 2087 | it.effect("rejects a server-allocated launch retry after the thread is deleted", () => { |
| 2088 | const harness = makeHarness(); |
| 2089 | return Effect.gen(function* () { |
| 2090 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2091 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 2092 | const { threadId: _unusedThreadId, ...rest } = launchInput({ |
| 2093 | command: "command:launch:allocated-deleted", |
| 2094 | thread: "unused", |
| 2095 | message: "Deleted before retry", |
| 2096 | }); |
| 2097 | const first = yield* launches.launch(rest); |
| 2098 | yield* threads.dispatch({ |
| 2099 | type: "thread.delete", |
| 2100 | commandId: CommandId.make("command:launch:allocated-deleted:delete"), |
| 2101 | threadId: first.threadId, |
| 2102 | }); |
| 2103 | const failed = yield* launches.launch(rest).pipe(Effect.flip); |
| 2104 | if (failed._tag !== "ThreadLaunchError") { |
| 2105 | assert.fail(`expected ThreadLaunchError, got ${failed._tag}`); |
| 2106 | } |
| 2107 | assert.equal(failed.operation, "create-thread"); |
| 2108 | assert.equal(failed.threadId, first.threadId); |
| 2109 | assert.equal(failed.cause, "Thread not found."); |
| 2110 | const shells = yield* threads.listProjectThreads({ projectId, includeSubagents: true }); |
| 2111 | assert.equal(shells.length, 0); |
| 2112 | }).pipe(Effect.provide(harness.layer)); |
| 2113 | }); |
| 2114 | |
| 2115 | it.effect("does not treat an unrelated accepted command receipt as a launch", () => { |
| 2116 | const harness = makeHarness(); |
| 2117 | return Effect.gen(function* () { |
| 2118 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2119 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 2120 | const threadId = ThreadId.make("thread:launch:unrelated-receipt"); |
| 2121 | yield* threads.dispatch({ |
| 2122 | type: "thread.create", |
| 2123 | commandId: CommandId.make("command:launch:unrelated-receipt:create"), |
| 2124 | threadId, |
| 2125 | projectId, |
| 2126 | title: "Existing", |
| 2127 | modelSelection, |
| 2128 | runtimeMode: "full-access", |
| 2129 | interactionMode: "default", |
| 2130 | branch: null, |
| 2131 | worktreePath: null, |
| 2132 | createdBy: "user", |
| 2133 | creationSource: "web", |
| 2134 | }); |
| 2135 | yield* threads.dispatch({ |
| 2136 | type: "thread.metadata.update", |
| 2137 | commandId: CommandId.make("command:launch:unrelated-receipt"), |
| 2138 | threadId, |
| 2139 | expectedEmpty: true, |
| 2140 | }); |
| 2141 | const { threadId: _unusedThreadId, ...rest } = launchInput({ |
| 2142 | command: "command:launch:unrelated-receipt", |
| 2143 | thread: "unused", |
| 2144 | message: "Should not become a launch", |
| 2145 | }); |
| 2146 | const failed = yield* launches.launch(rest).pipe(Effect.flip); |
| 2147 | if (failed._tag !== "ThreadLaunchError") { |
| 2148 | assert.fail(`expected ThreadLaunchError, got ${failed._tag}`); |
| 2149 | } |
| 2150 | assert.equal(failed.operation, "create-thread"); |
| 2151 | assert.include(String(failed.cause), "cannot be replayed"); |
| 2152 | const projection = yield* threads.getThreadProjection(threadId); |
| 2153 | assert.equal(projection.messages.length, 0); |
| 2154 | assert.equal(projection.runs.length, 0); |
| 2155 | }).pipe(Effect.provide(harness.layer)); |
| 2156 | }); |
| 2157 | |
| 2158 | it.effect("bounds concurrent first launches to one thread per command", () => |
| 2159 | Effect.gen(function* () { |
| 2160 | const setupEntered = yield* Deferred.make<void>(); |
| 2161 | const allowSetup = yield* Deferred.make<void>(); |
| 2162 | const harness = makeHarness({ |
| 2163 | runSetup: () => |
| 2164 | Deferred.succeed(setupEntered, undefined).pipe( |
| 2165 | Effect.andThen(Deferred.await(allowSetup)), |
| 2166 | Effect.as({ status: "no-script" as const }), |
| 2167 | ), |
| 2168 | }); |
| 2169 | yield* Effect.gen(function* () { |
| 2170 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2171 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 2172 | const { threadId: _unusedThreadId, ...rest } = launchInput({ |
| 2173 | command: "command:launch:concurrent-allocated", |
| 2174 | thread: "unused", |
| 2175 | message: "Race me", |
| 2176 | }); |
| 2177 | // The command receipt is reserved atomically with the winning create, so |
| 2178 | // a loser either replays the winner's stored events or surfaces a |
| 2179 | // transient replay conflict that the next attempt resolves — the race |
| 2180 | // can never persist a second thread. |
| 2181 | const results = yield* Effect.all( |
| 2182 | [launches.launch(rest).pipe(Effect.exit), launches.launch(rest).pipe(Effect.exit)], |
| 2183 | { concurrency: "unbounded" }, |
| 2184 | ); |
| 2185 | const winner = results.find(Exit.isSuccess); |
| 2186 | assert.isDefined(winner); |
| 2187 | const threadId = winner!.value.threadId; |
| 2188 | for (const result of results) { |
| 2189 | if (Exit.isSuccess(result)) { |
| 2190 | assert.equal(result.value.threadId, threadId); |
| 2191 | continue; |
| 2192 | } |
| 2193 | const error = Cause.findErrorOption(result.cause).pipe(Option.getOrThrow); |
| 2194 | if (error._tag !== "ThreadLaunchError") { |
| 2195 | assert.fail(`expected ThreadLaunchError, got ${error._tag}`); |
| 2196 | } |
| 2197 | assert.equal(error.operation, "create-thread"); |
| 2198 | assert.include(String(error.cause), "cannot be replayed"); |
| 2199 | const retried = yield* launches.launch(rest); |
| 2200 | assert.equal(retried.threadId, threadId); |
| 2201 | } |
| 2202 | const projectThreads = yield* threads.listProjectThreads({ |
| 2203 | projectId, |
| 2204 | includeSubagents: false, |
| 2205 | }); |
| 2206 | assert.equal(projectThreads.length, 1); |
| 2207 | const projection = yield* threads.getThreadProjection(threadId); |
| 2208 | assert.equal(projection.messages.length, 1); |
| 2209 | assert.equal(projection.runs.length, 1); |
| 2210 | yield* Deferred.await(setupEntered); |
| 2211 | assert.equal(harness.runSetup.mock.calls.length, 1); |
| 2212 | yield* Deferred.succeed(allowSetup, undefined); |
| 2213 | }).pipe(Effect.provide(harness.layer)); |
| 2214 | }), |
| 2215 | ); |
| 2216 | |
| 2217 | it.effect("deduplicates retried launch side effects in-process", () => |
| 2218 | Effect.gen(function* () { |
| 2219 | const setupEntered = yield* Deferred.make<void>(); |
| 2220 | const allowSetup = yield* Deferred.make<void>(); |
| 2221 | const harness = makeHarness({ |
| 2222 | runSetup: () => |
| 2223 | Deferred.succeed(setupEntered, undefined).pipe( |
| 2224 | Effect.andThen(Deferred.await(allowSetup)), |
| 2225 | Effect.as({ status: "no-script" as const }), |
| 2226 | ), |
| 2227 | }); |
| 2228 | yield* Effect.gen(function* () { |
| 2229 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2230 | const input = launchInput({ |
| 2231 | command: "command:launch:retry", |
| 2232 | thread: "thread:launch:retry", |
| 2233 | message: "Only once", |
| 2234 | }); |
| 2235 | const [first, retry] = yield* Effect.all([launches.launch(input), launches.launch(input)], { |
| 2236 | concurrency: "unbounded", |
| 2237 | }); |
| 2238 | yield* Deferred.await(setupEntered); |
| 2239 | assert.equal(first.threadId, retry.threadId); |
| 2240 | assert.isFalse(first.resumed); |
| 2241 | assert.isTrue(retry.resumed); |
| 2242 | assert.equal(harness.runSetup.mock.calls.length, 1); |
| 2243 | yield* Deferred.succeed(allowSetup, undefined); |
| 2244 | }).pipe(Effect.provide(harness.layer)); |
| 2245 | }), |
| 2246 | ); |
| 2247 | |
| 2248 | it.effect("does not let a failing same-command caller strand a concurrent durable launch", () => |
| 2249 | Effect.gen(function* () { |
| 2250 | const setupEntered = yield* Deferred.make<void>(); |
| 2251 | const allowSetup = yield* Deferred.make<void>(); |
| 2252 | const harness = makeHarness({ |
| 2253 | runSetup: () => |
| 2254 | Deferred.succeed(setupEntered, undefined).pipe( |
| 2255 | Effect.andThen(Deferred.await(allowSetup)), |
| 2256 | Effect.as({ status: "no-script" as const }), |
| 2257 | ), |
| 2258 | }); |
| 2259 | yield* Effect.gen(function* () { |
| 2260 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2261 | const command = "command:launch:failed-owner-race"; |
| 2262 | const [failed, launched] = yield* Effect.all( |
| 2263 | [ |
| 2264 | launches |
| 2265 | .launch({ |
| 2266 | ...launchInput({ |
| 2267 | command, |
| 2268 | thread: "thread:launch:failed-owner-race", |
| 2269 | message: "This invalid reuse fails", |
| 2270 | }), |
| 2271 | reuseExistingThread: true, |
| 2272 | }) |
| 2273 | .pipe(Effect.exit), |
| 2274 | launches.launch( |
| 2275 | launchInput({ |
| 2276 | command, |
| 2277 | thread: "thread:launch:successful-peer", |
| 2278 | message: "This peer persists", |
| 2279 | }), |
| 2280 | ), |
| 2281 | ], |
| 2282 | { concurrency: "unbounded" }, |
| 2283 | ); |
| 2284 | assert.isTrue(Exit.isFailure(failed)); |
| 2285 | assert.equal(launched.projection.runs[0]?.status, "preparing"); |
| 2286 | const entered = yield* Deferred.await(setupEntered).pipe( |
| 2287 | Effect.timeoutOption(Duration.seconds(2)), |
| 2288 | ); |
| 2289 | assert.isTrue(Option.isSome(entered)); |
| 2290 | assert.equal(harness.runSetup.mock.calls.length, 1); |
| 2291 | yield* Deferred.succeed(allowSetup, undefined); |
| 2292 | }).pipe(Effect.provide(harness.layer)); |
| 2293 | }), |
| 2294 | ); |
| 2295 | |
| 2296 | it.effect("schedules an accepted preparing message exactly once across concurrent retries", () => |
| 2297 | Effect.gen(function* () { |
| 2298 | const setupEntered = yield* Deferred.make<void>(); |
| 2299 | const allowSetup = yield* Deferred.make<void>(); |
| 2300 | const harness = makeHarness({ |
| 2301 | runSetup: () => |
| 2302 | Deferred.succeed(setupEntered, undefined).pipe( |
| 2303 | Effect.andThen(Deferred.await(allowSetup)), |
| 2304 | Effect.as({ status: "no-script" as const }), |
| 2305 | ), |
| 2306 | }); |
| 2307 | yield* Effect.gen(function* () { |
| 2308 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2309 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 2310 | const input = launchInput({ |
| 2311 | command: "command:launch:accepted-before-fork", |
| 2312 | thread: "thread:launch:accepted-before-fork", |
| 2313 | message: "Resume preparation", |
| 2314 | }); |
| 2315 | const messageId = MessageId.make("message:launch:accepted-before-fork"); |
| 2316 | |
| 2317 | yield* threads.dispatch({ |
| 2318 | type: "thread.create", |
| 2319 | commandId: input.commandId, |
| 2320 | threadId: input.threadId, |
| 2321 | projectId: input.projectId, |
| 2322 | title: input.title, |
| 2323 | modelSelection: input.modelSelection, |
| 2324 | runtimeMode: input.runtimeMode, |
| 2325 | interactionMode: input.interactionMode, |
| 2326 | branch: null, |
| 2327 | worktreePath: null, |
| 2328 | createdBy: input.createdBy, |
| 2329 | creationSource: input.creationSource, |
| 2330 | }); |
| 2331 | yield* threads.dispatch({ |
| 2332 | type: "message.dispatch", |
| 2333 | commandId: CommandId.make(`${input.commandId}:initial-message`), |
| 2334 | threadId: input.threadId, |
| 2335 | messageId, |
| 2336 | text: "Resume preparation", |
| 2337 | attachments: [], |
| 2338 | modelSelection: input.modelSelection, |
| 2339 | dispatchMode: { type: "defer_start" }, |
| 2340 | createdBy: input.createdBy, |
| 2341 | creationSource: input.creationSource, |
| 2342 | }); |
| 2343 | const preparing = yield* threads.getThreadProjection(input.threadId); |
| 2344 | assert.equal(preparing.runs[0]?.status, "preparing"); |
| 2345 | |
| 2346 | const [first, second] = yield* Effect.all([launches.launch(input), launches.launch(input)], { |
| 2347 | concurrency: "unbounded", |
| 2348 | }); |
| 2349 | yield* Deferred.await(setupEntered); |
| 2350 | assert.isTrue(first.resumed); |
| 2351 | assert.isTrue(second.resumed); |
| 2352 | assert.equal(harness.runSetup.mock.calls.length, 1); |
| 2353 | yield* Deferred.succeed(allowSetup, undefined); |
| 2354 | }).pipe(Effect.provide(harness.layer)); |
| 2355 | }), |
| 2356 | ); |
| 2357 | |
| 2358 | it.effect("creates a strong provider-thread mapping for an imported native session", () => { |
| 2359 | const harness = makeHarness(); |
| 2360 | return Effect.gen(function* () { |
| 2361 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2362 | const input = { |
| 2363 | ...launchInput({ |
| 2364 | command: "command:launch:imported-native-session", |
| 2365 | thread: "thread:launch:imported-native-session", |
| 2366 | }), |
| 2367 | importedNativeThread: { |
| 2368 | ref: { |
| 2369 | driver: ProviderDriverKind.make("codex"), |
| 2370 | nativeId: "native-session-42", |
| 2371 | strength: "strong" as const, |
| 2372 | }, |
| 2373 | metadata: { |
| 2374 | title: "Native session", |
| 2375 | updatedAt: "2026-08-23T00:00:00Z", |
| 2376 | }, |
| 2377 | }, |
| 2378 | }; |
| 2379 | |
| 2380 | const launched = yield* launches.launch(input); |
| 2381 | |
| 2382 | assert.deepInclude(launched.projection.providerThreads[0], { |
| 2383 | id: IdAllocator.deriveProviderThread({ |
| 2384 | driver: input.importedNativeThread.ref.driver, |
| 2385 | providerInstanceId: modelSelection.instanceId, |
| 2386 | nativeThreadId: input.importedNativeThread.ref.nativeId, |
| 2387 | }), |
| 2388 | driver: input.importedNativeThread.ref.driver, |
| 2389 | providerInstanceId: modelSelection.instanceId, |
| 2390 | appThreadId: input.threadId, |
| 2391 | nativeThreadRef: input.importedNativeThread.ref, |
| 2392 | status: "not_loaded", |
| 2393 | nativeMetadata: input.importedNativeThread.metadata, |
| 2394 | }); |
| 2395 | assert.equal( |
| 2396 | launched.projection.thread.activeProviderThreadId, |
| 2397 | launched.projection.providerThreads[0]?.id, |
| 2398 | ); |
| 2399 | }).pipe(Effect.provide(harness.layer)); |
| 2400 | }); |
| 2401 | |
| 2402 | it.effect("shared intake preserves durable attachment bytes after a lost launch result", () => { |
| 2403 | const harness = makeHarness(); |
| 2404 | const layerFiles = ServerConfig.layerTest(process.cwd(), { prefix: "t3-message-intake-" }).pipe( |
| 2405 | Layer.provideMerge(NodeServices.layer), |
| 2406 | ); |
| 2407 | return Effect.gen(function* () { |
| 2408 | const config = yield* ServerConfig.ServerConfig; |
| 2409 | const fs = yield* FileSystem.FileSystem; |
| 2410 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2411 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 2412 | const pendingId = createPendingAttachmentId(); |
| 2413 | assert.isNotNull(pendingId); |
| 2414 | const attachment: ChatAttachment = { |
| 2415 | type: "image", |
| 2416 | id: ChatAttachmentId.make(pendingId), |
| 2417 | name: "image.png", |
| 2418 | mimeType: "image/png", |
| 2419 | sizeBytes: 4, |
| 2420 | }; |
| 2421 | const pendingPath = resolveAttachmentPath({ |
| 2422 | attachmentsDir: config.attachmentsDir, |
| 2423 | attachment, |
| 2424 | }); |
| 2425 | assert.isNotNull(pendingPath); |
| 2426 | yield* fs.makeDirectory(config.attachmentsDir, { recursive: true }); |
| 2427 | yield* fs.writeFile(pendingPath, new Uint8Array([1, 2, 3, 4])); |
| 2428 | const input = { |
| 2429 | ...launchInput({ command: "intake-launch", thread: "intake-thread" }), |
| 2430 | initialMessage: { |
| 2431 | messageId: MessageId.make("intake-first"), |
| 2432 | text: "First [file](t3-context://v1/file/intake-file)", |
| 2433 | context: { |
| 2434 | version: 1 as const, |
| 2435 | records: [ |
| 2436 | { |
| 2437 | version: 1 as const, |
| 2438 | contextId: ComposerContextId.make("intake-file"), |
| 2439 | kind: "file" as const, |
| 2440 | label: attachment.name, |
| 2441 | attachmentId: attachment.id, |
| 2442 | name: attachment.name, |
| 2443 | mimeType: attachment.mimeType, |
| 2444 | sizeBytes: attachment.sizeBytes, |
| 2445 | }, |
| 2446 | ], |
| 2447 | }, |
| 2448 | attachments: [attachment], |
| 2449 | }, |
| 2450 | }; |
| 2451 | const failed = yield* ThreadMessageIntake.launchThread(input).pipe( |
| 2452 | Effect.provideService(ThreadLaunch.ThreadLaunchService, { |
| 2453 | launch: (request) => |
| 2454 | launches.launch(request).pipe( |
| 2455 | Effect.andThen( |
| 2456 | new ThreadLaunch.ThreadLaunchError({ |
| 2457 | operation: "create-thread", |
| 2458 | commandId: request.commandId, |
| 2459 | projectId, |
| 2460 | cause: "lost result after acceptance", |
| 2461 | }), |
| 2462 | ), |
| 2463 | ), |
| 2464 | checkWorktreeBase: launches.checkWorktreeBase, |
| 2465 | retryPreparation: launches.retryPreparation, |
| 2466 | prepareRun: launches.prepareRun, |
| 2467 | }), |
| 2468 | Effect.flip, |
| 2469 | ); |
| 2470 | assert.equal(failed._tag, "ThreadLaunchError"); |
| 2471 | // The observer failed, but the real V2 message and its bytes were accepted. |
| 2472 | const accepted = yield* threads.getThreadProjection(input.threadId); |
| 2473 | const stored = accepted.messages.find( |
| 2474 | (message) => message.id === input.initialMessage.messageId, |
| 2475 | ); |
| 2476 | assert.isDefined(stored); |
| 2477 | assert.notEqual(stored.attachments[0]?.id, attachment.id); |
| 2478 | assert.equal( |
| 2479 | (stored.context?.records[0] as { attachmentId: string }).attachmentId, |
| 2480 | stored.attachments[0]?.id, |
| 2481 | ); |
| 2482 | const userItem = accepted.turnItems.find( |
| 2483 | (item) => item.type === "user_message" && item.messageId === stored.id, |
| 2484 | ); |
| 2485 | assert.ok(userItem?.type === "user_message"); |
| 2486 | assert.deepEqual(userItem.context, stored.context); |
| 2487 | const storedPath = resolveAttachmentPath({ |
| 2488 | attachmentsDir: config.attachmentsDir, |
| 2489 | attachment: stored.attachments[0]!, |
| 2490 | }); |
| 2491 | assert.isNotNull(storedPath); |
| 2492 | assert.deepEqual(yield* fs.readFile(storedPath), new Uint8Array([1, 2, 3, 4])); |
| 2493 | assert.deepEqual(yield* fs.readFile(pendingPath), new Uint8Array([1, 2, 3, 4])); |
| 2494 | |
| 2495 | const replayed = yield* ThreadMessageIntake.launchThread(input); |
| 2496 | assert.equal(replayed.projection.messages[0]?.id, stored.id); |
| 2497 | assert.deepEqual(replayed.projection.messages[0]?.attachments, stored.attachments); |
| 2498 | const claimedFiles = Effect.map(fs.readDirectory(config.attachmentsDir), (files) => |
| 2499 | files.filter((name) => !name.startsWith("pending-")), |
| 2500 | ); |
| 2501 | assert.equal((yield* claimedFiles).length, 1); |
| 2502 | |
| 2503 | const missingProject = yield* ThreadMessageIntake.launchThread({ |
| 2504 | ...input, |
| 2505 | commandId: CommandId.make("intake-no-project"), |
| 2506 | projectId: ProjectId.make("missing-project"), |
| 2507 | }).pipe(Effect.flip); |
| 2508 | assert.equal(missingProject._tag, "ThreadLaunchError"); |
| 2509 | assert.equal((yield* claimedFiles).length, 1); |
| 2510 | const missingThread = yield* ThreadMessageIntake.dispatchCommand({ |
| 2511 | type: "message.dispatch", |
| 2512 | commandId: CommandId.make("intake-no-thread"), |
| 2513 | threadId: ThreadId.make("missing-thread"), |
| 2514 | messageId: MessageId.make("intake-missing"), |
| 2515 | text: "Missing", |
| 2516 | attachments: [attachment], |
| 2517 | dispatchMode: { type: "start_immediately" }, |
| 2518 | createdBy: "user", |
| 2519 | creationSource: "web", |
| 2520 | }).pipe(Effect.flip); |
| 2521 | assert.equal(missingThread._tag, "OrchestratorProjectionError"); |
| 2522 | assert.equal((yield* claimedFiles).length, 1); |
| 2523 | |
| 2524 | // Both ordinary command intake (RPC) and send intake (MCP) use the same store. |
| 2525 | const dispatch = ThreadMessageIntake.dispatchCommand({ |
| 2526 | type: "message.dispatch", |
| 2527 | commandId: CommandId.make("intake-dispatch"), |
| 2528 | threadId: input.threadId, |
| 2529 | messageId: MessageId.make("intake-second"), |
| 2530 | text: "Second", |
| 2531 | attachments: [attachment], |
| 2532 | dispatchMode: { type: "queue_after_active" }, |
| 2533 | createdBy: "user", |
| 2534 | creationSource: "web", |
| 2535 | }); |
| 2536 | yield* dispatch; |
| 2537 | yield* dispatch; |
| 2538 | const queuedProjection = yield* threads.getThreadProjection(input.threadId); |
| 2539 | const queuedRun = queuedProjection.runs.find( |
| 2540 | (run) => run.userMessageId === MessageId.make("intake-second"), |
| 2541 | ); |
| 2542 | assert.isDefined(queuedRun); |
| 2543 | assert.equal(queuedRun.status, "queued"); |
| 2544 | const queuedMessage = queuedProjection.messages.find( |
| 2545 | (message) => message.id === queuedRun.userMessageId, |
| 2546 | ); |
| 2547 | assert.isDefined(queuedMessage); |
| 2548 | const file: ChatAttachment = { |
| 2549 | type: "file", |
| 2550 | id: ChatAttachmentId.make(createPendingAttachmentId("pdf")), |
| 2551 | name: "queued.pdf", |
| 2552 | mimeType: "application/pdf", |
| 2553 | sizeBytes: 4, |
| 2554 | }; |
| 2555 | const filePath = resolveAttachmentPath({ |
| 2556 | attachmentsDir: config.attachmentsDir, |
| 2557 | attachment: file, |
| 2558 | }); |
| 2559 | assert.isNotNull(filePath); |
| 2560 | yield* fs.writeFile(filePath, new Uint8Array([5, 6, 7, 8])); |
| 2561 | const edit = ThreadMessageIntake.dispatchCommand({ |
| 2562 | type: "queued-run.edit", |
| 2563 | commandId: CommandId.make("intake-edit"), |
| 2564 | threadId: input.threadId, |
| 2565 | runId: queuedRun.id, |
| 2566 | text: "Edited with a file", |
| 2567 | attachments: [...queuedMessage.attachments, file], |
| 2568 | }); |
| 2569 | yield* edit; |
| 2570 | yield* edit; |
| 2571 | const editedProjection = yield* threads.getThreadProjection(input.threadId); |
| 2572 | const editedMessage = editedProjection.messages.find( |
| 2573 | (message) => message.id === queuedRun.userMessageId, |
| 2574 | ); |
| 2575 | assert.isDefined(editedMessage); |
| 2576 | assert.equal(editedMessage.attachments.length, 2); |
| 2577 | assert.deepEqual(editedMessage.attachments[0], queuedMessage.attachments[0]); |
| 2578 | assert.notEqual(editedMessage.attachments[1]?.id, file.id); |
| 2579 | const durableFilePath = resolveAttachmentPath({ |
| 2580 | attachmentsDir: config.attachmentsDir, |
| 2581 | attachment: editedMessage.attachments[1]!, |
| 2582 | }); |
| 2583 | assert.isNotNull(durableFilePath); |
| 2584 | assert.deepEqual(yield* fs.readFile(durableFilePath), new Uint8Array([5, 6, 7, 8])); |
| 2585 | assert.deepEqual(yield* fs.readFile(filePath), new Uint8Array([5, 6, 7, 8])); |
| 2586 | const beforeRejectedEdit = (yield* claimedFiles).length; |
| 2587 | const rejectedEdit = yield* ThreadMessageIntake.dispatchCommand({ |
| 2588 | type: "queued-run.edit", |
| 2589 | commandId: CommandId.make("intake-edit-rejected"), |
| 2590 | threadId: input.threadId, |
| 2591 | runId: queuedRun.id, |
| 2592 | text: "", |
| 2593 | attachments: [file], |
| 2594 | }).pipe(Effect.flip); |
| 2595 | assert.equal(rejectedEdit._tag, "OrchestratorCommandRejectedError"); |
| 2596 | assert.equal((yield* claimedFiles).length, beforeRejectedEdit); |
| 2597 | assert.deepEqual(yield* fs.readFile(filePath), new Uint8Array([5, 6, 7, 8])); |
| 2598 | const send = ThreadMessageIntake.sendToThread({ |
| 2599 | commandId: CommandId.make("intake-send"), |
| 2600 | projectId, |
| 2601 | threadId: input.threadId, |
| 2602 | messageId: MessageId.make("intake-third"), |
| 2603 | text: "Third", |
| 2604 | attachments: [attachment], |
| 2605 | mode: "queue", |
| 2606 | createdBy: "agent", |
| 2607 | creationSource: "mcp", |
| 2608 | }); |
| 2609 | yield* send; |
| 2610 | yield* send; |
| 2611 | assert.equal((yield* claimedFiles).length, 4); |
| 2612 | const final = yield* threads.getThreadProjection(input.threadId); |
| 2613 | assert.equal(final.messages.length, 3); |
| 2614 | for (const message of final.messages) { |
| 2615 | const path = resolveAttachmentPath({ |
| 2616 | attachmentsDir: config.attachmentsDir, |
| 2617 | attachment: message.attachments[0]!, |
| 2618 | }); |
| 2619 | assert.isNotNull(path); |
| 2620 | assert.deepEqual(yield* fs.readFile(path), new Uint8Array([1, 2, 3, 4])); |
| 2621 | } |
| 2622 | }).pipe(Effect.provide(Layer.mergeAll(harness.layer, layerFiles))); |
| 2623 | }); |
| 2624 | |
| 2625 | it.effect("cancels tracked setup before provider work is released", () => |
| 2626 | Effect.gen(function* () { |
| 2627 | const entered = yield* Deferred.make<void>(); |
| 2628 | const harness = makeHarness({ |
| 2629 | runSetup: () => Deferred.succeed(entered, undefined).pipe(Effect.andThen(Effect.never)), |
| 2630 | }); |
| 2631 | yield* Effect.gen(function* () { |
| 2632 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2633 | const tracker = yield* WorktreeSetupTracker.WorktreeSetupTracker; |
| 2634 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 2635 | const outbox = yield* EffectOutbox.EffectOutboxV2; |
| 2636 | const input = launchInput({ |
| 2637 | command: "launch:cancel-tracked", |
| 2638 | thread: "thread:cancel-tracked", |
| 2639 | message: "Start", |
| 2640 | workspace: { type: "worktree", baseRef: "main" }, |
| 2641 | }); |
| 2642 | const launched = yield* launches.launch(input); |
| 2643 | yield* Deferred.await(entered); |
| 2644 | assert.equal((yield* tracker.get(launched.threadId))?.phase, "running"); |
| 2645 | assert.isTrue(yield* tracker.cancel(launched.threadId)); |
| 2646 | assert.equal((yield* tracker.get(launched.threadId))?.phase, "cancelled"); |
| 2647 | const projection = yield* threads.getThreadProjection(launched.threadId); |
| 2648 | assert.equal(projection.runs[0]?.status, "failed"); |
| 2649 | assert.isNull(projection.thread.worktreePath); |
| 2650 | assert.isEmpty(yield* outbox.listByCommandId(CommandId.make(`${input.commandId}:release`))); |
| 2651 | }).pipe(Effect.provide(harness.layer)); |
| 2652 | }), |
| 2653 | ); |
| 2654 | |
| 2655 | it.effect.each([0, 1])("releases an async setup before its completion with exit %s", (exitCode) => |
| 2656 | Effect.gen(function* () { |
| 2657 | const completion = yield* Deferred.make<{ exitCode: number | null; durationMs: number }>(); |
| 2658 | const harness = makeHarness({ |
| 2659 | runSetup: () => |
| 2660 | Effect.succeed({ |
| 2661 | status: "started" as const, |
| 2662 | async: true, |
| 2663 | scriptId: "setup", |
| 2664 | scriptName: "Setup", |
| 2665 | scriptCommand: "vp install", |
| 2666 | terminalId: "setup", |
| 2667 | cwd: "/repo-worktrees/feature", |
| 2668 | completion: Deferred.await(completion), |
| 2669 | }), |
| 2670 | }); |
| 2671 | yield* Effect.gen(function* () { |
| 2672 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2673 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 2674 | const tracker = yield* WorktreeSetupTracker.WorktreeSetupTracker; |
| 2675 | const launched = yield* launches.launch( |
| 2676 | launchInput({ |
| 2677 | command: `command:launch:async-${exitCode}`, |
| 2678 | thread: `thread:launch:async-${exitCode}`, |
| 2679 | message: "Start during setup", |
| 2680 | workspace: { type: "worktree", baseRef: "main" }, |
| 2681 | }), |
| 2682 | ); |
| 2683 | yield* tracker.stream(launched.threadId).pipe( |
| 2684 | Stream.filter( |
| 2685 | (snapshot) => |
| 2686 | snapshot?.stages.some((stage) => stage.id === "agent" && stage.status === "done") === |
| 2687 | true, |
| 2688 | ), |
| 2689 | Stream.runHead, |
| 2690 | ); |
| 2691 | const running = yield* threads.getThreadProjection(launched.threadId); |
| 2692 | assert.equal(running.runs[0]?.status, "starting"); |
| 2693 | assert.equal((yield* tracker.get(launched.threadId))?.phase, "running"); |
| 2694 | yield* Deferred.succeed(completion, { exitCode, durationMs: 1 }); |
| 2695 | yield* tracker.stream(launched.threadId).pipe( |
| 2696 | Stream.filter((snapshot) => snapshot?.phase === "done"), |
| 2697 | Stream.runHead, |
| 2698 | ); |
| 2699 | const settled = yield* tracker.get(launched.threadId); |
| 2700 | assert.equal( |
| 2701 | settled?.stages.find((stage) => stage.id === "setup-script")?.status, |
| 2702 | exitCode === 0 ? "done" : "failed", |
| 2703 | ); |
| 2704 | assert.equal( |
| 2705 | (yield* threads.getThreadProjection(launched.threadId)).runs[0]?.status, |
| 2706 | "starting", |
| 2707 | ); |
| 2708 | }).pipe(Effect.provide(harness.layer)); |
| 2709 | }), |
| 2710 | ); |
| 2711 | |
| 2712 | it.effect.each([ |
| 2713 | { name: "is missing", baseRef: "t3/renamed", startFromOrigin: false, origin: true }, |
| 2714 | { |
| 2715 | name: "has no origin to come from", |
| 2716 | baseRef: "t3/renamed", |
| 2717 | startFromOrigin: true, |
| 2718 | origin: false, |
| 2719 | }, |
| 2720 | { |
| 2721 | name: "is a previous checkout there never was", |
| 2722 | baseRef: "-", |
| 2723 | startFromOrigin: false, |
| 2724 | origin: true, |
| 2725 | }, |
| 2726 | ])("refuses a new worktree whose base ref $name", (testCase) => |
| 2727 | Effect.gen(function* () { |
| 2728 | const harness = makeHarness({ |
| 2729 | hasCommit: ({ refName }) => Effect.succeed(refName === "HEAD"), |
| 2730 | git: { |
| 2731 | remoteExists: () => Effect.succeed(testCase.origin), |
| 2732 | hasRefNamed: () => Effect.succeed(false), |
| 2733 | }, |
| 2734 | }); |
| 2735 | yield* Effect.gen(function* () { |
| 2736 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2737 | const error = yield* launches |
| 2738 | .checkWorktreeBase({ |
| 2739 | projectId, |
| 2740 | workspaceStrategy: { |
| 2741 | type: "worktree", |
| 2742 | baseRef: testCase.baseRef, |
| 2743 | startFromOrigin: testCase.startFromOrigin, |
| 2744 | }, |
| 2745 | }) |
| 2746 | .pipe(Effect.flip); |
| 2747 | assert.deepInclude(error, { |
| 2748 | _tag: "ThreadLaunchBaseRefError", |
| 2749 | projectId, |
| 2750 | baseRef: testCase.baseRef, |
| 2751 | }); |
| 2752 | assert.equal( |
| 2753 | error.message, |
| 2754 | `Base ref "${testCase.baseRef}" does not resolve to a commit in project "Project" (/repo).`, |
| 2755 | ); |
| 2756 | }).pipe(Effect.provide(harness.layer)); |
| 2757 | }), |
| 2758 | ); |
| 2759 | |
| 2760 | it.effect.each([ |
| 2761 | { name: "a base ref that resolves", baseRef: "main", commits: ["main", "HEAD"] }, |
| 2762 | // Provisioning fetches it; the ref may exist nowhere else yet. |
| 2763 | { |
| 2764 | name: "a base ref only origin has", |
| 2765 | baseRef: "remote-only", |
| 2766 | startFromOrigin: true, |
| 2767 | commits: ["HEAD"], |
| 2768 | }, |
| 2769 | // `git worktree add` starts from a remote-tracking branch of that name. |
| 2770 | { |
| 2771 | name: "a base ref a remote branch carries", |
| 2772 | baseRef: "remote-only", |
| 2773 | commits: ["HEAD"], |
| 2774 | remoteBranch: true, |
| 2775 | }, |
| 2776 | // Provisioning runs such a repository without a worktree. |
| 2777 | { name: "a repository with no commits yet", baseRef: "main", commits: [] }, |
| 2778 | { name: "the previous checkout", baseRef: "-", commits: ["@{-1}", "HEAD"] }, |
| 2779 | { name: "a commit search", baseRef: ":/fix", commits: ["HEAD"] }, |
| 2780 | { name: "a folder that is not a repository", baseRef: "main", commits: null }, |
| 2781 | ])("leaves $name to the launch", (testCase) => |
| 2782 | Effect.gen(function* () { |
| 2783 | const { commits } = testCase; |
| 2784 | const harness = makeHarness({ |
| 2785 | git: { hasRefNamed: () => Effect.succeed(testCase.remoteBranch === true) }, |
| 2786 | hasCommit: ({ refName }) => |
| 2787 | commits === null |
| 2788 | ? Effect.fail( |
| 2789 | new GitCommandError({ |
| 2790 | operation: "GitWorkflowService.hasCommit", |
| 2791 | command: "git rev-parse", |
| 2792 | cwd: "/repo", |
| 2793 | detail: "Not a Git repository.", |
| 2794 | }), |
| 2795 | ) |
| 2796 | : Effect.succeed(commits.includes(refName)), |
| 2797 | }); |
| 2798 | yield* Effect.gen(function* () { |
| 2799 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2800 | yield* launches.checkWorktreeBase({ |
| 2801 | projectId, |
| 2802 | workspaceStrategy: { |
| 2803 | type: "worktree", |
| 2804 | baseRef: testCase.baseRef, |
| 2805 | ...(testCase.startFromOrigin === undefined |
| 2806 | ? {} |
| 2807 | : { startFromOrigin: testCase.startFromOrigin }), |
| 2808 | }, |
| 2809 | }); |
| 2810 | }).pipe(Effect.provide(harness.layer)); |
| 2811 | }), |
| 2812 | ); |
| 2813 | |
| 2814 | it.effect("replays an accepted send after the project disables the instance", () => { |
| 2815 | const harness = makeHarness(); |
| 2816 | const layerFiles = ServerConfig.layerTest(process.cwd(), { prefix: "t3-send-retry-" }).pipe( |
| 2817 | Layer.provideMerge(NodeServices.layer), |
| 2818 | ); |
| 2819 | return Effect.gen(function* () { |
| 2820 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2821 | const settings = yield* ServerSettings.ServerSettingsService; |
| 2822 | const created = yield* launches.launch( |
| 2823 | launchInput({ |
| 2824 | command: "command:send:retry-setup", |
| 2825 | thread: "thread:send:retry-after-disable", |
| 2826 | }), |
| 2827 | ); |
| 2828 | |
| 2829 | const sendInput = { |
| 2830 | commandId: CommandId.make("command:send:retry-after-disable"), |
| 2831 | projectId, |
| 2832 | threadId: created.threadId, |
| 2833 | messageId: MessageId.make("message:send:retry-after-disable"), |
| 2834 | text: "Hello again", |
| 2835 | attachments: [], |
| 2836 | mode: "auto" as const, |
| 2837 | createdBy: "user" as const, |
| 2838 | creationSource: "web" as const, |
| 2839 | }; |
| 2840 | const first = yield* ThreadMessageIntake.sendToThread(sendInput); |
| 2841 | |
| 2842 | yield* settings.updateSettings({ |
| 2843 | projectSettingsOverrides: { |
| 2844 | [projectId]: { providerInstances: { [modelSelection.instanceId]: { enabled: false } } }, |
| 2845 | }, |
| 2846 | }); |
| 2847 | |
| 2848 | const retry = yield* ThreadMessageIntake.sendToThread(sendInput); |
| 2849 | assert.equal(retry.message.id, first.message.id); |
| 2850 | |
| 2851 | const fresh = yield* ThreadMessageIntake.sendToThread({ |
| 2852 | ...sendInput, |
| 2853 | commandId: CommandId.make("command:send:fresh-after-disable"), |
| 2854 | messageId: MessageId.make("message:send:fresh-after-disable"), |
| 2855 | }).pipe(Effect.flip); |
| 2856 | if (fresh._tag !== "ThreadManagementProviderInstanceDisabledError") { |
| 2857 | assert.fail(`expected ThreadManagementProviderInstanceDisabledError, got ${fresh._tag}`); |
| 2858 | } |
| 2859 | }).pipe(Effect.provide(Layer.mergeAll(harness.layer, layerFiles))); |
| 2860 | }); |
| 2861 | |
| 2862 | it.effect("replays an accepted message.dispatch after the project disables the instance", () => { |
| 2863 | const harness = makeHarness(); |
| 2864 | const layerFiles = ServerConfig.layerTest(process.cwd(), { prefix: "t3-dispatch-retry-" }).pipe( |
| 2865 | Layer.provideMerge(NodeServices.layer), |
| 2866 | ); |
| 2867 | return Effect.gen(function* () { |
| 2868 | const launches = yield* ThreadLaunch.ThreadLaunchService; |
| 2869 | const settings = yield* ServerSettings.ServerSettingsService; |
| 2870 | const created = yield* launches.launch( |
| 2871 | launchInput({ |
| 2872 | command: "command:dispatch:retry-setup", |
| 2873 | thread: "thread:dispatch:retry-after-disable", |
| 2874 | }), |
| 2875 | ); |
| 2876 | |
| 2877 | const dispatchInput = { |
| 2878 | type: "message.dispatch" as const, |
| 2879 | commandId: CommandId.make("command:dispatch:retry-after-disable"), |
| 2880 | threadId: created.threadId, |
| 2881 | messageId: MessageId.make("message:dispatch:retry-after-disable"), |
| 2882 | text: "Hello again", |
| 2883 | attachments: [], |
| 2884 | dispatchMode: { type: "queue_after_active" as const }, |
| 2885 | createdBy: "user" as const, |
| 2886 | creationSource: "web" as const, |
| 2887 | }; |
| 2888 | const first = yield* ThreadMessageIntake.dispatchCommand(dispatchInput); |
| 2889 | |
| 2890 | yield* settings.updateSettings({ |
| 2891 | projectSettingsOverrides: { |
| 2892 | [projectId]: { providerInstances: { [modelSelection.instanceId]: { enabled: false } } }, |
| 2893 | }, |
| 2894 | }); |
| 2895 | |
| 2896 | const retry = yield* ThreadMessageIntake.dispatchCommand(dispatchInput); |
| 2897 | assert.deepEqual(retry, first); |
| 2898 | |
| 2899 | const fresh = yield* ThreadMessageIntake.dispatchCommand({ |
| 2900 | ...dispatchInput, |
| 2901 | commandId: CommandId.make("command:dispatch:fresh-after-disable"), |
| 2902 | messageId: MessageId.make("message:dispatch:fresh-after-disable"), |
| 2903 | }).pipe(Effect.flip); |
| 2904 | if (fresh._tag !== "ThreadManagementProviderInstanceDisabledError") { |
| 2905 | assert.fail(`expected ThreadManagementProviderInstanceDisabledError, got ${fresh._tag}`); |
| 2906 | } |
| 2907 | }).pipe(Effect.provide(Layer.mergeAll(harness.layer, layerFiles))); |
| 2908 | }); |
| 2909 | |
| 2910 | it.effect.each( |
| 2911 | (["codex", "claudeAgent"] as const).flatMap((driver) => |
| 2912 | [ |
| 2913 | { failSetup: false, failResolve: false }, |
| 2914 | { failSetup: true, failResolve: false }, |
| 2915 | { failSetup: false, failResolve: true }, |
| 2916 | { failSetup: false, failResolve: false, cancelSetup: true }, |
| 2917 | { failSetup: false, failResolve: false, cancelSetup: true, cancelAsync: true }, |
| 2918 | ].map((failures) => ({ driver, cancelSetup: false, cancelAsync: false, ...failures })), |
| 2919 | ), |
| 2920 | )( |
| 2921 | "prepares $driver delegated worktrees before execution, setup failure=$failSetup, commit failure=$failResolve, cancellation=$cancelSetup, async=$cancelAsync", |
| 2922 | ({ driver, failSetup, failResolve, cancelSetup, cancelAsync }) => |
| 2923 | Effect.gen(function* () { |
| 2924 | const setupEntered = yield* Deferred.make<void>(); |
| 2925 | const allowSetup = yield* Deferred.make<void>(); |
| 2926 | const terminalClosed = yield* Deferred.make<void>(); |
| 2927 | const childInstanceId = ProviderInstanceId.make(`child-${driver}`); |
| 2928 | const parentWorktree = driver === "codex" ? null : "/parent-worktree"; |
| 2929 | const resolveCommit = vi.fn(() => |
| 2930 | failResolve |
| 2931 | ? Effect.fail( |
| 2932 | new GitCommandError({ |
| 2933 | operation: "resolveCommit", |
| 2934 | command: "git", |
| 2935 | cwd: "/private-checkout", |
| 2936 | detail: "private Git failure details", |
| 2937 | }), |
| 2938 | ) |
| 2939 | : Effect.succeed({ commitSha: "parent-local-commit" }), |
| 2940 | ); |
| 2941 | const harness = makeHarness({ |
| 2942 | closeTerminal: () => Deferred.succeed(terminalClosed, undefined).pipe(Effect.asVoid), |
| 2943 | adapters: [ |
| 2944 | adapter, |
| 2945 | { ...adapter, instanceId: childInstanceId, driver: ProviderDriverKind.make(driver) }, |
| 2946 | ], |
| 2947 | serverSettings: { |
| 2948 | providerInstances: { |
| 2949 | [childInstanceId]: { driver: ProviderDriverKind.make(driver), config: {} }, |
| 2950 | }, |
| 2951 | }, |
| 2952 | hasCommit: () => Effect.succeed(true), |
| 2953 | providers: [ |
| 2954 | { |
| 2955 | instanceId: childInstanceId, |
| 2956 | driver: ProviderDriverKind.make(driver), |
| 2957 | enabled: true, |
| 2958 | installed: true, |
| 2959 | version: "test", |
| 2960 | checkedAt: "2026-10-13T00:00:00.000Z", |
| 2961 | slashCommands: [], |
| 2962 | skills: [], |
| 2963 | status: "ready", |
| 2964 | auth: { status: "authenticated" }, |
| 2965 | models: [{ slug: "child-model", name: "Child", isCustom: true, capabilities: null }], |
| 2966 | }, |
| 2967 | ], |
| 2968 | runSetup: () => |
| 2969 | cancelSetup |
| 2970 | ? Effect.succeed({ |
| 2971 | status: "started" as const, |
| 2972 | async: cancelAsync, |
| 2973 | scriptId: "setup", |
| 2974 | scriptName: "Setup", |
| 2975 | scriptCommand: "setup", |
| 2976 | terminalId: "setup", |
| 2977 | cwd: "/repo-worktrees/feature", |
| 2978 | completion: Deferred.succeed(setupEntered, undefined).pipe( |
| 2979 | Effect.andThen(Deferred.await(allowSetup)), |
| 2980 | Effect.as({ exitCode: 0, durationMs: 0 }), |
| 2981 | ), |
| 2982 | }) |
| 2983 | : Deferred.succeed(setupEntered, undefined).pipe( |
| 2984 | Effect.andThen(Deferred.await(allowSetup)), |
| 2985 | Effect.andThen( |
| 2986 | failSetup |
| 2987 | ? Effect.fail( |
| 2988 | new ProjectSetupScriptRunner.ProjectSetupScriptOperationError({ |
| 2989 | threadId: "child", |
| 2990 | worktreePath: "/repo-worktrees/feature", |
| 2991 | operation: "writeCommand", |
| 2992 | cause: "setup failed", |
| 2993 | }), |
| 2994 | ) |
| 2995 | : Effect.succeed({ status: "no-script" as const }), |
| 2996 | ), |
| 2997 | ), |
| 2998 | }); |
| 2999 | const layerService = OrchestratorMcpService.layer.pipe( |
| 3000 | Layer.provide(NodeCrypto.layer), |
| 3001 | Layer.provide(harness.layer), |
| 3002 | Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({ resolveCommit })), |
| 3003 | Layer.provide(Layer.mock(ScheduledTasks.ScheduledTaskService)({})), |
| 3004 | Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})), |
| 3005 | ); |
| 3006 | yield* Effect.gen(function* () { |
| 3007 | const service = yield* OrchestratorMcpService.OrchestratorMcpService; |
| 3008 | const threads = yield* ThreadManagement.ThreadManagementService; |
| 3009 | const parentThreadId = ThreadId.make("thread:delegated-worktree-parent"); |
| 3010 | yield* threads.dispatch({ |
| 3011 | type: "thread.create", |
| 3012 | commandId: CommandId.make("create-parent"), |
| 3013 | threadId: parentThreadId, |
| 3014 | projectId, |
| 3015 | title: "Parent", |
| 3016 | modelSelection, |
| 3017 | runtimeMode: "full-access", |
| 3018 | interactionMode: "default", |
| 3019 | branch: "parent-feature", |
| 3020 | worktreePath: parentWorktree, |
| 3021 | createdBy: "user", |
| 3022 | creationSource: "web", |
| 3023 | }); |
| 3024 | yield* threads.dispatch({ |
| 3025 | type: "message.dispatch", |
| 3026 | commandId: CommandId.make("start-parent"), |
| 3027 | threadId: parentThreadId, |
| 3028 | messageId: MessageId.make("parent-message"), |
| 3029 | text: "Delegate work", |
| 3030 | attachments: [], |
| 3031 | dispatchMode: { type: "start_immediately" }, |
| 3032 | createdBy: "user", |
| 3033 | creationSource: "web", |
| 3034 | }); |
| 3035 | const scope = { |
| 3036 | environmentId: EnvironmentId.make("delegated-worktree-test"), |
| 3037 | requestNamespace: "parent-session", |
| 3038 | thread: { |
| 3039 | threadId: parentThreadId, |
| 3040 | providerSessionId: "parent-session", |
| 3041 | providerInstanceId: modelSelection.instanceId, |
| 3042 | }, |
| 3043 | client: undefined, |
| 3044 | capabilities: new Set(["orchestration"] as const), |
| 3045 | issuedAt: 1, |
| 3046 | }; |
| 3047 | const input = { |
| 3048 | task: "Implement the child task", |
| 3049 | target: { providerInstanceId: childInstanceId, model: "child-model" }, |
| 3050 | isolation: "worktree" as const, |
| 3051 | clientRequestId: "isolated-child", |
| 3052 | }; |
| 3053 | if (failResolve) { |
| 3054 | const error = yield* service.delegateTask(scope, input).pipe(Effect.flip); |
| 3055 | assert.equal(error.code, "invalid_request"); |
| 3056 | assert.equal(error.message, "Unable to resolve the parent checkout commit."); |
| 3057 | assert.isEmpty((yield* threads.getThreadProjection(parentThreadId)).subagents); |
| 3058 | assert.equal(harness.createWorktree.mock.calls.length, 0); |
| 3059 | return; |
| 3060 | } |
| 3061 | const delegated = yield* service.delegateTask(scope, input); |
| 3062 | yield* Deferred.await(setupEntered); |
| 3063 | const child = yield* threads.getThreadProjection(delegated.childThreadId); |
| 3064 | assert.equal(child.thread.lineage.parentThreadId, parentThreadId); |
| 3065 | assert.equal(child.thread.worktreePath, "/repo-worktrees/feature"); |
| 3066 | assert.equal(child.runs[0]?.status, cancelAsync ? "starting" : "preparing"); |
| 3067 | assert.equal(child.thread.modelSelection.instanceId, childInstanceId); |
| 3068 | assert.equal(child.thread.modelSelection.model, "child-model"); |
| 3069 | assert.deepEqual(resolveCommit.mock.calls, [ |
| 3070 | [{ cwd: parentWorktree ?? "/repo", revision: "HEAD" }], |
| 3071 | ]); |
| 3072 | assert.equal(harness.createWorktree.mock.calls[0]?.[0].refName, "parent-local-commit"); |
| 3073 | assert.equal((yield* service.delegateTask(scope, input)).taskId, delegated.taskId); |
| 3074 | assert.equal(harness.createWorktree.mock.calls.length, 1); |
| 3075 | assert.equal(resolveCommit.mock.calls.length, 1); |
| 3076 | if (!cancelAsync) assert.isEmpty(child.checkpointScopes); |
| 3077 | const parent = yield* threads.getThreadProjection(parentThreadId); |
| 3078 | assert.equal(parent.thread.worktreePath, parentWorktree); |
| 3079 | assert.equal(parent.thread.branch, "parent-feature"); |
| 3080 | if (cancelSetup) { |
| 3081 | yield* service.cancelTask(scope, { taskId: delegated.taskId }); |
| 3082 | yield* Deferred.await(terminalClosed); |
| 3083 | if (!cancelAsync) { |
| 3084 | yield* threads.streamStoredEventsFrom({ threadId: delegated.childThreadId }).pipe( |
| 3085 | Stream.filter( |
| 3086 | (stored) => |
| 3087 | stored.event.type === "thread.metadata-updated" && |
| 3088 | stored.event.payload.worktreePath === null, |
| 3089 | ), |
| 3090 | Stream.runHead, |
| 3091 | ); |
| 3092 | } |
| 3093 | assert.equal(harness.closeTerminal.mock.calls.length, 1); |
| 3094 | assert.equal(harness.removeWorktree.mock.calls.length, cancelAsync ? 0 : 1); |
| 3095 | const cancelled = yield* threads.getThreadProjection(delegated.childThreadId); |
| 3096 | assert.equal(cancelled.runs[0]?.status, "interrupted"); |
| 3097 | if (cancelAsync) { |
| 3098 | assert.equal(cancelled.thread.worktreePath, "/repo-worktrees/feature"); |
| 3099 | assert.equal(cancelled.checkpointScopes[0]?.cwd, "/repo-worktrees/feature"); |
| 3100 | } else { |
| 3101 | assert.isNull(cancelled.thread.worktreePath); |
| 3102 | assert.isEmpty(cancelled.checkpointScopes); |
| 3103 | } |
| 3104 | return; |
| 3105 | } |
| 3106 | yield* Deferred.succeed(allowSetup, undefined); |
| 3107 | yield* threads |
| 3108 | .streamStoredEventsFrom({ threadId: delegated.childThreadId, eventType: "run.updated" }) |
| 3109 | .pipe( |
| 3110 | Stream.filter( |
| 3111 | (stored) => |
| 3112 | stored.event.type === "run.updated" && |
| 3113 | stored.event.payload.status === (failSetup ? "failed" : "starting"), |
| 3114 | ), |
| 3115 | Stream.runHead, |
| 3116 | ); |
| 3117 | const prepared = yield* threads.getThreadProjection(delegated.childThreadId); |
| 3118 | if (failSetup) { |
| 3119 | assert.isEmpty(prepared.checkpointScopes); |
| 3120 | } else { |
| 3121 | assert.equal(prepared.checkpointScopes[0]?.cwd, "/repo-worktrees/feature"); |
| 3122 | } |
| 3123 | }).pipe(Effect.provide(Layer.merge(harness.layer, layerService))); |
| 3124 | }), |
| 3125 | ); |