diff --git a/apps/server/src/orchestration-v2/ThreadSearch.test.ts b/apps/server/src/orchestration-v2/ThreadSearch.test.ts index 1dee6c9cfc0f7..ca93dfe46ead3 100644 --- a/apps/server/src/orchestration-v2/ThreadSearch.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSearch.test.ts @@ -3,9 +3,12 @@ import { EnvironmentId, EventId, MessageId, + NodeId, ProjectId, ProviderInstanceId, + RunId, ThreadId, + TurnItemId, type OrchestrationV2DomainEvent, type ThreadLinkedPullRequest, } from "@t3tools/contracts"; @@ -130,6 +133,7 @@ const message = ( readonly minute?: number; readonly streaming?: boolean; readonly createdBy?: "user" | "agent"; + readonly runId?: RunId; } = {}, ): OrchestrationV2DomainEvent => ({ id: EventId.make(`message:${id}`), @@ -142,7 +146,7 @@ const message = ( creationSource: role === "user" ? "web" : "provider", id: MessageId.make(id), threadId, - runId: null, + runId: options.runId ?? null, nodeId: null, role, text, @@ -757,22 +761,120 @@ 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 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 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}`), + 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; + 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( [ 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 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}`), + }), + 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, []); + assert.deepEqual( + (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 3bcdf9e84d4a2..8679dad2d070b 100644 --- a/apps/server/src/orchestration-v2/ThreadSearch.ts +++ b/apps/server/src/orchestration-v2/ThreadSearch.ts @@ -279,6 +279,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, 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 + 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 json_extract(items.payload_json, '$.messageId') = messages.message_id + AND json_extract(items.payload_json, '$.inputIntent') IS NOT 'queued_turn' + ) + ) + ) + ) AND NOT EXISTS ( -- The text read once, not once per check. SELECT 1 FROM (SELECT ${messageSearchText(sql)} AS text) AS message, diff --git a/packages/client-runtime/src/state/olderSettledThreads.ts b/packages/client-runtime/src/state/olderSettledThreads.ts index 8a7eb870c843d..a5b79beafe533 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 657a9cd62dab1..c265c28bca47a 100644 --- a/packages/client-runtime/src/state/orchestration.ts +++ b/packages/client-runtime/src/state/orchestration.ts @@ -233,8 +233,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",