Skip to content

Commit b97a371

Browse files
armanossilokoclaude
armanossiloko
and
claude
committed
fix(queue): a queued message being edited no longer sends when the turn ends
Opening a queued message for editing now holds it on the server until the edit is saved or cancelled, so a turn that finishes meanwhile no longer sends the original text. The messages behind it wait too, keeping order. Stop and failure pauses are unchanged, and Resume clears a hold left by a client that went away. Web and desktop set the hold; mobile is unchanged. Fixes #15611 Co-Authored-By: Claude Opus 5.5 <[email protected]>
1 parent a24ec90 commit b97a371

16 files changed

Lines changed: 270 additions & 16 deletions

File tree

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

Lines changed: 48 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -538,6 +538,7 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId {
538538
case "queued-run.reorder":
539539
case "queued-run.cancel":
540540
case "queued-run.edit":
541+
case "queued-run.editing":
541542
case "runtime-request.respond":
542543
case "runtime-request.delivery.fail":
543544
case "thread.user-input.dismiss":
@@ -1620,6 +1621,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
16201621
);
16211622
return;
16221623
}
1624+
// After the pause checks, so a failure during the edit still pauses the queue.
1625+
if (options?.queuedRunId === undefined && storedQueuedRun.editing === true) return;
16231626
// Move queued ordinals after runs that started while they waited to preserve transcript and checkpoint order.
16241627
// Started runs have prompts; cancelled queued messages do not.
16251628
const newerRunIds = projection.runs
@@ -9740,6 +9743,36 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
97409743
}
97419744
});
97429745

9746+
const dispatchQueuedRunEditing = Effect.fn("Orchestrator.dispatchQueuedRunEditing")(function* (
9747+
command: Extract<OrchestrationV2Command, { readonly type: "queued-run.editing" }>,
9748+
events: Ref.Ref<Array<OrchestrationV2DomainEvent>>,
9749+
) {
9750+
const projection = yield* projectionStore
9751+
.getThreadRecords(command.threadId, ["runs"])
9752+
.pipe(Effect.mapError(() => new OrchestratorProjectionError({ threadId: command.threadId })));
9753+
const queuedRun = projection.runs.find((candidate) => candidate.id === command.runId);
9754+
if (queuedRun === undefined || queuedRun.status !== "queued") {
9755+
return yield* new OrchestratorDispatchError({
9756+
commandId: command.commandId,
9757+
commandType: command.type,
9758+
cause: `Run ${command.runId} is not queued.`,
9759+
});
9760+
}
9761+
if ((queuedRun.editing === true) === command.editing) return;
9762+
yield* emit(
9763+
events,
9764+
command,
9765+
)({
9766+
type: "run.updated",
9767+
threadId: command.threadId,
9768+
runId: queuedRun.id,
9769+
...(queuedRun.rootNodeId === null ? {} : { nodeId: queuedRun.rootNodeId }),
9770+
providerInstanceId: queuedRun.providerInstanceId,
9771+
occurredAt: yield* DateTime.now,
9772+
payload: { ...queuedRun, editing: command.editing },
9773+
});
9774+
});
9775+
97439776
const subagentCount = (threadId: ThreadId) =>
97449777
projectionStore
97459778
.getSubagentCount(threadId)
@@ -13151,7 +13184,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1315113184
runId: run.id,
1315213185
providerInstanceId: run.providerInstanceId,
1315313186
occurredAt: now,
13154-
payload: { ...run, queueHeld: false },
13187+
// Also the way out of an edit hold left by a client that went away.
13188+
payload: { ...run, queueHeld: false, editing: false },
1315513189
});
1315613190
}
1315713191
if (queued.length === 0) {
@@ -13176,6 +13210,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1317613210
case "queued-run.edit":
1317713211
yield* dispatchQueuedRunEdit(command, events);
1317813212
break;
13213+
case "queued-run.editing":
13214+
yield* dispatchQueuedRunEditing(command, events);
13215+
break;
1317913216
case "checkpoint.rollback":
1318013217
yield* dispatchCheckpointRollback(command, events, effects);
1318113218
break;
@@ -13292,7 +13329,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1329213329
}),
1329313330
),
1329413331
);
13295-
if (command.type === "queue.resume" || command.type === "queued-run.start") {
13332+
if (
13333+
command.type === "queue.resume" ||
13334+
command.type === "queued-run.start" ||
13335+
(command.type === "queued-run.editing" && !command.editing)
13336+
) {
1329613337
yield* mapDispatchError(command)(
1329713338
startNextQueuedRun(
1329813339
command.threadId,
@@ -13439,7 +13480,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
1343913480
detail: committed.receipt.error ?? "Previously rejected.",
1344013481
});
1344113482
}
13442-
if (command.type === "queue.resume" || command.type === "queued-run.start") {
13483+
if (
13484+
command.type === "queue.resume" ||
13485+
command.type === "queued-run.start" ||
13486+
(command.type === "queued-run.editing" && !command.editing)
13487+
) {
1344313488
yield* mapDispatchError(command)(
1344413489
startNextQueuedRun(
1344513490
command.threadId,

‎apps/server/src/orchestration-v2/ProjectionStore.test.ts‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -247,6 +247,44 @@ it.effect("SQL queue readiness skips held runs without bypassing active work", (
247247
}).pipe(Effect.provide(layerTest)),
248248
);
249249

250+
it.effect("SQL and in-memory shells skip queued runs waiting on an open edit", () =>
251+
Effect.gen(function* () {
252+
const store = yield* ProjectionStore.ProjectionStoreV2;
253+
const threadId = yield* addRolledBackRecoveryCandidate("edited-queue");
254+
const run = (yield* store.getThreadProjection(threadId)).runs[0]!;
255+
const now = yield* DateTime.now;
256+
const queued = (suffix: string, ordinal: number, editing: boolean) =>
257+
store.apply({
258+
id: EventId.make(`edited-queue:${suffix}:${editing}`),
259+
type: "run.updated",
260+
threadId,
261+
occurredAt: now,
262+
payload: {
263+
...run,
264+
id: RunId.make(`edited-queue:${suffix}`),
265+
ordinal,
266+
status: "queued" as const,
267+
editing,
268+
startedAt: null,
269+
completedAt: null,
270+
},
271+
});
272+
const latestRunIds = Effect.gen(function* () {
273+
const sql = (yield* store.getShellSnapshot()).threads.find((row) => row.id === threadId)!;
274+
const memory = ProjectionStore.threadShellFromProjection(
275+
yield* store.getThreadProjection(threadId),
276+
);
277+
return [sql.latestRunId, memory.latestRunId];
278+
});
279+
yield* queued("edited", 2, true);
280+
yield* queued("behind", 3, false);
281+
assert.deepEqual(yield* latestRunIds, [run.id, run.id]);
282+
yield* queued("edited", 2, false);
283+
const behind = RunId.make("edited-queue:behind");
284+
assert.deepEqual(yield* latestRunIds, [behind, behind]);
285+
}).pipe(Effect.provide(layerTest)),
286+
);
287+
250288
it.effect("SQL shells flag only an active turn blocked solely on T3 thread waits", () =>
251289
Effect.gen(function* () {
252290
const store = yield* ProjectionStore.ProjectionStoreV2;

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

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6027,14 +6027,22 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
60276027
ORDER BY r.ordinal DESC, r.run_id DESC
60286028
LIMIT 1
60296029
)
6030-
-- The newest run not waiting in a held queue, matching latestUnheldRun.
6030+
-- The newest run not waiting in a held queue or behind an open edit, matching latestUnheldRun.
60316031
LEFT JOIN orchestration_v2_projection_runs presented ON presented.run_id = (
60326032
SELECT candidate.run_id
60336033
FROM orchestration_v2_projection_runs candidate
60346034
WHERE candidate.thread_id = t.thread_id
60356035
AND NOT (
60366036
candidate.status = 'queued'
6037-
AND json_extract(candidate.payload_json, '$.queueHeld') IS 1
6037+
AND (
6038+
json_extract(candidate.payload_json, '$.queueHeld') IS 1
6039+
OR EXISTS (
6040+
SELECT 1 FROM orchestration_v2_projection_runs edited
6041+
WHERE edited.thread_id = t.thread_id
6042+
AND edited.status = 'queued'
6043+
AND json_extract(edited.payload_json, '$.editing') IS 1
6044+
)
6045+
)
60386046
)
60396047
AND NOT ${isSilentShellRun("candidate")}
60406048
ORDER BY candidate.ordinal DESC, candidate.run_id DESC

‎apps/server/src/orchestration-v2/runtimeLayer.test.ts‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5766,6 +5766,76 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => {
57665766
}),
57675767
);
57685768

5769+
it.effect("holds a queued message and the ones behind it while it is edited", () =>
5770+
Effect.gen(function* () {
5771+
const orchestrator = yield* Orchestrator.OrchestratorV2;
5772+
const eventSink = yield* EventSink.EventSinkV2;
5773+
const threadId = ThreadId.make("runtime-layer-edited-queue");
5774+
yield* orchestrator.dispatch({
5775+
type: "thread.create",
5776+
createdBy: "user",
5777+
creationSource: "web",
5778+
commandId: CommandId.make(`${threadId}:create`),
5779+
threadId,
5780+
projectId: ProjectId.make(`${threadId}:project`),
5781+
title: "Edited queue",
5782+
modelSelection,
5783+
runtimeMode: "full-access",
5784+
interactionMode: "default",
5785+
branch: null,
5786+
worktreePath: process.cwd(),
5787+
});
5788+
for (const [index, text] of ["Active", "Edited", "Behind"].entries()) {
5789+
yield* orchestrator.dispatch({
5790+
type: "message.dispatch",
5791+
createdBy: "user",
5792+
creationSource: "web",
5793+
commandId: CommandId.make(`${threadId}:message:${index}`),
5794+
threadId,
5795+
messageId: MessageId.make(`${threadId}:message:${index}`),
5796+
text,
5797+
attachments: [],
5798+
modelSelection,
5799+
dispatchMode: { type: index === 0 ? "start_immediately" : "queue_after_active" },
5800+
});
5801+
}
5802+
const [active, edited, behind] = (yield* orchestrator.getThreadProjection(threadId)).runs;
5803+
const setEditing = (editing: boolean) =>
5804+
orchestrator.dispatch({
5805+
type: "queued-run.editing",
5806+
commandId: CommandId.make(`${threadId}:editing:${editing}`),
5807+
threadId,
5808+
runId: edited!.id,
5809+
editing,
5810+
});
5811+
yield* setEditing(true);
5812+
const now = yield* DateTime.now;
5813+
yield* eventSink.write({
5814+
events: [
5815+
{
5816+
id: EventId.make(`${threadId}:completed`),
5817+
type: "run.updated",
5818+
threadId,
5819+
runId: active!.id,
5820+
occurredAt: now,
5821+
payload: { ...active!, status: "completed", completedAt: now },
5822+
},
5823+
],
5824+
});
5825+
yield* orchestrator.resumeQueuedRuns;
5826+
const waiting = yield* orchestrator.getThreadProjection(threadId);
5827+
assert.deepEqual(
5828+
waiting.runs.map((run) => run.status),
5829+
["completed", "queued", "queued"],
5830+
);
5831+
5832+
yield* setEditing(false);
5833+
const after = yield* orchestrator.getThreadProjection(threadId);
5834+
assert.equal(after.runs.find((run) => run.id === edited!.id)?.status, "starting");
5835+
assert.equal(after.runs.find((run) => run.id === behind!.id)?.status, "queued");
5836+
}),
5837+
);
5838+
57695839
it.effect("places held queued messages after a continuation that started first", () =>
57705840
Effect.gen(function* () {
57715841
const orchestrator = yield* Orchestrator.OrchestratorV2;

‎apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -174,6 +174,7 @@ function commandThreadIds(command: OrchestrationV2Command): ReadonlyArray<Thread
174174
case "queued-run.reorder":
175175
case "queued-run.cancel":
176176
case "queued-run.edit":
177+
case "queued-run.editing":
177178
case "runtime-request.respond":
178179
case "thread.user-input.dismiss":
179180
case "checkpoint.rollback":

‎apps/web/src/components/ChatView.tsx‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -820,8 +820,9 @@ export default function ChatView(props: ChatViewProps) {
820820
const activeRuntime = isServerThread ? serverRuntime : (activeThread?.runtime ?? null);
821821
const hasHeldQueuedRuns =
822822
isServerThread &&
823-
serverProjection?.runs.some((run) => run.status === "queued" && run.queueHeld === true) ===
824-
true;
823+
serverProjection?.runs.some(
824+
(run) => run.status === "queued" && (run.queueHeld === true || run.editing === true),
825+
) === true;
825826
const resumableRunId = (() => {
826827
if (!isServerThread || serverProjection === null) return null;
827828
const run = latestExecutedRun(serverProjection.runs);

‎apps/web/src/components/chat/useQueuedRunEdit.browser.test.tsx‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,12 +46,16 @@ vi.mock("../../hooks/useSettings", () => ({
4646
const runsKey = (ref: ScopedThreadRef) => `${ref.environmentId}:${ref.threadId}`;
4747

4848
vi.mock("../../state/threads", async () => {
49-
const { Atom } = await import("effect/reactivity");
49+
const { AsyncResult, Atom } = await import("effect/reactivity");
5050
const key = (ref: ScopedThreadRef) => `${ref.environmentId}:${ref.threadId}`;
5151
return {
5252
threadEnvironment: {
5353
snapshotAtom: () => Atom.make(null),
5454
editQueuedRun: {},
55+
setQueuedRunEditing: {
56+
label: "set-queued-run-editing",
57+
run: async () => AsyncResult.success(undefined),
58+
},
5559
},
5660
environmentThreadDetails: {
5761
statusAtom: () => Atom.make("ready"),

‎apps/web/src/components/chat/useQueuedRunEdit.ts‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import {
1818
type ThreadId,
1919
} from "@t3tools/contracts";
2020
import { scopedThreadKey } from "@t3tools/client-runtime/environment";
21+
import { runAtomCommand } from "@t3tools/client-runtime/state/runtime";
2122
import {
2223
restoreAttachmentOnlyBootstrapPrompt,
2324
stripAttachmentOnlyBootstrapPrompt,
@@ -256,6 +257,17 @@ export function useQueuedRunEdit(input: {
256257
get(environmentThreadDetails.quietThreadAtom(threadRef))?.projection.runs ?? null,
257258
),
258259
}),
260+
// An older server rejects the hold; the edit then works as before.
261+
hold: (editing) =>
262+
void runAtomCommand(
263+
appAtomRegistry,
264+
threadEnvironment.setQueuedRunEditing,
265+
{
266+
environmentId,
267+
input: { threadId: threadRef.threadId, runId: request.runId, editing },
268+
},
269+
{ reportFailure: false },
270+
),
259271
// The run started or was cancelled from another client. A changed
260272
// edit moves into the thread's own draft when that draft is empty;
261273
// otherwise it is dropped. Edits outlive navigation, so the toast is

‎docs/user/composer.md‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -219,8 +219,9 @@ The pencil on a queued message opens it in the composer for editing. The origina
219219
stays in the queue until you save, and it is highlighted while you edit. The message's
220220
attachments appear above the text with a remove control, and new images can be added the usual way.
221221
The checkmark saves the queued message in place; **Cancel** next to it leaves it unchanged. Whatever
222-
you had typed in the composer before starting the edit is restored afterwards. If the queued
223-
message starts or is removed while you are editing, the edit ends: changed content moves into the
222+
you had typed in the composer before starting the edit is restored afterwards. The message and the
223+
ones behind it wait while you edit, even if the current turn finishes. If the queued message is
224+
sent or removed elsewhere while you are editing, the edit ends: changed content moves into the
224225
composer when it is empty, and is discarded otherwise.
225226

226227
## Commands and skills

‎packages/client-runtime/src/operations/commands.ts‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -248,6 +248,11 @@ export interface CancelQueuedRunInput extends ThreadCommandInput {
248248
readonly runId: RunId;
249249
}
250250

251+
export interface SetQueuedRunEditingInput extends ThreadCommandInput {
252+
readonly runId: RunId;
253+
readonly editing: boolean;
254+
}
255+
251256
export interface RetryWorkspacePreparationInput extends ThreadCommandInput {
252257
readonly runId: RunId;
253258
}
@@ -1057,6 +1062,18 @@ export const cancelQueuedRun = Effect.fn("EnvironmentCommands.cancelQueuedRun")(
10571062
});
10581063
});
10591064

1065+
export const setQueuedRunEditing = Effect.fn("EnvironmentCommands.setQueuedRunEditing")(function* (
1066+
input: SetQueuedRunEditingInput,
1067+
) {
1068+
return yield* dispatch({
1069+
type: "queued-run.editing",
1070+
commandId: yield* allocateCommandId(input),
1071+
threadId: input.threadId,
1072+
runId: input.runId,
1073+
editing: input.editing,
1074+
});
1075+
});
1076+
10601077
export const retryWorkspacePreparation = Effect.fn("EnvironmentCommands.retryWorkspacePreparation")(
10611078
function* (input: RetryWorkspacePreparationInput) {
10621079
return yield* dispatch({

0 commit comments

Comments
 (0)