From 15364b33dd2f274daef7d2720d310b4750419ec5 Mon Sep 17 00:00:00 2001 From: Alex Southwell <4596216+saphid@users.noreply.github.com> Date: Tue, 3 Nov 2026 13:42:08 +1100 Subject: [PATCH 1/4] fix(server): message search no longer finds turns that were rolled back Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/ThreadSearch.test.ts | 39 ++++++++++++++++++- .../src/orchestration-v2/ThreadSearch.ts | 5 +++ 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadSearch.test.ts b/apps/server/src/orchestration-v2/ThreadSearch.test.ts index b8d75340ec6bc..b285b20b9c444 100644 --- a/apps/server/src/orchestration-v2/ThreadSearch.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSearch.test.ts @@ -2,8 +2,10 @@ import { assert, it } from "@effect/vitest"; import { EventId, MessageId, + NodeId, ProjectId, ProviderInstanceId, + RunId, ThreadId, type OrchestrationV2DomainEvent, type ThreadLinkedPullRequest, @@ -127,6 +129,7 @@ const message = ( readonly minute?: number; readonly streaming?: boolean; readonly createdBy?: "user" | "agent"; + readonly runId?: RunId; } = {}, ): OrchestrationV2DomainEvent => ({ id: EventId.make(`message:${id}`), @@ -139,7 +142,7 @@ const message = ( creationSource: role === "user" ? "web" : "provider", id: MessageId.make(id), threadId, - runId: null, + runId: options.runId ?? null, nodeId: null, role, text, @@ -650,17 +653,49 @@ it.layer(layerTest)("ThreadSearch filters before it ranks", (it) => { }), ); - it.effect("leaves out answers to an agent's question, which show in its card", () => + it.effect("leaves out answers in their question card and rolled-back turns", () => Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; const search = yield* ThreadSearch.ThreadSearch; const project = ProjectId.make("project:answers"); const asked = ThreadId.make("thread:answers"); + const rewound = ThreadId.make("thread:rewound"); + const run = { + id: RunId.make("run:rewound"), + threadId: rewound, + ordinal: 1, + providerInstanceId, + modelSelection: { instanceId: providerInstanceId, model: "gpt-5" }, + providerThreadId: null, + userMessageId: MessageId.make("rewound-prompt"), + rootNodeId: NodeId.make("node:rewound"), + activeAttemptId: null, + status: "rolled_back" as const, + requestedAt: at(1), + startedAt: at(1), + completedAt: at(2), + checkpointId: null, + contextHandoffId: null, + }; yield* createProject(project); yield* Effect.forEach( [ thread(asked, project), message(asked, "async-answer:request-1", "user", "use the albatross drive"), + thread(rewound, project), + { + id: EventId.make("run:rewound"), + type: "run.created", + threadId: rewound, + runId: run.id, + nodeId: run.rootNodeId, + providerInstanceId, + occurredAt: at(2), + payload: run, + } satisfies OrchestrationV2DomainEvent, + message(rewound, "rewound-reply", "assistant", "the albatross drive is ready", { + runId: run.id, + }), ], projections.apply, { discard: true }, diff --git a/apps/server/src/orchestration-v2/ThreadSearch.ts b/apps/server/src/orchestration-v2/ThreadSearch.ts index 3c97122c1c63e..0bccafd368e0d 100644 --- a/apps/server/src/orchestration-v2/ThreadSearch.ts +++ b/apps/server/src/orchestration-v2/ThreadSearch.ts @@ -221,6 +221,11 @@ export const messageHitsQuery = ( -- Answers to an agent's question show inside its question card, where -- neither find nor a jump can reach them. AND messages.message_id NOT LIKE 'async-answer:%' + -- Nor can they reach a rolled-back turn, which the thread hides. + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_runs AS runs + WHERE runs.run_id = messages.run_id AND runs.status = 'rolled_back' + ) AND NOT EXISTS ( SELECT 1 FROM json_each(${substrings}) AS token WHERE instr(lower(json_extract(messages.payload_json, '$.text')), token.value) = 0 From 119992089396662870de2153169ac18069f9f10d Mon Sep 17 00:00:00 2001 From: Alex Southwell <4596216+saphid@users.noreply.github.com> Date: Tue, 3 Nov 2026 14:26:37 +1100 Subject: [PATCH 2/4] fix(server): message search no longer finds cancelled queued messages Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/ThreadSearch.test.ts | 105 +++++++++++++----- .../src/orchestration-v2/ThreadSearch.ts | 18 ++- 2 files changed, 92 insertions(+), 31 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadSearch.test.ts b/apps/server/src/orchestration-v2/ThreadSearch.test.ts index b285b20b9c444..bb9afbb8afc1a 100644 --- a/apps/server/src/orchestration-v2/ThreadSearch.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSearch.test.ts @@ -7,6 +7,7 @@ import { ProviderInstanceId, RunId, ThreadId, + TurnItemId, type OrchestrationV2DomainEvent, type ThreadLinkedPullRequest, } from "@t3tools/contracts"; @@ -653,54 +654,100 @@ it.layer(layerTest)("ThreadSearch filters before it ranks", (it) => { }), ); - it.effect("leaves out answers in their question card and rolled-back turns", () => + it.effect("leaves out answers in their question card and turns the thread hides", () => Effect.gen(function* () { const projections = yield* ProjectionStore.ProjectionStoreV2; const search = yield* ThreadSearch.ThreadSearch; const project = ProjectId.make("project:answers"); const asked = ThreadId.make("thread:answers"); const rewound = ThreadId.make("thread:rewound"); - const run = { - id: RunId.make("run:rewound"), - threadId: rewound, - ordinal: 1, - providerInstanceId, - modelSelection: { instanceId: providerInstanceId, model: "gpt-5" }, - providerThreadId: null, - userMessageId: MessageId.make("rewound-prompt"), - rootNodeId: NodeId.make("node:rewound"), - activeAttemptId: null, - status: "rolled_back" as const, - requestedAt: at(1), - startedAt: at(1), - completedAt: at(2), - checkpointId: null, - contextHandoffId: null, - }; + const queued = ThreadId.make("thread:queued"); + const stopped = ThreadId.make("thread:stopped"); + const run = (owner: ThreadId, status: "rolled_back" | "cancelled") => + ({ + id: EventId.make(`run:${owner}`), + type: "run.created", + threadId: owner, + runId: RunId.make(`run:${owner}`), + nodeId: NodeId.make(`node:${owner}`), + providerInstanceId, + occurredAt: at(2), + payload: { + id: RunId.make(`run:${owner}`), + threadId: owner, + ordinal: 1, + providerInstanceId, + modelSelection: { instanceId: providerInstanceId, model: "gpt-5" }, + providerThreadId: null, + userMessageId: MessageId.make(`${owner}:prompt`), + rootNodeId: NodeId.make(`node:${owner}`), + activeAttemptId: null, + status, + requestedAt: at(1), + startedAt: at(1), + completedAt: at(2), + checkpointId: null, + contextHandoffId: null, + }, + }) satisfies OrchestrationV2DomainEvent; yield* createProject(project); yield* Effect.forEach( [ thread(asked, project), message(asked, "async-answer:request-1", "user", "use the albatross drive"), thread(rewound, project), + run(rewound, "rolled_back"), + message(rewound, "rewound-reply", "assistant", "the albatross drive is ready", { + runId: RunId.make(`run:${rewound}`), + }), + thread(queued, project), + run(queued, "cancelled"), + message(queued, "queued-prompt", "user", "then mount the albatross drive", { + runId: RunId.make(`run:${queued}`), + }), + // A turn stopped while preparing still shows its prompt. + thread(stopped, project), + run(stopped, "cancelled"), + message(stopped, "stopped-prompt", "user", "format the albatross drive", { + runId: RunId.make(`run:${stopped}`), + }), { - id: EventId.make("run:rewound"), - type: "run.created", - threadId: rewound, - runId: run.id, - nodeId: run.rootNodeId, - providerInstanceId, + id: EventId.make("item:stopped-prompt"), + type: "turn-item.updated", + threadId: stopped, occurredAt: at(2), - payload: run, + payload: { + id: TurnItemId.make("stopped-prompt"), + threadId: stopped, + runId: RunId.make(`run:${stopped}`), + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: null, + startedAt: at(1), + completedAt: at(1), + updatedAt: at(1), + type: "user_message", + messageId: MessageId.make("stopped-prompt"), + text: "format the albatross drive", + inputIntent: "turn_start", + attachments: [], + createdBy: "user", + creationSource: "web", + }, } satisfies OrchestrationV2DomainEvent, - message(rewound, "rewound-reply", "assistant", "the albatross drive is ready", { - runId: run.id, - }), ], projections.apply, { discard: true }, ); - assert.deepEqual((yield* search.search({ query: "albatross" })).matches, []); + assert.deepEqual( + (yield* search.search({ query: "albatross" })).matches.map((match) => match.messageId), + [MessageId.make("stopped-prompt")], + ); }), ); diff --git a/apps/server/src/orchestration-v2/ThreadSearch.ts b/apps/server/src/orchestration-v2/ThreadSearch.ts index 0bccafd368e0d..a04a0602bf2a6 100644 --- a/apps/server/src/orchestration-v2/ThreadSearch.ts +++ b/apps/server/src/orchestration-v2/ThreadSearch.ts @@ -221,10 +221,24 @@ export const messageHitsQuery = ( -- Answers to an agent's question show inside its question card, where -- neither find nor a jump can reach them. AND messages.message_id NOT LIKE 'async-answer:%' - -- Nor can they reach a rolled-back turn, which the thread hides. + -- Nor can they reach a rolled-back turn or a cancelled queue row, which + -- the thread hides. A turn cancelled while preparing keeps its prompt. AND NOT EXISTS ( SELECT 1 FROM orchestration_v2_projection_runs AS runs - WHERE runs.run_id = messages.run_id AND runs.status = 'rolled_back' + WHERE runs.run_id = messages.run_id + AND ( + runs.status = 'rolled_back' + OR ( + runs.status = 'cancelled' + AND NOT EXISTS ( + SELECT 1 FROM orchestration_v2_projection_turn_items AS items + WHERE items.run_id = runs.run_id + AND items.type = 'user_message' + AND json_extract(items.payload_json, '$.messageId') = messages.message_id + AND json_extract(items.payload_json, '$.inputIntent') IS NOT 'queued_turn' + ) + ) + ) ) AND NOT EXISTS ( SELECT 1 FROM json_each(${substrings}) AS token From a96c53d3deb352d495a63bd295db53e071541508 Mon Sep 17 00:00:00 2001 From: Alex Southwell <4596216+saphid@users.noreply.github.com> Date: Tue, 3 Nov 2026 16:08:47 +1100 Subject: [PATCH 3/4] fix(server): search keeps replies from stopped turns A stopped turn's thread still shows every message an item carries, except a queued prompt. Search now leaves out only messages of a stopped turn that no shown item carries, instead of every message without a started prompt. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/ThreadSearch.test.ts | 84 ++++++++++++------- .../src/orchestration-v2/ThreadSearch.ts | 5 +- 2 files changed, 54 insertions(+), 35 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadSearch.test.ts b/apps/server/src/orchestration-v2/ThreadSearch.test.ts index bb9afbb8afc1a..ac1554390d505 100644 --- a/apps/server/src/orchestration-v2/ThreadSearch.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSearch.test.ts @@ -663,6 +663,7 @@ it.layer(layerTest)("ThreadSearch filters before it ranks", (it) => { const rewound = ThreadId.make("thread:rewound"); const queued = ThreadId.make("thread:queued"); const stopped = ThreadId.make("thread:stopped"); + const recovered = ThreadId.make("thread:recovered"); const run = (owner: ThreadId, status: "rolled_back" | "cancelled") => ({ id: EventId.make(`run:${owner}`), @@ -690,6 +691,40 @@ it.layer(layerTest)("ThreadSearch filters before it ranks", (it) => { contextHandoffId: null, }, }) satisfies OrchestrationV2DomainEvent; + const shown = (owner: ThreadId, id: string, as: "turn_start" | "queued_turn" | "reply") => + ({ + id: EventId.make(`item:${id}`), + type: "turn-item.updated", + threadId: owner, + occurredAt: at(2), + payload: { + id: TurnItemId.make(id), + threadId: owner, + runId: RunId.make(`run:${owner}`), + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: null, + startedAt: at(1), + completedAt: at(1), + updatedAt: at(1), + messageId: MessageId.make(id), + text: "", + ...(as === "reply" + ? { type: "assistant_message", streaming: false } + : { + type: "user_message", + inputIntent: as, + attachments: [], + createdBy: "user", + creationSource: "web", + }), + }, + }) satisfies OrchestrationV2DomainEvent; yield* createProject(project); yield* Effect.forEach( [ @@ -705,48 +740,33 @@ it.layer(layerTest)("ThreadSearch filters before it ranks", (it) => { message(queued, "queued-prompt", "user", "then mount the albatross drive", { runId: RunId.make(`run:${queued}`), }), - // A turn stopped while preparing still shows its prompt. + // A queued prompt usually has no item; an edited one has a queued item. + message(queued, "requeued-prompt", "user", "then eject the albatross drive", { + runId: RunId.make(`run:${queued}`), + }), + shown(queued, "requeued-prompt", "queued_turn"), + // A stopped turn keeps its prompt and whatever the agent already said. thread(stopped, project), run(stopped, "cancelled"), message(stopped, "stopped-prompt", "user", "format the albatross drive", { runId: RunId.make(`run:${stopped}`), }), - { - id: EventId.make("item:stopped-prompt"), - type: "turn-item.updated", - threadId: stopped, - occurredAt: at(2), - payload: { - id: TurnItemId.make("stopped-prompt"), - threadId: stopped, - runId: RunId.make(`run:${stopped}`), - nodeId: null, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: 1, - status: "completed", - title: null, - startedAt: at(1), - completedAt: at(1), - updatedAt: at(1), - type: "user_message", - messageId: MessageId.make("stopped-prompt"), - text: "format the albatross drive", - inputIntent: "turn_start", - attachments: [], - createdBy: "user", - creationSource: "web", - }, - } satisfies OrchestrationV2DomainEvent, + shown(stopped, "stopped-prompt", "turn_start"), + thread(recovered, project), + run(recovered, "cancelled"), + message(recovered, "recovered-reply", "assistant", "the albatross drive is formatted", { + runId: RunId.make(`run:${recovered}`), + }), + shown(recovered, "recovered-reply", "reply"), ], projections.apply, { discard: true }, ); assert.deepEqual( - (yield* search.search({ query: "albatross" })).matches.map((match) => match.messageId), - [MessageId.make("stopped-prompt")], + (yield* search.search({ query: "albatross" })).matches + .map((match) => match.messageId) + .sort(), + [MessageId.make("recovered-reply"), MessageId.make("stopped-prompt")], ); }), ); diff --git a/apps/server/src/orchestration-v2/ThreadSearch.ts b/apps/server/src/orchestration-v2/ThreadSearch.ts index a04a0602bf2a6..cdedc85ed7cd3 100644 --- a/apps/server/src/orchestration-v2/ThreadSearch.ts +++ b/apps/server/src/orchestration-v2/ThreadSearch.ts @@ -221,8 +221,8 @@ export const messageHitsQuery = ( -- Answers to an agent's question show inside its question card, where -- neither find nor a jump can reach them. AND messages.message_id NOT LIKE 'async-answer:%' - -- Nor can they reach a rolled-back turn or a cancelled queue row, which - -- the thread hides. A turn cancelled while preparing keeps its prompt. + -- Nor can they reach a rolled-back turn, or a message of a stopped turn + -- that no shown item carries, such as a queued prompt that never ran. AND NOT EXISTS ( SELECT 1 FROM orchestration_v2_projection_runs AS runs WHERE runs.run_id = messages.run_id @@ -233,7 +233,6 @@ export const messageHitsQuery = ( AND NOT EXISTS ( SELECT 1 FROM orchestration_v2_projection_turn_items AS items WHERE items.run_id = runs.run_id - AND items.type = 'user_message' AND json_extract(items.payload_json, '$.messageId') = messages.message_id AND json_extract(items.payload_json, '$.inputIntent') IS NOT 'queued_turn' ) From 9f600c6a3e87e9583a3354154d54c4550e3b7271 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Tue, 3 Nov 2026 06:15:37 +0000 Subject: [PATCH 4/4] fix(client): refresh message searches after rollback and cancellation --- packages/client-runtime/src/state/olderSettledThreads.ts | 3 ++- packages/client-runtime/src/state/orchestration.ts | 4 ++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/packages/client-runtime/src/state/olderSettledThreads.ts b/packages/client-runtime/src/state/olderSettledThreads.ts index 47779f110efdc..1895fd1891933 100644 --- a/packages/client-runtime/src/state/olderSettledThreads.ts +++ b/packages/client-runtime/src/state/olderSettledThreads.ts @@ -396,7 +396,8 @@ export function createOlderSettledThreadAtoms( ); const searchReadAtom = createEnvironmentQueryAtomFamily(runtime, { label: "environment-data:orchestration:older-settled-search", - idleTtlMs: 60_000, + staleTimeMs: 0, + idleTtlMs: 0, execute: Effect.fn("OlderSettledThreads.search")(function* (input: { readonly generation: number; readonly search: OrchestrationV2SearchThreadsInput; diff --git a/packages/client-runtime/src/state/orchestration.ts b/packages/client-runtime/src/state/orchestration.ts index 94dc18582a56e..e4d4e12cf4722 100644 --- a/packages/client-runtime/src/state/orchestration.ts +++ b/packages/client-runtime/src/state/orchestration.ts @@ -144,8 +144,8 @@ export function createOrchestrationEnvironmentAtoms( threadSearch: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:orchestration:thread-search", tag: ORCHESTRATION_V2_WS_METHODS.searchThreads, - staleTimeMs: 30_000, - idleTtlMs: 60_000, + staleTimeMs: 0, + idleTtlMs: 0, }), archivedShellSnapshot: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:orchestration:archived-shell-snapshot",