Skip to content

Commit fabedd5

Browse files
aneymaneym
committed
feat(server): option to start queued messages together as one turn
When a turn ends with several messages queued, each one starts its own turn, so the agent re-plans after every message and can act on the first before it sees a later one that changes the picture. Settings > Behavior > Send queued messages together (off by default) starts every compatible queued message as one turn when the agent finishes. The provider gets one message that says they were queued separately and lists them in order with their senders. Notifications, delegated completions, scheduled tasks, maintenance commands and messages for another provider or model still get their own turn. A completed turn also releases held messages. Written with Claude Opus 5.5 and GPT Sol in Claude Code.
1 parent dc84b09 commit fabedd5

8 files changed

Lines changed: 309 additions & 7 deletions

File tree

‎apps/server/src/orchestration-v2/Orchestrator.ts‎

Lines changed: 42 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -148,6 +148,7 @@ import * as RunExecutionService from "./RunExecutionService.ts";
148148
import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts";
149149
import { ProviderSwitchServiceV2 } from "./ProviderSwitchService.ts";
150150
import { isAutomaticCompletionRun, queuedRunsInDeliveryOrder } from "./QueuedRunOrder.ts";
151+
import { queuedRunBatch, queuedRunBatchEvents } from "./QueuedRunBatch.ts";
151152
import { RuntimePolicyV2 } from "./RuntimePolicy.ts";
152153
import {
153154
endOrphanedNativeSubagent,
@@ -1475,6 +1476,28 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
14751476
readonly queuedRunId?: RunId;
14761477
},
14771478
) {
1479+
if (options?.endedRunId !== undefined) {
1480+
const before = yield* readCommandProjection(threadId);
1481+
const ended = before.runs.find((run) => run.id === options.endedRunId);
1482+
const held =
1483+
ended?.status === "completed" &&
1484+
(yield* serverSettings.getSettings.pipe(Effect.orDie)).batchQueuedMessages
1485+
? before.runs.filter((run) => run.status === "queued" && run.queueHeld === true)
1486+
: [];
1487+
if (held.length > 0) {
1488+
const now = yield* DateTime.now;
1489+
yield* writeSystemEvents(
1490+
held.map((run) => ({
1491+
type: "run.updated" as const,
1492+
threadId,
1493+
runId: run.id,
1494+
providerInstanceId: run.providerInstanceId,
1495+
occurredAt: now,
1496+
payload: { ...run, queueHeld: false },
1497+
})),
1498+
);
1499+
}
1500+
}
14781501
if (!(yield* projectionStore.canStartQueuedRun(threadId))) return;
14791502
const projection = yield* readCommandProjection(threadId);
14801503
if (
@@ -1566,7 +1589,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
15661589
turnItemTypes: ["user_message", "notification"],
15671590
turnItemRunIds: newerRunIds,
15681591
})).turnItems.length > 0;
1569-
const ordinal = startedAhead ? nextRunOrdinal(projection) : queuedRun.ordinal;
1592+
const batchMembers =
1593+
options?.queuedRunId === undefined && settings.batchQueuedMessages
1594+
? queuedRunBatch(projection, storedQueuedRun)
1595+
: [];
1596+
const ordinal =
1597+
startedAhead || batchMembers.length > 0 ? nextRunOrdinal(projection) : queuedRun.ordinal;
15701598
const rootNodeId = queuedRun.rootNodeId;
15711599
const attemptId = queuedRun.activeAttemptId;
15721600
const providerThreadId = queuedRun.providerThreadId;
@@ -1941,6 +1969,18 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
19411969
completedAt: now,
19421970
updatedAt: now,
19431971
};
1972+
const batchEvents =
1973+
userTurnItem.type === "user_message"
1974+
? queuedRunBatchEvents({
1975+
members: batchMembers,
1976+
projection,
1977+
target: startingRun,
1978+
rootNodeId,
1979+
firstItem: userTurnItem,
1980+
itemId: (messageId) => idAllocator.derive.userTurnItem({ messageId }),
1981+
occurredAt: now,
1982+
})
1983+
: [];
19441984
const handoffTurnItem: OrchestrationV2TurnItem | null =
19451985
activeHandoff === null
19461986
? null
@@ -2150,6 +2190,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
21502190
occurredAt: now,
21512191
payload: notificationTurnItem(userTurnItem, queuedMessage, projection.subagents),
21522192
},
2193+
...batchEvents,
21532194
{
21542195
type: "run.updated",
21552196
threadId,

‎apps/server/src/orchestration-v2/ProviderTurnStartService.ts‎

Lines changed: 17 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@ import {
5858
pendingRestartCancelledBackgroundWork,
5959
restartCancelledBackgroundWorkNote,
6060
} from "./RestartBackgroundNote.ts";
61+
import { batchedProviderText, foldedQueuedTurnItems } from "./QueuedRunBatch.ts";
6162

6263
export class ProviderTurnStartError extends Schema.TaggedError<ProviderTurnStartError>()(
6364
"ProviderTurnStartError",
@@ -1094,10 +1095,20 @@ export const layer: Layer.Layer<
10941095
const routableSubagents = projection.subagents.filter((subagent) =>
10951096
RunExecutionService.canRouteRelatedSubagent(subagent.status),
10961097
);
1097-
const userText = projectComposerContextForProvider({
1098-
text: message.text,
1099-
records: message.context?.records ?? [],
1100-
});
1098+
const foldedItems = foldedQueuedTurnItems(projection.turnItems, run);
1099+
const userText = batchedProviderText(
1100+
[message, ...foldedItems].map((part) => ({
1101+
...part,
1102+
text: projectComposerContextForProvider({
1103+
text: part.text,
1104+
records: part.context?.records ?? [],
1105+
}),
1106+
})),
1107+
);
1108+
const attachments = [
1109+
...message.attachments,
1110+
...foldedItems.flatMap((item) => item.attachments),
1111+
];
11011112
// Codex and Claude parse /goal from the prompt, so it goes verbatim and history waits a turn.
11021113
const goalDriver = session.driver === "codex" || session.driver === "claudeAgent";
11031114
const goalCommand = goalDriver && parseGoalCommand(message) !== null;
@@ -1271,7 +1282,7 @@ export const layer: Layer.Layer<
12711282
tokenCap,
12721283
modelContextWindow,
12731284
userText: restartNote === "" ? userText : `${restartNote}\n\n${userText}`,
1274-
attachments: message.attachments,
1285+
attachments,
12751286
providerThread: budgetProviderThread,
12761287
nativeContextEstimate:
12771288
budgetProviderThread.contextUsage?.usedTokens === undefined
@@ -1403,7 +1414,7 @@ export const layer: Layer.Layer<
14031414
message: {
14041415
messageId: message.id,
14051416
text: userText,
1406-
attachments: message.attachments,
1417+
attachments,
14071418
createdBy: message.createdBy,
14081419
creationSource: message.creationSource,
14091420
...(message.scheduledTaskId === undefined

‎apps/server/src/orchestration-v2/QueuedRunBatch.ts‎

Lines changed: 203 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,203 @@
1+
import {
2+
parseGoalCommand,
3+
type NodeId,
4+
type OrchestrationV2ConversationMessage,
5+
type OrchestrationV2DomainEvent,
6+
type OrchestrationV2Run,
7+
type OrchestrationV2ThreadProjection,
8+
type OrchestrationV2TurnItem,
9+
} from "@t3tools/contracts";
10+
import { modelSelectionsEqual } from "@t3tools/shared/model";
11+
12+
import { isNativeMaintenanceCommand } from "./CommandPolicy.ts";
13+
import { queuedRunsInDeliveryOrder } from "./QueuedRunOrder.ts";
14+
15+
const MAX_BATCH_SIZE = 20;
16+
17+
export interface QueuedRunBatchMember {
18+
readonly run: OrchestrationV2Run;
19+
readonly message: OrchestrationV2ConversationMessage;
20+
}
21+
22+
type UserTurnItem = Extract<OrchestrationV2TurnItem, { readonly type: "user_message" }>;
23+
type DraftEvent = OrchestrationV2DomainEvent extends infer Event
24+
? Event extends OrchestrationV2DomainEvent
25+
? Omit<Event, "id">
26+
: never
27+
: never;
28+
29+
function batchableMessage(
30+
projection: Pick<OrchestrationV2ThreadProjection, "messages">,
31+
run: OrchestrationV2Run,
32+
): OrchestrationV2ConversationMessage | undefined {
33+
if (
34+
run.status !== "queued" ||
35+
run.queueHeld === true ||
36+
run.restartContinuationOfRunId !== undefined ||
37+
run.delegatedTaskId !== undefined ||
38+
run.sourcePlanRef !== undefined
39+
) {
40+
return undefined;
41+
}
42+
const message = projection.messages.find((candidate) => candidate.id === run.userMessageId);
43+
if (
44+
message === undefined ||
45+
message.notification !== undefined ||
46+
message.delegatedCompletion !== undefined ||
47+
message.workflowScript === true ||
48+
message.scheduledTaskId !== undefined ||
49+
message.idleCompaction !== undefined ||
50+
isNativeMaintenanceCommand(message) ||
51+
parseGoalCommand(message) !== null
52+
) {
53+
return undefined;
54+
}
55+
return message;
56+
}
57+
58+
export function queuedRunBatch(
59+
projection: Pick<OrchestrationV2ThreadProjection, "runs" | "messages">,
60+
first: OrchestrationV2Run,
61+
): ReadonlyArray<QueuedRunBatchMember> {
62+
if (batchableMessage(projection, first) === undefined) return [];
63+
const members: Array<QueuedRunBatchMember> = [];
64+
for (const run of queuedRunsInDeliveryOrder(projection)) {
65+
if (run.id === first.id) continue;
66+
const message = batchableMessage(projection, run);
67+
if (
68+
message === undefined ||
69+
run.providerInstanceId !== first.providerInstanceId ||
70+
!modelSelectionsEqual(run.modelSelection, first.modelSelection) ||
71+
run.workspacePreparation?.type !== first.workspacePreparation?.type ||
72+
members.length + 1 >= MAX_BATCH_SIZE
73+
) {
74+
break;
75+
}
76+
members.push({ run, message });
77+
}
78+
return members;
79+
}
80+
81+
export function foldedQueuedTurnItems(
82+
turnItems: ReadonlyArray<OrchestrationV2TurnItem>,
83+
run: Pick<OrchestrationV2Run, "id" | "userMessageId">,
84+
): ReadonlyArray<UserTurnItem> {
85+
return turnItems
86+
.filter(
87+
(item): item is UserTurnItem =>
88+
item.type === "user_message" &&
89+
item.runId === run.id &&
90+
item.inputIntent === "queued_turn" &&
91+
item.messageId !== run.userMessageId,
92+
)
93+
.toSorted((left, right) => left.ordinal - right.ordinal);
94+
}
95+
96+
export function queuedRunBatchEvents(input: {
97+
readonly members: ReadonlyArray<QueuedRunBatchMember>;
98+
readonly projection: Pick<OrchestrationV2ThreadProjection, "nodes" | "attempts">;
99+
readonly target: OrchestrationV2Run;
100+
readonly rootNodeId: NodeId;
101+
readonly firstItem: UserTurnItem;
102+
readonly itemId: (messageId: OrchestrationV2ConversationMessage["id"]) => UserTurnItem["id"];
103+
readonly occurredAt: UserTurnItem["updatedAt"];
104+
}): ReadonlyArray<DraftEvent> {
105+
const { target, rootNodeId, firstItem, occurredAt } = input;
106+
const into = {
107+
threadId: target.threadId,
108+
runId: target.id,
109+
nodeId: rootNodeId,
110+
providerInstanceId: target.providerInstanceId,
111+
occurredAt,
112+
};
113+
return input.members.flatMap(({ run, message }, index): ReadonlyArray<DraftEvent> => {
114+
const from = {
115+
threadId: run.threadId,
116+
runId: run.id,
117+
providerInstanceId: run.providerInstanceId,
118+
occurredAt,
119+
};
120+
const node = input.projection.nodes.find((candidate) => candidate.id === run.rootNodeId);
121+
const attempt = input.projection.attempts.find(
122+
(candidate) => candidate.id === run.activeAttemptId,
123+
);
124+
const item: UserTurnItem = {
125+
...firstItem,
126+
id: input.itemId(message.id),
127+
ordinal: firstItem.ordinal + index + 1,
128+
messageId: message.id,
129+
text: message.text,
130+
attachments: message.attachments,
131+
createdBy: message.createdBy,
132+
creationSource: message.creationSource,
133+
};
134+
const { context: _context, senderThreadId: _sender, ...base } = item;
135+
return [
136+
{
137+
...from,
138+
type: "run.updated",
139+
payload: { ...run, status: "cancelled", queuePosition: null, completedAt: occurredAt },
140+
},
141+
...(attempt === undefined
142+
? []
143+
: [
144+
{
145+
...from,
146+
type: "run-attempt.updated" as const,
147+
payload: { ...attempt, status: "cancelled" as const, completedAt: occurredAt },
148+
},
149+
]),
150+
...(node === undefined
151+
? []
152+
: [
153+
{
154+
...from,
155+
nodeId: node.id,
156+
type: "node.updated" as const,
157+
payload: { ...node, status: "cancelled" as const, completedAt: occurredAt },
158+
},
159+
]),
160+
{
161+
...into,
162+
type: "message.updated",
163+
payload: { ...message, runId: target.id, nodeId: rootNodeId },
164+
},
165+
{
166+
...into,
167+
type: "turn-item.updated",
168+
payload: {
169+
...base,
170+
...(message.context ? { context: message.context } : {}),
171+
...(message.senderThreadId === undefined
172+
? {}
173+
: { senderThreadId: message.senderThreadId }),
174+
},
175+
},
176+
];
177+
});
178+
}
179+
180+
export interface QueuedRunBatchPart {
181+
readonly text: string;
182+
readonly createdBy: OrchestrationV2ConversationMessage["createdBy"];
183+
readonly creationSource: OrchestrationV2ConversationMessage["creationSource"];
184+
readonly senderThreadId?: OrchestrationV2ConversationMessage["senderThreadId"];
185+
}
186+
187+
function batchSender(part: QueuedRunBatchPart): string {
188+
if (part.senderThreadId !== undefined) return `an agent in thread ${part.senderThreadId}`;
189+
return part.createdBy === "user"
190+
? `the user (${part.creationSource})`
191+
: `${part.createdBy} (${part.creationSource})`;
192+
}
193+
194+
export function batchedProviderText(parts: ReadonlyArray<QueuedRunBatchPart>): string {
195+
if (parts.length < 2) return parts[0]?.text ?? "";
196+
return [
197+
`These ${parts.length} messages were queued separately while you were working. They are listed in the order they were sent, with their senders. Handle each one accordingly.`,
198+
...parts.map(
199+
(part, index) =>
200+
`Message ${index + 1} of ${parts.length}, from ${batchSender(part)}:\n${part.text}`,
201+
),
202+
].join("\n\n");
203+
}

‎apps/web/src/components/settings/SettingsPanels.tsx‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -494,6 +494,9 @@ export function useSettingsRestore(onRestored?: () => void) {
494494
DEFAULT_UNIFIED_SETTINGS.continueThreadsAfterServerUpdate
495495
? ["Continue threads after restarts"]
496496
: []),
497+
...(settings.batchQueuedMessages !== DEFAULT_UNIFIED_SETTINGS.batchQueuedMessages
498+
? ["Send queued messages together"]
499+
: []),
497500
...(isBackgroundActivityDirty ? ["Background activity"] : []),
498501
...(settings.defaultThreadEnvMode !== DEFAULT_UNIFIED_SETTINGS.defaultThreadEnvMode
499502
? ["New thread mode"]
@@ -611,6 +614,7 @@ export function useSettingsRestore(onRestored?: () => void) {
611614
responseStreamingMode: DEFAULT_UNIFIED_SETTINGS.responseStreamingMode,
612615
enableProviderUpdateChecks: DEFAULT_UNIFIED_SETTINGS.enableProviderUpdateChecks,
613616
continueThreadsAfterServerUpdate: DEFAULT_UNIFIED_SETTINGS.continueThreadsAfterServerUpdate,
617+
batchQueuedMessages: DEFAULT_UNIFIED_SETTINGS.batchQueuedMessages,
614618
backgroundActivity: DEFAULT_UNIFIED_SETTINGS.backgroundActivity,
615619
backgroundActivityProfile: DEFAULT_UNIFIED_SETTINGS.backgroundActivityProfile,
616620
automaticGitFetchInterval: DEFAULT_UNIFIED_SETTINGS.automaticGitFetchInterval,
@@ -1965,6 +1969,35 @@ export function MaintenanceSettingsPanel() {
19651969
/>
19661970
}
19671971
/>
1972+
1973+
<SettingsRow
1974+
{...searchableSetting("batch-queued-messages")}
1975+
serverScoped
1976+
settingKeys={["batchQueuedMessages"]}
1977+
description="When a turn ends, start every queued message on the thread together as one turn instead of one turn each."
1978+
resetAction={
1979+
settings.batchQueuedMessages !== DEFAULT_UNIFIED_SETTINGS.batchQueuedMessages ? (
1980+
<SettingResetButton
1981+
label="send queued messages together"
1982+
onClick={() =>
1983+
updateSettings({
1984+
batchQueuedMessages: DEFAULT_UNIFIED_SETTINGS.batchQueuedMessages,
1985+
})
1986+
}
1987+
/>
1988+
) : null
1989+
}
1990+
control={
1991+
<ScopedSwitch
1992+
settingKeys={["batchQueuedMessages"]}
1993+
checked={settings.batchQueuedMessages}
1994+
onCheckedChange={(checked) =>
1995+
updateSettings({ batchQueuedMessages: Boolean(checked) })
1996+
}
1997+
aria-label="Send queued messages together"
1998+
/>
1999+
}
2000+
/>
19682001
</SettingsSection>
19692002

19702003
<SettingsSection id="background" title="Background activity">

0 commit comments

Comments
 (0)