Skip to content

Commit 92d04a8

Browse files
committed
feat(sdk): persist compaction and injected context through the transcript storage state
The model lane after a compaction cannot be rebuilt from the transcript, so every continuation used to re-read the whole conversation and summarise it again. The runtime now records the compacted lane in the storage's state slot, with the transcript id it covers and a fingerprint of that prefix, and rebuilds from it at boot when the prefix is unchanged. A rollback or edit that reconverts the lane clears the state in the same changeset as the truncate. Conversational messages added with chat.inject are recorded the same way, anchored to the transcript message they followed, so they survive a continuation instead of living only in the worker that received them. Adds an in-memory storage that logs the changesets it receives, and a test-only override for the storage the runtime persists through, so the exact changesets for a turn, a mid-turn steer, a compaction, a rollback and an injection are asserted.
1 parent a7e6626 commit 92d04a8

3 files changed

Lines changed: 611 additions & 16 deletions

File tree

packages/trigger-sdk/src/v3/ai.ts

Lines changed: 91 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -76,10 +76,29 @@ import {
7676
createTranscriptShadow,
7777
defaultStorage,
7878
diffTranscript,
79+
parseTranscriptRuntimeState,
80+
prefixFingerprint,
81+
restoreModelLane,
82+
type TranscriptChange,
7983
type TranscriptChangeReason,
84+
type TranscriptRuntimeState,
8085
type TranscriptShadow,
86+
type TranscriptStorage,
8187
type TranscriptStorageContext,
8288
} from "./transcriptStorage.js";
89+
90+
let transcriptStorageOverride: TranscriptStorage<unknown> | undefined;
91+
92+
/**
93+
* Test-only override for the storage `chat.agent` persists through, so a
94+
* test can capture the exact changesets the runtime produces.
95+
* @internal
96+
*/
97+
export function __setTranscriptStorageForTests(
98+
storage: TranscriptStorage<unknown> | undefined
99+
): void {
100+
transcriptStorageOverride = storage;
101+
}
83102
import {
84103
type ChatInputChunk,
85104
type ChatTaskWirePayload,
@@ -6870,8 +6889,18 @@ function chatAgent<
68706889
// collectively cost ~600ms on every first-message TTFC. Both reads
68716890
// swallow errors internally; the agent stays available either way.
68726891
const sessionIdForSnapshot = payload.sessionId ?? payload.chatId;
6873-
const transcriptStorage = defaultStorage;
6892+
const transcriptStorage = transcriptStorageOverride ?? defaultStorage;
68746893
let transcriptShadow: TranscriptShadow = createTranscriptShadow([]);
6894+
let bootTranscriptState: unknown = null;
6895+
/**
6896+
* True while the model lane holds a compaction summary, so it cannot be
6897+
* rebuilt from the transcript and has to be persisted as state. Reset
6898+
* wherever the lane is reconverted from the UI lane.
6899+
*/
6900+
let laneCompacted = false;
6901+
/** Conversational `chat.inject` messages in the lane, anchored to the transcript. */
6902+
let laneInjections: NonNullable<TranscriptRuntimeState["injections"]> = [];
6903+
let persistedStateSet = false;
68756904
let bootSnapshot:
68766905
| { messages: TUIMessage[]; lastOutEventId?: string; lastInEventId?: string }
68776906
| undefined;
@@ -6919,6 +6948,23 @@ function chatAgent<
69196948
const { changes, shadow } = diffTranscript(transcriptShadow, opts.messages, {
69206949
nonFinalIds: opts.nonFinalIds,
69216950
});
6951+
const lastId = opts.messages.at(-1)?.id;
6952+
const runtimeState: TranscriptRuntimeState | null =
6953+
laneCompacted && lastId !== undefined
6954+
? {
6955+
v: 1,
6956+
compaction: {
6957+
modelMessages: accumulatedMessages,
6958+
throughId: lastId,
6959+
fingerprint: prefixFingerprint(shadow, lastId),
6960+
},
6961+
}
6962+
: laneInjections.length > 0
6963+
? { v: 1, injections: laneInjections }
6964+
: null;
6965+
if (runtimeState !== null || persistedStateSet) {
6966+
changes.push({ op: "state", value: runtimeState } satisfies TranscriptChange);
6967+
}
69226968
const inCursor = chatInputRouter().resumeFloor();
69236969
await transcriptStorage.save(
69246970
{
@@ -6939,6 +6985,7 @@ function chatAgent<
69396985
}
69406986
);
69416987
transcriptShadow = shadow;
6988+
persistedStateSet = runtimeState !== null;
69426989
};
69436990

69446991
/**
@@ -7023,6 +7070,8 @@ function chatAgent<
70237070
clientData: bootClientData,
70247071
});
70257072
transcriptShadow = createTranscriptShadow(loaded.messages);
7073+
bootTranscriptState = loaded.state;
7074+
persistedStateSet = loaded.state !== null && loaded.state !== undefined;
70267075
bootSnapshot = {
70277076
messages: loaded.messages,
70287077
lastOutEventId: loaded.cursors?.lastOutEventId,
@@ -7367,7 +7416,14 @@ function chatAgent<
73677416
}
73687417
}
73697418
try {
7370-
accumulatedMessages = await toModelMessages(accumulatedUIMessages);
7419+
const restored = await restoreModelLane(
7420+
accumulatedUIMessages,
7421+
parseTranscriptRuntimeState(bootTranscriptState),
7422+
(messages) => toModelMessages(messages)
7423+
);
7424+
accumulatedMessages = restored.messages;
7425+
laneCompacted = restored.compacted;
7426+
laneInjections = restored.injections;
73717427
} catch (error) {
73727428
logger.warn("chat.agent: toModelMessages failed at boot; starting empty", {
73737429
error: error instanceof Error ? error.message : String(error),
@@ -8039,6 +8095,8 @@ function chatAgent<
80398095
);
80408096
accumulatedUIMessages = [...hydrated] as TUIMessage[];
80418097
accumulatedMessages = await toModelMessages(hydrated);
8098+
laneCompacted = false;
8099+
laneInjections = [];
80428100
locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages);
80438101
}
80448102

@@ -8076,6 +8134,8 @@ function chatAgent<
80768134
locals.set(chatOverrideMessagesKey, undefined);
80778135
accumulatedUIMessages = [...actionOverride] as TUIMessage[];
80788136
accumulatedMessages = await toModelMessages(actionOverride);
8137+
laneCompacted = false;
8138+
laneInjections = [];
80798139
locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages);
80808140

80818141
actionChangedHistory = true;
@@ -8208,6 +8268,8 @@ function chatAgent<
82088268

82098269
accumulatedUIMessages = merged;
82108270
accumulatedMessages = await toModelMessages(merged);
8271+
laneCompacted = false;
8272+
laneInjections = [];
82118273
locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages);
82128274

82138275
// Track new messages for onTurnComplete.newUIMessages.
@@ -8257,6 +8319,8 @@ function chatAgent<
82578319
accumulatedUIMessages.pop();
82588320
}
82598321
accumulatedMessages = await toModelMessages(accumulatedUIMessages);
8322+
laneCompacted = false;
8323+
laneInjections = [];
82608324
} else if (cleanedUIMessages.length > 0) {
82618325
// Submit-message (and the special-cased
82628326
// handover-prepare → submit-message rewrite earlier in
@@ -8310,6 +8374,8 @@ function chatAgent<
83108374
"chat.agent: replaced message not found at the model lane tail; reconverting the lane"
83118375
);
83128376
accumulatedMessages = await toModelMessages(accumulatedUIMessages);
8377+
laneCompacted = false;
8378+
laneInjections = [];
83138379
}
83148380
} else {
83158381
const incomingModelMessages = await toModelMessages(cleanedUIMessages);
@@ -8491,6 +8557,8 @@ function chatAgent<
84918557
locals.set(chatOverrideMessagesKey, undefined);
84928558
accumulatedUIMessages = [...turnStartOverride] as TUIMessage[];
84938559
accumulatedMessages = await toModelMessages(turnStartOverride);
8560+
laneCompacted = false;
8561+
laneInjections = [];
84948562
locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages);
84958563
}
84968564
},
@@ -8556,7 +8624,12 @@ function chatAgent<
85568624
const lastAccumulated = accumulatedMessages[accumulatedMessages.length - 1];
85578625
const bgQueue = locals.get(chatBackgroundQueueKey);
85588626
if (bgQueue && bgQueue.length > 0 && lastAccumulated?.role !== "tool") {
8559-
accumulatedMessages.push(...bgQueue.splice(0));
8627+
const injected = bgQueue.splice(0);
8628+
accumulatedMessages.push(...injected);
8629+
laneInjections.push({
8630+
afterId: accumulatedUIMessages.at(-1)?.id ?? "",
8631+
messages: injected,
8632+
});
85608633
}
85618634

85628635
if (isHeadStartFinalTurn) {
@@ -8749,6 +8822,8 @@ function chatAgent<
87498822
accumulatedMessages = await toModelMessages(
87508823
runOverride.filter((m) => !pendingIds.has(m.id))
87518824
);
8825+
laneCompacted = false;
8826+
laneInjections = [];
87528827
locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages);
87538828
}
87548829

@@ -8774,6 +8849,8 @@ function chatAgent<
87748849
accumulatedMessages = taskCompactionConfig?.compactModelMessages
87758850
? await taskCompactionConfig.compactModelMessages(compactEvent)
87768851
: modelOnlyOverride;
8852+
laneCompacted = true;
8853+
laneInjections = [];
87778854

87788855
// Apply UI messages: callback or default (preserve all)
87798856
if (taskCompactionConfig?.compactUIMessages) {
@@ -8866,6 +8943,8 @@ function chatAgent<
88668943
"chat.agent: replaced response not found at the model lane tail; reconverting the lane"
88678944
);
88688945
accumulatedMessages = await toModelMessages(accumulatedUIMessages);
8946+
laneCompacted = false;
8947+
laneInjections = [];
88698948
}
88708949
} else {
88718950
accumulatedMessages.push(...responseModelMessages);
@@ -8985,6 +9064,9 @@ function chatAgent<
89859064
},
89869065
];
89879066

9067+
laneCompacted = true;
9068+
laneInjections = [];
9069+
89889070
// UI messages: callback or default (preserve all)
89899071
if (outerCompaction.compactUIMessages) {
89909072
accumulatedUIMessages = (await outerCompaction.compactUIMessages(
@@ -9089,6 +9171,8 @@ function chatAgent<
90899171
locals.set(chatOverrideMessagesKey, undefined);
90909172
accumulatedUIMessages = [...override] as TUIMessage[];
90919173
accumulatedMessages = await toModelMessages(override);
9174+
laneCompacted = false;
9175+
laneInjections = [];
90929176
locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages);
90939177
// Update event so onTurnComplete sees compacted messages
90949178
turnCompleteEvent.messages = accumulatedMessages;
@@ -9148,6 +9232,8 @@ function chatAgent<
91489232
locals.set(chatOverrideMessagesKey, undefined);
91499233
accumulatedUIMessages = [...turnCompleteOverride] as TUIMessage[];
91509234
accumulatedMessages = await toModelMessages(turnCompleteOverride);
9235+
laneCompacted = false;
9236+
laneInjections = [];
91519237
locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages);
91529238
}
91539239
},
@@ -9512,6 +9598,8 @@ function chatAgent<
95129598
"chat.agent: replaced partial not found at the model lane tail; reconverting the lane"
95139599
);
95149600
accumulatedMessages = await toModelMessages(erroredUIMessagesWithPartial);
9601+
laneCompacted = false;
9602+
laneInjections = [];
95159603
}
95169604
}
95179605
accumulatedUIMessages = erroredUIMessagesWithPartial;

0 commit comments

Comments
 (0)