diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index fb087724dbdb7..9ac148895fece 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -2253,7 +2253,7 @@ describe("OrchestratorMcpService provider resolution", () => { cause: new Error("refused by the orchestrator"), }) as never, ) - : Effect.succeed({} as never), + : Effect.succeed({ storedEvents: [] } as never), }), ServerSettings.layerTest(), providerRegistryLayer([ @@ -2637,6 +2637,17 @@ describe("OrchestratorMcpService provider resolution", () => { }); const titled = yield* orchestrator.getThreadProjection(emojiAtLimit.threads[0]!.threadId); assert.equal(titled.thread.title, `${"a".repeat(76)}...`); + // One commit holds a prompted thread and its run, so no client sees it idle between. + const promptedCommits = yield* orchestrator + .streamStoredEventsFrom({ threadId: titled.thread.id }) + .pipe( + Stream.filter( + ({ event }) => event.type === "thread.created" || event.type === "run.created", + ), + Stream.take(2), + Stream.runCollect, + ); + assert.equal(promptedCommits[0]?.commandId, promptedCommits[1]?.commandId); yield* settings.updateSettings({ projectSettingsOverrides: { diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index d7ef5f788f0ba..165607994953f 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -2992,7 +2992,20 @@ const make = Effect.gen(function* () { title: request.title, index, }); - yield* threadManagement + const initialMessage = + request.prompt === undefined + ? undefined + : { + senderThreadId: scope.thread.threadId, + messageId: stableMessageId({ + scope, + requestKey: key, + index, + }), + text: request.prompt, + attachments: [], + }; + const created = yield* threadManagement .dispatch({ type: "thread.create", createdBy: "agent", @@ -3009,6 +3022,7 @@ const make = Effect.gen(function* () { ...(parent.thread.linkOrigin === undefined ? {} : { linkOrigin: parent.thread.linkOrigin }), + ...(initialMessage === undefined ? {} : { initialMessage }), }) .pipe( Effect.mapError((error) => @@ -3022,9 +3036,14 @@ const make = Effect.gen(function* () { const resolvedModelSelection = (yield* threadManagement .getThreadRecords(threadId, []) .pipe(Effect.mapError(threadManagementFailure))).thread.modelSelection; - if (request.prompt !== undefined) { + // A replayed creation that carried no prompt has no run yet. + if ( + initialMessage !== undefined && + !created.storedEvents.some(({ event }) => event.type === "run.created") + ) { yield* threadManagement .dispatch({ + ...initialMessage, type: "message.dispatch", createdBy: "agent", creationSource: "mcp", @@ -3035,14 +3054,6 @@ const make = Effect.gen(function* () { index, }), threadId, - senderThreadId: scope.thread.threadId, - messageId: stableMessageId({ - scope, - requestKey: key, - index, - }), - text: request.prompt, - attachments: [], modelSelection: resolvedModelSelection, dispatchMode: { type: "start_immediately" }, }) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index c6d1ab4654110..6049beffadcfe 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -795,6 +795,28 @@ function metadataUpdateContinuation( }; } +// A later command for the first message would leave the new thread committed with no run, +// and every client would show it idle until that command landed. +function threadCreateInitialMessage( + command: Extract, +): Extract | undefined { + if (command.initialMessage === undefined) return undefined; + const { workspaceStrategy, ...message } = command.initialMessage; + return { + ...message, + type: "message.dispatch", + commandId: command.commandId, + threadId: command.threadId, + modelSelection: command.modelSelection, + dispatchMode: + workspaceStrategy === undefined + ? { type: "start_immediately" } + : { type: "defer_start", workspaceStrategy }, + createdBy: command.createdBy, + creationSource: command.creationSource, + }; +} + // Provider wakes can start follow-up work; monitor updates only report progress. // Dispatch takes the parent's lock for follow-ups because intake writes the parent. function mayStartSubagentFollowup( @@ -6882,6 +6904,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ? projectionStore .getThreadRecords(command.threadId, ["messages"], { messageRoles: ["user"] }) .pipe( + // A thread this command creates has no stored messages yet. + Effect.catchTags({ + ProjectionStoreThreadNotFoundError: () => Effect.succeed({ messages: [] }), + }), Effect.map(({ messages }) => [...messages, ...userMessages]), Effect.map((all) => all.length > 0 && all.every(isNativeMaintenanceCommand)), mapDispatchError(command), @@ -14760,9 +14786,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | undefined; let updateDelegatedCompletionOnlyRunId: RunId | undefined; switch (command.type) { - case "thread.create": + case "thread.create": { yield* dispatchThreadCreate(command, events); + const initialMessage = threadCreateInitialMessage(command); + if (initialMessage !== undefined) { + cancelUnsettledEffects = yield* dispatchMessage( + withoutGuestAttribution(initialMessage), + events, + effects, + ); + } break; + } case "thread.visit": yield* dispatchThreadVisit(command, events); break; diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index 33d5fb657fa7b..ad255802013ec 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -633,7 +633,7 @@ it.effect.each([" /COMPACT "])( ); assert.isNull((yield* threads.getThreadProjection(launched.threadId)).thread.lastVisitedAt); assert.isFalse( - (yield* outbox.listByCommandId(CommandId.make("compact-title:initial-message"))).some( + (yield* outbox.listByCommandId(CommandId.make("compact-title"))).some( (effect) => effect.request.type === "thread-title.generate", ), ); @@ -776,7 +776,7 @@ it.effect("arms durable title generation after accepting the first message", () generateTitle: true, }; const launched = yield* launches.launch(input); - const generationCommandId = CommandId.make("command:launch:title-generation:initial-message"); + const generationCommandId = input.commandId; const projection = yield* threads.getThreadProjection(launched.threadId); assert.equal(projection.thread.title, "Generate my title"); @@ -850,9 +850,9 @@ it.effect("arms durable title generation after accepting the first message", () }), ); -it.effect("does not update a reused thread title when the initial message is rejected", () => +it.effect("a rejected initial message leaves no title change and no new thread", () => Effect.gen(function* () { - const harness = makeHarness(); + const harness = makeHarness({ adapters: [adapter, claudeAdapter] }); yield* Effect.gen(function* () { const launches = yield* ThreadLaunch.ThreadLaunchService; const threads = yield* ThreadManagement.ThreadManagementService; @@ -897,6 +897,24 @@ it.effect("does not update a reused thread title when the initial message is rej assert.isUndefined(projection.thread.titleRegeneration); assert.isEmpty(projection.messages); assert.isEmpty(yield* outbox.listByCommandId(CommandId.make(`${commandId}:initial-message`))); + + // The orchestrator refuses this goal command, after the launch's own checks pass. + const newThreadId = ThreadId.make("thread:launch:rejected-new-thread"); + const rejected = yield* launches + .launch({ + ...launchInput({ + command: "command:launch:rejected-new-thread", + thread: newThreadId, + message: "/goal pause", + }), + modelSelection: claudeModelSelection, + }) + .pipe(Effect.flip); + if (rejected._tag !== "ThreadLaunchError") { + assert.fail(`expected ThreadLaunchError, got ${rejected._tag}`); + } + assert.equal(rejected.operation, "create-thread"); + assert.isNull(yield* threads.getThreadShell(newThreadId)); }).pipe(Effect.provide(harness.layer)); }), ); @@ -1680,6 +1698,21 @@ it.effect("replays a server-allocated launch", () => workspace: { type: "worktree", baseRef: "main" }, }); const first = yield* launches.launch(rest); + // One commit holds the thread and its first run, so no client sees the thread idle between. + const launchCommits = yield* threads + .streamStoredEventsFrom({ threadId: first.threadId }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "thread.created" || stored.event.type === "run.created", + ), + Stream.take(2), + Stream.runCollect, + ); + assert.deepEqual( + Array.from(launchCommits, (stored) => stored.commandId), + [rest.commandId, rest.commandId], + ); yield* Deferred.await(setupEntered); const retry = yield* launches.launch(rest); assert.equal(first.threadId, retry.threadId); diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index 67f3b5402de78..bc29745e80065 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -1133,6 +1133,26 @@ const make = Effect.gen(function* () { const initialBranch = yield* launchBranch(workspaceStrategy, project.workspaceRoot); const initialWorktreePath = workspaceStrategy.type === "existing_worktree" ? workspaceStrategy.worktreePath : null; + const initialMessage = + input.initialMessage === undefined + ? undefined + : { + messageId: + input.initialMessage.messageId ?? + (yield* ids.allocate + .message({ threadId: candidateThreadId, ordinal: 1 }) + .pipe(Effect.mapError(mapError(input, "dispatch-message", candidateThreadId)))), + text: input.initialMessage.text, + ...(input.initialMessage.scheduledTaskId === undefined + ? {} + : { scheduledTaskId: input.initialMessage.scheduledTaskId }), + ...(input.initialMessage.senderThreadId === undefined + ? {} + : { senderThreadId: input.initialMessage.senderThreadId }), + attachments: input.initialMessage.attachments, + ...(input.initialMessage.context ? { context: input.initialMessage.context } : {}), + ...(input.generateTitle === true ? { titleSeed: input.title } : {}), + }; const claimDispatch = input.reuseExistingThread === true ? threads.dispatch({ @@ -1161,6 +1181,9 @@ const make = Effect.gen(function* () { : { delegatedFrom: input.delegatedFrom }), createdBy: input.createdBy, creationSource: input.creationSource, + ...(initialMessage === undefined + ? {} + : { initialMessage: { ...initialMessage, workspaceStrategy } }), }); const claimed = yield* claimDispatch.pipe( Effect.mapError( @@ -1180,53 +1203,43 @@ const make = Effect.gen(function* () { let runId: RunId | null = null; let messageSequence: number | undefined; let messageWasAlreadyAccepted = false; - if (input.initialMessage !== undefined) { - const messageCommandId = CommandId.make(`${input.commandId}:initial-message`); - const messageReceipt = yield* readReceipt(input, messageCommandId); - messageWasAlreadyAccepted = Option.isSome(messageReceipt); - const messageId = - input.initialMessage.messageId ?? - (yield* ids.allocate - .message({ threadId, ordinal: 1 }) - .pipe(Effect.mapError(mapError(input, "dispatch-message", threadId)))); - const dispatched = yield* threads - .dispatch({ - type: "message.dispatch", - commandId: messageCommandId, - threadId, - messageId, - text: input.initialMessage.text, - ...(input.initialMessage.scheduledTaskId === undefined - ? {} - : { scheduledTaskId: input.initialMessage.scheduledTaskId }), - ...(input.initialMessage.senderThreadId === undefined - ? {} - : { senderThreadId: input.initialMessage.senderThreadId }), - attachments: input.initialMessage.attachments, - ...(input.initialMessage.context ? { context: input.initialMessage.context } : {}), - ...(input.generateTitle === true ? { titleSeed: input.title } : {}), - modelSelection: input.modelSelection, - dispatchMode: { type: "defer_start", workspaceStrategy }, - createdBy: input.createdBy, - creationSource: input.creationSource, - }) - .pipe( - // An agent's launch can land on a thread that a replay, or a concurrent - // launch with the same key, created with broader modes. The orchestrator - // refuses the message under the thread's lock; the user's own clients - // launch into their own drafts, whatever modes those hold. - input.createdBy === "agent" - ? Effect.provideService(DispatchModeLimit, { - runtimeMode: input.runtimeMode, - interactionMode: input.interactionMode, - }) - : (effect) => effect, - Effect.mapError(mapError(input, "dispatch-message", threadId)), - ); + if (initialMessage !== undefined) { + // A new thread's claim carries its first message. A reused thread, or a replayed + // claim that carried none, takes it in a command of its own. + let accepted = claimed; + if (!claimed.storedEvents.some((stored) => stored.event.type === "run.created")) { + const messageCommandId = CommandId.make(`${input.commandId}:initial-message`); + const messageReceipt = yield* readReceipt(input, messageCommandId); + messageWasAlreadyAccepted = Option.isSome(messageReceipt); + accepted = yield* threads + .dispatch({ + ...initialMessage, + type: "message.dispatch", + commandId: messageCommandId, + threadId, + modelSelection: input.modelSelection, + dispatchMode: { type: "defer_start", workspaceStrategy }, + createdBy: input.createdBy, + creationSource: input.creationSource, + }) + .pipe( + // An agent's launch can land on a thread that a replay, or a concurrent + // launch with the same key, created with broader modes. The orchestrator + // refuses the message under the thread's lock; the user's own clients + // launch into their own drafts, whatever modes those hold. + input.createdBy === "agent" + ? Effect.provideService(DispatchModeLimit, { + runtimeMode: input.runtimeMode, + interactionMode: input.interactionMode, + }) + : (effect) => effect, + Effect.mapError(mapError(input, "dispatch-message", threadId)), + ); + } // A concurrent retry can read no receipt and still replay the dispatch. - messageWasAlreadyAccepted ||= dispatched.replayed === true; - messageSequence = dispatched.sequence; - const runCreated = dispatched.storedEvents.find( + messageWasAlreadyAccepted ||= accepted.replayed === true; + messageSequence = accepted.sequence; + const runCreated = accepted.storedEvents.find( (stored) => stored.event.type === "run.created", ); runId = runCreated?.event.type === "run.created" ? runCreated.event.payload.id : null; diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index c2502902ef8e3..d1ae8ab555161 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -68,6 +68,12 @@ it("stamps authoritative provenance on commands that create threads or messages" threadId: ThreadId.make("thread:elsewhere"), title: "Somewhere else", }, + // A client's message goes through intake; only the server folds one into the creation. + initialMessage: { + messageId: MessageId.make("message:thread-management:forged"), + text: "Forged", + attachments: [], + }, }; const stamped = ThreadManagementService.withCreationProvenance(command, { @@ -80,6 +86,7 @@ it("stamps authoritative provenance on commands that create threads or messages" }); expect(stamped).not.toHaveProperty("linkOrigin"); expect(stamped).not.toHaveProperty("delegatedFrom"); + expect(stamped).not.toHaveProperty("initialMessage"); }); const layerReceiptsTest = CommandReceiptStore.layer.pipe( diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 482674b772d43..a6aad880cd88a 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -84,8 +84,13 @@ export function withCreationProvenance( ): OrchestrationV2Command { switch (command.type) { case "thread.create": { - // Only the server stamps a link's origin, so a client cannot set one. - const { linkOrigin: _linkOrigin, delegatedFrom: _delegatedFrom, ...rest } = command; + // Only the server stamps a link's origin or folds in a first message, so a client cannot. + const { + linkOrigin: _linkOrigin, + delegatedFrom: _delegatedFrom, + initialMessage: _initialMessage, + ...rest + } = command; return { ...rest, ...provenance }; } case "message.dispatch": diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index e4a57262d12dc..2a758467b5d8b 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3094,6 +3094,22 @@ export const OrchestrationV2Command = Schema.Union([ metadata: Schema.optional(OrchestrationV2ProviderThreadNativeMetadata), }), ), + /** + * Only the server sets this. The thread and its first run commit together, so no client sees + * the thread idle before its run exists. The run waits for `workspaceStrategy` when given one. + */ + initialMessage: Schema.optional( + Schema.Struct({ + messageId: MessageId, + text: Schema.String, + attachments: Schema.Array(ChatAttachment), + context: Schema.optional(OrchestrationMessageContext), + titleSeed: Schema.optional(TrimmedNonEmptyString), + scheduledTaskId: Schema.optional(ScheduledTaskId), + senderThreadId: Schema.optional(ThreadId), + workspaceStrategy: Schema.optional(OrchestrationV2ThreadLaunchWorkspaceStrategy), + }), + ), }), Schema.Struct({ type: Schema.Literal("thread.archive"),