Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions apps/server/src/orchestration-v2/ThreadFind.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -372,6 +372,28 @@ describe("V2 thread find", () => {
}).pipe(Effect.provide(layerTest)),
);

it.effect("skips the empty-response fallback the timeline hides beside an HTML render", () =>
Effect.gen(function* () {
const projection = yield* setup;
const shownRunId = RunId.make("run:shown");
yield* commit([run(threadId, shownRunId, 2)]);
yield* putItems([
{
...item("render", 0, ""),
type: "dynamic_tool",
toolName: "html_render",
input: {},
output: { htmlRender: { attachmentId: "visual", title: "Result", height: 120 } },
},
item("hidden", 1, ""),
{ ...item("shown", 2, ""), runId: shownRunId },
]);
const result = yield* projection.searchThread({ threadId, query: "empty" });
assert.equal(result.totalMatches, 1);
assert.equal(result.match?.entryId, "shown");
}).pipe(Effect.provide(layerTest)),
);

it.effect("searches displayed skill labels and invalidates parsed text when labels change", () =>
Effect.gen(function* () {
const projection = yield* setup;
Expand Down
63 changes: 55 additions & 8 deletions apps/server/src/orchestration-v2/ThreadFind.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import {
} from "@t3tools/contracts";
import { searchableMessageSegments, searchablePlanSegments } from "@t3tools/shared/threadFindText";
import { countThreadSearchOccurrences } from "@t3tools/shared/threadSearch";
import { htmlRenderFromToolItem, mcpAppFromToolItem } from "@t3tools/shared/toolOutput";
import * as Cache from "effect/Cache";
import * as Data from "effect/Data";
import * as Effect from "effect/Effect";
Expand All @@ -34,6 +35,7 @@ interface FindPayload extends FindRow {
readonly entryId: string;
readonly runId: OrchestrationV2TurnItem["runId"];
readonly type: OrchestrationV2TurnItem["type"];
readonly emptyText: boolean;
}
interface FindDocument extends FindRow {
readonly entryId: string;
Expand Down Expand Up @@ -177,7 +179,7 @@ function resultForSelection(
export const makeThreadFind = Effect.fn("makeThreadFind")(function* (
readIndex: (threadId: ThreadId) => Effect.Effect<
readonly (Omit<FindRow, "position"> & {
readonly item: Pick<OrchestrationV2TurnItem, "type">;
readonly item: Pick<OrchestrationV2TurnItem, "type" | "runId">;
})[],
ProjectionStoreV2Error
>,
Expand Down Expand Up @@ -216,11 +218,13 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* (
entry_id: string;
run_id: OrchestrationV2TurnItem["runId"];
item_type: OrchestrationV2TurnItem["type"];
empty_text: number | null;
}>`
SELECT item.thread_id, item.turn_item_id, item.payload_json,
COALESCE(json_extract(item.payload_json, '$.messageId'), item.turn_item_id) AS entry_id,
json_extract(item.payload_json, '$.runId') AS run_id,
json_extract(item.payload_json, '$.type') AS item_type
json_extract(item.payload_json, '$.type') AS item_type,
json_extract(item.payload_json, '$.text') = '' AS empty_text
FROM orchestration_v2_projection_turn_items AS item
JOIN json_each(${sources}) AS wanted
ON item.thread_id = json_extract(wanted.value, '$.threadId')
Expand All @@ -242,6 +246,7 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* (
entryId: payload.entry_id,
runId: payload.run_id,
type: payload.item_type,
emptyText: payload.empty_text === 1,
});
});
});
Expand Down Expand Up @@ -315,11 +320,33 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* (
row.item.type,
),
);
const payloads: FindPayload[] = [];
for (let start = 0; start < candidates.length; start += 128) {
payloads.push(...(yield* load(threadId, candidates.slice(start, start + 128))));
}
return { payloads, cwd, sequence, cacheKey, sourceThreadIds };
const loadAll = Effect.fnUntraced(function* (rows: readonly FindRow[]) {
const loaded: FindPayload[] = [];
for (let start = 0; start < rows.length; start += 128) {
loaded.push(...(yield* load(threadId, rows.slice(start, start + 128))));
}
return loaded;
});
const payloads = yield* loadAll(candidates);
// Only runs with an empty reply can hide its fallback, so other tool payloads stay unread.
const emptyReplyRunIds = new Set(
payloads.flatMap((row) =>
row.type === "assistant_message" && row.emptyText && row.runId !== null
? [row.runId]
: [],
),
);
const toolPayloads = yield* loadAll(
index
.map((row, position) => ({ ...row, position }))
.filter(
(row) =>
row.item.type === "dynamic_tool" &&
row.item.runId !== null &&
emptyReplyRunIds.has(row.item.runId),
),
);
return { payloads, toolPayloads, cwd, sequence, cacheKey, sourceThreadIds };
},
(effect) => sql.withTransaction(effect).pipe(NodeSqliteClient.readOnly),
(effect, input) =>
Expand All @@ -338,7 +365,7 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* (
// Release SQLite before decoding and parsing. Both frames describe one snapshot.
const data = yield* snapshot(input, scans);
if (data.result) return Stream.succeed(data.result);
const { payloads, cwd, sequence, cacheKey, sourceThreadIds } = data;
const { payloads, toolPayloads, cwd, sequence, cacheKey, sourceThreadIds } = data;
const decode = (row: FindPayload) =>
decodeItem(row.payload).pipe(
Effect.map((item) => ({ ...row, item })),
Expand All @@ -354,6 +381,17 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* (
if (item.type === "user_input_request" && item.questionAnswer)
foldedAnswerRunIds.set(`async-answer:${item.questionAnswer.requestId}`, item.runId);
}
// The timeline hides an empty reply's fallback when its run shows an HTML render or MCP app.
const embeddedResultRunIds = new Set<OrchestrationV2TurnItem["runId"]>();
for (const payload of toolPayloads) {
const { item } = yield* decode(payload);
if (
item.type === "dynamic_tool" &&
((item.status === "completed" && htmlRenderFromToolItem(item)) ||
mcpAppFromToolItem(item))
)
embeddedResultRunIds.add(item.runId);
}
const rows = payloads.filter(
(row) =>
row.type !== "user_input_request" &&
Expand All @@ -364,6 +402,15 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* (
const cached = documents.get(row.position);
if (cached) return cached;
const decoded = yield* decode(row);
if (
decoded.item.type === "assistant_message" &&
decoded.item.text === "" &&
embeddedResultRunIds.has(decoded.item.runId)
) {
const document = documentFor(decoded, 0);
documents.set(row.position, document);
return document;
}
const key = textKey(decoded.item, cwd, input.skills ?? []);
const segments = yield* Cache.get(
key.text.length <= 32_768 ? textCache.text : textCache.largeText,
Expand Down
Loading