,
@@ -65,22 +84,36 @@ vi.mock("../../state/conversationSummaries", () => ({
expand: (request: { environmentId: EnvironmentId; input: ConversationSummaryExpandInput }) => {
state.expansions.push(request);
const offset = request.input.offset ?? 0;
+ const textOffset = request.input.textOffset ?? 0;
+ const maxChars = request.input.maxChars ?? 12_000;
const summary = state.list?.summaries[0];
if (summary?.state !== "available") throw new Error("Expected summary");
+ const firstPart = offset === 0 && textOffset === 0;
return {
summary,
- sources: summary.summary.sources.slice(offset, offset + 6).map((ref, index) => ({
- ref,
- result:
- offset + index === 0
- ? { state: "unavailable", reason: "changed" }
- : {
- state: "available",
- role: "assistant",
- text: "Exact source " + (offset + index + 1) + ": 日本語 🧪\n unchanged.",
- },
- })),
- nextOffset: offset === 0 ? 6 : null,
+ sources: summary.summary.sources
+ .slice(offset, offset + (firstPart ? 1 : 6))
+ .map((ref, index) => ({
+ ref,
+ result:
+ offset + index === 0
+ ? {
+ state: "available",
+ role: "assistant",
+ text: state.longSource.slice(textOffset, textOffset + maxChars),
+ textOffset,
+ totalChars: state.longSource.length,
+ }
+ : offset + index === 1
+ ? { state: "unavailable", reason: "changed" }
+ : {
+ state: "available",
+ role: ref.kind === "turn_item" ? "tool" : "assistant",
+ text: "Exact source " + (offset + index + 1) + ": 日本語 🧪\n unchanged.",
+ },
+ })),
+ nextOffset: firstPart ? 0 : offset === 0 ? 6 : null,
+ nextTextOffset: firstPart ? maxChars : 0,
};
},
},
@@ -99,6 +132,8 @@ function setup() {
state.lists = [];
state.expansions = [];
state.list = null;
+ state.longSource =
+ "Exact source 1: 日本語 🧪\n unchanged.\n" + "x".repeat(12_000) + "\nFinal evidence.";
vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true);
const container = document.createElement("div");
const root = createRoot(container);
@@ -141,20 +176,39 @@ it("reads only while open and supported, shows pass/citations/stale reasons and
screen.setSupported(false);
expect(state.lists).toEqual([]);
expect(screen.container.textContent).toContain("Update the server");
+ const sources = Array.from({ length: 10 }, (_, index): ConversationSummarySourceRef =>
+ index === 9
+ ? {
+ kind: "turn_item",
+ threadId: ThreadId.make("thread"),
+ itemId: TurnItemId.make("tool-output"),
+ digest: "a".repeat(64),
+ }
+ : {
+ kind: "message",
+ threadId: ThreadId.make("thread"),
+ messageId: MessageId.make("message-" + (index < 2 ? 0 : index)),
+ digest: "a".repeat(64),
+ ...(index < 2
+ ? {
+ range: {
+ start: index * state.longSource.length,
+ end: (index + 1) * state.longSource.length,
+ total: state.longSource.length * 2,
+ },
+ }
+ : {}),
+ },
+ );
const summary: ConversationSummary = {
id: "summary:" + "a".repeat(64),
projectId: ProjectId.make("project"),
threadId: ThreadId.make("thread"),
- sources: Array.from({ length: 10 }, (_, index) => ({
- kind: "message",
- threadId: ThreadId.make("thread"),
- messageId: MessageId.make("message-" + index),
- digest: "a".repeat(64),
- })),
+ sources,
statements: [
{
text: "Keep the canonical sources.",
- citations: [MessageId.make("message-0"), MessageId.make("message-9")],
+ citations: [sources[0]!, sources[1]!, sources[9]!].map(conversationSummarySourceKey),
},
],
modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" },
@@ -190,9 +244,9 @@ it("reads only while open and supported, shows pass/citations/stale reasons and
});
expect(state.expansions).toEqual([]);
expect(screen.container.textContent).toContain(
- "Last pass failed. No model 10 older messages wait",
+ "Last pass failed. No model 10 older sources wait",
);
- expect(screen.container.textContent).toContain("Keep the canonical sources. [#1, #10]");
+ expect(screen.container.textContent).toContain("Keep the canonical sources. [#1, #2, #10]");
expect(screen.container.textContent).toContain(
"Outdated: source changed. The summary text is hidden",
);
@@ -200,27 +254,45 @@ it("reads only while open and supported, shows pass/citations/stale reasons and
expect(state.lists.at(-1)?.input.cursor).toBe(staleId);
await screen.click("Earlier summaries");
expect(state.lists.at(-1)?.input.cursor).toBeUndefined();
- await screen.click("Show source messages");
+ await screen.click("Show sources");
expect(state.expansions.at(-1)).toEqual({
environmentId: "remote",
- input: { projectId: "project", threadId: "thread", summaryId: summary.id, offset: 0, limit: 6 },
+ input: {
+ projectId: "project",
+ threadId: "thread",
+ summaryId: summary.id,
+ offset: 0,
+ textOffset: 0,
+ limit: 6,
+ maxChars: 12_000,
+ },
});
+ expect(screen.container.textContent).toContain("Exact source 1: 日本語 🧪\n unchanged.");
+ expect(screen.container.textContent).not.toContain("Final evidence.");
+ await screen.click("Next sources");
+ expect(state.expansions.at(-1)?.input).toMatchObject({ offset: 0, textOffset: 12_000 });
+ expect(screen.container.textContent).toContain("Final evidence.");
+ expect(screen.container.textContent).not.toContain("Exact source 1:");
expect(screen.container.textContent).toContain("Source changed.");
- expect(screen.container.textContent).toContain("Exact source 2: 日本語 🧪\n unchanged.");
- await screen.click("Next messages");
+ await screen.click("Previous sources");
+ expect(state.expansions.at(-1)?.input).toMatchObject({ offset: 0, textOffset: 0 });
+ expect(screen.container.textContent).toContain("Exact source 1:");
+ await screen.click("Next sources");
+ await screen.click("Next sources");
expect(state.expansions.at(-1)?.input.offset).toBe(6);
expect(screen.container.textContent).toContain("#7 · Assistant");
- expect(screen.container.textContent).not.toContain("Exact source 2:");
+ expect(screen.container.textContent).toContain("#10 · Tool");
+ expect(screen.container.textContent).not.toContain("Exact source 1:");
expect(
[...screen.container.querySelectorAll("button")].find(
- (button) => button.textContent === "Next messages",
+ (button) => button.textContent === "Next sources",
)?.disabled,
).toBe(true);
- await screen.click("Previous messages");
- expect(state.expansions.at(-1)?.input.offset).toBe(0);
+ await screen.click("Previous sources");
+ expect(state.expansions.at(-1)?.input).toMatchObject({ offset: 0, textOffset: 12_000 });
await screen.click("Back to summaries");
expect(screen.container.textContent).not.toContain("Exact source");
- await screen.click("Show source messages");
+ await screen.click("Show sources");
const listsBeforeCapabilityLoss = state.lists.length;
const expansionsBeforeCapabilityLoss = state.expansions.length;
screen.setSupported(false);
diff --git a/apps/mobile/src/features/conversation-summaries/ConversationSummariesRouteScreen.tsx b/apps/mobile/src/features/conversation-summaries/ConversationSummariesRouteScreen.tsx
index 5c2d6e32d5757..29ddbed53c99c 100644
--- a/apps/mobile/src/features/conversation-summaries/ConversationSummariesRouteScreen.tsx
+++ b/apps/mobile/src/features/conversation-summaries/ConversationSummariesRouteScreen.tsx
@@ -1,11 +1,16 @@
import { useAtomValue } from "@effect/atom-react";
import { useIsFocused, useNavigation, type StaticScreenProps } from "@react-navigation/native";
-import { conversationSummaryPassLabel } from "@t3tools/client-runtime/conversationSummaries";
+import {
+ conversationSummaryPassLabel,
+ conversationSummarySourceRangeLabel,
+ type ConversationSummarySourceCursor,
+} from "@t3tools/client-runtime/conversationSummaries";
import { KNOWLEDGE_SOURCE_UNAVAILABLE_LABEL } from "@t3tools/client-runtime/projectKnowledge";
import {
EnvironmentId,
ProjectId,
ThreadId,
+ conversationSummarySourceKey,
type ConversationSummaryId,
type ConversationSummaryRead,
} from "@t3tools/contracts";
@@ -13,7 +18,7 @@ import { useState } from "react";
import { ScrollView, View } from "react-native";
import { AppBar } from "../../components/ui/AppBar";
-import { AppText as Text } from "../../components/ui/AppText";
+import { AppText as Text, AppTextInput } from "../../components/ui/AppText";
import { Button } from "../../components/ui/Button";
import { ScreenContent } from "../../components/ui/ScreenContent";
import {
@@ -71,7 +76,7 @@ function SummaryText({ read }: { readonly read: ConversationSummaryRead }) {
);
const position = new Map(
- read.summary.sources.map((source, index) => [source.messageId, index + 1]),
+ read.summary.sources.map((source, index) => [conversationSummarySourceKey(source), index + 1]),
);
return (
@@ -91,9 +96,16 @@ function SummaryText({ read }: { readonly read: ConversationSummaryRead }) {
function SummaryPages(props: Scope) {
const [cursors, setCursors] = useState>([]);
const [selected, setSelected] = useState(null);
+ const [searchText, setSearchText] = useState("");
+ const [query, setQuery] = useState("");
+ const search = () => {
+ setQuery(searchText.trim());
+ setCursors([]);
+ };
return (
@@ -103,14 +115,42 @@ function SummaryPages(props: Scope) {
>
) : (
- setCursors((current) => current.slice(0, -1))}
- onLater={(cursor) => setCursors((current) => [...current, cursor])}
- onSelect={setSelected}
- />
+ <>
+
+
+
+
+ {searchText || query ? (
+
+
+ setCursors((current) => current.slice(0, -1))}
+ onLater={(cursor) => setCursors((current) => [...current, cursor])}
+ onSelect={setSelected}
+ />
+ >
)}
);
@@ -118,6 +158,7 @@ function SummaryPages(props: Scope) {
function SummaryList(
props: Scope & {
+ readonly query: string;
readonly cursors: ReadonlyArray;
readonly onEarlier: () => void;
readonly onLater: (cursor: ConversationSummaryId) => void;
@@ -133,6 +174,7 @@ function SummaryList(
threadId: props.threadId,
limit: 20,
...(cursor === undefined ? {} : { cursor }),
+ ...(props.query ? { query: props.query } : {}),
},
}),
);
@@ -147,8 +189,8 @@ function SummaryList(
return (
<>
- Model-written summaries of older messages. They can replace history that would not fit in a
- handoff; source messages are kept unchanged.
+ Model-written summaries of older messages and tool output. They can replace history that
+ would not fit in a handoff; sources are kept unchanged.
{!enabled ? (
@@ -160,7 +202,11 @@ function SummaryList(
? conversationSummaryPassLabel(lastPass)
: "No summary pass has run yet. Summaries are written after a turn completes and enough older messages have settled."}
- {summaries.length === 0 ? This thread has no summaries yet. : null}
+ {summaries.length === 0 ? (
+
+ {props.query ? "No summaries match your search." : "This thread has no summaries yet."}
+
+ ) : null}
{summaries.map((read) => {
const id = read.state === "available" ? read.summary.id : read.id;
const sourceCount =
@@ -169,10 +215,10 @@ function SummaryList(
return (
- {sourceCount} messages · written {new Date(createdAt).toLocaleString()}
+ {sourceCount} sources · written {new Date(createdAt).toLocaleString()}
-
);
})}
@@ -197,15 +243,16 @@ function SummaryList(
}
function SummarySources(props: Scope & { readonly summaryId: ConversationSummaryId }) {
- const [offsets, setOffsets] = useState>([]);
- const offset = offsets.at(-1) ?? 0;
+ const [cursors, setCursors] = useState>([]);
+ const cursor = cursors.at(-1) ?? { offset: 0, textOffset: 0 };
return (
setOffsets((current) => current.slice(0, -1))}
- onNext={(nextOffset) => setOffsets((current) => [...current, nextOffset])}
+ cursor={cursor}
+ hasPrevious={cursors.length > 0}
+ onPrevious={() => setCursors((current) => current.slice(0, -1))}
+ onNext={(nextCursor) => setCursors((current) => [...current, nextCursor])}
/>
);
}
@@ -213,12 +260,13 @@ function SummarySources(props: Scope & { readonly summaryId: ConversationSummary
function SummarySourcePage(
props: Scope & {
readonly summaryId: ConversationSummaryId;
- readonly offset: number;
+ readonly cursor: ConversationSummarySourceCursor;
+ readonly hasPrevious: boolean;
readonly onPrevious: () => void;
- readonly onNext: (offset: number) => void;
+ readonly onNext: (cursor: ConversationSummarySourceCursor) => void;
},
) {
- const { offset } = props;
+ const { offset, textOffset } = props.cursor;
const query = useEnvironmentQuery(
conversationSummaries.expand({
environmentId: props.environmentId,
@@ -227,7 +275,9 @@ function SummarySourcePage(
threadId: props.threadId,
summaryId: props.summaryId,
offset,
+ textOffset,
limit: 6,
+ maxChars: 12_000,
},
}),
);
@@ -237,21 +287,31 @@ function SummarySourcePage(
{query.error}
);
- if (!query.data) return Loading messages…;
- const { summary, sources, nextOffset } = query.data;
+ if (!query.data) return Loading sources…;
+ const { summary, sources, nextOffset, nextTextOffset } = query.data;
return (
<>
{sources.map(({ ref, result }, index) => (
-
+
#{offset + index + 1} ·{" "}
{result.state === "available"
? result.role === "user"
? "User"
- : "Assistant"
+ : result.role === "tool"
+ ? "Tool"
+ : "Assistant"
: "Unavailable"}
+ {conversationSummarySourceRangeLabel(ref, result) ? (
+
+ {conversationSummarySourceRangeLabel(ref, result)}
+
+ ) : null}
{result.state === "available"
? result.text
@@ -260,12 +320,13 @@ function SummarySourcePage(
))}
-
+ {
- if (nextOffset !== null) props.onNext(nextOffset);
+ if (nextOffset !== null)
+ props.onNext({ offset: nextOffset, textOffset: nextTextOffset ?? 0 });
}}
/>
diff --git a/apps/server/src/mcp/toolkits/conversationSummaries/tools.ts b/apps/server/src/mcp/toolkits/conversationSummaries/tools.ts
index d7559e093e8ce..071212e8104f2 100644
--- a/apps/server/src/mcp/toolkits/conversationSummaries/tools.ts
+++ b/apps/server/src/mcp/toolkits/conversationSummaries/tools.ts
@@ -28,7 +28,7 @@ export const ConversationSummariesToolkit = Toolkit.make(
Tool.make("t3_conversation_summary_list", {
...shared,
description:
- "List the generated summaries of a thread's older messages, oldest first, with whether summarizing is on for the project and what the latest background pass did. Use when a handoff mentions a conversation summary or you need the gist of a long thread before reading it with t3_thread_read. A summary whose source messages changed or were removed is stale and returns no text. Paginate with cursor=nextCursor. Does not generate summaries.",
+ "List stored thread summaries (messages/tool evidence), oldest first, for long-thread recall or handoffs. Includes enabled status and last pass. query matches statement text as a literal case-insensitive substring. Follow cursor=nextCursor with the same query. Stale summaries expose no text; this never generates summaries.",
parameters: Schema.Struct({
...ConversationSummaryListInput.fields,
environmentId: OrchestratorMcpEnvironmentTarget,
@@ -40,7 +40,7 @@ export const ConversationSummariesToolkit = Toolkit.make(
Tool.make("t3_conversation_summary_describe", {
...shared,
description:
- "Read one conversation summary: its statements, the message ids each statement cites, and its exact source messages (ids and digests). The text is a model-written summary, not the conversation; check anything you rely on with t3_conversation_summary_expand. Stale summaries return no text.",
+ "Read a generated summary's statements, citation keys and canonical message/tool source refs with digests and optional UTF-16 ranges (end exclusive). Verify claims with t3_conversation_summary_expand. Stale summaries expose no text.",
parameters: Schema.Struct({
...ConversationSummaryDescribeInput.fields,
environmentId: OrchestratorMcpEnvironmentTarget,
@@ -52,7 +52,7 @@ export const ConversationSummariesToolkit = Toolkit.make(
Tool.make("t3_conversation_summary_expand", {
...shared,
description:
- "Return the exact original messages a conversation summary was written from, in order, as they are now. A message that changed, was removed, or sits in a deleted or archived thread returns a typed unavailable reason instead of text. Paginate with offset=nextOffset.",
+ "Read canonical source ranges; changed, removed, archived or unfinished sources are unavailable. maxChars bounds page text (default 12000, max 48000 UTF-16 units); limit bounds source count. Continue with offset=nextOffset AND textOffset=nextTextOffset even when offset repeats. null nextOffset ends paging.",
parameters: Schema.Struct({
...ConversationSummaryExpandInput.fields,
environmentId: OrchestratorMcpEnvironmentTarget,
diff --git a/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts b/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts
index d2dcd07a62d29..a3a6df9bb745f 100644
--- a/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts
+++ b/apps/server/src/orchestration-v2/ContextHandoffBudget.test.ts
@@ -2,6 +2,7 @@ import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAda
import { assert, describe, it } from "@effect/vitest";
import {
ContextHandoffId,
+ NodeId,
PROVIDER_SEND_TURN_MAX_ATTACHMENTS,
ProviderDriverKind,
ProviderInstanceId,
@@ -111,7 +112,7 @@ describe("handoff budget", () => {
assert.equal(decodeHandoff(projected).id, handoff.id);
assert.deepEqual(handoff.history?.messages, messages);
});
- it("carries command outcomes as attributed activity and excludes reasoning", () => {
+ it("carries complete tool evidence as attributed activity and excludes reasoning", () => {
const base = {
id: TurnItemId.make("item:command"),
threadId,
@@ -139,6 +140,71 @@ describe("handoff budget", () => {
assert.equal(command?.kind, "command_execution");
assert.include(command!.text, "Exit code: 1");
assert.include(command!.text, "界".repeat(300));
+ const evidence = [
+ {
+ item: {
+ ...base,
+ type: "dynamic_tool",
+ toolName: "verify_health",
+ input: { endpoint: "/health" },
+ output: { failure: "Expected HTTP 200, received 503" },
+ },
+ text: "Expected HTTP 200, received 503",
+ },
+ {
+ item: {
+ ...base,
+ type: "file_change",
+ fileName: "src/health.ts",
+ diffStr: "-return 503;\n+return 200;",
+ },
+ text: "+return 200;",
+ },
+ {
+ item: {
+ ...base,
+ type: "file_search",
+ pattern: "health",
+ results: [{ fileName: "src/health.ts", line: 7, preview: "export const health = 200" }],
+ },
+ text: "export const health = 200",
+ },
+ {
+ item: {
+ ...base,
+ type: "web_search",
+ patterns: ["HTTP 503"],
+ results: [{ url: "https://example.com/status/503", title: "Service unavailable" }],
+ },
+ text: "Service unavailable",
+ },
+ {
+ item: {
+ ...base,
+ type: "subagent",
+ subagentId: NodeId.make("node:health-check"),
+ origin: "app_owned",
+ driver: ProviderDriverKind.make("codex"),
+ providerInstanceId: ProviderInstanceId.make("codex"),
+ childThreadId: null,
+ prompt: "Check the health endpoint",
+ result: "The upstream service is unavailable",
+ },
+ text: "The upstream service is unavailable",
+ },
+ ] satisfies ReadonlyArray<{ item: OrchestrationV2TurnItem; text: string }>;
+ for (const { item, text } of evidence) {
+ const historical = historicalMessage(item);
+ assert.isNotNull(
+ historical,
+ `${item.type} must be eligible for handoff and summary coverage`,
+ );
+ assert.include(historical!.text, text);
+ assert.equal(historical!.kind, item.type);
+ assert.equal(historical!.itemId, item.id);
+ if (item.type === "dynamic_tool")
+ assert.isNull(historicalMessage({ ...item, outputOmitted: true }));
+ }
assert.isNull(
historicalMessage({
...base,
diff --git a/apps/server/src/orchestration-v2/ContextHandoffDelivery.ts b/apps/server/src/orchestration-v2/ContextHandoffDelivery.ts
index 68212918d0031..501fc61377937 100644
--- a/apps/server/src/orchestration-v2/ContextHandoffDelivery.ts
+++ b/apps/server/src/orchestration-v2/ContextHandoffDelivery.ts
@@ -177,7 +177,7 @@ export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextH
return {
summaryIds: chosen
.filter((summary) => summary.coveredItemIds.some((id) => own.has(id)))
- .map((summary) => summary.id),
+ .flatMap((summary) => summary.summaryIds ?? [summary.id]),
summarizedItemIds,
};
};
diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts
index 421299e7f8b02..b47d454258786 100644
--- a/apps/server/src/orchestration-v2/EffectOutbox.ts
+++ b/apps/server/src/orchestration-v2/EffectOutbox.ts
@@ -3,6 +3,7 @@ import {
CheckpointScopeId,
CommandId,
MessageId,
+ NonNegativeInt,
ProjectId,
NodeId,
ProviderSessionId,
@@ -134,6 +135,7 @@ export const OrchestrationEffectRequestV2 = Schema.Union([
Schema.Struct({
type: Schema.Literal("conversation-summary.generate"),
runId: RunId,
+ pass: Schema.optional(NonNegativeInt),
}),
]);
export type OrchestrationEffectRequestV2 = typeof OrchestrationEffectRequestV2.Type;
diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts
index 1362adad9e59c..34de62625c347 100644
--- a/apps/server/src/orchestration-v2/EffectWorker.test.ts
+++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts
@@ -38,6 +38,10 @@ import * as ThreadManagementService from "./ThreadManagementService.ts";
import * as ServerSettings from "../serverSettings.ts";
import * as SqlitePersistence from "../persistence/Sqlite.ts";
import { AssistedLearningRunner } from "../project/knowledge/assistedLearningSchedule.ts";
+import {
+ ConversationSummaryPassError,
+ ConversationSummaryRuntime,
+} from "../project/conversationSummaries/conversationSummarySchedule.ts";
import {
DeliveryScrubRunner,
deliveryScrubEffect,
@@ -89,6 +93,102 @@ it.effect("coalesces delivery scrub roots and resumes bounded pages after proces
const threadId = ThreadId.make("thread:effect-worker-restart");
+it.effect(
+ "continues bounded summary passes without a new turn and resumes after process loss",
+ () =>
+ Effect.gen(function* () {
+ const calls = yield* Ref.make(0);
+ const root = {
+ id: "effect:conversation-summary.generate:run:summary-pages",
+ commandId: CommandId.make("command:effect:conversation-summary.generate:run:summary-pages"),
+ threadId,
+ request: {
+ type: "conversation-summary.generate",
+ runId: RunId.make("run:summary-pages"),
+ },
+ } as const;
+ const runner = Layer.succeed(ConversationSummaryRuntime, {
+ summarizeThread: () =>
+ Ref.updateAndGet(calls, (count) => count + 1).pipe(
+ Effect.map((count) => ({ hasMore: count < 3 })),
+ ),
+ forHandoff: () => Effect.succeed([]),
+ });
+ const dependencies = Layer.merge(
+ EffectOutbox.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
+ Layer.succeed(EffectWorker.OrchestrationEffectExecutorV2, { execute: () => Effect.void }),
+ );
+ const layer = EffectWorker.layer.pipe(
+ Layer.provideMerge(dependencies),
+ Layer.provide(runner),
+ );
+ yield* Effect.gen(function* () {
+ const outbox = yield* EffectOutbox.EffectOutboxV2;
+ const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
+ yield* outbox.enqueue([root, root]);
+ assert.isTrue(yield* worker.runMaintenanceOnce);
+ const continuation = yield* outbox.get(`${root.id}:pass:1`);
+ assert.isTrue(Option.isSome(continuation));
+ const claimed = yield* outbox.claimNext({ workerId: "lost-worker", lane: "maintenance" });
+ assert.isTrue(Option.isSome(claimed));
+ assert.deepEqual(yield* outbox.reconcileAfterProcessLoss, { requeued: 1, cancelled: 0 });
+ assert.equal(yield* worker.drain(), 2);
+ assert.equal(yield* Ref.get(calls), 3);
+ assert.isTrue(Option.isNone(yield* outbox.get(`${root.id}:pass:3`)));
+ yield* outbox.enqueue([root]);
+ assert.equal(yield* worker.drain(), 0);
+ }).pipe(Effect.provide(layer));
+ }),
+);
+
+it.effect("retries a failed summary pass before scheduling its continuation", () =>
+ Effect.gen(function* () {
+ const calls = yield* Ref.make(0);
+ const root = {
+ id: "effect:conversation-summary.generate:run:summary-retry",
+ commandId: CommandId.make("command:effect:conversation-summary.generate:run:summary-retry"),
+ threadId,
+ request: {
+ type: "conversation-summary.generate",
+ runId: RunId.make("run:summary-retry"),
+ },
+ } as const;
+ const runner = Layer.succeed(ConversationSummaryRuntime, {
+ summarizeThread: ({ runId }) =>
+ Ref.updateAndGet(calls, (count) => count + 1).pipe(
+ Effect.flatMap((count) =>
+ count === 1
+ ? Effect.fail(new ConversationSummaryPassError({ runId, cause: "model unavailable" }))
+ : Effect.succeed({ hasMore: count === 2 }),
+ ),
+ ),
+ forHandoff: () => Effect.succeed([]),
+ });
+ const dependencies = Layer.merge(
+ EffectOutbox.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
+ Layer.succeed(EffectWorker.OrchestrationEffectExecutorV2, { execute: () => Effect.void }),
+ );
+ const layer = EffectWorker.layer.pipe(Layer.provideMerge(dependencies), Layer.provide(runner));
+ yield* Effect.gen(function* () {
+ const outbox = yield* EffectOutbox.EffectOutboxV2;
+ const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
+ yield* outbox.enqueue([root]);
+ assert.isTrue(yield* worker.runMaintenanceOnce);
+ const retrying = yield* outbox.get(root.id);
+ assert.isTrue(Option.isSome(retrying));
+ if (Option.isSome(retrying)) {
+ assert.equal(retrying.value.status, "pending");
+ assert.equal(retrying.value.attemptCount, 1);
+ }
+ assert.isTrue(Option.isNone(yield* outbox.get(`${root.id}:pass:1`)));
+ assert.equal(yield* worker.drain(), 0);
+ yield* TestClock.adjust(100);
+ assert.equal(yield* worker.drain(), 2);
+ assert.equal(yield* Ref.get(calls), 3);
+ }).pipe(Effect.provide(layer));
+ }),
+);
+
it.effect.each(["turn", "maintenance"] as const)(
"wakes the %s lane when the other lane is waiting first",
(lane) =>
@@ -201,50 +301,60 @@ const providerTurnId = ProviderTurnId.make("provider-turn:effect-worker-restart"
const attemptId = RunAttemptId.make("run-attempt:effect-worker-restart");
const runId = RunId.make("run:effect-worker-restart");
-it.effect("starts a follow-up turn while assisted learning is held", () =>
- Effect.gen(function* () {
- const entered = yield* Deferred.make();
- const release = yield* Deferred.make();
- const events = yield* Ref.make>([]);
- const outboxLayer = EffectOutbox.layer.pipe(Layer.provide(SqlitePersistence.layerMemory));
- const executorLayer = layerExecutorFor({ events }).pipe(
- Layer.provide(
- Layer.succeed(AssistedLearningRunner, {
- extractRun: () =>
- Deferred.succeed(entered, undefined).pipe(Effect.andThen(Deferred.await(release))),
+it.effect.each(["project-knowledge.extract", "conversation-summary.generate"] as const)(
+ "starts a follow-up turn while %s is held",
+ (maintenance) =>
+ Effect.gen(function* () {
+ const entered = yield* Deferred.make();
+ const release = yield* Deferred.make();
+ const events = yield* Ref.make>([]);
+ const outboxLayer = EffectOutbox.layer.pipe(Layer.provide(SqlitePersistence.layerMemory));
+ const hold = Deferred.succeed(entered, undefined).pipe(
+ Effect.andThen(Deferred.await(release)),
+ );
+ const runners = Layer.merge(
+ Layer.succeed(AssistedLearningRunner, { extractRun: () => hold }),
+ Layer.succeed(ConversationSummaryRuntime, {
+ summarizeThread: () => hold.pipe(Effect.as({ hasMore: false })),
+ forHandoff: () => Effect.succeed([]),
}),
- ),
- );
- const dependencies = Layer.merge(outboxLayer, executorLayer);
- const layer = EffectWorker.layer.pipe(Layer.provideMerge(dependencies));
- yield* Effect.gen(function* () {
- const outbox = yield* EffectOutbox.EffectOutboxV2;
- const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
- yield* outbox.enqueue([
- {
- id: "effect:held-learning",
- commandId: CommandId.make("command:held-learning"),
- threadId,
- request: { type: "project-knowledge.extract", runId, runOutcome: "completed" },
- },
- ]);
- yield* worker.runMaintenanceOnce.pipe(Effect.forkScoped);
- yield* Deferred.await(entered);
- yield* outbox.enqueue([
- {
- id: "effect:follow-up",
- commandId: CommandId.make("command:follow-up"),
- threadId,
- request: { type: "provider-turn.start", runId: RunId.make("run:follow-up") },
- },
- ]);
- assert.isTrue(yield* worker.runTurnOnce);
- assert.deepEqual(yield* Ref.get(events), ["start"]);
- assert.isFalse(yield* Deferred.isDone(release));
- yield* Deferred.succeed(release, undefined);
- yield* outbox.awaitSettled("effect:held-learning");
- }).pipe(Effect.provide(layer));
- }),
+ );
+ const dependencies = Layer.merge(outboxLayer, layerExecutorFor({ events }));
+ const layer = EffectWorker.layer.pipe(
+ Layer.provideMerge(dependencies),
+ Layer.provide(runners),
+ );
+ yield* Effect.gen(function* () {
+ const outbox = yield* EffectOutbox.EffectOutboxV2;
+ const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
+ yield* outbox.enqueue([
+ {
+ id: "effect:held-maintenance",
+ commandId: CommandId.make("command:held-maintenance"),
+ threadId,
+ request:
+ maintenance === "project-knowledge.extract"
+ ? { type: maintenance, runId, runOutcome: "completed" }
+ : { type: maintenance, runId },
+ },
+ ]);
+ yield* worker.runMaintenanceOnce.pipe(Effect.forkScoped);
+ yield* Deferred.await(entered);
+ yield* outbox.enqueue([
+ {
+ id: "effect:follow-up",
+ commandId: CommandId.make("command:follow-up"),
+ threadId,
+ request: { type: "provider-turn.start", runId: RunId.make("run:follow-up") },
+ },
+ ]);
+ assert.isTrue(yield* worker.runTurnOnce);
+ assert.deepEqual(yield* Ref.get(events), ["start"]);
+ assert.isFalse(yield* Deferred.isDone(release));
+ yield* Deferred.succeed(release, undefined);
+ yield* outbox.awaitSettled("effect:held-maintenance");
+ }).pipe(Effect.provide(layer));
+ }),
);
it.effect.each(["provider-turn.start", "thread-title.generate"] as const)(
diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts
index 9e133a639bf94..02306714a5ac0 100644
--- a/apps/server/src/orchestration-v2/EffectWorker.ts
+++ b/apps/server/src/orchestration-v2/EffectWorker.ts
@@ -33,7 +33,10 @@ import {
deliveryScrubEffect,
} from "../project/knowledge/deliveryScrubSchedule.ts";
import { AssistedLearningRunner } from "../project/knowledge/assistedLearningSchedule.ts";
-import { ConversationSummaryRuntime } from "../project/conversationSummaries/conversationSummarySchedule.ts";
+import {
+ ConversationSummaryRuntime,
+ conversationSummaryEffect,
+} from "../project/conversationSummaries/conversationSummarySchedule.ts";
export class OrchestrationEffectExecutionError extends Schema.TaggedError()(
"OrchestrationEffectExecutionError",
@@ -596,6 +599,7 @@ export const layerExecutor: Layer.Layer<
return conversationSummaries
.summarizeThread({ threadId: effect.threadId, runId: effect.request.runId })
.pipe(
+ Effect.asVoid,
Effect.mapError(
(cause) =>
new OrchestrationEffectExecutionError({
@@ -698,6 +702,7 @@ export const layerWithOptions = (
const outbox = yield* EffectOutbox.EffectOutboxV2;
const executor = yield* OrchestrationEffectExecutorV2;
const deliveryScrubber = yield* DeliveryScrubRunner;
+ const conversationSummaries = yield* ConversationSummaryRuntime;
const workerId = options.workerId ?? `orchestration-v2:${process.pid}`;
const maxAttempts = Math.max(1, options.maxAttempts ?? 5);
const wasCancelled = (effectId: string) =>
@@ -865,7 +870,35 @@ export const layerWithOptions = (
}),
),
)
- : executor.execute(effect, { willRetry: effect.attemptCount < maxAttempts })
+ : effect.request.type === "conversation-summary.generate"
+ ? Effect.gen(function* () {
+ const request = effect.request;
+ if (request.type !== "conversation-summary.generate") return;
+ const result = yield* conversationSummaries.summarizeThread({
+ threadId: effect.threadId,
+ runId: request.runId,
+ });
+ if (result.hasMore) {
+ yield* outbox.enqueue([
+ conversationSummaryEffect({
+ threadId: effect.threadId,
+ runId: request.runId,
+ pass: (request.pass ?? 0) + 1,
+ }),
+ ]);
+ yield* outbox.notifyAvailable();
+ }
+ }).pipe(
+ Effect.mapError(
+ (cause) =>
+ new OrchestrationEffectExecutionError({
+ effectId: effect.id,
+ effectType: effect.request.type,
+ cause,
+ }),
+ ),
+ )
+ : executor.execute(effect, { willRetry: effect.attemptCount < maxAttempts })
).pipe(boundExecution(effect), Effect.as("executed" as const));
const exit = yield* Effect.exit(Effect.raceFirst(execution, cancellation)).pipe(
Effect.ensuring(outbox.clearCancellation(effect.id)),
diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts
index 975ce7b0b9c50..7ca38d9401f40 100644
--- a/apps/server/src/orchestration-v2/ProjectionStore.ts
+++ b/apps/server/src/orchestration-v2/ProjectionStore.ts
@@ -333,6 +333,7 @@ export interface ProjectionRecordFilter {
readonly monitorNotificationsOnly?: boolean;
readonly turnItemRunId?: RunId;
readonly messageIds?: ReadonlyArray;
+ readonly turnItemIds?: ReadonlyArray;
readonly messageRunIds?: ReadonlyArray;
readonly turnItemRunIds?: ReadonlyArray;
readonly turnItemNodeIds?: ReadonlyArray;
@@ -3041,6 +3042,7 @@ export const layer: Layer.Layer =
(fields !== undefined && !fields.includes("turnItems")) ||
(window === undefined &&
matchesNothing(
+ filter?.turnItemIds,
filter?.turnItemTypes,
filter?.turnItemStatuses,
filter?.turnItemRunIds,
@@ -3052,6 +3054,7 @@ export const layer: Layer.Layer =
SELECT payload_json
FROM orchestration_v2_projection_turn_items
WHERE thread_id = ${threadId}
+ ${filter?.turnItemIds === undefined ? sql`` : sql`AND turn_item_id IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemIds)}))`}
${filter?.turnItemTypes === undefined ? sql`` : sql`AND type IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemTypes)}))`}
${filter?.turnItemStatuses === undefined ? sql`` : sql`AND status IN (SELECT value FROM json_each(${encodeIdList(filter.turnItemStatuses)}))`}
${filter?.turnItemRunId === undefined ? sql`` : sql`AND run_id = ${filter.turnItemRunId}`}
diff --git a/apps/server/src/project/conversationSummaries/ConversationSummaries.test.ts b/apps/server/src/project/conversationSummaries/ConversationSummaries.test.ts
index 0bd11c30c6a21..865f50a041105 100644
--- a/apps/server/src/project/conversationSummaries/ConversationSummaries.test.ts
+++ b/apps/server/src/project/conversationSummaries/ConversationSummaries.test.ts
@@ -4,10 +4,12 @@ import {
CommandId,
ComposerContextId,
ConversationSummary,
+ conversationSummarySourceKey,
DEFAULT_SERVER_SETTINGS,
EnvironmentId,
EventId,
MessageId,
+ NodeId,
ProjectFactId,
ProjectId,
ProviderDriverKind,
@@ -56,7 +58,10 @@ import * as HolographicMemory from "../knowledge/HolographicMemory.ts";
import * as ProjectKnowledge from "../knowledge/ProjectKnowledge.ts";
import { deliveredKnowledgeForHistory } from "../knowledge/composer.ts";
import * as ConversationSummaries from "./ConversationSummaries.ts";
-import { conversationSummaryEffects } from "./conversationSummarySchedule.ts";
+import {
+ ConversationSummaryPassError,
+ conversationSummaryEffects,
+} from "./conversationSummarySchedule.ts";
type Generate = TextGeneration.TextGeneration["Service"]["generateConversationSummary"];
@@ -298,7 +303,10 @@ const persistSummary = Effect.fn("test.persistSummary")(function* (
return summary;
});
-const prepareHandoff = Effect.fn("test.prepareSummaryHandoff")(function* (messageCount: number) {
+const prepareHandoff = Effect.fn("test.prepareSummaryHandoff")(function* (
+ messageCount: number,
+ extraItems: ReadonlyArray = [],
+) {
const handoffs = yield* ContextHandoffService.ContextHandoffServiceV2;
return yield* handoffs.prepareProviderHandoff({
threadId,
@@ -310,7 +318,7 @@ const prepareHandoff = Effect.fn("test.prepareSummaryHandoff")(function* (messag
toProviderInstanceId: instanceId,
coveredRunOrdinals: { from: 1, to: 1 },
strategy: "full_thread_summary",
- items: Array.from({ length: messageCount }, (_, index) => item(index)),
+ items: [...extraItems, ...Array.from({ length: messageCount }, (_, index) => item(index))],
createdAt: at,
});
});
@@ -344,7 +352,7 @@ it.effect(
);
const summary = yield* onlySummary;
assert.deepEqual(
- summary.sources.map((source) => source.messageId),
+ summary.sources.map((source) => (source.kind === "message" ? source.messageId : null)),
Array.from({ length: 10 }, (_, index) => messageId(index)),
);
// Only cited refs survive, mapped to the exact message ids; the invented ref is dropped.
@@ -399,7 +407,7 @@ it.effect("records an unavailable model as a failed pass without publishing a su
);
return Effect.gen(function* () {
yield* seed;
- yield* summarize;
+ assert.instanceOf(yield* summarize.pipe(Effect.flip), ConversationSummaryPassError);
const result = yield* list;
assert.deepEqual(result.summaries, []);
assert.deepInclude(result.lastPass, {
@@ -410,6 +418,432 @@ it.effect("records an unavailable model as a failed pass without publishing a su
}).pipe(Effect.provide(makeLayer(model.generate, optedIn)));
});
+it.effect(
+ "covers every range of a long source and offers it only after all ranges are summarized",
+ () => {
+ const model = citing();
+ return Effect.gen(function* () {
+ yield* seed;
+ const projections = yield* ProjectionStore.ProjectionStoreV2;
+ const longText = `${"aaaaaa🧪".repeat(17_000)}TAIL-EXACT-EVIDENCE`;
+ yield* projections.apply(messageEvent(0, longText));
+ const service = yield* ConversationSummaries.ConversationSummaries;
+ const first = yield* summarize;
+ const handoff = yield* prepareHandoff(MESSAGE_COUNT);
+ const input = {
+ projectId,
+ handoffs: [handoff],
+ itemIds: new Set(Array.from({ length: 10 }, (_, index) => itemId(index))),
+ };
+ assert.deepEqual(yield* service.forHandoff(input), []);
+ assert.isTrue(first.hasMore);
+ let hasMore = first.hasMore;
+ for (let pass = 0; pass < 8 && hasMore; pass++) hasMore = (yield* summarize).hasMore;
+ assert.isFalse(hasMore);
+ const ordinary = new Set(Array.from({ length: MESSAGE_COUNT }, (_, index) => rawText(index)));
+ const supplied = model.calls
+ .flatMap((call) => call.messages)
+ .filter((message) => !ordinary.has(message.text));
+ assert.equal(supplied.map((message) => message.text).join(""), longText);
+ assert.isTrue(
+ supplied.every((message) => message.text.length <= 6_000 && message.text.isWellFormed()),
+ );
+ assert.isTrue(
+ model.calls.every(
+ (call) => call.messages.reduce((sum, message) => sum + message.text.length, 0) <= 32_000,
+ ),
+ );
+ const summaries = (yield* list).summaries.flatMap((read) =>
+ read.state === "available" ? [read.summary] : [],
+ );
+ assert.isAbove(summaries.length, 3);
+ const starts = summaries.map((summary) => summary.sources[0]!.range?.start ?? 0);
+ assert.deepEqual(
+ starts,
+ starts.toSorted((a, b) => a - b),
+ );
+ const pagedIds: Array = [];
+ let cursor: string | undefined;
+ do {
+ const page = yield* service.list({
+ ...scope,
+ limit: 2,
+ ...(cursor === undefined ? {} : { cursor }),
+ });
+ pagedIds.push(
+ ...page.summaries.flatMap((read) =>
+ read.state === "available" ? [read.summary.id] : [],
+ ),
+ );
+ cursor = page.nextCursor ?? undefined;
+ } while (cursor !== undefined);
+ assert.deepEqual(
+ pagedIds,
+ summaries.map((summary) => summary.id),
+ );
+ assert.isTrue(
+ summaries.every((summary) =>
+ summary.statements.every((statement) =>
+ statement.citations.every((citation) =>
+ summary.sources.some((source) => conversationSummarySourceKey(source) === citation),
+ ),
+ ),
+ ),
+ );
+ const offers = yield* service.forHandoff(input);
+ assert.equal(offers.length, 1);
+ assert.deepEqual(
+ offers[0]!.summaryIds ?? [],
+ summaries.map((summary) => summary.id),
+ );
+ assert.sameMembers(offers[0]!.coveredItemIds, [...input.itemIds]);
+ }).pipe(Effect.provide(makeLayer(model.generate, optedIn)));
+ },
+);
+
+it.effect(
+ "keeps scrubbed summaries stale after a source reverts and regenerates their text",
+ () => {
+ const model = citing();
+ return Effect.gen(function* () {
+ yield* seed;
+ yield* summarize;
+ const summary = yield* onlySummary;
+ const service = yield* ConversationSummaries.ConversationSummaries;
+ const projections = yield* ProjectionStore.ProjectionStoreV2;
+ yield* projections.apply(messageEvent(0, "Changed source"));
+ assert.equal((yield* service.describe({ ...scope, summaryId: summary.id })).state, "stale");
+ yield* projections.apply(messageEvent(0, rawText(0)));
+ assert.equal((yield* service.describe({ ...scope, summaryId: summary.id })).state, "stale");
+ yield* summarize;
+ assert.equal(model.calls.length, 2);
+ const regenerated = yield* onlySummary;
+ assert.equal(regenerated.statements[0]!.text, summary.statements[0]!.text);
+ assert.notInclude(regenerated.statements[0]!.text, "Source unavailable");
+ }).pipe(Effect.provide(makeLayer(model.generate, optedIn)));
+ },
+);
+
+it.effect("regenerates legacy prefix summaries while keeping their full source recoverable", () => {
+ const model = citing();
+ return Effect.gen(function* () {
+ yield* seed;
+ yield* summarize;
+ const previous = yield* onlySummary;
+ const text = `${"Legacy text. ".repeat(1_000)}EXACT-TAIL`;
+ const projections = yield* ProjectionStore.ProjectionStoreV2;
+ yield* projections.apply(messageEvent(0, text));
+ const crypto = yield* Crypto.Crypto;
+ const digest = Hex.encode(yield* crypto.digest("SHA-256", new TextEncoder().encode(text)));
+ const legacy = {
+ ...previous,
+ sources: previous.sources.map((source, index) =>
+ index === 0 ? { ...source, digest } : source,
+ ),
+ };
+ const sql = yield* SqlClient.SqlClient;
+ yield* sql`UPDATE conversation_summaries SET payload_json = ${yield* encodeSummary(legacy)} WHERE summary_id = ${legacy.id}`;
+ const service = yield* ConversationSummaries.ConversationSummaries;
+ assert.equal((yield* service.describe({ ...scope, summaryId: legacy.id })).state, "stale");
+ const original = yield* service.expand({
+ ...scope,
+ summaryId: legacy.id,
+ limit: 1,
+ maxChars: 257,
+ });
+ assert.equal(
+ original.sources[0]?.result.state === "available" ? original.sources[0].result.text : null,
+ text.slice(0, 257),
+ );
+ assert.equal(original.nextOffset, 0);
+ assert.equal(original.nextTextOffset, 257);
+ yield* summarize;
+ const replacement = yield* onlySummary;
+ assert.notEqual(replacement.id, legacy.id);
+ assert.isTrue(replacement.sources.some((source) => source.range !== undefined));
+ assert.include(
+ model.calls
+ .at(-1)
+ ?.messages.map((message) => message.text)
+ .join(""),
+ "EXACT-TAIL",
+ );
+ }).pipe(Effect.provide(makeLayer(model.generate, optedIn)));
+});
+
+it.effect(
+ "includes canonical command evidence and invalidates it when command output changes",
+ () => {
+ const model = citing();
+ return Effect.gen(function* () {
+ yield* seed;
+ const projections = yield* ProjectionStore.ProjectionStoreV2;
+ const command: OrchestrationV2TurnItem = {
+ ...item(0),
+ id: TurnItemId.make("test-command"),
+ type: "command_execution",
+ ordinal: 20,
+ input: "vp test run regression.test.ts",
+ output: "3 tests failed: suffix regression",
+ exitCode: 1,
+ completedAt: DateTime.add(at, { minutes: 3, seconds: 30 }),
+ };
+ const event: OrchestrationV2DomainEvent = {
+ id: EventId.make("test-command-complete"),
+ type: "turn-item.updated",
+ threadId,
+ providerInstanceId: instanceId,
+ occurredAt: at,
+ payload: command,
+ };
+ yield* projections.apply(event);
+ yield* summarize;
+ const supplied = model.calls
+ .flatMap((call) => call.messages)
+ .find((message) => message.role === "tool");
+ assert.include(supplied?.text, "3 tests failed: suffix regression");
+ const summary = yield* onlySummary;
+ const index = summary.sources.findIndex(
+ (source) => source.kind === "turn_item" && source.itemId === command.id,
+ );
+ assert.isAtLeast(index, 0);
+ const service = yield* ConversationSummaries.ConversationSummaries;
+ const expansion = yield* service.expand({
+ ...scope,
+ summaryId: summary.id,
+ offset: index,
+ limit: 1,
+ });
+ assert.equal(
+ expansion.sources[0]?.result.state === "available"
+ ? expansion.sources[0].result.text
+ : null,
+ supplied?.text,
+ );
+ yield* projections.apply({
+ ...event,
+ id: EventId.make("test-command-edited"),
+ payload: { ...command, output: "Now different" },
+ });
+ assert.equal((yield* service.describe({ ...scope, summaryId: summary.id })).state, "stale");
+ assert.deepEqual(
+ (yield* service.expand({ ...scope, summaryId: summary.id, offset: index, limit: 1 }))
+ .sources[0]?.result,
+ { state: "unavailable", reason: "changed" },
+ );
+ }).pipe(Effect.provide(makeLayer(model.generate, optedIn)));
+ },
+);
+
+it.effect(
+ "excludes cancelled queued input from generation and previously generated summaries",
+ () => {
+ const model = citing();
+ return Effect.gen(function* () {
+ yield* seed;
+ const projections = yield* ProjectionStore.ProjectionStoreV2;
+ const queued = item(0);
+ if (queued.type !== "user_message") return yield* Effect.die("Expected a user message");
+ yield* projections.apply({
+ id: EventId.make("queued-source-item"),
+ type: "turn-item.updated",
+ threadId,
+ providerInstanceId: instanceId,
+ occurredAt: at,
+ payload: { ...queued, inputIntent: "queued_turn" },
+ });
+ const run: Extract = {
+ id: EventId.make("queued-source-run"),
+ type: "run.updated",
+ threadId,
+ runId,
+ providerInstanceId: instanceId,
+ occurredAt: at,
+ payload: {
+ id: runId,
+ threadId,
+ ordinal: 1,
+ providerInstanceId: instanceId,
+ modelSelection: { instanceId, model: "gpt-5" },
+ providerThreadId: null,
+ userMessageId: messageId(0),
+ rootNodeId: NodeId.make("queued-source-root"),
+ activeAttemptId: null,
+ status: "completed",
+ requestedAt: at,
+ startedAt: at,
+ completedAt: at,
+ checkpointId: null,
+ contextHandoffId: null,
+ },
+ };
+ yield* projections.apply(run);
+ yield* summarize;
+ const summary = yield* onlySummary;
+ yield* projections.apply({
+ ...run,
+ id: EventId.make("queued-source-cancelled"),
+ payload: { ...run.payload, status: "cancelled" },
+ });
+ const service = yield* ConversationSummaries.ConversationSummaries;
+ assert.equal((yield* service.describe({ ...scope, summaryId: summary.id })).state, "stale");
+ assert.deepEqual(
+ (yield* service.expand({ ...scope, summaryId: summary.id, limit: 1 })).sources[0]?.result,
+ { state: "unavailable", reason: "missing" },
+ );
+ yield* summarize;
+ assert.equal(model.calls.length, 2);
+ assert.isTrue(model.calls[1]!.messages.every((message) => message.text !== rawText(0)));
+ const current = (yield* list).summaries.flatMap((read) =>
+ read.state === "available" ? [read.summary] : [],
+ );
+ assert.equal(current.length, 1);
+ assert.equal(current[0]!.sources.length, 9);
+ assert.isFalse(
+ current[0]!.sources.some(
+ (source) => source.kind === "message" && source.messageId === messageId(0),
+ ),
+ );
+ }).pipe(Effect.provide(makeLayer(model.generate, optedIn)));
+ },
+);
+
+it.effect("does not offer tool-only evidence from a deleted project", () => {
+ const model = citing();
+ return Effect.gen(function* () {
+ yield* seedThread(6);
+ const projections = yield* ProjectionStore.ProjectionStoreV2;
+ const command: OrchestrationV2TurnItem = {
+ ...item(0),
+ id: TurnItemId.make("tool-only-command"),
+ type: "command_execution",
+ input: "vp test run",
+ output: "Check output. ".repeat(600),
+ exitCode: 0,
+ completedAt: DateTime.subtract(at, { minutes: 1 }),
+ };
+ yield* projections.apply({
+ id: EventId.make("tool-only-command-completed"),
+ type: "turn-item.updated",
+ threadId,
+ providerInstanceId: instanceId,
+ occurredAt: at,
+ payload: command,
+ });
+ yield* summarize;
+ const summary = yield* onlySummary;
+ assert.isTrue(summary.sources.every((source) => source.kind === "turn_item"));
+ const service = yield* ConversationSummaries.ConversationSummaries;
+ const input = {
+ projectId,
+ handoffs: [yield* prepareHandoff(6, [command])],
+ itemIds: new Set([command.id]),
+ };
+ assert.equal((yield* service.forHandoff(input)).length, 1);
+ const projects = yield* ProjectStore.ProjectStoreV2;
+ yield* projects.apply({
+ sequence: 1,
+ eventId: EventId.make("summary-project-deleted"),
+ aggregateKind: "project",
+ aggregateId: projectId,
+ occurredAt: DateTime.formatIso(at),
+ commandId: null,
+ causationEventId: null,
+ correlationId: null,
+ metadata: {},
+ type: "project.deleted",
+ payload: { projectId, deletedAt: DateTime.formatIso(at) },
+ });
+ assert.deepEqual(yield* service.forHandoff(input), []);
+ }).pipe(Effect.provide(makeLayer(model.generate, optedIn)));
+});
+
+it.effect(
+ "searches statements literally and expands source text within a resumable character budget",
+ () => {
+ const model = citing();
+ return Effect.gen(function* () {
+ yield* seed;
+ yield* summarize;
+ const summary = yield* onlySummary;
+ const service = yield* ConversationSummaries.ConversationSummaries;
+ assert.equal((yield* service.list({ ...scope, query: "EXACT TEXT" })).summaries.length, 1);
+ assert.equal((yield* service.list({ ...scope, query: "%" })).summaries.length, 0);
+ assert.equal((yield* service.list({ ...scope, query: "unrelated" })).summaries.length, 0);
+ const emojiOffset = rawText(0).indexOf("🧪");
+ const emoji = yield* service.expand({
+ ...scope,
+ summaryId: summary.id,
+ textOffset: emojiOffset,
+ maxChars: 2,
+ });
+ assert.equal(
+ emoji.sources[0]?.result.state === "available" ? emoji.sources[0].result.text : null,
+ "🧪",
+ );
+ assert.equal(emoji.nextOffset, 0);
+ assert.equal(emoji.nextTextOffset, emojiOffset + 2);
+ const invalid = yield* service
+ .expand({ ...scope, summaryId: summary.id, textOffset: emojiOffset + 1 })
+ .pipe(Effect.flip);
+ assert.equal(invalid.reason, "invalid_cursor");
+ let offset: number | null = 0;
+ let textOffset = 0;
+ const restored = new Map();
+ let pages = 0;
+ while (offset !== null && pages < 100) {
+ const page = yield* service.expand({
+ ...scope,
+ summaryId: summary.id,
+ offset,
+ textOffset,
+ maxChars: 257,
+ limit: 3,
+ });
+ const text = page.sources.flatMap((source) =>
+ source.result.state === "available" ? [source.result.text] : [],
+ );
+ assert.isAtMost(text.join("").length, 257);
+ assert.isTrue(text.every((part) => part.isWellFormed()));
+ for (const source of page.sources)
+ if (source.result.state === "available") {
+ const key = conversationSummarySourceKey(source.ref);
+ restored.set(key, (restored.get(key) ?? "") + source.result.text);
+ }
+ offset = page.nextOffset;
+ textOffset = page.nextTextOffset ?? 0;
+ pages += 1;
+ }
+ assert.isNull(offset);
+ assert.isAbove(pages, 10);
+ for (let index = 0; index < 10; index++)
+ assert.equal(restored.get(messageId(index)), rawText(index));
+ }).pipe(Effect.provide(makeLayer(model.generate, optedIn)));
+ },
+);
+
+it.effect("records and retries a source change during model generation", () => {
+ let changeSource: Effect.Effect = Effect.void;
+ const model = scripted(() =>
+ changeSource.pipe(
+ Effect.as({
+ statements: [{ text: "A statement about the prior source.", citations: ["M1"] }],
+ }),
+ ),
+ );
+ return Effect.gen(function* () {
+ yield* seed;
+ const projections = yield* ProjectionStore.ProjectionStoreV2;
+ changeSource = projections
+ .apply(messageEvent(0, "Changed during the model call"))
+ .pipe(Effect.orDie);
+ assert.instanceOf(yield* summarize.pipe(Effect.flip), ConversationSummaryPassError);
+ const result = yield* list;
+ assert.deepEqual(result.summaries, []);
+ assert.deepInclude(result.lastPass, { result: "failed", generated: 0, waitingMessages: 10 });
+ }).pipe(Effect.provide(makeLayer(model.generate, optedIn)));
+});
+
it.effect("continues coverage and handoff lookup beyond the first 500 summaries", () => {
const model = citing();
return Effect.gen(function* () {
@@ -429,6 +863,28 @@ it.effect("continues coverage and handoff lookup beyond the first 500 summaries"
assert.deepInclude((yield* list).lastPass, { generated: 0, waitingMessages: 0 });
const service = yield* ConversationSummaries.ConversationSummaries;
+ const sql = yield* SqlClient.SqlClient;
+ const lastId = `summary:${(summaryCount - 1).toString(16).padStart(64, "0")}`;
+ yield* sql`UPDATE conversation_summaries SET payload_json = json_set(payload_json, '$.statements[0].text', 'Summary 500 RÉSUMÉ') WHERE summary_id = ${lastId}`;
+ const unicode = yield* service.list({ ...scope, query: "résumé" });
+ assert.equal(unicode.summaries.length, 1);
+ assert.equal(
+ unicode.summaries[0]?.state === "available" ? unicode.summaries[0].summary.id : null,
+ lastId,
+ );
+ const firstMatch = yield* service.list({ ...scope, query: "SUMMARY 50", limit: 1 });
+ assert.isNotNull(firstMatch.nextCursor);
+ const secondMatch = yield* service.list({
+ ...scope,
+ query: "SUMMARY 50",
+ limit: 1,
+ cursor: firstMatch.nextCursor!,
+ });
+ assert.equal(
+ secondMatch.summaries[0]?.state === "available" ? secondMatch.summaries[0].summary.id : null,
+ lastId,
+ );
+ assert.isNull(secondMatch.nextCursor);
const handoff = yield* prepareHandoff(messageCount);
const offered = yield* service.forHandoff({
projectId,
@@ -816,7 +1272,7 @@ it.effect.each([0, 2])(
const delivery = yield* deliver(durable);
yield* delivery.delivered;
assert.include(delivery.context, `[Conversation summary ${summary.id}`);
- assert.include(delivery.context, "It is a summary, not the original messages.");
+ assert.include(delivery.context, "It is a summary, not the original evidence.");
assert.include(delivery.context, "t3_conversation_summary_expand");
assert.include(delivery.context, fact.content!);
assert.equal(delivery.context.split(";
+type SummarySources = Pick;
+interface SummaryCursor {
+ readonly first_source_at: string;
+ readonly first_source_id: string;
+ readonly first_source_offset: number;
+ readonly summary_id: string;
+}
const byConversationOrder = (left: FinalizedMessage, right: FinalizedMessage) =>
DateTime.toEpochMillis(left.createdAt) - DateTime.toEpochMillis(right.createdAt) ||
left.id.localeCompare(right.id);
-/**
- * Consecutive open stretches of settled messages, cut into chunks a single model call can
- * summarize. A chunk that is cut short ends on a reply so an exchange stays together.
- */
-function planLeafChunks(input: {
- readonly settled: ReadonlyArray;
- readonly covered: ReadonlySet;
-}): ReadonlyArray> {
- const chunks: Array> = [];
- let index = 0;
- while (index < input.settled.length) {
- if (input.covered.has(input.settled[index]!.id)) {
- index += 1;
- continue;
- }
- let end = index;
- while (end < input.settled.length && !input.covered.has(input.settled[end]!.id)) end += 1;
- // A stretch with covered messages after it is closed: it can never grow, so any size counts.
- const closed = end < input.settled.length;
- let start = index;
- while (start < end) {
- const chunk: Array = [];
- let chars = 0;
- for (let cursor = start; cursor < end; cursor++) {
- const message = input.settled[cursor]!;
- const cost = Math.min(message.text.length, MAX_MESSAGE_CHARS);
- if (
- chunk.length >= CONVERSATION_SUMMARY_MAX_SOURCES ||
- (chunk.length > 0 && chars + cost > MAX_PROMPT_CHARS)
- )
- break;
- chunk.push(message);
- chars += cost;
- }
- const cut = start + chunk.length < end;
- const lastReply = chunk.findLastIndex((message) => message.role === "assistant");
- const kept = cut && lastReply > 0 ? chunk.slice(0, lastReply + 1) : chunk;
- if (kept.length < MIN_LEAF_MESSAGES && !cut && !closed) break;
- chunks.push(kept);
- start += kept.length;
- }
- index = end;
- }
- return chunks;
-}
-
const make = Effect.gen(function* () {
const sql = yield* SqlClient.SqlClient;
const crypto = yield* Crypto.Crypto;
@@ -150,6 +128,20 @@ const make = Effect.gen(function* () {
const textGeneration = yield* TextGeneration.TextGeneration;
const digest = (text: string) =>
crypto.digest("SHA-256", new TextEncoder().encode(text)).pipe(Effect.map(Hex.encode));
+ // Several leaves of a long source share a timestamp. Its identity and range offset keep
+ // those leaves in source order, independently of their content-addressed summary ids.
+ const firstSourceId = sql`CASE json_extract(payload_json, '$.sources[0].kind')
+ WHEN 'message' THEN 'message:' || json_extract(payload_json, '$.sources[0].messageId')
+ ELSE 'turn_item:' || json_extract(payload_json, '$.sources[0].itemId') END`;
+ const firstSourceOffset = sql`coalesce(json_extract(payload_json, '$.sources[0].range.start'), 0)`;
+ const summaryOrder = sql`first_source_at, ${firstSourceId}, ${firstSourceOffset}, summary_id`;
+ const afterSummary = (after: SummaryCursor | undefined) =>
+ after === undefined
+ ? sql``
+ : sql`
+ AND (first_source_at, ${firstSourceId}, ${firstSourceOffset}, summary_id) >
+ (${after.first_source_at}, ${after.first_source_id}, ${after.first_source_offset}, ${after.summary_id})
+ `;
const storageError = (projectId: ProjectId) => (cause: unknown) =>
isConversationSummaryError(cause)
@@ -171,43 +163,50 @@ const make = Effect.gen(function* () {
threadId: ThreadId,
// Return false when the caller has enough results; true continues through every page.
visit: (page: ReadonlyArray) => Effect.Effect,
- messageIds?: ReadonlyArray,
+ sourceIds?: ReadonlyArray,
) {
- const selectedMessages = messageIds === undefined ? undefined : encodeIds(messageIds);
- let after: { readonly first_source_at: string; readonly summary_id: string } | undefined;
+ const selectedSources = sourceIds === undefined ? undefined : encodeIds(sourceIds);
+ let after: SummaryCursor | undefined;
while (true) {
const rows = yield* sql<{
readonly first_source_at: string;
readonly summary_id: string;
+ readonly first_source_id: string;
+ readonly first_source_offset: number;
readonly sources_json: string;
+ readonly scrubbed: number;
}>`
SELECT first_source_at, summary_id,
- json_extract(payload_json, '$.sources') AS sources_json
+ ${firstSourceId} AS first_source_id, ${firstSourceOffset} AS first_source_offset,
+ json_extract(payload_json, '$.sources') AS sources_json,
+ (coalesce(json_extract(payload_json, '$.scrubbed'), 0) = 1
+ OR json_extract(payload_json, '$.statements[0].text') = ${SCRUBBED_TEXT}) AS scrubbed
FROM conversation_summaries
WHERE project_id = ${projectId} AND thread_id = ${threadId}
+ ${afterSummary(after)}
${
- after === undefined
- ? sql``
- : sql`AND (first_source_at > ${after.first_source_at}
- OR (first_source_at = ${after.first_source_at} AND summary_id > ${after.summary_id}))`
- }
- ${
- selectedMessages === undefined
+ selectedSources === undefined
? sql``
: sql`AND NOT EXISTS (
SELECT 1 FROM json_each(payload_json, '$.sources') AS source
- WHERE json_extract(source.value, '$.messageId') NOT IN (
- SELECT value FROM json_each(${selectedMessages})
+ WHERE CASE json_extract(source.value, '$.kind')
+ WHEN 'message' THEN 'message:' || json_extract(source.value, '$.messageId')
+ ELSE 'turn_item:' || json_extract(source.value, '$.itemId') END NOT IN (
+ SELECT value FROM json_each(${selectedSources})
)
)`
}
- ORDER BY first_source_at, summary_id
+ ORDER BY ${summaryOrder}
LIMIT ${THREAD_SUMMARY_PAGE_SIZE}
`;
if (rows.length === 0) return;
const page = yield* Effect.forEach(rows, (row) =>
decodeSources(row.sources_json).pipe(
- Effect.map((sources) => ({ id: row.summary_id, sources })),
+ Effect.map((sources) => ({
+ id: row.summary_id,
+ sources,
+ scrubbed: row.scrubbed === 1,
+ })),
),
);
if (!(yield* visit(page)) || rows.length < THREAD_SUMMARY_PAGE_SIZE) return;
@@ -234,33 +233,200 @@ const make = Effect.gen(function* () {
);
});
- // The same canonical check citations use: thread scope, deletion, archive, streaming, digest.
+ const invisibleMessageIds = Effect.fn("ConversationSummaries.invisibleMessageIds")(function* (
+ threadId: ThreadId,
+ messageIds?: ReadonlyArray,
+ ) {
+ // Read identity/visibility metadata only. The canonical message reader still owns body
+ // retrieval, and the timeline's shared rule decides whether queued input was ever sent.
+ const rows = yield* sql<{
+ readonly message_id: MessageId;
+ readonly run_id: RunId | null;
+ readonly status: OrchestrationV2RunStatus | null;
+ readonly input_intent: OrchestrationV2UserMessageInputIntent | null;
+ }>`
+ SELECT json_extract(item.payload_json, '$.messageId') AS message_id, item.run_id, run.status,
+ json_extract(item.payload_json, '$.inputIntent') AS input_intent
+ FROM orchestration_v2_projection_turn_items AS item
+ LEFT JOIN orchestration_v2_projection_runs AS run ON run.run_id = item.run_id
+ WHERE item.thread_id = ${threadId} AND item.type = 'user_message'
+ ${messageIds === undefined ? sql`` : sql`AND json_extract(item.payload_json, '$.messageId') IN (SELECT value FROM json_each(${encodeIds(messageIds)}))`}
+ `;
+ const visible = createOrchestrationV2TurnItemVisibility({
+ runs: rows.flatMap((row) =>
+ row.run_id === null || row.status === null ? [] : [{ id: row.run_id, status: row.status }],
+ ),
+ // These are message identities; the superseded-interrupt rule does not apply.
+ attempts: [],
+ items: [],
+ });
+ return new Set(
+ rows
+ .filter(
+ (row) =>
+ !visible({
+ type: "user_message",
+ runId: row.run_id,
+ nodeId: null,
+ ...(row.input_intent === null ? {} : { inputIntent: row.input_intent }),
+ }),
+ )
+ .map((row) => row.message_id),
+ );
+ });
+
+ const readSources = Effect.fn("ConversationSummaries.readSources")(function* (input: {
+ readonly projectId: ProjectId;
+ readonly threadId: ThreadId;
+ readonly sources: ReadonlyArray;
+ }) {
+ // Read/hash each canonical body once, even if a leaf cites several ranges of it.
+ const unique = [
+ ...new Map(
+ input.sources.map((source) => [conversationSummarySourceIdentity(source), source]),
+ ).values(),
+ ];
+ const messages = unique.filter(
+ (source): source is Extract =>
+ source.kind === "message",
+ );
+ const messageResults = yield* knowledge.readSources({ ...input, sources: messages });
+ const invisible =
+ messages.length === 0
+ ? new Set()
+ : yield* invisibleMessageIds(
+ input.threadId,
+ messages.map((source) => source.messageId),
+ );
+ const current = new Map();
+ messages.forEach((source, index) =>
+ current.set(
+ conversationSummarySourceIdentity(source),
+ invisible.has(source.messageId)
+ ? { state: "unavailable", reason: "missing" }
+ : messageResults[index]!,
+ ),
+ );
+ const tools = unique.filter(
+ (source): source is Extract =>
+ source.kind === "turn_item",
+ );
+ if (tools.length > 0) {
+ const records = yield* projections
+ .getThreadRecords(input.threadId, ["turnItems"], {
+ turnItemIds: tools.map((source) => source.itemId),
+ })
+ .pipe(Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }));
+ const rolledBack = yield* sql<{ readonly item_id: string }>`
+ SELECT item.turn_item_id AS item_id FROM orchestration_v2_projection_turn_items AS item
+ JOIN orchestration_v2_projection_runs AS run ON run.run_id = item.run_id
+ WHERE item.thread_id = ${input.threadId} AND run.status = 'rolled_back'
+ AND item.turn_item_id IN (SELECT value FROM json_each(${encodeIds(tools.map((source) => source.itemId))}))
+ `;
+ const excluded = new Set(rolledBack.map((row) => row.item_id));
+ const items = new Map(records?.turnItems.map((item) => [item.id, item]));
+ for (const source of tools) {
+ const item = items.get(source.itemId);
+ const text = item === undefined ? null : summaryToolText(item);
+ const reason =
+ records === null
+ ? "missing"
+ : records.thread.projectId !== input.projectId
+ ? "wrong_project"
+ : records.thread.deletedAt !== null
+ ? "deleted"
+ : records.thread.archivedAt !== null
+ ? "archived"
+ : item === undefined || excluded.has(item.id)
+ ? "missing"
+ : text === null
+ ? "streaming"
+ : source.digest !== (yield* digest(text))
+ ? "changed"
+ : null;
+ current.set(
+ conversationSummarySourceIdentity(source),
+ reason === null && item !== undefined && text !== null
+ ? {
+ state: "available",
+ source,
+ text,
+ role: "tool",
+ title: records!.thread.title,
+ createdAt: DateTime.formatIso(item.completedAt ?? item.updatedAt),
+ archived: false,
+ }
+ : { state: "unavailable", reason: reason ?? "missing" },
+ );
+ }
+ }
+ return input.sources.map((source): ConversationSummarySourceResult => {
+ const result = current.get(conversationSummarySourceIdentity(source));
+ if (source.threadId !== input.threadId) return { state: "unavailable", reason: "missing" };
+ if (result === undefined) return { state: "unavailable", reason: "missing" };
+ if (result.state !== "available") return result;
+ if (result.source.digest !== source.digest)
+ return { state: "unavailable", reason: "changed" };
+ const range = source.range;
+ if (
+ range !== undefined &&
+ (range.total !== result.text.length ||
+ range.start >= range.end ||
+ range.end > range.total ||
+ !isTextBoundary(result.text, range.start) ||
+ !isTextBoundary(result.text, range.end))
+ )
+ return { state: "unavailable", reason: "changed" };
+ return {
+ ...result,
+ source,
+ text: range === undefined ? result.text : result.text.slice(range.start, range.end),
+ };
+ });
+ });
+
+ // Validate canonical scope, deletion, archive, final status, full digest and exact range.
const read = Effect.fn("ConversationSummaries.read")(function* (summary: ConversationSummary) {
- const current = yield* knowledge.readSources({
+ const current = yield* readSources({
projectId: summary.projectId,
threadId: summary.threadId,
sources: summary.sources,
});
const unavailable = current.find((source) => source.state === "unavailable");
- if (unavailable?.state === "unavailable" && unavailable.reason !== "archived") {
+ const scrubbed = summary.scrubbed === true || summary.statements[0]?.text === SCRUBBED_TEXT;
+ // Old leaves claimed a complete source while their model only saw the first 6,000 units.
+ const legacyTruncated = current.some(
+ (result, index) =>
+ result.state === "available" &&
+ summary.sources[index]!.range === undefined &&
+ result.text.length > MAX_SOURCE_CHARS,
+ );
+ const reason =
+ unavailable?.state === "unavailable"
+ ? unavailable.reason
+ : scrubbed || legacyTruncated
+ ? "changed"
+ : null;
+ if (reason !== null && reason !== "archived" && !summary.scrubbed) {
// Handoff references keep the summary skeleton, not text derived from a removed source.
yield* sql`UPDATE conversation_summaries SET payload_json = ${encodeSummary({
...summary,
+ scrubbed: true,
statements: [
{
- text: "[Source unavailable: removed or changed.]",
- citations: [summary.sources[0]!.messageId],
+ text: SCRUBBED_TEXT,
+ citations: [conversationSummarySourceKey(summary.sources[0]!)],
},
],
})} WHERE summary_id = ${summary.id}`;
}
- return unavailable?.state === "unavailable"
+ return reason !== null
? ({
state: "stale",
id: summary.id,
sourceCount: summary.sources.length,
createdAt: summary.createdAt,
- reason: unavailable.reason,
+ reason,
} satisfies ConversationSummaryRead)
: ({ state: "available", summary } satisfies ConversationSummaryRead);
});
@@ -275,10 +441,12 @@ const make = Effect.gen(function* () {
`;
const lastPass =
passRows[0] === undefined ? null : yield* decodePass(passRows[0].payload_json);
- let after: { readonly firstSourceAt: string; readonly id: string } | null = null;
+ let after: SummaryCursor | undefined;
if (input.cursor !== undefined) {
- const cursorRows = yield* sql<{ readonly first_source_at: string }>`
- SELECT first_source_at FROM conversation_summaries
+ const cursorRows = yield* sql`
+ SELECT first_source_at, summary_id,
+ ${firstSourceId} AS first_source_id, ${firstSourceOffset} AS first_source_offset
+ FROM conversation_summaries
WHERE project_id = ${input.projectId} AND thread_id = ${input.threadId}
AND summary_id = ${input.cursor}
`;
@@ -287,30 +455,49 @@ const make = Effect.gen(function* () {
reason: "invalid_cursor",
projectId: input.projectId,
});
- after = { firstSourceAt: cursorRows[0].first_source_at, id: input.cursor };
+ after = cursorRows[0];
}
const limit = input.limit ?? 20;
- const rows = yield* sql<{ readonly payload_json: string }>`
- SELECT payload_json FROM conversation_summaries
- WHERE project_id = ${input.projectId} AND thread_id = ${input.threadId}
- ${
- after === null
- ? sql``
- : sql`AND (first_source_at > ${after.firstSourceAt}
- OR (first_source_at = ${after.firstSourceAt} AND summary_id > ${after.id}))`
+ const query = input.query?.trim().toLowerCase() ?? "";
+ const matching: Array = [];
+ const scanLimit = query === "" ? limit + 1 : THREAD_SUMMARY_PAGE_SIZE;
+ // Fold Unicode in application code; SQLite lower() only folds ASCII. Each read is
+ // bounded, and the cursor remains the last returned summary, even across sparse matches.
+ while (matching.length <= limit) {
+ const rows = yield* sql<{
+ readonly payload_json: string;
+ readonly summary_id: string;
+ readonly first_source_at: string;
+ readonly first_source_id: string;
+ readonly first_source_offset: number;
+ }>`
+ SELECT payload_json, summary_id, first_source_at,
+ ${firstSourceId} AS first_source_id, ${firstSourceOffset} AS first_source_offset
+ FROM conversation_summaries
+ WHERE project_id = ${input.projectId} AND thread_id = ${input.threadId}
+ ${afterSummary(after)}
+ ORDER BY ${summaryOrder} LIMIT ${scanLimit}
+ `;
+ for (const row of rows) {
+ const summary = yield* decodeSummary(row.payload_json);
+ if (
+ query === "" ||
+ summary.statements.some((statement) => statement.text.toLowerCase().includes(query))
+ )
+ matching.push(summary);
+ if (matching.length > limit) break;
}
- ORDER BY first_source_at, summary_id
- LIMIT ${limit + 1}
- `;
- const page = yield* Effect.forEach(rows.slice(0, limit), (row) =>
- decodeSummary(row.payload_json),
- );
+ if (rows.length < scanLimit) break;
+ const last = rows.at(-1)!;
+ after = last;
+ }
+ const page = matching.slice(0, limit);
return {
enabled: resolveProjectSettings(settings, input.projectId).settings
.enableConversationSummaries,
lastPass,
summaries: yield* Effect.forEach(page, read),
- nextCursor: rows.length > limit ? (page.at(-1)?.id ?? null) : null,
+ nextCursor: matching.length > limit ? (page.at(-1)?.id ?? null) : null,
};
},
(effect, input) => effect.pipe(Effect.mapError(storageError(input.projectId))),
@@ -333,7 +520,11 @@ const make = Effect.gen(function* () {
yield* requireProject(input.projectId);
const summary = yield* findSummary(input);
const offset = input.offset ?? 0;
- if (offset > summary.sources.length)
+ const textOffset = input.textOffset ?? 0;
+ if (
+ offset > summary.sources.length ||
+ (offset === summary.sources.length && textOffset !== 0)
+ )
return yield* new ConversationSummaryError({
reason: "invalid_cursor",
projectId: input.projectId,
@@ -342,15 +533,62 @@ const make = Effect.gen(function* () {
const page = summary.sources.slice(offset, offset + limit);
// Each source is re-read with its recorded digest, so a changed message comes back as
// unavailable instead of as text the summary was not made from.
- const results = yield* knowledge.readSources({
+ const results = yield* readSources({
projectId: input.projectId,
threadId: input.threadId,
sources: page,
});
+ let remaining = input.maxChars ?? 12_000;
+ // Contracts enforce this for RPC/MCP; service callers get the same hard upper bound.
+ remaining = Math.max(2, Math.min(remaining, CONVERSATION_SUMMARY_MAX_EXPANSION_CHARS));
+ const sources: Array = [];
+ let nextOffset: number | null = null;
+ let nextTextOffset: number | null = null;
+ for (let index = 0; index < page.length; index++) {
+ const ref = page[index]!;
+ const result = results[index]!;
+ const start = index === 0 ? textOffset : 0;
+ if (result.state === "available") {
+ if (start > result.text.length || !isTextBoundary(result.text, start))
+ return yield* new ConversationSummaryError({
+ reason: "invalid_cursor",
+ projectId: input.projectId,
+ });
+ const end = boundedTextEnd(result.text, start, remaining);
+ if (end === start && start < result.text.length) {
+ nextOffset = offset + index;
+ nextTextOffset = start;
+ break;
+ }
+ sources.push({
+ ref,
+ result: {
+ ...result,
+ text: result.text.slice(start, end),
+ textOffset: start,
+ totalChars: result.text.length,
+ },
+ });
+ remaining -= end - start;
+ if (end < result.text.length) {
+ nextOffset = offset + index;
+ nextTextOffset = end;
+ break;
+ }
+ } else sources.push({ ref, result });
+ if (offset + index + 1 < summary.sources.length) {
+ nextOffset = offset + index + 1;
+ nextTextOffset = 0;
+ } else {
+ nextOffset = null;
+ nextTextOffset = null;
+ }
+ }
return {
summary: yield* read(summary),
- sources: page.map((ref, index) => ({ ref, result: results[index]! })),
- nextOffset: offset + page.length < summary.sources.length ? offset + page.length : null,
+ sources,
+ nextOffset,
+ nextTextOffset,
};
},
(effect, input) => effect.pipe(Effect.mapError(storageError(input.projectId))),
@@ -397,6 +635,7 @@ const make = Effect.gen(function* () {
threadId: ThreadId,
digests: ReadonlyMap,
covered: ReadonlySet,
+ validSourceKeys: ReadonlySet,
) {
// Reuse the pass's validated coverage. Re-reading every source here would hash the whole
// history again on every completed turn, even when no summary needs replacing.
@@ -408,8 +647,15 @@ const make = Effect.gen(function* () {
Effect.fnUntraced(function* (summaries) {
const replaced = summaries.filter(
(summary) =>
- summary.sources.every((source) => covered.has(source.messageId)) &&
- summary.sources.some((source) => digests.get(source.messageId) !== source.digest),
+ summary.sources.every((source) =>
+ covered.has(conversationSummarySourceIdentity(source)),
+ ) &&
+ (summary.scrubbed ||
+ summary.sources.some(
+ (source) =>
+ digests.get(conversationSummarySourceIdentity(source)) !== source.digest ||
+ !validSourceKeys.has(conversationSummarySourceKey(source)),
+ )),
);
if (replaced.length > 0) {
const removed = yield* sql<{ readonly summary_id: string }>`
@@ -438,21 +684,45 @@ const make = Effect.gen(function* () {
function* ({ threadId, runId }) {
yield* cleanupDeleted();
const records = yield* projections
- .getThreadRecords(threadId, ["messages"], { messageRoles: ["user", "assistant"] })
+ .getThreadRecords(threadId, ["messages", "turnItems"], {
+ messageRoles: ["user", "assistant"],
+ turnItemTypes: summaryToolTypes,
+ turnItemStatuses: finalizedToolStatuses,
+ })
.pipe(Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }));
- if (records === null) return;
+ if (records === null) return { hasMore: false };
const projectId = records.thread.projectId;
const digests = new Map();
const covered = new Set();
+ const validSources = new Map();
let hasStaleSummaries = false;
+ const coveredOrigins = () => {
+ const all = new Map>();
+ for (const [key, source] of validSources) {
+ const identity = conversationSummarySourceIdentity(source.ref);
+ const keys = all.get(identity) ?? [];
+ keys.push(key);
+ all.set(identity, keys);
+ }
+ return new Set(
+ [...all].filter(([, keys]) => keys.every((key) => covered.has(key))).map(([id]) => id),
+ );
+ };
const finish = (
result: ConversationSummaryPass["result"],
reason: string | null,
generated: number,
waitingMessages: number,
+ hasMore = false,
) =>
(hasStaleSummaries
- ? cleanupSuperseded(projectId, threadId, digests, covered)
+ ? cleanupSuperseded(
+ projectId,
+ threadId,
+ digests,
+ coveredOrigins(),
+ new Set(validSources.keys()),
+ )
: Effect.void
).pipe(
Effect.andThen(DateTime.now),
@@ -466,6 +736,7 @@ const make = Effect.gen(function* () {
at: DateTime.formatIso(now),
}),
),
+ Effect.as({ hasMore }),
);
if (records.thread.deletedAt !== null)
return yield* finish("skipped", "The thread was deleted.", 0, 0);
@@ -475,64 +746,120 @@ const make = Effect.gen(function* () {
yield* serverSettings.getSettings,
projectId,
).settings;
- // The setting can change between scheduling and running; the latest choice wins.
if (!settings.enableConversationSummaries)
return yield* finish("skipped", "Conversation summaries are off for this project.", 0, 0);
const project = yield* projects.get(projectId);
- if (Option.isNone(project) || project.value.deletedAt !== null) return;
+ if (Option.isNone(project) || project.value.deletedAt !== null) return { hasMore: false };
const cwd = records.thread.worktreePath ?? project.value.workspaceRoot;
- // Match canonical source eligibility without reloading and rehashing each summary's
- // sources. Deleted/archived threads and non-final messages are excluded above/below.
- const rolledBack = yield* sql<{ readonly message_id: string }>`
- SELECT message.message_id
- FROM orchestration_v2_projection_messages AS message
- JOIN orchestration_v2_projection_runs AS run ON run.run_id = message.run_id
- WHERE message.thread_id = ${threadId} AND run.status = 'rolled_back'
+ const rolledBack = yield* sql<{ readonly run_id: string }>`
+ SELECT run_id FROM orchestration_v2_projection_runs
+ WHERE thread_id = ${threadId} AND status = 'rolled_back'
`;
- const excluded = new Set(rolledBack.map((row) => row.message_id));
+ const excluded = new Set(rolledBack.map((row) => row.run_id));
+ const invisible = yield* invisibleMessageIds(threadId);
const finalized = records.messages
.filter(
(message): message is FinalizedMessage =>
(message.role === "user" || message.role === "assistant") &&
- !excluded.has(message.id) &&
+ (message.runId === null || !excluded.has(message.runId)) &&
+ !invisible.has(message.id) &&
!message.streaming &&
message.text.trim() !== "",
)
.toSorted(byConversationOrder);
- for (const message of finalized) digests.set(message.id, yield* digest(message.text));
- // Only summaries whose every source is unchanged count as coverage; a stale range is
- // summarized again from the messages as they are now.
+ const freshIndex = Math.max(0, finalized.length - FRESH_TAIL_MESSAGES);
+ const freshAt = finalized[freshIndex]?.createdAt;
+ const settledMessageIds = new Set(
+ finalized.slice(0, freshIndex).map((message) => message.id),
+ );
+ const all: Array = [];
+ for (const message of finalized) {
+ all.push({
+ ref: {
+ kind: "message",
+ threadId,
+ messageId: message.id,
+ digest: yield* digest(message.text),
+ },
+ role: message.role,
+ text: message.text,
+ createdAt: message.createdAt,
+ });
+ }
+ for (const item of records.turnItems) {
+ if (item.runId !== null && excluded.has(item.runId)) continue;
+ const text = summaryToolText(item);
+ if (text === null) continue;
+ all.push({
+ ref: { kind: "turn_item", threadId, itemId: item.id, digest: yield* digest(text) },
+ role: "tool",
+ text,
+ createdAt: item.completedAt ?? item.updatedAt,
+ });
+ }
+ all.sort(
+ (a, b) =>
+ DateTime.toEpochMillis(a.createdAt) - DateTime.toEpochMillis(b.createdAt) ||
+ conversationSummarySourceIdentity(a.ref).localeCompare(
+ conversationSummarySourceIdentity(b.ref),
+ ),
+ );
+ const settled: Array = [];
+ for (const source of all) {
+ digests.set(conversationSummarySourceIdentity(source.ref), source.ref.digest);
+ const ranges = sourceRanges(source);
+ for (const range of ranges)
+ validSources.set(conversationSummarySourceKey(range.ref), range);
+ if (
+ source.ref.kind === "message"
+ ? settledMessageIds.has(source.ref.messageId)
+ : freshAt === undefined ||
+ DateTime.toEpochMillis(source.createdAt) < DateTime.toEpochMillis(freshAt)
+ ) {
+ settled.push(...ranges);
+ }
+ }
yield* scanThreadSummarySources(projectId, threadId, (summaries) =>
Effect.sync(() => {
for (const summary of summaries) {
if (
- summary.sources.every((source) => digests.get(source.messageId) === source.digest)
+ !summary.scrubbed &&
+ summary.sources.every(
+ (source) =>
+ source.threadId === threadId &&
+ digests.get(conversationSummarySourceIdentity(source)) === source.digest &&
+ validSources.has(conversationSummarySourceKey(source)),
+ )
) {
- for (const source of summary.sources) covered.add(source.messageId);
- } else {
- hasStaleSummaries = true;
- }
+ for (const source of summary.sources)
+ covered.add(conversationSummarySourceKey(source));
+ } else hasStaleSummaries = true;
}
return true;
}),
);
- const settled = finalized.slice(0, Math.max(0, finalized.length - FRESH_TAIL_MESSAGES));
- const waiting = () => settled.filter((message) => !covered.has(message.id)).length;
-
+ const waiting = () =>
+ new Set(
+ settled
+ .filter((source) => !covered.has(conversationSummarySourceKey(source.ref)))
+ .map((source) => conversationSummarySourceIdentity(source.ref)),
+ ).size;
let generated = 0;
+ const failPass = (reason: string) =>
+ finish("failed", reason.slice(0, 300), generated, waiting()).pipe(
+ Effect.andThen(Effect.fail(new ConversationSummaryPassError({ runId, cause: reason }))),
+ );
for (const chunk of planLeafChunks({ settled, covered }).slice(0, MAX_LEAVES_PER_PASS)) {
- const refs = new Map();
- const messages = chunk.map((message, index) => {
+ const refs = new Map();
+ const messages = chunk.map((source, index) => {
const ref = `M${index + 1}`;
- refs.set(ref, message.id);
+ refs.set(ref, conversationSummarySourceKey(source.ref));
return {
ref,
- role: message.role,
- text:
- message.text.length > MAX_MESSAGE_CHARS
- ? `${message.text.slice(0, MAX_MESSAGE_CHARS)}\n[truncated]`
- : message.text,
+ role: source.role,
+ text: source.text,
+ ...(source.ref.range === undefined ? {} : { range: source.ref.range }),
};
});
const reply = yield* textGeneration
@@ -548,9 +875,7 @@ const make = Effect.gen(function* () {
Effect.succeed({ ok: false as const, detail: error.detail }),
}),
);
- if (!reply.ok)
- return yield* finish("failed", reply.detail.slice(0, 300), generated, waiting());
- // The model only names refs; citations become the message ids read above.
+ if (!reply.ok) return yield* failPass(reply.detail);
const statements: Array = [];
for (const statement of reply.statements.slice(0, MAX_STATEMENTS)) {
const text = statement.text.trim().slice(0, MAX_STATEMENT_CHARS).trim();
@@ -560,29 +885,17 @@ const make = Effect.gen(function* () {
if (text !== "" && citations.length > 0) statements.push({ text, citations });
}
if (statements.length === 0)
- return yield* finish(
- "failed",
- "The model returned no statements citing the summarized messages.",
- generated,
- waiting(),
- );
- const sources: ReadonlyArray = chunk.map((message) => ({
- kind: "message",
- threadId,
- messageId: message.id,
- digest: digests.get(message.id)!,
- }));
- // A source edited or removed during generation must not be published as covered.
- const current = yield* knowledge.readSources({ projectId, threadId, sources });
+ return yield* failPass("The model returned no statements citing the summarized sources.");
+ const sources = chunk.map((source) => source.ref);
+ const current = yield* readSources({ projectId, threadId, sources });
if (current.some((source) => source.state !== "available"))
- return yield* finish(
- "skipped",
- "A message changed while it was being summarized.",
- generated,
- waiting(),
- );
+ return yield* failPass("A source changed while it was being summarized.");
const id = `summary:${yield* digest(
- [threadId, ...sources.map((source) => `${source.messageId}:${source.digest}`)].join("\n"),
+ [
+ "v2",
+ threadId,
+ ...sources.map((source) => `${conversationSummarySourceKey(source)}:${source.digest}`),
+ ].join("\n"),
)}`;
const summary: ConversationSummary = {
id,
@@ -597,31 +910,42 @@ const make = Effect.gen(function* () {
INSERT INTO conversation_summaries (
summary_id, project_id, thread_id, first_source_at, created_at, payload_json
) VALUES (
- ${id}, ${projectId}, ${threadId},
- ${DateTime.formatIso(chunk[0]!.createdAt)}, ${summary.createdAt},
- ${encodeSummary(summary)}
+ ${id}, ${projectId}, ${threadId}, ${DateTime.formatIso(chunk[0]!.createdAt)},
+ ${summary.createdAt}, ${encodeSummary(summary)}
)
- ON CONFLICT (summary_id) DO NOTHING
+ ON CONFLICT (summary_id) DO UPDATE SET
+ created_at = excluded.created_at, payload_json = excluded.payload_json
+ WHERE coalesce(json_extract(conversation_summaries.payload_json, '$.scrubbed'), 0) = 1
+ OR json_extract(conversation_summaries.payload_json, '$.statements[0].text') = ${SCRUBBED_TEXT}
`;
- for (const message of chunk) covered.add(message.id);
+ for (const source of sources) covered.add(conversationSummarySourceKey(source));
generated += 1;
}
- yield* finish(
+ return yield* finish(
generated > 0 ? "ran" : "skipped",
generated > 0 ? null : "No settled stretch is long enough to summarize yet.",
generated,
waiting(),
+ planLeafChunks({ settled, covered }).length > 0,
);
},
(effect, { runId }) =>
- effect.pipe(Effect.mapError((cause) => new ConversationSummaryPassError({ runId, cause }))),
+ effect.pipe(
+ Effect.mapError((cause) =>
+ cause instanceof ConversationSummaryPassError
+ ? cause
+ : new ConversationSummaryPassError({ runId, cause }),
+ ),
+ ),
);
const renderForHandoff = (summary: ConversationSummary) => {
- const position = new Map(summary.sources.map((source, index) => [source.messageId, index + 1]));
+ const position = new Map(
+ summary.sources.map((source, index) => [conversationSummarySourceKey(source), index + 1]),
+ );
return [
- `[Conversation summary ${summary.id}: written by a model from ${summary.sources.length} earlier messages of thread ${summary.threadId}. It is a summary, not the original messages.]`,
- `Read the exact messages with t3_conversation_summary_expand({projectId:"${summary.projectId}",threadId:"${summary.threadId}",summaryId:"${summary.id}"}); #n is the nth source it returns.`,
+ `[Conversation summary ${summary.id}: written by a model from ${summary.sources.length} earlier source ranges of thread ${summary.threadId}. It is a summary, not the original evidence.]`,
+ `Read the exact source ranges with t3_conversation_summary_expand({projectId:"${summary.projectId}",threadId:"${summary.threadId}",summaryId:"${summary.id}"}); #n is the nth source. Follow nextOffset and nextTextOffset together to retrieve all text.`,
...summary.statements.map(
(statement) =>
`- ${statement.text} [${statement.citations.map((id) => `#${position.get(id)}`).join(", ")}]`,
@@ -633,6 +957,7 @@ const make = Effect.gen(function* () {
"ConversationSummaries.forHandoff",
)(
function* ({ projectId, handoffs, itemIds }) {
+ yield* requireProject(projectId);
if (
!resolveProjectSettings(yield* serverSettings.getSettings, projectId).settings
.enableConversationSummaries
@@ -647,62 +972,105 @@ const make = Effect.gen(function* () {
const requestedItems = encodeIds([...itemIds]);
for (const threadId of threadIds) {
if (offered.length >= HANDOFF_CANDIDATE_LIMIT) break;
- // Look up only the requested item identities, not every command and message body in
- // the thread. json_each avoids SQLite's bind limit for a long omitted-history range.
+ // Resolve exact requested canonical item identities without loading their bodies.
const items = yield* sql<{
readonly itemId: TurnItemId;
- readonly messageId: MessageId;
+ readonly messageId: MessageId | null;
readonly runId: RunId | null;
- readonly kind: "user_message" | "assistant_message";
+ readonly kind: string;
}>`
SELECT item.turn_item_id AS "itemId", item.run_id AS "runId", item.type AS kind,
json_extract(item.payload_json, '$.messageId') AS "messageId"
FROM json_each(${requestedItems}) AS requested
JOIN orchestration_v2_projection_turn_items AS item ON item.turn_item_id = requested.value
- WHERE item.thread_id = ${threadId} AND item.type IN ('user_message', 'assistant_message')
+ WHERE item.thread_id = ${threadId}
`;
- const itemByMessage = new Map(items.map((item) => [item.messageId, item]));
- if (itemByMessage.size === 0) continue;
+ const itemBySource = new Map(
+ items.map((item) => [
+ (item.kind === "user_message" || item.kind === "assistant_message") &&
+ item.messageId !== null
+ ? `message:${item.messageId}`
+ : `turn_item:${item.itemId}`,
+ item,
+ ]),
+ );
+ const itemByMessage = new Map(
+ items.flatMap((item) =>
+ item.messageId === null ? [] : [[item.messageId, item] as const],
+ ),
+ );
+ if (itemBySource.size === 0) continue;
+ const candidates: Array = [];
yield* scanThreadSummarySources(
projectId,
threadId,
Effect.fnUntraced(function* (summaries) {
for (const metadata of summaries) {
- const sourceItems = metadata.sources.flatMap(
- (source) => itemByMessage.get(source.messageId) ?? [],
- );
- if (sourceItems.length !== metadata.sources.length) continue;
- // Decode statement bodies only for summaries covering the handoff's requested
- // messages; source validation still runs immediately before offering any text.
+ if (
+ metadata.scrubbed ||
+ !metadata.sources.every((source) =>
+ itemBySource.has(conversationSummarySourceIdentity(source)),
+ )
+ )
+ continue;
const summary = yield* findSummary({ projectId, threadId, summaryId: metadata.id });
- // Stale rows must not consume the offer limit and hide later usable summaries.
if ((yield* read(summary)).state !== "available") continue;
- // Summary input excludes attached knowledge; replay its receipts even when the
- // source messages were omitted before this handoff was prepared.
- const evidence = yield* deliveredKnowledgeForHistory({
- projectId,
- messages: sourceItems.map((item) => ({ ...item, threadId })),
- userItemId: (messageId) => itemByMessage.get(messageId)?.itemId,
- }).pipe(Effect.provideService(ProjectKnowledge.ProjectKnowledge, knowledge));
- offered.push({
- id: summary.id,
- text: [
- renderForHandoff(summary),
- ...sourceItems.flatMap((item, index) => {
- const attached = evidence.get(item.itemId);
- return attached === undefined
- ? []
- : [`[Summary source #${index + 1}; item=${item.itemId}]\n${attached}`];
- }),
- ].join("\n\n"),
- coveredItemIds: sourceItems.map((item) => item.itemId),
- });
- if (offered.length >= HANDOFF_CANDIDATE_LIMIT) return false;
+ candidates.push(summary);
+ if (candidates.length >= HANDOFF_CANDIDATE_LIMIT) return false;
}
return true;
}),
- [...itemByMessage.keys()],
+ [...itemBySource.keys()],
);
+ // Leaves sharing a split source form one atomic offer. Omitting the item after only
+ // one of its ranges was summarized would silently discard the remaining evidence.
+ const groups: Array<{ summaries: Array; identities: Set }> =
+ [];
+ for (const summary of candidates) {
+ const group = {
+ summaries: [summary],
+ identities: new Set(summary.sources.map(conversationSummarySourceIdentity)),
+ };
+ for (let index = groups.length - 1; index >= 0; index--) {
+ const previous = groups[index]!;
+ if (![...previous.identities].some((id) => group.identities.has(id))) continue;
+ group.summaries.unshift(...previous.summaries);
+ for (const id of previous.identities) group.identities.add(id);
+ groups.splice(index, 1);
+ }
+ groups.push(group);
+ }
+ for (const group of groups) {
+ if (!sourcesCoverWholeItems(group.summaries.flatMap((summary) => summary.sources)))
+ continue;
+ const sourceItems = [...group.identities].map((id) => itemBySource.get(id)!);
+ const evidence = yield* deliveredKnowledgeForHistory({
+ projectId,
+ messages: sourceItems.flatMap((item) =>
+ item.kind === "user_message" || item.kind === "assistant_message"
+ ? [{ ...item, kind: item.kind, threadId }]
+ : [],
+ ),
+ userItemId: (messageId) => itemByMessage.get(messageId)?.itemId,
+ }).pipe(Effect.provideService(ProjectKnowledge.ProjectKnowledge, knowledge));
+ offered.push({
+ id: group.summaries[0]!.id,
+ ...(group.summaries.length > 1
+ ? { summaryIds: group.summaries.map((summary) => summary.id) }
+ : {}),
+ text: [
+ ...group.summaries.map(renderForHandoff),
+ ...sourceItems.flatMap((item) => {
+ const attached = evidence.get(item.itemId);
+ return attached === undefined
+ ? []
+ : [`[Summary source item=${item.itemId}]\n${attached}`];
+ }),
+ ].join("\n\n"),
+ coveredItemIds: sourceItems.map((item) => item.itemId),
+ });
+ if (offered.length >= HANDOFF_CANDIDATE_LIMIT) break;
+ }
}
return offered;
},
diff --git a/apps/server/src/project/conversationSummaries/conversationSummarySchedule.ts b/apps/server/src/project/conversationSummaries/conversationSummarySchedule.ts
index aedb431e39d52..bc6d31c5fbd44 100644
--- a/apps/server/src/project/conversationSummaries/conversationSummarySchedule.ts
+++ b/apps/server/src/project/conversationSummaries/conversationSummarySchedule.ts
@@ -27,17 +27,29 @@ export function conversationSummaryEffects(input: {
}): ReadonlyArray {
if (!resolveProjectSettings(input.settings, input.projectId).settings.enableConversationSummaries)
return [];
- return [
- {
- id: `effect:conversation-summary.generate:${input.runId}`,
- commandId: CommandId.make(`command:effect:conversation-summary.generate:${input.runId}`),
- threadId: input.threadId,
- request: { type: "conversation-summary.generate", runId: input.runId },
+ return [conversationSummaryEffect(input)];
+}
+
+export function conversationSummaryEffect(input: {
+ readonly threadId: ThreadId;
+ readonly runId: RunId;
+ readonly pass?: number;
+}): PendingOrchestrationEffectV2 {
+ const pass = input.pass ?? 0;
+ const id = `effect:conversation-summary.generate:${input.runId}${pass === 0 ? "" : `:pass:${pass}`}`;
+ return {
+ id,
+ commandId: CommandId.make(`command:${id}`),
+ threadId: input.threadId,
+ request: {
+ type: "conversation-summary.generate",
+ runId: input.runId,
+ ...(pass === 0 ? {} : { pass }),
},
- ];
+ };
}
-/** Storage or projection failures; the effect worker retries the pass on these. */
+/** A failed pass retains its durable work item for the effect worker's bounded retries. */
export class ConversationSummaryPassError extends Schema.TaggedError()(
"ConversationSummaryPassError",
{ runId: Schema.String, cause: Schema.Defect() },
@@ -46,6 +58,7 @@ export class ConversationSummaryPassError extends Schema.TaggedError;
/** Labelled as a generated summary, with the ids needed to expand its sources. */
readonly text: string;
readonly coveredItemIds: ReadonlyArray;
@@ -55,7 +68,7 @@ export interface ConversationSummaryRuntimeShape {
readonly summarizeThread: (input: {
readonly threadId: ThreadId;
readonly runId: RunId;
- }) => Effect.Effect;
+ }) => Effect.Effect<{ readonly hasMore: boolean }, ConversationSummaryPassError>;
/**
* Current summaries, oldest first, whose sources are all among `itemIds` of the pending
* handoffs. Never fails: a summary that cannot be verified is simply not offered.
@@ -75,7 +88,7 @@ export class ConversationSummaryRuntime extends Context.Reference ({
- summarizeThread: () => Effect.void,
+ summarizeThread: () => Effect.succeed({ hasMore: false }),
forHandoff: () => Effect.succeed([]),
}),
},
diff --git a/apps/server/src/project/conversationSummaries/conversationSummarySources.ts b/apps/server/src/project/conversationSummaries/conversationSummarySources.ts
new file mode 100644
index 0000000000000..5536d5a0d3369
--- /dev/null
+++ b/apps/server/src/project/conversationSummaries/conversationSummarySources.ts
@@ -0,0 +1,135 @@
+import {
+ CONVERSATION_SUMMARY_MAX_SOURCES,
+ conversationSummarySourceIdentity,
+ conversationSummarySourceKey,
+ type ConversationSummarySourceRef,
+} from "@t3tools/contracts";
+import * as DateTime from "effect/DateTime";
+
+export const MAX_SOURCE_CHARS = 6_000;
+const MAX_PROMPT_CHARS = 32_000;
+const MIN_LEAF_SOURCES = 8;
+
+export interface SummarySource {
+ readonly ref: ConversationSummarySourceRef;
+ readonly role: "user" | "assistant" | "tool";
+ readonly text: string;
+ readonly createdAt: DateTime.Utc;
+}
+
+export const isTextBoundary = (text: string, offset: number): boolean =>
+ offset === 0 ||
+ offset === text.length ||
+ !(
+ text.charCodeAt(offset - 1) >= 0xd800 &&
+ text.charCodeAt(offset - 1) <= 0xdbff &&
+ text.charCodeAt(offset) >= 0xdc00 &&
+ text.charCodeAt(offset) <= 0xdfff
+ );
+
+/** Never split a surrogate pair. A caller with fewer than two units left may make no progress. */
+export const boundedTextEnd = (text: string, start: number, maxChars: number): number => {
+ const end = Math.min(text.length, start + maxChars);
+ return isTextBoundary(text, end) ? end : end - 1;
+};
+
+/** All of a source is represented, including long command output and message suffixes. */
+export function sourceRanges(source: SummarySource): ReadonlyArray {
+ if (source.text.length <= MAX_SOURCE_CHARS) return [source];
+ const ranges: Array = [];
+ for (let start = 0; start < source.text.length;) {
+ const end = boundedTextEnd(source.text, start, MAX_SOURCE_CHARS);
+ ranges.push({
+ ...source,
+ ref: { ...source.ref, range: { start, end, total: source.text.length } },
+ text: source.text.slice(start, end),
+ });
+ start = end;
+ }
+ return ranges;
+}
+
+/** Consecutive uncovered ranges, with bounded model input and a small open-tail threshold. */
+export function planLeafChunks(input: {
+ readonly settled: ReadonlyArray;
+ readonly covered: ReadonlySet;
+}): ReadonlyArray> {
+ const chunks: Array> = [];
+ let index = 0;
+ while (index < input.settled.length) {
+ if (input.covered.has(conversationSummarySourceKey(input.settled[index]!.ref))) {
+ index += 1;
+ continue;
+ }
+ let end = index;
+ while (
+ end < input.settled.length &&
+ !input.covered.has(conversationSummarySourceKey(input.settled[end]!.ref))
+ )
+ end += 1;
+ const closed = end < input.settled.length;
+ let start = index;
+ while (start < end) {
+ const chunk: Array = [];
+ let chars = 0;
+ for (let cursor = start; cursor < end; cursor++) {
+ const source = input.settled[cursor]!;
+ // Include a bound for ref/role framing, in addition to the complete source range.
+ const cost = source.text.length + 128;
+ if (
+ chunk.length >= CONVERSATION_SUMMARY_MAX_SOURCES ||
+ (chunk.length > 0 && chars + cost > MAX_PROMPT_CHARS)
+ )
+ break;
+ chunk.push(source);
+ chars += cost;
+ }
+ const cut = start + chunk.length < end;
+ // Prefer a completed reply as a cut boundary, preserving an exchange when it fits.
+ const lastReply = chunk.findLastIndex(
+ (source) =>
+ source.role === "assistant" &&
+ (source.ref.range === undefined || source.ref.range.end === source.ref.range.total),
+ );
+ const kept = cut && lastReply > 0 ? chunk.slice(0, lastReply + 1) : chunk;
+ // A partial long source must always be finishable, even if its last range is short.
+ if (
+ kept.length < MIN_LEAF_SOURCES &&
+ !cut &&
+ !closed &&
+ !kept.some((source) => source.ref.range !== undefined)
+ )
+ break;
+ chunks.push(kept);
+ start += kept.length;
+ }
+ index = end;
+ }
+ return chunks;
+}
+
+/** A handoff can omit a canonical item only when the union contains its complete source. */
+export function sourcesCoverWholeItems(
+ sources: ReadonlyArray,
+): boolean {
+ const groups = new Map>();
+ for (const source of sources) {
+ const id = conversationSummarySourceIdentity(source);
+ const group = groups.get(id) ?? [];
+ group.push(source);
+ groups.set(id, group);
+ }
+ return [...groups.values()].every((group) => {
+ if (group.some((source) => source.digest !== group[0]!.digest)) return false;
+ if (group.some((source) => source.range === undefined)) return true;
+ const ranges = group.map((source) => source.range!).toSorted((a, b) => a.start - b.start);
+ const total = ranges[0]!.total;
+ let coveredUntil = 0;
+ for (const range of ranges) {
+ if (range.total !== total || range.start > coveredUntil || range.end <= range.start)
+ return false;
+ coveredUntil = Math.max(coveredUntil, range.end);
+ }
+ return coveredUntil === total;
+ });
+}
diff --git a/apps/web/src/components/conversationSummaries/ConversationSummariesDialog.browser.test.tsx b/apps/web/src/components/conversationSummaries/ConversationSummariesDialog.browser.test.tsx
index 5605f581547fe..31325568d38b7 100644
--- a/apps/web/src/components/conversationSummaries/ConversationSummariesDialog.browser.test.tsx
+++ b/apps/web/src/components/conversationSummaries/ConversationSummariesDialog.browser.test.tsx
@@ -8,9 +8,12 @@ import {
ProviderInstanceId,
RunId,
ThreadId,
+ TurnItemId,
+ conversationSummarySourceKey,
type ConversationSummary,
type ConversationSummaryExpansion,
type ConversationSummaryList,
+ type ConversationSummaryListInput,
} from "@t3tools/contracts";
import { Atom, AtomRegistry, AsyncResult } from "effect/reactivity";
import { expect, it, vi } from "vite-plus/test";
@@ -21,6 +24,7 @@ const state = vi.hoisted(() => ({
config: null as unknown,
list: null as unknown,
expand: null as unknown,
+ lists: [] as { environmentId: EnvironmentId; input: ConversationSummaryListInput }[],
updateSettings: vi.fn(),
}));
vi.mock(
@@ -38,7 +42,13 @@ vi.mock(
import("../../state/conversationSummaries"),
() =>
({
- conversationSummaries: { list: () => state.list, expand: () => state.expand },
+ conversationSummaries: {
+ list: (request: { environmentId: EnvironmentId; input: ConversationSummaryListInput }) => {
+ state.lists.push(request);
+ return state.list;
+ },
+ expand: () => state.expand,
+ },
}) as never,
);
vi.mock(
@@ -59,15 +69,22 @@ const ref = (id: string) => ({
messageId: MessageId.make(id),
digest: "a".repeat(64),
});
+const firstSource = { ...ref("first"), range: { start: 0, end: 33, total: 66 } };
+const toolSource = {
+ kind: "turn_item" as const,
+ threadId: scope.threadId,
+ itemId: TurnItemId.make("release-tool"),
+ digest: "a".repeat(64),
+};
const summary: ConversationSummary = {
id: `summary:${"b".repeat(64)}`,
projectId: scope.projectId,
threadId: scope.threadId,
- sources: [ref("first"), ref("second")],
+ sources: [firstSource, ref("second"), toolSource],
statements: [
{
text: "The user wants releases signed.",
- citations: [MessageId.make("first"), MessageId.make("second")],
+ citations: [firstSource, ref("second"), toolSource].map(conversationSummarySourceKey),
},
],
modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" },
@@ -76,6 +93,7 @@ const summary: ConversationSummary = {
function setup() {
const registry = AtomRegistry.make();
+ state.lists = [];
state.config = Atom.make({
environment: { capabilities: { conversationSummaries: true, projectSettingsOverrides: true } },
settings: DEFAULT_SERVER_SETTINGS,
@@ -100,17 +118,23 @@ function setup() {
reason: "changed",
},
],
- nextCursor: null,
+ nextCursor: `summary:${"c".repeat(64)}`,
};
state.list = Atom.make(AsyncResult.success(list));
const expansion: ConversationSummaryExpansion = {
- summary: { state: "available", summary },
+ summary: {
+ state: "stale",
+ id: summary.id,
+ sourceCount: summary.sources.length,
+ createdAt: summary.createdAt,
+ reason: "deleted",
+ },
sources: [
{
- ref: ref("first"),
+ ref: firstSource,
result: {
state: "available",
- source: ref("first"),
+ source: firstSource,
text: "Please sign every release bundle.",
title: "Thread",
role: "user",
@@ -119,6 +143,18 @@ function setup() {
},
},
{ ref: ref("second"), result: { state: "unavailable", reason: "deleted" } },
+ {
+ ref: toolSource,
+ result: {
+ state: "available",
+ source: toolSource,
+ text: "Release signature verified.",
+ title: "Verification",
+ role: "tool",
+ createdAt: summary.createdAt,
+ archived: false,
+ },
+ },
],
nextOffset: null,
};
@@ -136,15 +172,20 @@ it("shows current summaries with their sources, hides outdated text, and opts th
);
await expect.element(page.getByText(/The user wants releases signed\./)).toBeVisible();
- await expect.element(page.getByText("[#1, #2]")).toBeVisible();
+ await expect.element(page.getByText("[#1, #2, #3]")).toBeVisible();
await expect
- .element(page.getByText("Last pass failed. No model 3 older messages wait for a summary."))
+ .element(page.getByText("Last pass failed. No model 3 older sources wait for a summary."))
.toBeVisible();
await expect.element(page.getByText(/Outdated: source changed\./)).toBeVisible();
- await userEvent.click(page.getByRole("button", { name: "Show source messages" }).first());
+ await userEvent.click(page.getByRole("button", { name: "Show sources" }).first());
await expect.element(page.getByText("Please sign every release bundle.")).toBeVisible();
await expect.element(page.getByText("Source deleted.")).toBeVisible();
+ await expect.element(page.getByText("#3 · Tool")).toBeVisible();
+ await expect.element(page.getByText("Release signature verified.")).toBeVisible();
+ await expect.element(page.getByText(/The user wants releases signed\./)).not.toBeInTheDocument();
+ await userEvent.click(page.getByRole("button", { name: "Hide sources" }));
+ await expect.element(page.getByText(/The user wants releases signed\./)).not.toBeInTheDocument();
await userEvent.click(page.getByRole("switch", { name: "Summarize older messages" }));
expect(state.updateSettings).toHaveBeenCalledWith({
@@ -155,4 +196,17 @@ it("shows current summaries with their sources, hides outdated text, and opts th
},
},
});
+
+ await userEvent.click(page.getByRole("button", { name: "Later", exact: true }));
+ expect(state.lists.at(-1)?.input.cursor).toBe(`summary:${"c".repeat(64)}`);
+ await userEvent.fill(page.getByRole("textbox", { name: "Search summaries" }), "signed");
+ await userEvent.click(page.getByRole("button", { name: "Search", exact: true }));
+ expect(state.lists.at(-1)?.input).toEqual({
+ projectId: scope.projectId,
+ threadId: scope.threadId,
+ limit: 20,
+ query: "signed",
+ });
+ await userEvent.click(page.getByRole("button", { name: "Clear", exact: true }));
+ expect(state.lists.at(-1)?.input.query).toBeUndefined();
});
diff --git a/apps/web/src/components/conversationSummaries/ConversationSummariesDialog.tsx b/apps/web/src/components/conversationSummaries/ConversationSummariesDialog.tsx
index ede878e2b639a..3903371a9ac42 100644
--- a/apps/web/src/components/conversationSummaries/ConversationSummariesDialog.tsx
+++ b/apps/web/src/components/conversationSummaries/ConversationSummariesDialog.tsx
@@ -1,14 +1,20 @@
import { useAtomValue } from "@effect/atom-react";
import {
AuthSettingsWriteScope,
+ conversationSummarySourceKey,
type ConversationSummary,
+ type ConversationSummaryExpansion,
type ConversationSummaryId,
type ConversationSummaryRead,
type EnvironmentId,
type ProjectId,
type ThreadId,
} from "@t3tools/contracts";
-import { conversationSummaryPassLabel } from "@t3tools/client-runtime/conversationSummaries";
+import {
+ conversationSummaryPassLabel,
+ conversationSummarySourceRangeLabel,
+ type ConversationSummarySourceCursor,
+} from "@t3tools/client-runtime/conversationSummaries";
import { KNOWLEDGE_SOURCE_UNAVAILABLE_LABEL } from "@t3tools/client-runtime/projectKnowledge";
import { squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime";
import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
@@ -29,9 +35,11 @@ import {
DialogTitle,
} from "../ui/dialog";
import { Switch } from "../ui/switch";
+import { Input } from "../ui/input";
const PAGE_SIZE = 20;
const SOURCE_PAGE_SIZE = 6;
+const SOURCE_PAGE_CHARS = 12_000;
interface Scope {
readonly environmentId: EnvironmentId;
@@ -85,7 +93,7 @@ function SummariesToggle(props: Scope) {
After turns complete in this project, your text generation model summarizes settled
- stretches of up to 24 older messages. Each summary costs one model call.
+ stretches of older messages and tool output. Each summary costs one model call.
+ {expansion.error
+ ? "Summary unavailable until its sources can be checked."
+ : "Checking sources…"}
+
+ ) : read.state === "available" ? (
+
) : (
- Outdated: {KNOWLEDGE_SOURCE_UNAVAILABLE_LABEL[props.read.reason].toLowerCase()}. The
- summary text is hidden and agents no longer receive it.
+ Outdated: {KNOWLEDGE_SOURCE_UNAVAILABLE_LABEL[read.reason].toLowerCase()}. The summary
+ text is hidden and agents no longer receive it.
setCursors((current) => current.slice(0, -1))}
+ disabled={props.cursors.length === 0}
+ onClick={props.onEarlier}
>
Earlier
@@ -256,7 +383,7 @@ function SummaryList(props: Scope) {
size="sm"
disabled={nextCursor === null}
onClick={() => {
- if (nextCursor !== null) setCursors((current) => [...current, nextCursor]);
+ if (nextCursor !== null) props.onLater(nextCursor);
}}
>
Later
@@ -276,15 +403,20 @@ export function ConversationSummariesDialog(
Conversation summaries
- Model-written summaries of this thread's older messages. When the thread moves to
- another provider or is forked, agents receive them in place of messages that no longer
- fit, and can open the exact messages behind each one.
+ Model-written summaries of this thread's older messages and tool output. When the thread
+ moves to another provider or is forked, agents receive them in place of messages that no
+ longer fit, and can open the exact sources behind each one.