Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 25 additions & 1 deletion packages/agent-modules/context-manager/src/token-estimator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -199,8 +199,15 @@ function getLastAssistantUsageInfo(
return undefined;
}

const MAX_CACHED_TEXT_UNITS = 1_048_576;
const MAX_CACHED_TEXT_ENTRIES = 2_048;

export class BpeTokenEstimator implements TokenEstimator {
private readonly countExactTokens: (text: string) => number;
// Content keys remain valid across detached histories, edits and model
// projections. Bound retained UTF-16 text as well as entry count.
private readonly textTokens = new Map<string, number>();
private cachedTextUnits = 0;

/**
* @param encoder Override the default o200k_base tokenizer. Tests inject
Expand All @@ -222,12 +229,29 @@ export class BpeTokenEstimator implements TokenEstimator {

estimateTextTokens(text: string): number {
if (!text) return 0;
const cached = this.textTokens.get(text);
if (cached !== undefined) {
this.textTokens.delete(text);
this.textTokens.set(text, cached);
return cached;
}
try {
if (text.length > MAX_EXACT_TOKENIZER_CHARS && hasOversizedAlphanumericRun(text)) {
return estimateTextTokensUpperBound(text);
}
const tokens = this.countExactTokens(text);
return Number.isFinite(tokens) && tokens >= 0 ? tokens : estimateTextTokensUpperBound(text);
if (!Number.isFinite(tokens) || tokens < 0) return estimateTextTokensUpperBound(text);
if (text.length <= MAX_CACHED_TEXT_UNITS) {
while (this.textTokens.size >= MAX_CACHED_TEXT_ENTRIES || this.cachedTextUnits + text.length > MAX_CACHED_TEXT_UNITS) {
const oldest = this.textTokens.keys().next().value;
if (oldest === undefined) break;
this.textTokens.delete(oldest);
this.cachedTextUnits -= oldest.length;
}
this.textTokens.set(text, tokens);
this.cachedTextUnits += text.length;
}
return tokens;
} catch {
return estimateTextTokensUpperBound(text);
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -305,7 +305,7 @@ export function createSessionSystemCanonicalHistoryProvider(
`Canonical compaction generation verification failed: ${change.sessionId}`,
);
}
await syncIndex(change.sessionId, paths, true);
await syncIndex(change.sessionId, paths, true, committed);
const result = {
revision: canonicalActiveHistoryRevision(committed),
generation: committedGeneration,
Expand Down Expand Up @@ -335,7 +335,7 @@ export function createSessionSystemCanonicalHistoryProvider(
}
options.activity?.notify(sessionId);
}
await syncIndex(sessionId, paths, true);
await syncIndex(sessionId, paths, true, records);
return historySnapshot(records, allowPendingToolCallTail);
}

Expand Down Expand Up @@ -372,8 +372,9 @@ export function createSessionSystemCanonicalHistoryProvider(
sessionId: string,
paths: SessionHistoryPaths,
allowMissingParent = false,
verifiedRecords?: readonly CanonicalHistoryEnvelope[],
): Promise<void> {
const active = await files.readActiveStrict(paths.messages);
const active = verifiedRecords ?? (await files.readActiveStrict(paths.messages));
const adapter = indexes.get(sessionId) ?? createCanonicalHistoryIndexAdapter(paths.sessionDir);
indexes.set(sessionId, adapter);
try {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -27,9 +27,10 @@ export interface DurableCanonicalHistoryProvider {
read(sessionId: string): Promise<CanonicalHistorySnapshot>;
/** Active post-mutation history; the final tool round may still await results. */
readActive(sessionId: string): Promise<CanonicalHistorySnapshot>;
append(change: CanonicalHistoryChange): Promise<unknown>;
replace(change: CanonicalHistoryChange): Promise<unknown>;
compact?(change: CanonicalHistoryCompactionChange): Promise<unknown>;
/** Return the verified post-write snapshot from the same lane, or request a reread with void. */
append(change: CanonicalHistoryChange): Promise<CanonicalHistorySnapshot | void>;
replace(change: CanonicalHistoryChange): Promise<CanonicalHistorySnapshot | void>;
compact?(change: CanonicalHistoryCompactionChange): Promise<CanonicalHistorySnapshot | void>;
settleTurnTail?(mutation: SettleTurnTailMutation): Promise<TurnHistoryMutationCommit>;
retractTurn?(mutation: RetractTurnMutation): Promise<TurnHistoryMutationCommit>;
}
Expand Down Expand Up @@ -102,13 +103,17 @@ export class DurableCanonicalHistoryStore implements CanonicalHistoryStore {
private commit(
kind: 'append' | 'replace',
change: CanonicalHistoryChange,
mutate: (snapshot: CanonicalHistoryChange) => Promise<unknown>,
mutate: (snapshot: CanonicalHistoryChange) => Promise<CanonicalHistorySnapshot | void>,
): Promise<CanonicalHistoryCommit> {
const snapshot = captureSemanticSnapshot(change).value;
validateCanonicalHistoryChange(snapshot, kind);
return this.lane.run(snapshot.sessionId, async () => {
await mutate(snapshot);
return this.readActiveCommitted(snapshot.sessionId);
const committed = await mutate(snapshot);
// Session-owned providers already reread and validate inside their lane.
// Legacy adapters returning void retain the strict post-write read.
return committed === undefined
? this.readActiveCommitted(snapshot.sessionId)
: this.readProviderSnapshot(committed);
});
}

Expand Down
Original file line number Diff line number Diff line change
@@ -1,139 +1,26 @@
const BLOCK_BYTES = 64;
const UTF8_CHUNK_CODE_UNITS = 8 * 1_024;

const INITIAL_STATE = [
0x6a09e667, 0xbb67ae85, 0x3c6ef372, 0xa54ff53a, 0x510e527f, 0x9b05688c, 0x1f83d9ab, 0x5be0cd19,
] as const;
import { createHash } from 'node:crypto';

const ROUND_CONSTANTS = new Uint32Array([
0x428a2f98, 0x71374491, 0xb5c0fbcf, 0xe9b5dba5, 0x3956c25b, 0x59f111f1, 0x923f82a4, 0xab1c5ed5,
0xd807aa98, 0x12835b01, 0x243185be, 0x550c7dc3, 0x72be5d74, 0x80deb1fe, 0x9bdc06a7, 0xc19bf174,
0xe49b69c1, 0xefbe4786, 0x0fc19dc6, 0x240ca1cc, 0x2de92c6f, 0x4a7484aa, 0x5cb0a9dc, 0x76f988da,
0x983e5152, 0xa831c66d, 0xb00327c8, 0xbf597fc7, 0xc6e00bf3, 0xd5a79147, 0x06ca6351, 0x14292967,
0x27b70a85, 0x2e1b2138, 0x4d2c6dfc, 0x53380d13, 0x650a7354, 0x766a0abb, 0x81c2c92e, 0x92722c85,
0xa2bfe8a1, 0xa81a664b, 0xc24b8b70, 0xc76c51a3, 0xd192e819, 0xd6990624, 0xf40e3585, 0x106aa070,
0x19a4c116, 0x1e376c08, 0x2748774c, 0x34b0bcb5, 0x391c0cb3, 0x4ed8aa4a, 0x5b9cca4f, 0x682e6ff3,
0x748f82ee, 0x78a5636f, 0x84c87814, 0x8cc70208, 0x90befffa, 0xa4506ceb, 0xbef9a3f7, 0xc67178f2,
]);
const UTF8_CHUNK_CODE_UNITS = 8 * 1_024;

/**
* Dependency-free incremental SHA-256 for AgentHost semantic identity.
* Text is encoded in bounded chunks so hashing one large History string does
* not allocate another payload-sized Buffer. This is intentionally internal;
* persistence digests continue to use their owning adapters.
*/
/** Native SHA-256 with bounded UTF-8 chunks and the existing update boundaries. */
export class IncrementalSha256 {
private readonly state = new Uint32Array(INITIAL_STATE);
private readonly block = new Uint8Array(BLOCK_BYTES);
private readonly schedule = new Uint32Array(64);
private readonly textEncoder = new TextEncoder();
private bufferedBytes = 0;
private totalBytes = 0;
private readonly hash = createHash('sha256');
private finalized = false;

update(value: string): void {
if (this.finalized) throw new Error('SHA-256 digest is already finalized.');
let offset = 0;
while (offset < value.length) {
const end = safeUtf8ChunkEnd(value, offset);
this.updateBytes(this.textEncoder.encode(value.slice(offset, end)));
this.hash.update(value.slice(offset, end), 'utf8');
offset = end;
}
}

digestHex(): string {
if (this.finalized) throw new Error('SHA-256 digest is already finalized.');
this.finalized = true;
const messageBytes = this.totalBytes;
this.block[this.bufferedBytes] = 0x80;
this.bufferedBytes += 1;
if (this.bufferedBytes > 56) {
this.block.fill(0, this.bufferedBytes);
this.compress();
this.bufferedBytes = 0;
}
this.block.fill(0, this.bufferedBytes, 56);
const messageBits = messageBytes * 8;
writeUint32BigEndian(this.block, 56, Math.floor(messageBits / 0x1_0000_0000));
writeUint32BigEndian(this.block, 60, messageBits >>> 0);
this.compress();
return [...this.state].map((word) => word.toString(16).padStart(8, '0')).join('');
}

private updateBytes(bytes: Uint8Array): void {
this.totalBytes += bytes.length;
let offset = 0;
while (offset < bytes.length) {
const copied = Math.min(BLOCK_BYTES - this.bufferedBytes, bytes.length - offset);
this.block.set(bytes.subarray(offset, offset + copied), this.bufferedBytes);
this.bufferedBytes += copied;
offset += copied;
if (this.bufferedBytes !== BLOCK_BYTES) continue;
this.compress();
this.bufferedBytes = 0;
}
}

private compress(): void {
const words = this.schedule;
for (let index = 0; index < 16; index += 1) {
const offset = index * 4;
words[index] =
(arrayValue(this.block, offset) << 24) |
(arrayValue(this.block, offset + 1) << 16) |
(arrayValue(this.block, offset + 2) << 8) |
arrayValue(this.block, offset + 3);
}
for (let index = 16; index < 64; index += 1) {
const prior15 = arrayValue(words, index - 15);
const prior2 = arrayValue(words, index - 2);
const sigma0 = rotateRight(prior15, 7) ^ rotateRight(prior15, 18) ^ (prior15 >>> 3);
const sigma1 = rotateRight(prior2, 17) ^ rotateRight(prior2, 19) ^ (prior2 >>> 10);
words[index] =
(arrayValue(words, index - 16) + sigma0 + arrayValue(words, index - 7) + sigma1) >>> 0;
}
this.compressRounds(words);
}

private compressRounds(words: Uint32Array): void {
let a = arrayValue(this.state, 0);
let b = arrayValue(this.state, 1);
let c = arrayValue(this.state, 2);
let d = arrayValue(this.state, 3);
let e = arrayValue(this.state, 4);
let f = arrayValue(this.state, 5);
let g = arrayValue(this.state, 6);
let h = arrayValue(this.state, 7);
for (let index = 0; index < 64; index += 1) {
const upperSigma1 = rotateRight(e, 6) ^ rotateRight(e, 11) ^ rotateRight(e, 25);
const choose = (e & f) ^ (~e & g);
const temporary1 =
(h +
upperSigma1 +
choose +
arrayValue(ROUND_CONSTANTS, index) +
arrayValue(words, index)) >>>
0;
const upperSigma0 = rotateRight(a, 2) ^ rotateRight(a, 13) ^ rotateRight(a, 22);
const majority = (a & b) ^ (a & c) ^ (b & c);
const temporary2 = (upperSigma0 + majority) >>> 0;
h = g;
g = f;
f = e;
e = (d + temporary1) >>> 0;
d = c;
c = b;
b = a;
a = (temporary1 + temporary2) >>> 0;
}
this.state[0] = (arrayValue(this.state, 0) + a) >>> 0;
this.state[1] = (arrayValue(this.state, 1) + b) >>> 0;
this.state[2] = (arrayValue(this.state, 2) + c) >>> 0;
this.state[3] = (arrayValue(this.state, 3) + d) >>> 0;
this.state[4] = (arrayValue(this.state, 4) + e) >>> 0;
this.state[5] = (arrayValue(this.state, 5) + f) >>> 0;
this.state[6] = (arrayValue(this.state, 6) + g) >>> 0;
this.state[7] = (arrayValue(this.state, 7) + h) >>> 0;
return this.hash.digest('hex');
}
}

Expand All @@ -152,20 +39,3 @@ function isHighSurrogate(value: number): boolean {
function isLowSurrogate(value: number): boolean {
return value >= 0xdc00 && value <= 0xdfff;
}

function rotateRight(value: number, bits: number): number {
return (value >>> bits) | (value << (32 - bits));
}

function writeUint32BigEndian(target: Uint8Array, offset: number, value: number): void {
target[offset] = (value >>> 24) & 0xff;
target[offset + 1] = (value >>> 16) & 0xff;
target[offset + 2] = (value >>> 8) & 0xff;
target[offset + 3] = value & 0xff;
}

function arrayValue(values: Uint8Array | Uint32Array, index: number): number {
const value = values[index];
if (value === undefined) throw new RangeError(`SHA-256 array index is out of bounds: ${index}.`);
return value;
}
Original file line number Diff line number Diff line change
Expand Up @@ -5,19 +5,39 @@ export interface SemanticSnapshot<T> {
readonly fingerprint: string;
}

const ownedValues = new WeakSet<object>();
const fingerprints = new WeakMap<object, string>();

/**
* Detach callback-owned History/event data before it becomes an in-run
* identity or deferred delivery payload. Unsupported or cyclic values fail
* closed. The digest is streamed so identity memory does not scale with the
* encoded payload size.
*/
export function captureSemanticSnapshot<T>(value: T): SemanticSnapshot<T> {
const snapshot = structuredClone(value);
const encoder = new SemanticIdentityEncoder(new IncrementalSha256());
encodeValue(snapshot, new WeakSet<object>(), encoder);
// Clone an external graph in one operation: independent subtree clones can
// change aliasing and getter order. Reuse only an already-owned root.
const snapshot =
typeof value === 'object' && value !== null && ownedValues.has(value)
? value
: freezeSemanticValue(structuredClone(value), new WeakSet<object>());
let fingerprint: string | undefined;
return {
value: freezeSemanticValue(snapshot, new WeakSet<object>()),
fingerprint: encoder.digest(),
value: snapshot,
// Value-only consumers still validate and detach eagerly, but never encode
// or hash the history. Only values frozen by this module may be reused.
get fingerprint() {
if (fingerprint !== undefined) return fingerprint;
const object = typeof snapshot === 'object' && snapshot !== null ? snapshot : undefined;
fingerprint = object ? fingerprints.get(object) : undefined;
if (fingerprint === undefined) {
const encoder = new SemanticIdentityEncoder(new IncrementalSha256());
encodeValue(snapshot, new WeakSet<object>(), encoder);
fingerprint = encoder.digest();
if (object) fingerprints.set(object, fingerprint);
}
return fingerprint;
},
};
}

Expand Down Expand Up @@ -54,13 +74,28 @@ class SemanticIdentityEncoder {
}
}

function freezeSemanticValue<T>(value: T, seen: WeakSet<object>): T {
if (typeof value !== 'object' || value === null || seen.has(value)) return value;
seen.add(value);
Reflect.ownKeys(value).forEach((key) => {
freezeSemanticValue(Reflect.get(value, key), seen);
});
return Object.freeze(value);
function freezeSemanticValue<T>(value: T, ancestors: WeakSet<object>): T {
if (typeof value !== 'object' || value === null) {
if (!['undefined', 'string', 'boolean', 'number'].includes(typeof value) && value !== null) {
throw new TypeError(`Unsupported semantic identity value: ${typeof value}.`);
}
return value;
}
if (ownedValues.has(value)) return value;
if (ancestors.has(value)) throw new TypeError('Cyclic semantic identity values are unsupported.');
const prototype = Object.getPrototypeOf(value);
if (!Array.isArray(value) && prototype !== Object.prototype && prototype !== null) {
throw new TypeError('Semantic identity values must contain only plain objects and arrays.');
}
if (Object.getOwnPropertySymbols(value).length > 0) {
throw new TypeError('Symbol-keyed semantic identity values are unsupported.');
}
ancestors.add(value);
for (const key of Object.keys(value)) freezeSemanticValue(Reflect.get(value, key), ancestors);
ancestors.delete(value);
Object.freeze(value);
ownedValues.add(value);
return value;
}

function encodeValue(
Expand Down
Loading
Loading