| author | |
| committer | |
| log | 477282263a9115f911380b5aec7d4973267e4edc |
| tree | 3fd42280314dc853955e9eba9aeedd3adbea1f73 |
| parent | 5f7294d452a5c6d6e0b7bd2c992478f830fbe212 |
| signature | Signed by PGP key B5690EEEBB952194 |
8 files changed, 312 insertions(+), 45 deletions(-)
apps/mobile/src/features/threads/ThreadDetailScreen.tsx+14-5| ... | @@ -173,9 +173,15 @@ export interface ThreadDetailScreenProps { | ... | @@ -173,9 +173,15 @@ export interface ThreadDetailScreenProps { |
| 173 | * The server has not created this thread yet. "preparing" runs while the | 173 | * The server has not created this thread yet. "preparing" runs while the |
| 174 | * queued creation is delivered (a worktree may be checking out); "failed" | 174 | * queued creation is delivered (a worktree may be checking out); "failed" |
| 175 | * is a rejected creation whose content went back to the project draft. | 175 | * is a rejected creation whose content went back to the project draft. |
| 176 | * `serverOwned` marks a thread the server already created that is still | ||
| 177 | * setting up its worktree; follow-ups queue on the server behind that setup. | ||
| 176 | */ | 178 | */ |
| 177 | readonly creationState: | 179 | readonly creationState: |
| 178 | | { readonly kind: "preparing"; readonly preparingWorktree: boolean } | 180 | | { |
| 181 | readonly kind: "preparing"; | ||
| 182 | readonly preparingWorktree: boolean; | ||
| 183 | readonly serverOwned: boolean; | ||
| 184 | } | ||
| 179 | | { readonly kind: "failed"; readonly reason: string; readonly onEditTask: () => void } | 185 | | { readonly kind: "failed"; readonly reason: string; readonly onEditTask: () => void } |
| 180 | | null; | 186 | | null; |
| 181 | readonly activePendingApproval: PendingApproval | null; | 187 | readonly activePendingApproval: PendingApproval | null; |
| ... | @@ -1439,11 +1445,14 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread | ... | @@ -1439,11 +1445,14 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread |
| 1439 | canStopThread={props.canStopThread} | 1445 | canStopThread={props.canStopThread} |
| 1440 | environmentId={props.environmentId} | 1446 | environmentId={props.environmentId} |
| 1441 | projectCwd={props.threadCwd ?? props.projectWorkspaceRoot} | 1447 | projectCwd={props.threadCwd ?? props.projectWorkspaceRoot} |
| 1442 | // Follow-ups typed during setup wait in the draft: queueing | 1448 | // Follow-ups typed before the server creates the thread wait in |
| 1443 | // them against a thread id the server may still reject | 1449 | // the draft: queueing them against a thread id the server may |
| 1444 | // would strand them in the outbox. | 1450 | // still reject would strand them in the outbox. |
| 1445 | sendBlockedReason={ | 1451 | sendBlockedReason={ |
| 1446 | props.creationState?.kind === "preparing" ? "Starting the task…" : null | 1452 | props.creationState?.kind === "preparing" && |
| 1453 | !props.creationState.serverOwned | ||
| 1454 | ? "Starting the task…" | ||
| 1455 | : null | ||
| 1447 | } | 1456 | } |
| 1448 | draftKey={props.composerDraftKey ?? undefined} | 1457 | draftKey={props.composerDraftKey ?? undefined} |
| 1449 | followUpBehavior={props.followUpBehavior} | 1458 | followUpBehavior={props.followUpBehavior} |
apps/mobile/src/features/threads/ThreadRouteScreen.tsx+4-1| ... | @@ -889,7 +889,9 @@ function ThreadRouteContent( | ... | @@ -889,7 +889,9 @@ function ThreadRouteContent( |
| 889 | ]); | 889 | ]); |
| 890 | const creationState = ((): ThreadDetailScreenProps["creationState"] => { | 890 | const creationState = ((): ThreadDetailScreenProps["creationState"] => { |
| 891 | if (selectedThreadCreation === null) { | 891 | if (selectedThreadCreation === null) { |
| 892 | return awaitingBootstrapTurn ? { kind: "preparing", preparingWorktree: true } : null; | 892 | return awaitingBootstrapTurn |
| 893 | ? { kind: "preparing", preparingWorktree: true, serverOwned: true } | ||
| 894 | : null; | ||
| 893 | } | 895 | } |
| 894 | if (selectedThreadCreation.outcome?.kind === "failed") { | 896 | if (selectedThreadCreation.outcome?.kind === "failed") { |
| 895 | return { | 897 | return { |
| ... | @@ -901,6 +903,7 @@ function ThreadRouteContent( | ... | @@ -901,6 +903,7 @@ function ThreadRouteContent( |
| 901 | return { | 903 | return { |
| 902 | kind: "preparing", | 904 | kind: "preparing", |
| 903 | preparingWorktree: selectedThreadCreation.message.creation?.workspaceMode === "worktree", | 905 | preparingWorktree: selectedThreadCreation.message.creation?.workspaceMode === "worktree", |
| 906 | serverOwned: false, | ||
| 904 | }; | 907 | }; |
| 905 | })(); | 908 | })(); |
| 906 | if (!environmentId || !threadId) { | 909 | if (!environmentId || !threadId) { |
apps/mobile/src/state/use-thread-composer-state.ts+16-2| ... | @@ -677,8 +677,20 @@ export function useThreadComposerState() { | ... | @@ -677,8 +677,20 @@ export function useThreadComposerState() { |
| 677 | alternateModifier: followUpOverride !== undefined && followUpOverride !== followUpBehavior, | 677 | alternateModifier: followUpOverride !== undefined && followUpOverride !== followUpBehavior, |
| 678 | activeTurnDefault: followUpBehavior, | 678 | activeTurnDefault: followUpBehavior, |
| 679 | }); | 679 | }); |
| 680 | const followUpDispatchMode = | 680 | // A send while the worktree is still being set up waits behind that setup, |
| 681 | followUpAction === "auto" ? null : followUpAction === "queue" ? "queue" : "auto"; | 681 | // so the server can hold it if the setup ends without a worktree. |
| 682 | // Before detail loads, the shell's latest run can be a queued follow-up | ||
| 683 | // while its runtime still reports the setup. | ||
| 684 | const setupRunning = [selectedThreadActivityRun?.status, selectedThreadRuntime?.status].some( | ||
| 685 | (status) => status === "preparing" || status === "starting", | ||
| 686 | ); | ||
| 687 | const followUpDispatchMode = setupRunning | ||
| 688 | ? "queue" | ||
| 689 | : followUpAction === "auto" | ||
| 690 | ? null | ||
| 691 | : followUpAction === "queue" | ||
| 692 | ? "queue" | ||
| 693 | : "auto"; | ||
| 682 | 694 | ||
| 683 | const metadata = makeQueuedMessageMetadata(); | 695 | const metadata = makeQueuedMessageMetadata(); |
| 684 | const messageId = MessageId.make(metadata.messageId); | 696 | const messageId = MessageId.make(metadata.messageId); |
| ... | @@ -735,7 +747,9 @@ export function useThreadComposerState() { | ... | @@ -735,7 +747,9 @@ export function useThreadComposerState() { |
| 735 | saveQueuedRunEdit, | 747 | saveQueuedRunEdit, |
| 736 | selectedEnvironmentRuntime?.connectionState, | 748 | selectedEnvironmentRuntime?.connectionState, |
| 737 | selectedEnvironmentRuntime?.serverConfig, | 749 | selectedEnvironmentRuntime?.serverConfig, |
| 750 | selectedThreadActivityRun?.status, | ||
| 738 | selectedThreadCreation, | 751 | selectedThreadCreation, |
| 752 | selectedThreadRuntime?.status, | ||
| 739 | selectedThreadShell, | 753 | selectedThreadShell, |
| 740 | uploadThreadFeedback, | 754 | uploadThreadFeedback, |
| 741 | ], | 755 | ], |
apps/server/src/orchestration-v2/Orchestrator.ts+54-11| ... | @@ -1269,7 +1269,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | ... | @@ -1269,7 +1269,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 1269 | ]); | 1269 | ]); |
| 1270 | }); | 1270 | }); |
| 1271 | 1271 | ||
| 1272 | const startNextQueuedRun = (threadId: ThreadId, options?: { readonly failedRunId?: RunId }) => | 1272 | const startNextQueuedRun = ( |
| 1273 | threadId: ThreadId, | ||
| 1274 | options?: { readonly failedRunId?: RunId; readonly endedRunId?: RunId }, | ||
| 1275 | ) => | ||
| 1273 | Effect.gen(function* () { | 1276 | Effect.gen(function* () { |
| 1274 | // Every terminal run checks the queue. Only a deliverable queued run | 1277 | // Every terminal run checks the queue. Only a deliverable queued run |
| 1275 | // needs the transcript for provider handoff and legacy import context. | 1278 | // needs the transcript for provider handoff and legacy import context. |
| ... | @@ -1304,15 +1307,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | ... | @@ -1304,15 +1307,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 1304 | // Hold the queue so the user decides when to resume it. Validation | 1307 | // Hold the queue so the user decides when to resume it. Validation |
| 1305 | // failures (setup, unsupported handoff) belong to that message alone, | 1308 | // failures (setup, unsupported handoff) belong to that message alone, |
| 1306 | // and a message queued for another provider is how users recover. | 1309 | // and a message queued for another provider is how users recover. |
| 1310 | // A message queued behind a worktree that was cancelled or never created | ||
| 1311 | // would run in the project checkout instead, so it waits for the user. | ||
| 1307 | const failedRun = latestExecutedRun(projection.runs); | 1312 | const failedRun = latestExecutedRun(projection.runs); |
| 1308 | const failureClass = | 1313 | const failureClass = |
| 1309 | failedRun?.id === options?.failedRunId | 1314 | failedRun?.id === options?.failedRunId |
| 1310 | ? latestRootProviderFailure(failedRun, projection.turnItems)?.class | 1315 | ? latestRootProviderFailure(failedRun, projection.turnItems)?.class |
| 1311 | : undefined; | 1316 | : undefined; |
| 1317 | const worktreeMissing = | ||
| 1318 | failedRun?.id === options?.endedRunId && | ||
| 1319 | failedRun?.workspacePreparation?.type === "worktree" && | ||
| 1320 | projection.thread.worktreePath === null; | ||
| 1312 | if ( | 1321 | if ( |
| 1313 | failureClass !== undefined && | 1322 | worktreeMissing || |
| 1314 | failureClass !== "validation_error" && | 1323 | (failureClass !== undefined && |
| 1315 | failedRun?.providerInstanceId === queuedRun.providerInstanceId | 1324 | failureClass !== "validation_error" && |
| 1325 | failedRun?.providerInstanceId === queuedRun.providerInstanceId) | ||
| 1316 | ) { | 1326 | ) { |
| 1317 | const now = yield* DateTime.now; | 1327 | const now = yield* DateTime.now; |
| 1318 | yield* writeSystemEvents( | 1328 | yield* writeSystemEvents( |
| ... | @@ -4871,9 +4881,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | ... | @@ -4871,9 +4881,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 4871 | (candidate) => candidate.id === projection.thread.activeProviderThreadId, | 4881 | (candidate) => candidate.id === projection.thread.activeProviderThreadId, |
| 4872 | ); | 4882 | ); |
| 4873 | const activeRun = projection.runs.find(isBlockingRun); | 4883 | const activeRun = projection.runs.find(isBlockingRun); |
| 4884 | // A message queued during worktree setup can arrive after that setup failed | ||
| 4885 | // without a worktree. It waits, held, like the messages queued before the | ||
| 4886 | // failure, instead of starting in the project checkout. | ||
| 4887 | const latestRun = latestExecutedRun(projection.runs); | ||
| 4888 | const worktreeMissingRun = | ||
| 4889 | activeRun === undefined && | ||
| 4890 | dispatchMode.type === "queue_after_active" && | ||
| 4891 | (latestRun?.status === "failed" || latestRun?.status === "interrupted") && | ||
| 4892 | latestRun.workspacePreparation?.type === "worktree" && | ||
| 4893 | projection.thread.worktreePath === null | ||
| 4894 | ? latestRun | ||
| 4895 | : undefined; | ||
| 4896 | const queueBehindRun = activeRun ?? worktreeMissingRun; | ||
| 4874 | const pendingMergeBackTransfers = pendingMergeBackTransfersForThread(projection); | 4897 | const pendingMergeBackTransfers = pendingMergeBackTransfersForThread(projection); |
| 4875 | const shouldQueue = | 4898 | const shouldQueue = |
| 4876 | activeRun !== undefined && | 4899 | queueBehindRun !== undefined && |
| 4877 | (dispatchMode.type === "defer_start" || | 4900 | (dispatchMode.type === "defer_start" || |
| 4878 | dispatchMode.type === "start_immediately" || | 4901 | dispatchMode.type === "start_immediately" || |
| 4879 | dispatchMode.type === "queue_after_active"); | 4902 | dispatchMode.type === "queue_after_active"); |
| ... | @@ -4888,13 +4911,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | ... | @@ -4888,13 +4911,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 4888 | const queueProviderThread = | 4911 | const queueProviderThread = |
| 4889 | activeProviderThread ?? | 4912 | activeProviderThread ?? |
| 4890 | projection.providerThreads.find( | 4913 | projection.providerThreads.find( |
| 4891 | (candidate) => candidate.id === activeRun.providerThreadId, | 4914 | (candidate) => candidate.id === queueBehindRun.providerThreadId, |
| 4892 | ); | 4915 | ); |
| 4893 | if (queueProviderThread === undefined) { | 4916 | if (queueProviderThread === undefined) { |
| 4894 | return yield* new OrchestratorDispatchError({ | 4917 | return yield* new OrchestratorDispatchError({ |
| 4895 | commandId: command.commandId, | 4918 | commandId: command.commandId, |
| 4896 | commandType: command.type, | 4919 | commandType: command.type, |
| 4897 | cause: `Active run ${activeRun.id} has no provider thread for queued dispatch.`, | 4920 | cause: `Active run ${queueBehindRun.id} has no provider thread for queued dispatch.`, |
| 4898 | }); | 4921 | }); |
| 4899 | } | 4922 | } |
| 4900 | const now = yield* DateTime.now; | 4923 | const now = yield* DateTime.now; |
| ... | @@ -4948,7 +4971,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | ... | @@ -4948,7 +4971,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 4948 | const attemptId = idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }); | 4971 | const attemptId = idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }); |
| 4949 | const rootNodeId = idAllocator.derive.rootNode({ runId }); | 4972 | const rootNodeId = idAllocator.derive.rootNode({ runId }); |
| 4950 | const checkpointScope = | 4973 | const checkpointScope = |
| 4951 | activeRun.status === "preparing" | 4974 | queueBehindRun.status === "preparing" || worktreeMissingRun !== undefined |
| 4952 | ? null | 4975 | ? null |
| 4953 | : yield* runtimePolicy.resolve({ thread: projection.thread, modelSelection }).pipe( | 4976 | : yield* runtimePolicy.resolve({ thread: projection.thread, modelSelection }).pipe( |
| 4954 | Effect.flatMap((resolvedRuntimePolicy) => | 4977 | Effect.flatMap((resolvedRuntimePolicy) => |
| ... | @@ -4985,7 +5008,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | ... | @@ -4985,7 +5008,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 4985 | rootNodeId, | 5008 | rootNodeId, |
| 4986 | activeAttemptId: attemptId, | 5009 | activeAttemptId: attemptId, |
| 4987 | status: "queued", | 5010 | status: "queued", |
| 4988 | ...(projection.runs.some( | 5011 | ...(worktreeMissingRun !== undefined || |
| 5012 | projection.runs.some( | ||
| 4989 | (candidate) => candidate.status === "queued" && candidate.queueHeld === true, | 5013 | (candidate) => candidate.status === "queued" && candidate.queueHeld === true, |
| 4990 | ) | 5014 | ) |
| 4991 | ? { queueHeld: true } | 5015 | ? { queueHeld: true } |
| ... | @@ -8292,6 +8316,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | ... | @@ -8292,6 +8316,20 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 8292 | type: "run.updated", | 8316 | type: "run.updated", |
| 8293 | payload: { ...state.run, status: "preparing", completedAt: null }, | 8317 | payload: { ...state.run, status: "preparing", completedAt: null }, |
| 8294 | }); | 8318 | }); |
| 8319 | // Messages held when this preparation failed follow the run again. A hold | ||
| 8320 | // from a later run (a stop, say) is not this failure's to release. | ||
| 8321 | const failureOwnsQueue = latestExecutedRun(projection.runs)?.id === state.run.id; | ||
| 8322 | for (const run of projection.runs) { | ||
| 8323 | if (!failureOwnsQueue || run.status !== "queued" || run.queueHeld !== true) continue; | ||
| 8324 | yield* emitEvent({ | ||
| 8325 | type: "run.updated", | ||
| 8326 | threadId: command.threadId, | ||
| 8327 | runId: run.id, | ||
| 8328 | providerInstanceId: run.providerInstanceId, | ||
| 8329 | occurredAt: now, | ||
| 8330 | payload: { ...run, queueHeld: false }, | ||
| 8331 | }); | ||
| 8332 | } | ||
| 8295 | }); | 8333 | }); |
| 8296 | 8334 | ||
| 8297 | /** | 8335 | /** |
| ... | @@ -11056,8 +11094,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | ... | @@ -11056,8 +11094,13 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio |
| 11056 | threadId, | 11094 | threadId, |
| 11057 | startNextQueuedRun( | 11095 | startNextQueuedRun( |
| 11058 | threadId, | 11096 | threadId, |
| 11059 | stored.event.type === "run.updated" && stored.event.payload.status === "failed" | 11097 | stored.event.type === "run.updated" |
| 11060 | ? { failedRunId: stored.event.payload.id } | 11098 | ? { |
| 11099 | endedRunId: stored.event.payload.id, | ||
| 11100 | ...(stored.event.payload.status === "failed" | ||
| 11101 | ? { failedRunId: stored.event.payload.id } | ||
| 11102 | : {}), | ||
| 11103 | } | ||
| 11061 | : undefined, | 11104 | : undefined, |
| 11062 | ), | 11105 | ), |
| 11063 | ) | 11106 | ) |
apps/server/src/orchestration-v2/ThreadLaunchService.test.ts+156| ... | @@ -1424,6 +1424,162 @@ it.effect("retries a failed workspace preparation on the same run", () => { | ... | @@ -1424,6 +1424,162 @@ it.effect("retries a failed workspace preparation on the same run", () => { |
| 1424 | }).pipe(Effect.provide(harness.layer)); | 1424 | }).pipe(Effect.provide(harness.layer)); |
| 1425 | }); | 1425 | }); |
| 1426 | 1426 | ||
| 1427 | it.effect.each(["fails", "is interrupted"] as const)( | ||
| 1428 | "holds a message queued during setup when the workspace preparation %s", | ||
| 1429 | (ending) => | ||
| 1430 | Effect.gen(function* () { | ||
| 1431 | const fetchEntered = yield* Deferred.make<void>(); | ||
| 1432 | const allowFetch = yield* Deferred.make<void>(); | ||
| 1433 | let fetchFailures = 1; | ||
| 1434 | const harness = makeHarness({ | ||
| 1435 | fetchRemote: () => | ||
| 1436 | fetchFailures-- <= 0 | ||
| 1437 | ? Effect.void | ||
| 1438 | : Deferred.succeed(fetchEntered, undefined).pipe( | ||
| 1439 | Effect.andThen(Deferred.await(allowFetch)), | ||
| 1440 | Effect.andThen( | ||
| 1441 | Effect.fail( | ||
| 1442 | new GitCommandError({ | ||
| 1443 | operation: "GitVcsDriver.fetchRemote", | ||
| 1444 | command: "git", | ||
| 1445 | cwd: project.workspaceRoot, | ||
| 1446 | detail: "Git could not reach the remote.", | ||
| 1447 | exitCode: 128, | ||
| 1448 | }), | ||
| 1449 | ), | ||
| 1450 | ), | ||
| 1451 | ), | ||
| 1452 | }); | ||
| 1453 | yield* Effect.gen(function* () { | ||
| 1454 | const launches = yield* ThreadLaunch.ThreadLaunchService; | ||
| 1455 | const threads = yield* ThreadManagement.ThreadManagementService; | ||
| 1456 | const launched = yield* launches.launch( | ||
| 1457 | launchInput({ | ||
| 1458 | command: `command:launch:queued-setup-${ending}`, | ||
| 1459 | thread: `thread:launch:queued-setup-${ending}`, | ||
| 1460 | message: "First message", | ||
| 1461 | workspace: { type: "worktree", baseRef: "main", startFromOrigin: true }, | ||
| 1462 | }), | ||
| 1463 | ); | ||
| 1464 | yield* Deferred.await(fetchEntered); | ||
| 1465 | const queued = yield* threads.sendToThread({ | ||
| 1466 | projectId, | ||
| 1467 | commandId: CommandId.make(`command:launch:queued-setup-${ending}:follow-up`), | ||
| 1468 | threadId: launched.threadId, | ||
| 1469 | messageId: MessageId.make(`message:launch:queued-setup-${ending}:follow-up`), | ||
| 1470 | text: "Sent during setup", | ||
| 1471 | attachments: [], | ||
| 1472 | mode: "queue", | ||
| 1473 | createdBy: "user", | ||
| 1474 | creationSource: "web", | ||
| 1475 | }); | ||
| 1476 | assert.equal(queued.delivery, "queued"); | ||
| 1477 | if (ending === "fails") { | ||
| 1478 | yield* Deferred.succeed(allowFetch, undefined); | ||
| 1479 | } else { | ||
| 1480 | // An agent stop does not ask to hold the queue. | ||
| 1481 | yield* threads.dispatch({ | ||
| 1482 | type: "run.interrupt", | ||
| 1483 | commandId: CommandId.make("command:launch:queued-setup-interrupt"), | ||
| 1484 | threadId: launched.threadId, | ||
| 1485 | runId: launched.projection.runs[0]!.id, | ||
| 1486 | }); | ||
| 1487 | } | ||
| 1488 | |||
| 1489 | // Without its worktree the follow-up would start in the project checkout. | ||
| 1490 | yield* threads.streamStoredEventsFrom({ threadId: launched.threadId }).pipe( | ||
| 1491 | Stream.filter( | ||
| 1492 | (stored) => | ||
| 1493 | stored.event.type === "run.updated" && stored.event.payload.queueHeld === true, | ||
| 1494 | ), | ||
| 1495 | Stream.runHead, | ||
| 1496 | ); | ||
| 1497 | const projection = yield* threads.getThreadProjection(launched.threadId); | ||
| 1498 | assert.equal(projection.runs.at(-1)?.status, "queued"); | ||
| 1499 | assert.equal(projection.thread.worktreePath, null); | ||
| 1500 | |||
| 1501 | if (ending === "fails") { | ||
| 1502 | // A successful retry lets the held message follow the first turn again. | ||
| 1503 | yield* launches.retryPreparation({ | ||
| 1504 | commandId: CommandId.make("command:launch:queued-setup-retry"), | ||
| 1505 | threadId: launched.threadId, | ||
| 1506 | runId: launched.projection.runs[0]!.id, | ||
| 1507 | }); | ||
| 1508 | const retried = yield* threads.getThreadProjection(launched.threadId); | ||
| 1509 | assert.equal(retried.runs.at(-1)?.queueHeld, false); | ||
| 1510 | } | ||
| 1511 | }).pipe(Effect.provide(harness.layer)); | ||
| 1512 | }), | ||
| 1513 | ); | ||
| 1514 | |||
| 1515 | it.effect.each(["failed", "interrupted"] as const)( | ||
| 1516 | "holds a message queued during setup that arrives after the setup %s", | ||
| 1517 | (ending) => { | ||
| 1518 | let fetchEntered: Deferred.Deferred<void> | null = null; | ||
| 1519 | const harness = makeHarness({ | ||
| 1520 | fetchRemote: () => | ||
| 1521 | ending === "interrupted" | ||
| 1522 | ? Deferred.succeed(fetchEntered!, undefined).pipe(Effect.andThen(Effect.never)) | ||
| 1523 | : Effect.fail( | ||
| 1524 | new GitCommandError({ | ||
| 1525 | operation: "GitVcsDriver.fetchRemote", | ||
| 1526 | command: "git", | ||
| 1527 | cwd: project.workspaceRoot, | ||
| 1528 | detail: "Git could not reach the remote.", | ||
| 1529 | exitCode: 128, | ||
| 1530 | }), | ||
| 1531 | ), | ||
| 1532 | }); | ||
| 1533 | return Effect.gen(function* () { | ||
| 1534 | fetchEntered = yield* Deferred.make<void>(); | ||
| 1535 | const launches = yield* ThreadLaunch.ThreadLaunchService; | ||
| 1536 | const threads = yield* ThreadManagement.ThreadManagementService; | ||
| 1537 | const launched = yield* launches.launch( | ||
| 1538 | launchInput({ | ||
| 1539 | command: `command:launch:late-queued-follow-up-${ending}`, | ||
| 1540 | thread: `thread:launch:late-queued-follow-up-${ending}`, | ||
| 1541 | message: "First message", | ||
| 1542 | workspace: { type: "worktree", baseRef: "main", startFromOrigin: true }, | ||
| 1543 | }), | ||
| 1544 | ); | ||
| 1545 | if (ending === "interrupted") { | ||
| 1546 | yield* Deferred.await(fetchEntered); | ||
| 1547 | // An agent stop does not ask to hold the queue. | ||
| 1548 | yield* threads.dispatch({ | ||
| 1549 | type: "run.interrupt", | ||
| 1550 | commandId: CommandId.make("command:launch:late-queued-follow-up:interrupt"), | ||
| 1551 | threadId: launched.threadId, | ||
| 1552 | runId: launched.projection.runs[0]!.id, | ||
| 1553 | }); | ||
| 1554 | } | ||
| 1555 | yield* threads.streamStoredEventsFrom({ threadId: launched.threadId }).pipe( | ||
| 1556 | Stream.filter( | ||
| 1557 | (stored) => stored.event.type === "run.updated" && stored.event.payload.status === ending, | ||
| 1558 | ), | ||
| 1559 | Stream.runHead, | ||
| 1560 | ); | ||
| 1561 | |||
| 1562 | // Queued while setup ran, delivered only after it failed: it must not start | ||
| 1563 | // in the project checkout. | ||
| 1564 | const queued = yield* threads.sendToThread({ | ||
| 1565 | projectId, | ||
| 1566 | commandId: CommandId.make(`command:launch:late-queued-follow-up-${ending}:send`), | ||
| 1567 | threadId: launched.threadId, | ||
| 1568 | messageId: MessageId.make(`message:launch:late-queued-follow-up-${ending}:send`), | ||
| 1569 | text: "Sent during setup", | ||
| 1570 | attachments: [], | ||
| 1571 | mode: "queue", | ||
| 1572 | createdBy: "user", | ||
| 1573 | creationSource: "web", | ||
| 1574 | }); | ||
| 1575 | assert.equal(queued.delivery, "queued"); | ||
| 1576 | const projection = yield* threads.getThreadProjection(launched.threadId); | ||
| 1577 | assert.equal(projection.runs.at(-1)?.status, "queued"); | ||
| 1578 | assert.equal(projection.runs.at(-1)?.queueHeld, true); | ||
| 1579 | }).pipe(Effect.provide(harness.layer)); | ||
| 1580 | }, | ||
| 1581 | ); | ||
| 1582 | |||
| 1427 | it.effect("a retry reuses a recorded worktree without undoing its branch rename", () => { | 1583 | it.effect("a retry reuses a recorded worktree without undoing its branch rename", () => { |
| 1428 | let setupFailures = 1; | 1584 | let setupFailures = 1; |
| 1429 | const harness = makeHarness({ | 1585 | const harness = makeHarness({ |
apps/web/src/components/BranchToolbarBranchSelector.tsx+14-1| ... | @@ -122,6 +122,10 @@ export function BranchToolbarBranchSelector({ | ... | @@ -122,6 +122,10 @@ export function BranchToolbarBranchSelector({ |
| 122 | threadEnvironment.updateMetadata, | 122 | threadEnvironment.updateMetadata, |
| 123 | "thread metadata update", | 123 | "thread metadata update", |
| 124 | ); | 124 | ); |
| 125 | // An automatic default loses to any launch that lands first; that rejection is expected. | ||
| 126 | const updateEmptyThreadMetadata = useAtomCommand(threadEnvironment.updateMetadata, { | ||
| 127 | reportFailure: false, | ||
| 128 | }); | ||
| 125 | const switchRef = useAtomCommand(vcsEnvironment.switchRef, { | 129 | const switchRef = useAtomCommand(vcsEnvironment.switchRef, { |
| 126 | reportFailure: false, | 130 | reportFailure: false, |
| 127 | }); | 131 | }); |
| ... | @@ -188,12 +192,14 @@ export function BranchToolbarBranchSelector({ | ... | @@ -188,12 +192,14 @@ export function BranchToolbarBranchSelector({ |
| 188 | }); | 192 | }); |
| 189 | } | 193 | } |
| 190 | if (hasServerThread) { | 194 | if (hasServerThread) { |
| 191 | void updateThreadMetadata({ | 195 | void (automatic ? updateEmptyThreadMetadata : updateThreadMetadata)({ |
| 192 | environmentId, | 196 | environmentId, |
| 193 | input: { | 197 | input: { |
| 194 | threadId: activeThreadId, | 198 | threadId: activeThreadId, |
| 195 | branch, | 199 | branch, |
| 196 | worktreePath, | 200 | worktreePath, |
| 201 | // A launched thread records its own worktree; a late default must not reset it. | ||
| 202 | ...(automatic ? { expectedEmpty: true } : {}), | ||
| 197 | }, | 203 | }, |
| 198 | }); | 204 | }); |
| 199 | } | 205 | } |
| ... | @@ -228,6 +234,7 @@ export function BranchToolbarBranchSelector({ | ... | @@ -228,6 +234,7 @@ export function BranchToolbarBranchSelector({ |
| 228 | effectiveEnvMode, | 234 | effectiveEnvMode, |
| 229 | draftThread?.environmentSelection, | 235 | draftThread?.environmentSelection, |
| 230 | stopThreadSession, | 236 | stopThreadSession, |
| 237 | updateEmptyThreadMetadata, | ||
| 231 | updateThreadMetadata, | 238 | updateThreadMetadata, |
| 232 | ], | 239 | ], |
| 233 | ); | 240 | ); |
| ... | @@ -538,8 +545,13 @@ export function BranchToolbarBranchSelector({ | ... | @@ -538,8 +545,13 @@ export function BranchToolbarBranchSelector({ |
| 538 | ? null | 545 | ? null |
| 539 | : (defaultBranchName ?? currentGitBranch); | 546 | : (defaultBranchName ?? currentGitBranch); |
| 540 | 547 | ||
| 548 | // A started thread's server is creating its worktree and records the branch | ||
| 549 | // itself; defaulting it here would reset the thread to the base checkout and | ||
| 550 | // detach the agent that just started. | ||
| 551 | const threadStarted = serverSession !== null; | ||
| 541 | useEffect(() => { | 552 | useEffect(() => { |
| 542 | if ( | 553 | if ( |
| 554 | threadStarted || | ||
| 543 | effectiveEnvMode !== "worktree" || | 555 | effectiveEnvMode !== "worktree" || |
| 544 | activeWorktreePath || | 556 | activeWorktreePath || |
| 545 | activeThreadBranch || | 557 | activeThreadBranch || |
| ... | @@ -553,6 +565,7 @@ export function BranchToolbarBranchSelector({ | ... | @@ -553,6 +565,7 @@ export function BranchToolbarBranchSelector({ |
| 553 | activeWorktreePath, | 565 | activeWorktreePath, |
| 554 | effectiveEnvMode, | 566 | effectiveEnvMode, |
| 555 | setThreadBranch, | 567 | setThreadBranch, |
| 568 | threadStarted, | ||
| 556 | worktreeBaseBranchCandidate, | 569 | worktreeBaseBranchCandidate, |
| 557 | ]); | 570 | ]); |
| 558 | 571 |
apps/web/src/components/ChatView.tsx+51-25| ... | @@ -3664,6 +3664,14 @@ export default function ChatView(props: ChatViewProps) { | ... | @@ -3664,6 +3664,14 @@ export default function ChatView(props: ChatViewProps) { |
| 3664 | // Server-side workspace preparation: unlike the local-dispatch flag this | 3664 | // Server-side workspace preparation: unlike the local-dispatch flag this |
| 3665 | // survives reloads and shows on remote viewers of the same thread. | 3665 | // survives reloads and shows on remote viewers of the same thread. |
| 3666 | const activeRunPreparing = activeActivityRun?.status === "preparing"; | 3666 | const activeRunPreparing = activeActivityRun?.status === "preparing"; |
| 3667 | // The server queues a message behind a run that is still preparing its | ||
| 3668 | // worktree or starting its agent, so a send during setup waits there and | ||
| 3669 | // starts on its own instead of being refused. | ||
| 3670 | const sendQueuesBehindSetup = | ||
| 3671 | isServerThread && (activeRunPreparing || activeActivityRun?.status === "starting"); | ||
| 3672 | // Such a send skips the local dispatch, so this keeps its thread's composer busy while it is | ||
| 3673 | // in flight. | ||
| 3674 | const [setupQueueSendThreadKey, setSetupQueueSendThreadKey] = useState<string | null>(null); | ||
| 3667 | useEffect(() => { | 3675 | useEffect(() => { |
| 3668 | attachmentPreviewHandoffByMessageIdRef.current = attachmentPreviewHandoffByMessageId; | 3676 | attachmentPreviewHandoffByMessageIdRef.current = attachmentPreviewHandoffByMessageId; |
| 3669 | }, [attachmentPreviewHandoffByMessageId]); | 3677 | }, [attachmentPreviewHandoffByMessageId]); |
| ... | @@ -4017,11 +4025,13 @@ export default function ChatView(props: ChatViewProps) { | ... | @@ -4017,11 +4025,13 @@ export default function ChatView(props: ChatViewProps) { |
| 4017 | // Sends wait for the agent handoff, not for the setup script: an async | 4025 | // Sends wait for the agent handoff, not for the setup script: an async |
| 4018 | // script keeps the snapshot running while the agent already works, and a | 4026 | // script keeps the snapshot running while the agent already works, and a |
| 4019 | // follow-up must not be held behind a slow install. Before the first | 4027 | // follow-up must not be held behind a slow install. Before the first |
| 4020 | // snapshot arrives the starting session stands in for it. | 4028 | // snapshot arrives the starting session stands in for it. Once the server |
| 4029 | // has the thread, a send queues behind the setup run instead. | ||
| 4021 | const worktreeSetupBlocksSend = | 4030 | const worktreeSetupBlocksSend = |
| 4022 | worktreeSetup !== null | 4031 | !sendQueuesBehindSetup && |
| 4032 | (worktreeSetup !== null | ||
| 4023 | ? worktreeSetup.phase === "running" && !worktreeSetupAgentStarted(worktreeSetup) | 4033 | ? worktreeSetup.phase === "running" && !worktreeSetupAgentStarted(worktreeSetup) |
| 4024 | : isPreparingWorktree; | 4034 | : isPreparingWorktree); |
| 4025 | const cancelWorktreeSetup = useAtomCommand(vcsEnvironment.cancelWorktreeSetup, { | 4035 | const cancelWorktreeSetup = useAtomCommand(vcsEnvironment.cancelWorktreeSetup, { |
| 4026 | reportFailure: false, | 4036 | reportFailure: false, |
| 4027 | }); | 4037 | }); |
| ... | @@ -8808,7 +8818,7 @@ export default function ChatView(props: ChatViewProps) { | ... | @@ -8808,7 +8818,7 @@ export default function ChatView(props: ChatViewProps) { |
| 8808 | if ( | 8818 | if ( |
| 8809 | !readEnvironmentScope(environmentId, AuthOrchestrationOperateScope) || | 8819 | !readEnvironmentScope(environmentId, AuthOrchestrationOperateScope) || |
| 8810 | !activeThread || | 8820 | !activeThread || |
| 8811 | isSendBusy || | 8821 | (isSendBusy && !sendQueuesBehindSetup) || |
| 8812 | isConnecting || | 8822 | isConnecting || |
| 8813 | isRevertingCheckpoint || | 8823 | isRevertingCheckpoint || |
| 8814 | !clientSettingsHydrated || | 8824 | !clientSettingsHydrated || |
| ... | @@ -9252,7 +9262,7 @@ export default function ChatView(props: ChatViewProps) { | ... | @@ -9252,7 +9262,7 @@ export default function ChatView(props: ChatViewProps) { |
| 9252 | return; | 9262 | return; |
| 9253 | } | 9263 | } |
| 9254 | const threadIdForSend = activeThread.id; | 9264 | const threadIdForSend = activeThread.id; |
| 9255 | const isFirstMessage = !isServerThread || activeMessageCount === 0; | 9265 | const isFirstMessage = !sendQueuesBehindSetup && (!isServerThread || activeMessageCount === 0); |
| 9256 | const baseBranchForWorktree = | 9266 | const baseBranchForWorktree = |
| 9257 | isFirstMessage && sendEnvMode === "worktree" && !activeThread.worktreePath | 9267 | isFirstMessage && sendEnvMode === "worktree" && !activeThread.worktreePath |
| 9258 | ? activeThreadBranch | 9268 | ? activeThreadBranch |
| ... | @@ -9309,9 +9319,11 @@ export default function ChatView(props: ChatViewProps) { | ... | @@ -9309,9 +9319,11 @@ export default function ChatView(props: ChatViewProps) { |
| 9309 | resumeCompactionTokens !== null && | 9319 | resumeCompactionTokens !== null && |
| 9310 | !keepFullHistory && | 9320 | !keepFullHistory && |
| 9311 | messageTextForSend.toLowerCase() !== "/compact"; | 9321 | messageTextForSend.toLowerCase() !== "/compact"; |
| 9312 | const turnDispatchMode = compactBeforeSend ? "queue" : dispatchMode; | 9322 | const turnDispatchMode = compactBeforeSend || sendQueuesBehindSetup ? "queue" : dispatchMode; |
| 9313 | const shouldQueueBehindActiveRun = | 9323 | const shouldQueueBehindActiveRun = |
| 9314 | compactBeforeSend || (phase === "running" && dispatchMode === "queue"); | 9324 | compactBeforeSend || |
| 9325 | sendQueuesBehindSetup || | ||
| 9326 | (phase === "running" && dispatchMode === "queue"); | ||
| 9315 | const outgoingMessageText = formatOutgoingPrompt({ | 9327 | const outgoingMessageText = formatOutgoingPrompt({ |
| 9316 | provider: ctxSelectedProvider, | 9328 | provider: ctxSelectedProvider, |
| 9317 | model: ctxSelectedModel, | 9329 | model: ctxSelectedModel, |
| ... | @@ -9460,21 +9472,28 @@ export default function ChatView(props: ChatViewProps) { | ... | @@ -9460,21 +9472,28 @@ export default function ChatView(props: ChatViewProps) { |
| 9460 | ); | 9472 | ); |
| 9461 | return; | 9473 | return; |
| 9462 | } | 9474 | } |
| 9463 | beginLocalDispatch({ | 9475 | // A send queued behind setup leaves the first send's dispatch and setup card alone. |
| 9464 | preparingWorktree: multipleModelSelections !== null || Boolean(baseBranchForWorktree), | 9476 | if (sendQueuesBehindSetup) { |
| 9465 | // Only a draft has a background submission to hide behind its hero. | 9477 | setSetupQueueSendThreadKey(activeThreadKey); |
| 9466 | submissionIntent: | 9478 | } else { |
| 9467 | submissionIntent === "background" && !isLocalDraftThread ? "foreground" : submissionIntent, | 9479 | beginLocalDispatch({ |
| 9468 | }); | 9480 | preparingWorktree: multipleModelSelections !== null || Boolean(baseBranchForWorktree), |
| 9469 | setWorktreeSetupRef( | 9481 | // Only a draft has a background submission to hide behind its hero. |
| 9470 | multipleModelSelections === null && baseBranchForWorktree | 9482 | submissionIntent: |
| 9471 | ? { | 9483 | submissionIntent === "background" && !isLocalDraftThread |
| 9472 | environmentId: activeThread.environmentId, | 9484 | ? "foreground" |
| 9473 | threadId: threadIdForSend, | 9485 | : submissionIntent, |
| 9474 | ownerKey: worktreeSetupOwnerKey, | 9486 | }); |
| 9475 | } | 9487 | setWorktreeSetupRef( |
| 9476 | : null, | 9488 | multipleModelSelections === null && baseBranchForWorktree |
| 9477 | ); | 9489 | ? { |
| 9490 | environmentId: activeThread.environmentId, | ||
| 9491 | threadId: threadIdForSend, | ||
| 9492 | ownerKey: worktreeSetupOwnerKey, | ||
| 9493 | } | ||
| 9494 | : null, | ||
| 9495 | ); | ||
| 9496 | } | ||
| 9478 | 9497 | ||
| 9479 | const turnAttachmentsPromise = Promise.all( | 9498 | const turnAttachmentsPromise = Promise.all( |
| 9480 | composerAttachmentsSnapshot.map(async (attachment) => { | 9499 | composerAttachmentsSnapshot.map(async (attachment) => { |
| ... | @@ -10141,11 +10160,12 @@ export default function ChatView(props: ChatViewProps) { | ... | @@ -10141,11 +10160,12 @@ export default function ChatView(props: ChatViewProps) { |
| 10141 | } | 10160 | } |
| 10142 | } | 10161 | } |
| 10143 | sendInFlightRef.current = false; | 10162 | sendInFlightRef.current = false; |
| 10163 | setSetupQueueSendThreadKey((current) => (current === activeThreadKey ? null : current)); | ||
| 10144 | if (!turnStartSucceeded) { | 10164 | if (!turnStartSucceeded) { |
| 10145 | setDockedDraftHeroThreadKey((currentThreadKey) => | 10165 | setDockedDraftHeroThreadKey((currentThreadKey) => |
| 10146 | currentThreadKey === activeThreadKey ? null : currentThreadKey, | 10166 | currentThreadKey === activeThreadKey ? null : currentThreadKey, |
| 10147 | ); | 10167 | ); |
| 10148 | resetLocalDispatch(); | 10168 | if (!sendQueuesBehindSetup) resetLocalDispatch(); |
| 10149 | } | 10169 | } |
| 10150 | }; | 10170 | }; |
| 10151 | 10171 | ||
| ... | @@ -11652,7 +11672,13 @@ export default function ChatView(props: ChatViewProps) { | ... | @@ -11652,7 +11672,13 @@ export default function ChatView(props: ChatViewProps) { |
| 11652 | phase={phase} | 11672 | phase={phase} |
| 11653 | canInterrupt={canInterruptRunningThread} | 11673 | canInterrupt={canInterruptRunningThread} |
| 11654 | isConnecting={isConnecting} | 11674 | isConnecting={isConnecting} |
| 11655 | isSendBusy={isSendBusy || isSavingQueuedEdit || isResuming} | 11675 | isSendBusy={ |
| 11676 | (isSendBusy && !sendQueuesBehindSetup) || | ||
| 11677 | (setupQueueSendThreadKey !== null && | ||
| 11678 | setupQueueSendThreadKey === activeThreadKey) || | ||
| 11679 | isSavingQueuedEdit || | ||
| 11680 | isResuming | ||
| 11681 | } | ||
| 11656 | canResume={resumableRunId !== null || hasHeldQueuedRuns} | 11682 | canResume={resumableRunId !== null || hasHeldQueuedRuns} |
| 11657 | isRevertingCheckpoint={isRevertingCheckpoint} | 11683 | isRevertingCheckpoint={isRevertingCheckpoint} |
| 11658 | sendDisabledReason={ | 11684 | sendDisabledReason={ |
| ... | @@ -11670,7 +11696,7 @@ export default function ChatView(props: ChatViewProps) { | ... | @@ -11670,7 +11696,7 @@ export default function ChatView(props: ChatViewProps) { |
| 11670 | ? "Preparing worktree" | 11696 | ? "Preparing worktree" |
| 11671 | : projectCloneSendBlockReason | 11697 | : projectCloneSendBlockReason |
| 11672 | } | 11698 | } |
| 11673 | isPreparingWorktree={isPreparingWorktree} | 11699 | isPreparingWorktree={isPreparingWorktree && !sendQueuesBehindSetup} |
| 11674 | queuedRunsControl={ | 11700 | queuedRunsControl={ |
| 11675 | isServerThread && activeThread ? ( | 11701 | isServerThread && activeThread ? ( |
| 11676 | <QueuedRunsControl | 11702 | <QueuedRunsControl |
packages/client-runtime/src/operations/commands.ts+3| ... | @@ -132,6 +132,8 @@ export interface UpdateThreadMetadataInput extends ThreadCommandInput { | ... | @@ -132,6 +132,8 @@ export interface UpdateThreadMetadataInput extends ThreadCommandInput { |
| 132 | readonly regenerateTitle?: boolean; | 132 | readonly regenerateTitle?: boolean; |
| 133 | /** Link (object) or unlink (null) a pull request (#8160). */ | 133 | /** Link (object) or unlink (null) a pull request (#8160). */ |
| 134 | readonly linkedPullRequest?: ThreadLinkedPullRequest | null; | 134 | readonly linkedPullRequest?: ThreadLinkedPullRequest | null; |
| 135 | /** Apply only while no message or run has landed on the thread. */ | ||
| 136 | readonly expectedEmpty?: boolean; | ||
| 135 | } | 137 | } |
| 136 | 138 | ||
| 137 | export interface SetThreadRuntimeModeInput extends ThreadCommandInput { | 139 | export interface SetThreadRuntimeModeInput extends ThreadCommandInput { |
| ... | @@ -579,6 +581,7 @@ export const updateThreadMetadata = Effect.fn("EnvironmentCommands.updateThreadM | ... | @@ -579,6 +581,7 @@ export const updateThreadMetadata = Effect.fn("EnvironmentCommands.updateThreadM |
| 579 | ...(input.title === undefined ? {} : { title: input.title }), | 581 | ...(input.title === undefined ? {} : { title: input.title }), |
| 580 | ...(input.branch === undefined ? {} : { branch: input.branch }), | 582 | ...(input.branch === undefined ? {} : { branch: input.branch }), |
| 581 | ...(input.worktreePath === undefined ? {} : { worktreePath: input.worktreePath }), | 583 | ...(input.worktreePath === undefined ? {} : { worktreePath: input.worktreePath }), |
| 584 | ...(input.expectedEmpty === true ? { expectedEmpty: true } : {}), | ||
| 582 | ...(input.regenerateTitle === undefined ? {} : { regenerateTitle: input.regenerateTitle }), | 585 | ...(input.regenerateTitle === undefined ? {} : { regenerateTitle: input.regenerateTitle }), |
| 583 | ...(input.linkedPullRequest === undefined | 586 | ...(input.linkedPullRequest === undefined |
| 584 | ? {} | 587 | ? {} |