Open History to see the last commit
1import type {
2 OrchestrationV2ContextHandoff,
3 OrchestrationV2ProviderThread,
4} from "@t3tools/contracts";
5import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
6import { ContextHandoffBudgetError } from "@t3tools/provider-core/server/failure";
7import * as Effect from "effect/Effect";
8import * as Schema from "effect/Schema";
9import {
10 historyCost,
11 renderHistory,
12 selectHistory,
13} from "@t3tools/provider-core/server/handoffBudget";
14
15/** The provider prompt for a turn that carries inline context ahead of the user's text. */
16export const withInlineContext = (context: string, userText: string): string =>
17 context === "" ? userText : `${context}\n\nUser message:\n${userText}`;
18
19/** Persist around injection; ambiguous delivery requires a fresh native thread. Call unsent if refused before prompt consumption. */
20export const deliverContextHandoffs = Effect.fn("orchestrationV2.deliverContextHandoffs")(
21 function* <InjectError = never, PersistError = never, BudgetError = never>(input: {
22 readonly handoffs: ReadonlyArray<OrchestrationV2ContextHandoff>;
23 readonly providerThread: OrchestrationV2ProviderThread;
24 readonly budget: number | Effect.Effect<number, BudgetError>;
25 readonly deferInline?: boolean;
26 readonly alreadyDeliveredItemIds: ReadonlySet<string>;
27 readonly inject?: (
28 history: ProviderAdapter.ProviderAdapterV2HistoricalContext,
29 ) => Effect.Effect<boolean, InjectError>;
30 readonly persist: (handoff: OrchestrationV2ContextHandoff) => Effect.Effect<void, PersistError>;
31 }) {
32 const nativeThreadId = input.providerThread.nativeThreadRef?.nativeId ?? undefined;
33 const pending = input.handoffs.filter(
34 (handoff) =>
35 nativeThreadId === undefined ||
36 handoff.delivery?.nativeThreadId !== nativeThreadId ||
37 handoff.delivery.status === "pending",
38 );
39 if (pending.length === 0 || (input.deferInline && input.inject === undefined))
40 return { context: "", delivered: Effect.void, unsent: Effect.void };
41 const budget = typeof input.budget === "number" ? input.budget : yield* input.budget;
42 let coverage = pending
43 .map(
44 (handoff) =>
45 `Context handoff (${handoff.strategy === "fork_delta_summary" ? "merge_back / fork_delta_summary" : handoff.strategy}):\n${
46 handoff.history?.coverage ??
47 `From thread ${handoff.threadId}, runs ${handoff.coveredRunOrdinals.from}-${handoff.coveredRunOrdinals.to}. Recover history with t3_thread_read, view=activity; paginate with afterPosition, and use itemId/textOffset for long items.`
48 }`,
49 )
50 .join("\n");
51 if (historyCost([], coverage) > Math.min(4_000, budget / 2)) {
52 const strategies = Array.from(new Set(pending.map((handoff) => handoff.strategy)));
53 coverage = `Context handoff (${strategies.join(", ")}). ${pending.length} handoff records; detailed coverage references omitted. Recover an omitted item only when the request needs it, with t3_thread_read({threadId:"${input.providerThread.appThreadId ?? pending[0]!.threadId}",view:"activity",itemId:"<omitted item ID>",maxCharsPerItem:4000}); continue long text with textOffset=nextTextOffset until null. Follow fork/handoff source references in activity. Items included below need no read.`;
54 }
55 const seen = new Set(input.alreadyDeliveredItemIds);
56 const messages = pending
57 .flatMap((handoff) => handoff.history?.messages ?? [])
58 .filter((message) => {
59 if (seen.has(message.itemId)) return false;
60 seen.add(message.itemId);
61 return true;
62 });
63 const oldContext = pending
64 .filter((handoff) => handoff.history === undefined)
65 .map((handoff) => handoff.summaryText)
66 .join("\n\n");
67 const fullCoverage =
68 oldContext && historyCost([], `${coverage}\n${oldContext}`) + 512 <= budget
69 ? `${coverage}\n${oldContext}`
70 : coverage;
71 const selected = selectHistory({
72 messages,
73 coverage: fullCoverage,
74 omittedItems: pending.reduce((sum, handoff) => sum + (handoff.history?.omittedItems ?? 0), 0),
75 omittedItemIds: pending.flatMap((handoff) => handoff.history?.omittedItemIds ?? []),
76 budget,
77 });
78 if (historyCost(selected.messages, selected.context) > budget) {
79 if (input.deferInline) return { context: "", delivered: Effect.void, unsent: Effect.void };
80 return yield* new ContextHandoffBudgetError();
81 }
82 const omittedItemIds = new Set(selected.omittedItemIds);
83 const persist = (status: "pending" | "injected" | "inline") =>
84 Effect.forEach(
85 pending,
86 (handoff) =>
87 input.persist({
88 ...handoff,
89 ...(nativeThreadId === undefined
90 ? {}
91 : {
92 delivery: {
93 nativeThreadId,
94 status,
95 omittedItemIds: [
96 ...(handoff.history?.omittedItemIds ?? []),
97 ...(handoff.history?.messages ?? [])
98 .filter((message) => omittedItemIds.has(message.itemId))
99 .map((message) => message.itemId),
100 ],
101 itemIds: selected.messages
102 .filter((message) =>
103 handoff.history?.messages.some(
104 (candidate) => candidate.itemId === message.itemId,
105 ),
106 )
107 .map((message) => message.itemId),
108 },
109 }),
110 }),
111 { discard: true },
112 );
113 if (input.inject !== undefined && nativeThreadId !== undefined) {
114 if (
115 pending.some(
116 (handoff) =>
117 handoff.delivery?.nativeThreadId === nativeThreadId &&
118 handoff.delivery.status === "pending",
119 )
120 ) {
121 return yield* new ContextHandoffDeliveryUncertainError();
122 }
123 yield* persist("pending");
124 const injected = yield* input.inject({
125 messages: selected.messages,
126 context: selected.context,
127 });
128 if (injected) {
129 yield* persist("injected");
130 return { context: "", delivered: Effect.void, unsent: Effect.void };
131 }
132 } else {
133 yield* persist("pending");
134 }
135 if (input.deferInline) {
136 // Compaction cannot carry inline transcripts; retain for the next turn if native injection is unsupported.
137 yield* Effect.forEach(pending, input.persist, { discard: true });
138 return { context: "", delivered: Effect.void, unsent: Effect.void };
139 }
140 return {
141 context: renderHistory(selected.messages, selected.context),
142 delivered: persist("inline"),
143 unsent: Effect.forEach(pending, input.persist, { discard: true }),
144 };
145 },
146);
147
148export class ContextHandoffDeliveryUncertainError extends Schema.TaggedError<ContextHandoffDeliveryUncertainError>()(
149 "ContextHandoffDeliveryUncertainError",
150 {},
151) {
152 override get message() {
153 return "Historical context delivery is uncertain; replace the native thread before retrying.";
154 }
155}