Skip to content

Commit 7a020c4

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 15f04f2 commit 7a020c4

3 files changed

Lines changed: 656 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
{
@@ -6947,6 +6993,7 @@ function chatAgent<
69476993
}
69486994
);
69496995
transcriptShadow = shadow;
6996+
persistedStateSet = runtimeState !== null;
69506997
};
69516998

69526999
/**
@@ -7031,6 +7078,8 @@ function chatAgent<
70317078
clientData: bootClientData,
70327079
});
70337080
transcriptShadow = createTranscriptShadow(loaded.messages);
7081+
bootTranscriptState = loaded.state;
7082+
persistedStateSet = loaded.state !== null && loaded.state !== undefined;
70347083
bootSnapshot = {
70357084
messages: loaded.messages,
70367085
lastOutEventId: loaded.cursors?.lastOutEventId,
@@ -7375,7 +7424,14 @@ function chatAgent<
73757424
}
73767425
}
73777426
try {
7378-
accumulatedMessages = await toModelMessages(accumulatedUIMessages);
7427+
const restored = await restoreModelLane(
7428+
accumulatedUIMessages,
7429+
parseTranscriptRuntimeState(bootTranscriptState),
7430+
(messages) => toModelMessages(messages)
7431+
);
7432+
accumulatedMessages = restored.messages;
7433+
laneCompacted = restored.compacted;
7434+
laneInjections = restored.injections;
73797435
} catch (error) {
73807436
logger.warn("chat.agent: toModelMessages failed at boot; starting empty", {
73817437
error: error instanceof Error ? error.message : String(error),
@@ -8047,6 +8103,8 @@ function chatAgent<
80478103
);
80488104
accumulatedUIMessages = [...hydrated] as TUIMessage[];
80498105
accumulatedMessages = await toModelMessages(hydrated);
8106+
laneCompacted = false;
8107+
laneInjections = [];
80508108
locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages);
80518109
}
80528110

@@ -8084,6 +8142,8 @@ function chatAgent<
80848142
locals.set(chatOverrideMessagesKey, undefined);
80858143
accumulatedUIMessages = [...actionOverride] as TUIMessage[];
80868144
accumulatedMessages = await toModelMessages(actionOverride);
8145+
laneCompacted = false;
8146+
laneInjections = [];
80878147
locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages);
80888148

80898149
actionChangedHistory = true;
@@ -8216,6 +8276,8 @@ function chatAgent<
82168276

82178277
accumulatedUIMessages = merged;
82188278
accumulatedMessages = await toModelMessages(merged);
8279+
laneCompacted = false;
8280+
laneInjections = [];
82198281
locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages);
82208282

82218283
// Track new messages for onTurnComplete.newUIMessages.
@@ -8265,6 +8327,8 @@ function chatAgent<
82658327
accumulatedUIMessages.pop();
82668328
}
82678329
accumulatedMessages = await toModelMessages(accumulatedUIMessages);
8330+
laneCompacted = false;
8331+
laneInjections = [];
82688332
} else if (cleanedUIMessages.length > 0) {
82698333
// Submit-message (and the special-cased
82708334
// handover-prepare → submit-message rewrite earlier in
@@ -8318,6 +8382,8 @@ function chatAgent<
83188382
"chat.agent: replaced message not found at the model lane tail; reconverting the lane"
83198383
);
83208384
accumulatedMessages = await toModelMessages(accumulatedUIMessages);
8385+
laneCompacted = false;
8386+
laneInjections = [];
83218387
}
83228388
} else {
83238389
const incomingModelMessages = await toModelMessages(cleanedUIMessages);
@@ -8499,6 +8565,8 @@ function chatAgent<
84998565
locals.set(chatOverrideMessagesKey, undefined);
85008566
accumulatedUIMessages = [...turnStartOverride] as TUIMessage[];
85018567
accumulatedMessages = await toModelMessages(turnStartOverride);
8568+
laneCompacted = false;
8569+
laneInjections = [];
85028570
locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages);
85038571
}
85048572
},
@@ -8564,7 +8632,12 @@ function chatAgent<
85648632
const lastAccumulated = accumulatedMessages[accumulatedMessages.length - 1];
85658633
const bgQueue = locals.get(chatBackgroundQueueKey);
85668634
if (bgQueue && bgQueue.length > 0 && lastAccumulated?.role !== "tool") {
8567-
accumulatedMessages.push(...bgQueue.splice(0));
8635+
const injected = bgQueue.splice(0);
8636+
accumulatedMessages.push(...injected);
8637+
laneInjections.push({
8638+
afterId: accumulatedUIMessages.at(-1)?.id ?? "",
8639+
messages: injected,
8640+
});
85688641
}
85698642

85708643
if (isHeadStartFinalTurn) {
@@ -8757,6 +8830,8 @@ function chatAgent<
87578830
accumulatedMessages = await toModelMessages(
87588831
runOverride.filter((m) => !pendingIds.has(m.id))
87598832
);
8833+
laneCompacted = false;
8834+
laneInjections = [];
87608835
locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages);
87618836
}
87628837

@@ -8782,6 +8857,8 @@ function chatAgent<
87828857
accumulatedMessages = taskCompactionConfig?.compactModelMessages
87838858
? await taskCompactionConfig.compactModelMessages(compactEvent)
87848859
: modelOnlyOverride;
8860+
laneCompacted = true;
8861+
laneInjections = [];
87858862

87868863
// Apply UI messages: callback or default (preserve all)
87878864
if (taskCompactionConfig?.compactUIMessages) {
@@ -8874,6 +8951,8 @@ function chatAgent<
88748951
"chat.agent: replaced response not found at the model lane tail; reconverting the lane"
88758952
);
88768953
accumulatedMessages = await toModelMessages(accumulatedUIMessages);
8954+
laneCompacted = false;
8955+
laneInjections = [];
88778956
}
88788957
} else {
88798958
accumulatedMessages.push(...responseModelMessages);
@@ -8993,6 +9072,9 @@ function chatAgent<
89939072
},
89949073
];
89959074

9075+
laneCompacted = true;
9076+
laneInjections = [];
9077+
89969078
// UI messages: callback or default (preserve all)
89979079
if (outerCompaction.compactUIMessages) {
89989080
accumulatedUIMessages = (await outerCompaction.compactUIMessages(
@@ -9097,6 +9179,8 @@ function chatAgent<
90979179
locals.set(chatOverrideMessagesKey, undefined);
90989180
accumulatedUIMessages = [...override] as TUIMessage[];
90999181
accumulatedMessages = await toModelMessages(override);
9182+
laneCompacted = false;
9183+
laneInjections = [];
91009184
locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages);
91019185
// Update event so onTurnComplete sees compacted messages
91029186
turnCompleteEvent.messages = accumulatedMessages;
@@ -9156,6 +9240,8 @@ function chatAgent<
91569240
locals.set(chatOverrideMessagesKey, undefined);
91579241
accumulatedUIMessages = [...turnCompleteOverride] as TUIMessage[];
91589242
accumulatedMessages = await toModelMessages(turnCompleteOverride);
9243+
laneCompacted = false;
9244+
laneInjections = [];
91599245
locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages);
91609246
}
91619247
},
@@ -9524,6 +9610,8 @@ function chatAgent<
95249610
"chat.agent: replaced partial not found at the model lane tail; reconverting the lane"
95259611
);
95269612
accumulatedMessages = await toModelMessages(erroredUIMessagesWithPartial);
9613+
laneCompacted = false;
9614+
laneInjections = [];
95279615
}
95289616
}
95299617
accumulatedUIMessages = erroredUIMessagesWithPartial;

0 commit comments

Comments
 (0)