diff --git a/apps/server/src/orchestration-v2/ThreadSearch.test.ts b/apps/server/src/orchestration-v2/ThreadSearch.test.ts index b8d75340ec6bc..ac1554390d505 100644 --- a/apps/server/src/orchestration-v2/ThreadSearch.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSearch.test.ts @@ -2,9 +2,12 @@ import { assert, it } from "@effect/vitest"; import { EventId, MessageId, + NodeId, ProjectId, ProviderInstanceId, + RunId, ThreadId, + TurnItemId, type OrchestrationV2DomainEvent, type ThreadLinkedPullRequest, } from "@t3tools/contracts"; @@ -127,6 +130,7 @@ const message = ( readonly minute?: number; readonly streaming?: boolean; readonly createdBy?: "user" | "agent"; + readonly runId?: RunId; } = {}, ): OrchestrationV2DomainEvent => ({ id: EventId.make(`message:${id}`), @@ -139,7 +143,7 @@ const message = ( creationSource: role === "user" ? "web" : "provider", id: MessageId.make(id), threadId, - runId: null, + runId: options.runId ?? null, nodeId: null, role, text, @@ -650,22 +654,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 3c97122c1c63e..cdedc85ed7cd3 100644 --- a/apps/server/src/orchestration-v2/ThreadSearch.ts +++ b/apps/server/src/orchestration-v2/ThreadSearch.ts @@ -221,6 +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, 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 ( SELECT 1 FROM json_each(${substrings}) AS token WHERE instr(lower(json_extract(messages.payload_json, '$.text')), token.value) = 0 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",