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/**
16 * Persist before/after injection: an ambiguous pending delivery requires a fresh native thread.
17 * Run `unsent` when the provider refused the turn before reading the prompt, so that
18 * refusal does not leave an ambiguous marker behind.
19 */
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 // Repeated failures can accumulate many recovery markers. Keep a single
52 // thread-level entry point when detailed coverage would crowd out history;
53 // its activity includes the original handoff/fork source references.
54 if (historyCost([], coverage) > Math.min(4_000, budget / 2)) {
55 const strategies = Array.from(new Set(pending.map((handoff) => handoff.strategy)));
56 coverage = `Context handoff (${strategies.join(", ")}). ${pending.length} handoff records; detailed coverage references omitted. Recover history with t3_thread_read({threadId:"${input.providerThread.appThreadId ?? pending[0]!.threadId}",view:"activity",limit:20,maxCharsPerItem:4000}); paginate with afterPosition=nextPosition. Follow fork/handoff source references in activity. For long items use itemId and textOffset=nextTextOffset until null.`;
57 }
58 const seen = new Set(input.alreadyDeliveredItemIds);
59 const messages = pending
60 .flatMap((handoff) => handoff.history?.messages ?? [])
61 .filter((message) => {
62 if (seen.has(message.itemId)) return false;
63 seen.add(message.itemId);
64 return true;
65 });
66 // Old preview handoffs remain readable. Their preformatted context is included
67 // as a whole when it fits, otherwise the coverage marker points to retrieval.
68 const oldContext = pending
69 .filter((handoff) => handoff.history === undefined)
70 .map((handoff) => handoff.summaryText)
71 .join("\n\n");
72 const fullCoverage =
73 oldContext && historyCost([], `${coverage}\n${oldContext}`) + 512 <= budget
74 ? `${coverage}\n${oldContext}`
75 : coverage;
76 const selected = selectHistory({
77 messages,
78 coverage: fullCoverage,
79 omittedItems: pending.reduce((sum, handoff) => sum + (handoff.history?.omittedItems ?? 0), 0),
80 budget,
81 });
82 if (historyCost(selected.messages, selected.context) > budget) {
83 if (input.deferInline) return { context: "", delivered: Effect.void, unsent: Effect.void };
84 return yield* new ContextHandoffBudgetError();
85 }
86 const omittedItemIds = new Set(selected.omittedItemIds);
87 const persist = (status: "pending" | "injected" | "inline") =>
88 Effect.forEach(
89 pending,
90 (handoff) =>
91 input.persist({
92 ...handoff,
93 ...(nativeThreadId === undefined
94 ? {}
95 : {
96 delivery: {
97 nativeThreadId,
98 status,
99 omittedItemIds: [
100 ...(handoff.history?.omittedItemIds ?? []),
101 ...(handoff.history?.messages ?? [])
102 .filter((message) => omittedItemIds.has(message.itemId))
103 .map((message) => message.itemId),
104 ],
105 itemIds: selected.messages
106 .filter((message) =>
107 handoff.history?.messages.some(
108 (candidate) => candidate.itemId === message.itemId,
109 ),
110 )
111 .map((message) => message.itemId),
112 },
113 }),
114 }),
115 { discard: true },
116 );
117 if (input.inject !== undefined && nativeThreadId !== undefined) {
118 if (
119 pending.some(
120 (handoff) =>
121 handoff.delivery?.nativeThreadId === nativeThreadId &&
122 handoff.delivery.status === "pending",
123 )
124 ) {
125 return yield* new ContextHandoffDeliveryUncertainError();
126 }
127 yield* persist("pending");
128 const injected = yield* input.inject({
129 messages: selected.messages,
130 context: selected.context,
131 });
132 if (injected) {
133 yield* persist("injected");
134 return { context: "", delivered: Effect.void, unsent: Effect.void };
135 }
136 } else {
137 // Text-only delivery can also be accepted before a connection drops.
138 yield* persist("pending");
139 }
140 if (input.deferInline) {
141 // Compaction APIs cannot accept an inline transcript. Keep it available
142 // for the next ordinary turn when native injection is unsupported.
143 yield* Effect.forEach(pending, input.persist, { discard: true });
144 return { context: "", delivered: Effect.void, unsent: Effect.void };
145 }
146 return {
147 context: renderHistory(selected.messages, selected.context),
148 delivered: persist("inline"),
149 unsent: Effect.forEach(pending, input.persist, { discard: true }),
150 };
151 },
152);
153
154export class ContextHandoffDeliveryUncertainError extends Schema.TaggedError<ContextHandoffDeliveryUncertainError>()(
155 "ContextHandoffDeliveryUncertainError",
156 {},
157) {
158 override get message() {
159 return "Historical context delivery is uncertain; replace the native thread before retrying.";
160 }
161}