From 024f54a9b04fcec55478553e30d2e38bc386c209 Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Mon, 21 Sep 2026 15:57:29 +0800 Subject: [PATCH 01/10] perf: reduce identity framing and repeated repository query work --- .../session-system/messages/repo/drizzle.ts | 58 +++++-- .../session-system/sessions/repo/drizzle.ts | 23 ++- .../agent-host/history/semantic-identity.ts | 37 +++-- test/history-processing.test.ts | 145 +++++++++++++++++- 4 files changed, 234 insertions(+), 29 deletions(-) diff --git a/packages/local-runtime-v2/src/service/session-system/messages/repo/drizzle.ts b/packages/local-runtime-v2/src/service/session-system/messages/repo/drizzle.ts index 4b7ed7a2..eb62483a 100644 --- a/packages/local-runtime-v2/src/service/session-system/messages/repo/drizzle.ts +++ b/packages/local-runtime-v2/src/service/session-system/messages/repo/drizzle.ts @@ -1,4 +1,4 @@ -import { and, asc, desc, eq, gt, gte, inArray, lt, lte, sql } from 'drizzle-orm'; +import { and, asc, desc, eq, gt, gte, inArray, lt, lte, placeholder, sql } from 'drizzle-orm'; import type { AppDb } from '../../../../infra/db/client.js'; import { runWithWriteLock } from '../../../../infra/db/write-transaction.js'; @@ -37,6 +37,36 @@ export function createMessageRepository(options: MessageRepositoryOptions): Mess return new DrizzleMessageRepository(options); } +function prepareMessageRead(db: AppDb) { + return db + .select() + .from(messageRows) + .where( + and( + eq(messageRows.sessionId, placeholder('sessionId')), + eq(messageRows.messageId, placeholder('msgId')), + ), + ) + .prepare(); +} + +function prepareTurnRead(db: AppDb) { + return db + .select() + .from(messageRows) + .where( + and( + eq(messageRows.sessionId, placeholder('sessionId')), + eq(messageRows.turnId, placeholder('turnId')), + ), + ) + .orderBy(asc(messageRows.id)) + .prepare(); +} + +const messageReads = new WeakMap>(); +const turnReads = new WeakMap>(); + class DrizzleMessageRepository implements MessageRepository { private readonly nowMs: () => number; constructor(private readonly options: MessageRepositoryOptions) { @@ -45,11 +75,13 @@ class DrizzleMessageRepository implements MessageRepository { async get(sessionId: string, msgId: string): Promise { this.ensureReady(sessionId); - const row = this.options.db - .select() - .from(messageRows) - .where(and(eq(messageRows.sessionId, sessionId), eq(messageRows.messageId, msgId))) - .get(); + const db = this.options.db; + let query = messageReads.get(db); + if (!query) { + query = prepareMessageRead(db); + messageReads.set(db, query); + } + const row = query.get({ sessionId, msgId }); return row ? decodeDisplayMessage(row) : undefined; } @@ -97,13 +129,13 @@ class DrizzleMessageRepository implements MessageRepository { async listTurn(sessionId: string, turnId: string): Promise { this.ensureReady(sessionId); - return this.options.db - .select() - .from(messageRows) - .where(and(eq(messageRows.sessionId, sessionId), eq(messageRows.turnId, turnId))) - .orderBy(asc(messageRows.id)) - .all() - .map(decodeDisplayMessage); + const db = this.options.db; + let query = turnReads.get(db); + if (!query) { + query = prepareTurnRead(db); + turnReads.set(db, query); + } + return query.all({ sessionId, turnId }).map(decodeDisplayMessage); } async listRecent( diff --git a/packages/local-runtime-v2/src/service/session-system/sessions/repo/drizzle.ts b/packages/local-runtime-v2/src/service/session-system/sessions/repo/drizzle.ts index 3829a6aa..007f23af 100644 --- a/packages/local-runtime-v2/src/service/session-system/sessions/repo/drizzle.ts +++ b/packages/local-runtime-v2/src/service/session-system/sessions/repo/drizzle.ts @@ -14,6 +14,7 @@ import { ne, notInArray, or, + placeholder, sql, type SQL, } from 'drizzle-orm'; @@ -118,6 +119,16 @@ const DEFAULT_TREE_FILTER: SessionChildrenOptions = { excludeInternalTreeSessions: true, }; +function prepareSessionRead(db: AppDb) { + return db + .select() + .from(sessions) + .where(and(eq(sessions.sessionId, placeholder('sessionId')), eq(sessions.columnarVersion, 3))) + .prepare(); +} + +const sessionReads = new WeakMap>(); + export function createSessionRepository(options: SessionRepositoryOptions): SessionRepository { return new DrizzleSessionRepository(options); } @@ -214,11 +225,13 @@ class DrizzleSessionRepository implements SessionRepository { } async get(sessionId: string): Promise { - const row = this.options.db - .select() - .from(sessions) - .where(and(eq(sessions.sessionId, sessionId), eq(sessions.columnarVersion, 3))) - .get(); + const db = this.options.db; + let query = sessionReads.get(db); + if (!query) { + query = prepareSessionRead(db); + sessionReads.set(db, query); + } + const row = query.get({ sessionId }); return row ? decodeSessionRow(row) : undefined; } diff --git a/packages/local-runtime-v2/src/service/turn-system/agent-host/history/semantic-identity.ts b/packages/local-runtime-v2/src/service/turn-system/agent-host/history/semantic-identity.ts index f1a3ca7f..ee9f51f4 100644 --- a/packages/local-runtime-v2/src/service/turn-system/agent-host/history/semantic-identity.ts +++ b/packages/local-runtime-v2/src/service/turn-system/agent-host/history/semantic-identity.ts @@ -118,27 +118,44 @@ export function estimateSemanticValueSize(value: unknown): number { return encoder.byteSize; } +const SEMANTIC_TAGS = [ + 'null', + 'undefined', + 'string', + 'boolean', + 'number', + 'begin', + 'length', + 'end', + 'key', +] as const; +type SemanticTag = (typeof SEMANTIC_TAGS)[number]; +const FRAME_PREFIXES = Object.fromEntries( + SEMANTIC_TAGS.map((tag) => [tag, `${tag.length}:${tag}`]), +) as Record; + class SemanticIdentityEncoder { byteSize = 0; constructor(private readonly hash?: IncrementalSha256) {} - frame(tag: string, payload: string): void { - this.write(`${Buffer.byteLength(tag)}:`); - this.write(tag); - this.write(`${Buffer.byteLength(payload)}:`); - this.write(payload); + frame(tag: SemanticTag, payload: string): void { + // Only string values and object/array keys can contain non-ASCII text. + const payloadBytes = + tag === 'string' || tag === 'key' ? Buffer.byteLength(payload) : payload.length; + const prefix = FRAME_PREFIXES[tag]; + const payloadLength = `${payloadBytes}:`; + this.byteSize += prefix.length + payloadLength.length + payloadBytes; + // Tags and length fields are ASCII. Keep the payload in its own update so + // UTF-8 surrogate handling remains identical at each frame boundary. + this.hash?.update(`${prefix}${payloadLength}`); + this.hash?.update(payload); } digest(): string { if (!this.hash) throw new Error('Semantic identity digest was not requested.'); return this.hash.digestHex(); } - - private write(value: string): void { - this.byteSize += Buffer.byteLength(value); - this.hash?.update(value); - } } function freezeSemanticValue(value: T, ancestors: WeakSet): T { diff --git a/test/history-processing.test.ts b/test/history-processing.test.ts index d9270fcf..4fc8e1a7 100644 --- a/test/history-processing.test.ts +++ b/test/history-processing.test.ts @@ -15,11 +15,18 @@ import { DatabaseClient } from '../packages/local-runtime-v2/src/infra/db/client import { initializeDatabase } from '../packages/local-runtime-v2/src/infra/db/initialize.js'; import { queryCollapseViewStates } from '../packages/local-runtime-v2/src/infra/db/schema/query-collapse.js'; import { turnIngress } from '../packages/local-runtime-v2/src/infra/db/schema/turn.js'; +import { sessions } from '../packages/local-runtime-v2/src/infra/db/schema/sessions.js'; +import { messageRows } from '../packages/local-runtime-v2/src/infra/db/schema/messages.js'; +import { createSessionRepository } from '../packages/local-runtime-v2/src/service/session-system/sessions/repo/drizzle.js'; +import { createMessageRepository } from '../packages/local-runtime-v2/src/service/session-system/messages/repo/drizzle.js'; import { createQueryCollapseState } from '../packages/local-runtime-v2/src/service/session-system/query-collapse-state.js'; import { createQueueTurnAdmissionPriorityFence } from '../packages/local-runtime-v2/src/service/session-system/index.js'; import { createTurnRepository } from '../packages/local-runtime-v2/src/service/turn-system/persistence/turn.repository.js'; import { IncrementalSha256 } from '../packages/local-runtime-v2/src/service/turn-system/agent-host/history/incremental-sha256.js'; -import { captureSemanticSnapshot } from '../packages/local-runtime-v2/src/service/turn-system/agent-host/history/semantic-identity.js'; +import { + captureSemanticSnapshot, + estimateSemanticValueSize, +} from '../packages/local-runtime-v2/src/service/turn-system/agent-host/history/semantic-identity.js'; import { DurableCanonicalHistoryStore } from '../packages/local-runtime-v2/src/service/turn-system/agent-host/history/durable-canonical-history-store.js'; import type { CanonicalHistoryChange } from '../packages/local-runtime-v2/src/service/turn-system/agent-host/history/contracts.js'; import { BpeTokenEstimator } from '../packages/agent-modules/context-manager/src/token-estimator.js'; @@ -49,6 +56,105 @@ describe('prepared runtime reads', () => { } } + it('keeps session lookups fresh and enforces the columnar version after reuse', async () => { + await withDatabase(async (client, writer) => { + const repository = createSessionRepository({ db: client.db }); + expect(await repository.get('s1')).toBeUndefined(); + for (const sessionId of ['s1', 's2']) { + await repository.create({ + sessionId, + agentName: 'test', + workspaceDir: '/tmp', + runtime: 'pi-agent', + title: sessionId, + }); + } + expect((await repository.get('s1'))?.title).toBe('s1'); + expect((await repository.get('s2'))?.title).toBe('s2'); + writer.db + .update(sessions) + .set({ title: 'updated' }) + .where(eq(sessions.sessionId, 's1')) + .run(); + expect((await repository.get('s1'))?.title).toBe('updated'); + writer.db + .update(sessions) + .set({ columnarVersion: 2 }) + .where(eq(sessions.sessionId, 's1')) + .run(); + expect(await repository.get('s1')).toBeUndefined(); + writer.db + .update(sessions) + .set({ columnarVersion: 3 }) + .where(eq(sessions.sessionId, 's1')) + .run(); + client.close(); + const reopened = createSessionRepository({ db: client.db }); + expect((await reopened.get('s1'))?.title).toBe('updated'); + writer.db.delete(sessions).where(eq(sessions.sessionId, 's1')).run(); + expect(await reopened.get('s1')).toBeUndefined(); + expect(await reopened.get("s2' OR 1=1 --")).toBeUndefined(); + }); + }); + + it('keeps message and turn reads fresh, isolated and ordered after reuse', async () => { + await withDatabase(async (client, writer) => { + const sessionRepository = createSessionRepository({ db: client.db }); + for (const sessionId of ['s1', 's2']) { + await sessionRepository.create({ + sessionId, + agentName: 'test', + workspaceDir: '/tmp', + runtime: 'pi-agent', + }); + } + const repository = createMessageRepository({ db: client.db }); + const writerRepository = createMessageRepository({ db: writer.db }); + expect(await repository.get('s1', 'm1')).toBeUndefined(); + expect(await repository.listTurn('s1', 't1')).toEqual([]); + for (const [sessionId, turnId, msgId] of [ + ['s1', 't1', 'm2'], + ['s1', 't1', 'm1'], + ['s1', 't2', 'm3'], + ['s2', 't1', 'm1'], + ]) { + await writerRepository.upsert({ + sessionId: sessionId!, + turnId, + message: { + msg_id: msgId, + role: 'assistant', + text: `${sessionId}/${msgId}`, + timestamp: 1, + }, + }); + } + expect((await repository.listTurn('s1', 't1')).map((m) => m.msg_id)).toEqual(['m2', 'm1']); + expect((await repository.listTurn('s1', 't2')).map((m) => m.msg_id)).toEqual(['m3']); + expect((await repository.get('s2', 'm1'))?.text).toBe('s2/m1'); + await writerRepository.upsert({ + sessionId: 's1', + turnId: 't2', + message: { + msg_id: 'm1', + role: 'assistant', + text: 'updated', + timestamp: 2, + }, + }); + expect((await repository.get('s1', 'm1'))?.text).toBe('updated'); + expect((await repository.listTurn('s1', 't1')).map((m) => m.msg_id)).toEqual(['m2']); + client.close(); + const reopened = createMessageRepository({ db: client.db }); + expect((await reopened.listTurn('s1', 't2')).map((m) => m.msg_id)).toEqual(['m1', 'm3']); + writer.db.delete(messageRows).where(eq(messageRows.sessionId, 's1')).run(); + expect(await reopened.get('s1', 'm1')).toBeUndefined(); + expect(await reopened.listTurn('s1', 't2')).toEqual([]); + expect(await reopened.get("s2' OR 1=1 --", 'm1')).toBeUndefined(); + expect(await reopened.listTurn('s2', "t1' OR 1=1 --")).toEqual([]); + }); + }); + it('keeps processing reads fresh across sessions, completion and another connection', async () => { await withDatabase(async (client, writer) => { const state = createQueryCollapseState({ db: client.db, nowMs: () => 1 }); @@ -212,6 +318,43 @@ describe('streamed canonical history revisions', () => { }); describe('semantic snapshots', () => { + it('preserves baseline digest bytes and replay byte counts for Unicode and special values', () => { + const sparse = new Array(12); + sparse[2] = undefined; + sparse[10] = '\ud800πŸ™‚'; + Object.defineProperty(sparse, 'extra', { + value: 'δΈ­ζ–‡\udc00', + enumerable: true, + }); + // Golden values from the pre-optimization encoder, including its framing. + const cases = [ + { + value: { + z: 'δΈ­ζ–‡πŸ™‚\ud800', + a: [null, undefined, true, false, NaN, Infinity, -Infinity, -0, 0, 1.25], + ['\ud800']: 'tail\udc00', + ['__proto__']: { '10': true, '2': null }, + }, + fingerprint: 'f4961c05ea68951c93239df0af388afe47951d7e293f9982597bb9313723a1c8', + bytes: 436, + }, + { + value: sparse, + fingerprint: 'b29cce838437e6e8aee783feba4a57bc3c58edd363e5b8abd51525bd07bc1df8', + bytes: 116, + }, + { + value: 'a'.repeat(8191) + 'πŸ™‚δΈ­\ud800' + 'b'.repeat(16385) + '\udc00', + fingerprint: 'b9a02642e610f3591d56337e788f7214c7d9ccbd51edf447bfd5539438b3c11f', + bytes: 24603, + }, + ]; + for (const { value, fingerprint, bytes } of cases) { + const snapshot = captureSemanticSnapshot(value); + expect(snapshot.fingerprint).toBe(fingerprint); + expect(estimateSemanticValueSize(snapshot.value)).toBe(bytes); + } + }); it('detaches and freezes eagerly but hashes only when identity is requested', () => { const hash = vi.spyOn(IncrementalSha256.prototype, 'update'); try { From 18bee224e9e57befa11ed953b61f2f68b7ea98f6 Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Mon, 21 Sep 2026 18:05:52 +0800 Subject: [PATCH 02/10] perf: reuse validated immutable history rows by exact file content --- .../src/infra/file/canonical-history-jsonl.ts | 36 +++- .../local-runtime-v2/src/infra/file/jsonl.ts | 23 ++- .../history/canonical-history-provider.ts | 2 +- .../representation/canonical-history.ts | 15 +- test/history-processing.test.ts | 178 ++++++++++++++++++ 5 files changed, 247 insertions(+), 7 deletions(-) diff --git a/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts b/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts index 8040765b..5157d034 100644 --- a/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts +++ b/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts @@ -24,6 +24,9 @@ export type { } from './canonical-history-source.js'; export type { CanonicalHistoryArtifact } from './canonical-history-artifact.js'; +// Only records decoded from file contents and recursively frozen here are trusted. +const ownedEnvelopeJson = new WeakMap(); + const ENVELOPE_KEYS = new Set([ 'message_id', 'turn_id', @@ -268,6 +271,8 @@ export interface CanonicalHistoryEnvelope { export interface CanonicalHistoryJsonlDataSourceOptions { readonly activePath: string; + /** Internal readers may share frozen records; public readers retain detached values. */ + readonly reuseDecodedRecords?: boolean; readonly onMalformedLine?: (line: JsonlMalformedLine) => void; } @@ -294,6 +299,9 @@ export type CanonicalHistorySequenceInspection = }; export function decodeCanonicalHistoryEnvelope(value: unknown): CanonicalHistoryEnvelope { + if (ownedEnvelopeJson.has(value as CanonicalHistoryEnvelope)) { + return value as CanonicalHistoryEnvelope; + } const envelope = requirePlainRecord(value, 'envelope'); assertExactKeys(envelope, ENVELOPE_KEYS, ['message_id', 'turn_id', 'message'], 'envelope'); assertJsonCompatible(envelope, 'envelope'); @@ -389,6 +397,8 @@ export function inspectCanonicalHistorySequence( * not provide a cross-process writer lock. */ export class CanonicalHistoryJsonlDataSource { + private readonly decodedLines = new Map(); + constructor(private readonly options: CanonicalHistoryJsonlDataSourceOptions) {} async readActive(): Promise { @@ -398,7 +408,7 @@ export class CanonicalHistoryJsonlDataSource { } async readActiveStrict(filePath = this.options.activePath): Promise { - const records = await readStrictEnvelopeFile(filePath); + const records = await this.readEnvelopesStrict(filePath); inspectCanonicalHistorySequence(records); return records; } @@ -407,7 +417,8 @@ export class CanonicalHistoryJsonlDataSource { async readEnvelopesStrict( filePath = this.options.activePath, ): Promise { - return readStrictEnvelopeFile(filePath); + if (!this.options.reuseDecodedRecords) return readStrictEnvelopeFile(filePath); + return readJsonl(filePath, decodeOwnedEnvelope, undefined, this.decodedLines); } async readStrict(filePath = this.options.activePath): Promise { @@ -569,11 +580,30 @@ function revisionOfNormalized(records: readonly CanonicalHistoryEnvelope[]): str const hash = createHash('sha256').update('['); for (let index = 0; index < records.length; index += 1) { if (index > 0) hash.update(','); - hash.update(canonicalJson(records[index]), 'utf8'); + const record = records[index]!; + let serialized = ownedEnvelopeJson.get(record); + if (serialized === undefined) { + serialized = canonicalJson(record); + if (ownedEnvelopeJson.has(record)) ownedEnvelopeJson.set(record, serialized); + } + hash.update(serialized, 'utf8'); } return `sha256:${hash.update(']').digest('hex')}`; } +function decodeOwnedEnvelope(value: unknown): CanonicalHistoryEnvelope { + const record = decodeCanonicalHistoryEnvelope(value); + freezeDecodedJson(record); + ownedEnvelopeJson.set(record, undefined); + return record; +} + +function freezeDecodedJson(value: unknown): void { + if (value === null || typeof value !== 'object' || Object.isFrozen(value)) return; + for (const child of Object.values(value)) freezeDecodedJson(child); + Object.freeze(value); +} + function decodeMessage(value: unknown): CanonicalHistoryMessage { const message = requirePlainRecord(value, 'message'); const role = requireNonEmptyString(message['role'], 'message.role'); diff --git a/packages/local-runtime-v2/src/infra/file/jsonl.ts b/packages/local-runtime-v2/src/infra/file/jsonl.ts index 5f0bdb52..9edc106a 100644 --- a/packages/local-runtime-v2/src/infra/file/jsonl.ts +++ b/packages/local-runtime-v2/src/infra/file/jsonl.ts @@ -36,17 +36,32 @@ export async function readJsonl( filePath: string, decode: (value: unknown) => T, onMalformedLine?: (line: JsonlMalformedLine) => void, + /** Only supply a cache when decoded values are immutable and privately owned. */ + decodedLines?: Map, ): Promise { const contents = await readFile(filePath, 'utf-8'); - if (contents.length === 0) return []; + if (contents.length === 0) { + decodedLines?.clear(); + return []; + } const lines = contents.split('\n'); if (lines.at(-1) === '') lines.pop(); const records: T[] = []; + const nextLines = decodedLines ? new Map() : undefined; + let cachedTextUnits = 0; for (const [index, line] of lines.entries()) { try { if (line.trim().length === 0) throw new Error('blank line'); - records.push(decode(parseJsonLine(line))); + const record = decodedLines?.has(line) + ? decodedLines.get(line)! + : decode(parseJsonLine(line)); + records.push(record); + // Retain only this read's rows, with a bounded text budget per reader. + if (nextLines && !nextLines.has(line) && cachedTextUnits + line.length <= 4 * 1024 * 1024) { + nextLines.set(line, record); + cachedTextUnits += line.length; + } } catch (error) { const malformed = { path: filePath, @@ -57,6 +72,10 @@ export async function readJsonl( onMalformedLine(malformed); } } + if (decodedLines && nextLines) { + decodedLines.clear(); + for (const [line, record] of nextLines) decodedLines.set(line, record); + } return records; } diff --git a/packages/local-runtime-v2/src/service/session-system/messages/history/canonical-history-provider.ts b/packages/local-runtime-v2/src/service/session-system/messages/history/canonical-history-provider.ts index 68be9a7a..853d210f 100644 --- a/packages/local-runtime-v2/src/service/session-system/messages/history/canonical-history-provider.ts +++ b/packages/local-runtime-v2/src/service/session-system/messages/history/canonical-history-provider.ts @@ -204,7 +204,7 @@ async function scanCompactionLineageAllowingMissingParent( export function createSessionSystemCanonicalHistoryProvider( options: SessionSystemCanonicalHistoryProviderOptions, ): SessionSystemCanonicalHistoryProvider { - const files = options.files ?? createCanonicalHistoryFileAdapter(); + const files = options.files ?? createCanonicalHistoryFileAdapter({ reuseDecodedRecords: true }); const inspectionFiles = options.files ?? createCanonicalHistoryFileAdapter(); const nowMs = options.nowMs ?? Date.now; const retryDelay = options.retryDelay ?? ((delayMs: number) => delay(delayMs)); diff --git a/packages/local-runtime-v2/src/service/session-system/sessions/representation/canonical-history.ts b/packages/local-runtime-v2/src/service/session-system/sessions/representation/canonical-history.ts index 74516dfe..6953f016 100644 --- a/packages/local-runtime-v2/src/service/session-system/sessions/representation/canonical-history.ts +++ b/packages/local-runtime-v2/src/service/session-system/sessions/representation/canonical-history.ts @@ -8,6 +8,8 @@ import type { import type { CanonicalHistoryFileAdapter } from './canonical-history-contract.js'; export interface CreateCanonicalHistoryFileAdapterOptions { + /** Internal provider only: returned records remain private and immutable. */ + readonly reuseDecodedRecords?: boolean; readonly onMalformedLine?: CanonicalHistoryJsonlDataSourceOptions['onMalformedLine']; } @@ -18,6 +20,11 @@ export function createCanonicalHistoryFileAdapter( } class JsonlCanonicalHistoryFileAdapter implements CanonicalHistoryFileAdapter { + private cachedSource?: { + path: string; + source: CanonicalHistoryJsonlDataSource; + }; + constructor(private readonly options: CreateCanonicalHistoryFileAdapterOptions) {} async targetExists(path: string) { @@ -63,10 +70,16 @@ class JsonlCanonicalHistoryFileAdapter implements CanonicalHistoryFileAdapter { return this.source(snapshotPath).publishSnapshot(snapshotPath, records); } private source(path: string) { - return new CanonicalHistoryJsonlDataSource({ + if (this.options.reuseDecodedRecords && this.cachedSource?.path === path) { + return this.cachedSource.source; + } + const source = new CanonicalHistoryJsonlDataSource({ activePath: path, + reuseDecodedRecords: this.options.reuseDecodedRecords, ...(this.options.onMalformedLine ? { onMalformedLine: this.options.onMalformedLine } : {}), }); + if (this.options.reuseDecodedRecords) this.cachedSource = { path, source }; + return source; } } async function exists(path: string) { diff --git a/test/history-processing.test.ts b/test/history-processing.test.ts index 4fc8e1a7..f03cd0cd 100644 --- a/test/history-processing.test.ts +++ b/test/history-processing.test.ts @@ -871,3 +871,181 @@ describe('committed history read reuse', () => { expect(readActive).toHaveBeenCalledTimes(1); }); }); + +describe('owned decoded history rows', () => { + async function withReaders( + run: (path: string, reader: CanonicalHistoryJsonlDataSource) => Promise, + ) { + const dir = await mkdtemp(join(tmpdir(), 'mcode-owned-rows-')); + const path = join(dir, 'messages.jsonl'); + try { + await run( + path, + new CanonicalHistoryJsonlDataSource({ + activePath: path, + reuseDecodedRecords: true, + }), + ); + } finally { + await rm(dir, { recursive: true, force: true }); + } + } + const row = (id: string, content: unknown = 'δΈ­ζ–‡πŸ™‚\ud800') => ({ + message_id: `msg-${id}`, + turn_id: `turn-${id}`, + message: { role: 'user', timestamp: 1, content }, + }); + const encode = (rows: unknown[]) => rows.map((value) => JSON.stringify(value)).join('\n') + '\n'; + + it('reuses unchanged rows across append while matching uncached revisions', async () => { + await withReaders(async (path, reader) => { + const records = [row('a'), row('b', { '2': 'two', z: [-0, null, true], __proto__: null })]; + await writeFile(path, encode(records.slice(0, 1))); + const first = await reader.readActiveStrict(); + const firstRevision = canonicalActiveHistoryRevision(first); + await reader.append([records[1]!], first); + const second = await reader.readActiveStrict(); + expect(second[0]).toBe(first[0]); + const plain = await new CanonicalHistoryJsonlDataSource({ + activePath: path, + }).readActiveStrict(); + expect(second).toEqual(plain); + expect(canonicalActiveHistoryRevision(second)).toBe(canonicalActiveHistoryRevision(plain)); + expect(canonicalHistoryRevision(second)).toBe(canonicalHistoryRevision(plain)); + expect(canonicalActiveHistoryRevision(first)).toBe(firstRevision); + expect(Object.isFrozen(second[1]!.message.content)).toBe(true); + }); + }); + + it('observes same-length edits, truncation, deletion and recreation', async () => { + await withReaders(async (path, reader) => { + await writeFile(path, encode([row('a', 'first')])); + const first = await reader.readActiveStrict(); + const revision = canonicalActiveHistoryRevision(first); + await writeFile(path, encode([row('a', 'other')])); + const edited = await reader.readActiveStrict(); + expect(edited[0]!.message.content).toBe('other'); + expect(canonicalActiveHistoryRevision(edited)).not.toBe(revision); + await writeFile(path, ''); + expect(await reader.readActiveStrict()).toEqual([]); + await rm(path); + await expect(reader.readActiveStrict()).rejects.toMatchObject({ + code: 'ENOENT', + }); + await writeFile(path, encode([row('b')])); + expect((await reader.readActiveStrict())[0]!.message_id).toBe('msg-b'); + }); + }); + + it('still rejects corrupt and duplicate rows after warming the cache', async () => { + await withReaders(async (path, reader) => { + const valid = row('a'); + await writeFile(path, encode([valid])); + await reader.readActiveStrict(); + await writeFile(path, encode([valid, valid])); + await expect(reader.readActiveStrict()).rejects.toThrow('duplicate'); + await writeFile(path, encode([valid]) + '{"message_id":"private-payload"\n'); + await expect(reader.readActiveStrict()).rejects.toThrow('invalid JSON'); + await writeFile(path, encode([valid]) + '\n'); + await expect(reader.readActiveStrict()).rejects.toThrow('blank line'); + await writeFile(path, encode([{ ...valid, message: { ...valid.message, timestamp: null } }])); + await expect(reader.readActiveStrict()).rejects.toThrow('timestamp'); + }); + }); + + it('checks pending and settled sequence rules on every cached read', async () => { + await withReaders(async (path, reader) => { + const pending = { + message_id: 'msg-assistant', + turn_id: 'turn-a', + message: { + role: 'assistant', + timestamp: 1, + content: [ + { + type: 'toolCall', + id: 'call-a', + name: 'bash', + arguments: { command: 'pwd' }, + }, + ], + }, + }; + await writeFile(path, encode([pending])); + const records = await reader.readActiveStrict(); + canonicalActiveHistoryRevision(records); + expect(() => canonicalHistoryRevision(records)).toThrow('tool results'); + await expect(reader.readStrict()).rejects.toThrow('tool results'); + expect((await reader.readActiveStrict())[0]).toBe(records[0]); + await writeFile(path, encode([pending, row('b')])); + await expect(reader.readActiveStrict()).rejects.toThrow(); + }); + }); + + it('does not trust caller-frozen records or leak mutable state from ordinary readers', async () => { + await withReaders(async (path) => { + const input = Object.freeze(row('a', { nested: ['original'] })); + const before = canonicalHistoryRevision([input]); + (input.message.content as { nested: string[] }).nested[0] = 'changed'; + expect(canonicalHistoryRevision([input])).not.toBe(before); + await writeFile(path, encode([input])); + const ordinary = new CanonicalHistoryJsonlDataSource({ + activePath: path, + }); + const first = await ordinary.readActiveStrict(); + (first[0]!.message.content as { nested: string[] }).nested[0] = 'caller edit'; + expect((await ordinary.readActiveStrict())[0]!.message.content).toEqual({ + nested: ['changed'], + }); + }); + }); + + it('keeps default provider snapshots mutable and detached from its private rows', async () => { + const dataDir = await mkdtemp(join(tmpdir(), 'mcode-owned-provider-')); + try { + const session: SessionRecord = { + sessionId: 'owned-session', + agentName: 'test', + workspaceDir: dataDir, + runtime: 'pi-agent', + sessionType: 'root', + sessionKind: 'conversation', + archived: false, + status: 'idle', + createdAtMs: 0, + updatedAtMs: 0, + historyRelativeDir: utcSessionHistoryRelativeDir('owned-session', 0), + }; + const provider = createSessionSystemCanonicalHistoryProvider({ + dataDir, + sessions: { get: async () => session }, + }); + const input = { + role: 'user', + timestamp: 1, + content: [{ type: 'text', text: 'original' }], + }; + const committed = await provider.append({ + sessionId: session.sessionId, + turnId: 't1', + reason: 'messageDelta', + messages: [input], + operation: { id: 'a1', kind: 'append' }, + }); + input.content[0]!.text = 'caller input edit'; + (committed.messages[0] as typeof input).content[0]!.text = 'caller output edit'; + const next = await provider.readActive(session.sessionId); + expect((next.messages[0] as typeof input).content[0]!.text).toBe('original'); + expect(next.revision).toBe(committed.revision); + const path = resolveSessionHistoryPaths(dataDir, session).messages; + await writeFile(path, encode([row('external', 'outside')])); + expect((await provider.readActive(session.sessionId)).messages[0]).toMatchObject({ + content: 'outside', + }); + await writeFile(path, '{bad}\n'); + await expect(provider.readActive(session.sessionId)).rejects.toThrow(); + } finally { + await rm(dataDir, { recursive: true, force: true }); + } + }); +}); From 74af093768eff042e3d5a5d7dd726da38c458ddb Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Mon, 21 Sep 2026 18:10:20 +0800 Subject: [PATCH 03/10] fix: detach cached history text from whole file strings --- .../src/infra/file/canonical-history-jsonl.ts | 3 ++- .../local-runtime-v2/src/infra/file/jsonl.ts | 19 +++++++++++++------ 2 files changed, 15 insertions(+), 7 deletions(-) diff --git a/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts b/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts index 5157d034..554394d1 100644 --- a/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts +++ b/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts @@ -6,6 +6,7 @@ import { readJsonl, writeJsonlAtomically, type JsonlMalformedLine, + type JsonlDecodedLine, } from './jsonl.js'; import { decodeCanonicalHistoryArtifact, @@ -397,7 +398,7 @@ export function inspectCanonicalHistorySequence( * not provide a cross-process writer lock. */ export class CanonicalHistoryJsonlDataSource { - private readonly decodedLines = new Map(); + private readonly decodedLines = new Map>(); constructor(private readonly options: CanonicalHistoryJsonlDataSourceOptions) {} diff --git a/packages/local-runtime-v2/src/infra/file/jsonl.ts b/packages/local-runtime-v2/src/infra/file/jsonl.ts index 9edc106a..276b1318 100644 --- a/packages/local-runtime-v2/src/infra/file/jsonl.ts +++ b/packages/local-runtime-v2/src/infra/file/jsonl.ts @@ -32,12 +32,17 @@ export class JsonlAppendCommitUncertainError extends Error { } } +export interface JsonlDecodedLine { + readonly text: string; + readonly value: T; +} + export async function readJsonl( filePath: string, decode: (value: unknown) => T, onMalformedLine?: (line: JsonlMalformedLine) => void, /** Only supply a cache when decoded values are immutable and privately owned. */ - decodedLines?: Map, + decodedLines?: Map>, ): Promise { const contents = await readFile(filePath, 'utf-8'); if (contents.length === 0) { @@ -48,18 +53,20 @@ export async function readJsonl( const lines = contents.split('\n'); if (lines.at(-1) === '') lines.pop(); const records: T[] = []; - const nextLines = decodedLines ? new Map() : undefined; + const nextLines = decodedLines ? new Map>() : undefined; let cachedTextUnits = 0; for (const [index, line] of lines.entries()) { try { if (line.trim().length === 0) throw new Error('blank line'); - const record = decodedLines?.has(line) - ? decodedLines.get(line)! - : decode(parseJsonLine(line)); + const cached = decodedLines?.get(line); + // File slices can keep the entire read buffer alive. Both the retained key + // and parsed values must originate from an independently stored string. + const text = cached?.text ?? (decodedLines ? Buffer.from(line, 'utf8').toString('utf8') : line); + const record = cached ? cached.value : decode(parseJsonLine(text)); records.push(record); // Retain only this read's rows, with a bounded text budget per reader. if (nextLines && !nextLines.has(line) && cachedTextUnits + line.length <= 4 * 1024 * 1024) { - nextLines.set(line, record); + nextLines.set(text, cached ?? { text, value: record }); cachedTextUnits += line.length; } } catch (error) { From f95a1a20c5e93fe976fb6cef91ee65cc548537d9 Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Mon, 21 Sep 2026 18:37:14 +0800 Subject: [PATCH 04/10] perf: reuse verified file prefixes across history and index reads --- .../src/infra/file/canonical-history-jsonl.ts | 6 +- .../local-runtime-v2/src/infra/file/jsonl.ts | 63 +++++++++---------- .../history/canonical-history-provider.ts | 4 +- .../mutation/canonical-history-scanner.ts | 2 +- test/history-processing.test.ts | 37 +++++++++++ 5 files changed, 74 insertions(+), 38 deletions(-) diff --git a/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts b/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts index 554394d1..271677f9 100644 --- a/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts +++ b/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts @@ -6,7 +6,7 @@ import { readJsonl, writeJsonlAtomically, type JsonlMalformedLine, - type JsonlDecodedLine, + type JsonlReadCache, } from './jsonl.js'; import { decodeCanonicalHistoryArtifact, @@ -398,7 +398,7 @@ export function inspectCanonicalHistorySequence( * not provide a cross-process writer lock. */ export class CanonicalHistoryJsonlDataSource { - private readonly decodedLines = new Map>(); + private readonly readCache: JsonlReadCache = { text: '', records: [] }; constructor(private readonly options: CanonicalHistoryJsonlDataSourceOptions) {} @@ -419,7 +419,7 @@ export class CanonicalHistoryJsonlDataSource { filePath = this.options.activePath, ): Promise { if (!this.options.reuseDecodedRecords) return readStrictEnvelopeFile(filePath); - return readJsonl(filePath, decodeOwnedEnvelope, undefined, this.decodedLines); + return readJsonl(filePath, decodeOwnedEnvelope, undefined, this.readCache); } async readStrict(filePath = this.options.activePath): Promise { diff --git a/packages/local-runtime-v2/src/infra/file/jsonl.ts b/packages/local-runtime-v2/src/infra/file/jsonl.ts index 276b1318..8a7b2421 100644 --- a/packages/local-runtime-v2/src/infra/file/jsonl.ts +++ b/packages/local-runtime-v2/src/infra/file/jsonl.ts @@ -32,56 +32,53 @@ export class JsonlAppendCommitUncertainError extends Error { } } -export interface JsonlDecodedLine { - readonly text: string; - readonly value: T; +export interface JsonlReadCache { + text: string; + records: readonly T[]; } export async function readJsonl( filePath: string, decode: (value: unknown) => T, onMalformedLine?: (line: JsonlMalformedLine) => void, - /** Only supply a cache when decoded values are immutable and privately owned. */ - decodedLines?: Map>, + /** Only reuse privately owned immutable values from strict reads. */ + readCache?: JsonlReadCache, ): Promise { const contents = await readFile(filePath, 'utf-8'); - if (contents.length === 0) { - decodedLines?.clear(); - return []; - } - - const lines = contents.split('\n'); + // Tolerant readers must still report every malformed line on every read. + const cache = onMalformedLine ? undefined : readCache; + const reuse = cache && (contents === cache.text || + (cache.text.endsWith('\n') && contents.startsWith(cache.text))); + const records: T[] = reuse ? [...cache.records] : []; + const prefixLength = reuse ? cache.text.length : 0; + const firstLine = records.length; + const lines = contents.slice(prefixLength).split('\n'); if (lines.at(-1) === '') lines.pop(); - const records: T[] = []; - const nextLines = decodedLines ? new Map>() : undefined; - let cachedTextUnits = 0; for (const [index, line] of lines.entries()) { try { if (line.trim().length === 0) throw new Error('blank line'); - const cached = decodedLines?.get(line); - // File slices can keep the entire read buffer alive. Both the retained key - // and parsed values must originate from an independently stored string. - const text = cached?.text ?? (decodedLines ? Buffer.from(line, 'utf8').toString('utf8') : line); - const record = cached ? cached.value : decode(parseJsonLine(text)); - records.push(record); - // Retain only this read's rows, with a bounded text budget per reader. - if (nextLines && !nextLines.has(line) && cachedTextUnits + line.length <= 4 * 1024 * 1024) { - nextLines.set(text, cached ?? { text, value: record }); - cachedTextUnits += line.length; - } + // Parsed values must not retain a slice of an older whole-file string. + const text = cache ? Buffer.from(line, 'utf8').toString('utf8') : line; + records.push(decode(parseJsonLine(text))); } catch (error) { - const malformed = { - path: filePath, - lineNo: index + 1, - reason: errorReason(error), - }; + const malformed = { path: filePath, lineNo: firstLine + index + 1, reason: errorReason(error) }; if (!onMalformedLine) throw malformedLineError(malformed); onMalformedLine(malformed); } } - if (decodedLines && nextLines) { - decodedLines.clear(); - for (const [line, record] of nextLines) decodedLines.set(line, record); + if (cache) { + const limit = 4 * 1024 * 1024; + if (contents.length <= limit) { + cache.text = contents; + cache.records = records.slice(); + } else { + // Keep a bounded complete-line prefix. It remains reusable when the file + // grows beyond the budget, without cycling through and evicting every row. + const end = contents.lastIndexOf('\n', limit - 1) + 1; + const text = Buffer.from(contents.slice(0, end), 'utf8').toString('utf8'); + cache.text = text; + cache.records = records.slice(0, text.split('\n').length - 1); + } } return records; } diff --git a/packages/local-runtime-v2/src/service/session-system/messages/history/canonical-history-provider.ts b/packages/local-runtime-v2/src/service/session-system/messages/history/canonical-history-provider.ts index 853d210f..f95f241e 100644 --- a/packages/local-runtime-v2/src/service/session-system/messages/history/canonical-history-provider.ts +++ b/packages/local-runtime-v2/src/service/session-system/messages/history/canonical-history-provider.ts @@ -384,7 +384,9 @@ export function createSessionSystemCanonicalHistoryProvider( activeGeneration: activeGeneration(active), activeRevision: canonicalActiveHistoryRevision(active), }, - (scannerPaths) => scanCanonicalHistoryArtifacts(scannerPaths), + // Only the default reader owns reusable immutable records. Preserve the + // independent on-disk scanner for externally supplied adapters. + (scannerPaths) => scanCanonicalHistoryArtifacts(scannerPaths, options.files ? undefined : files), { activePath: paths.messages, snapshotsPath: paths.snapshots, sessionId }, ); } catch (error) { diff --git a/packages/local-runtime-v2/src/service/session-system/messages/history/mutation/canonical-history-scanner.ts b/packages/local-runtime-v2/src/service/session-system/messages/history/mutation/canonical-history-scanner.ts index 1292cc61..05def70e 100644 --- a/packages/local-runtime-v2/src/service/session-system/messages/history/mutation/canonical-history-scanner.ts +++ b/packages/local-runtime-v2/src/service/session-system/messages/history/mutation/canonical-history-scanner.ts @@ -74,8 +74,8 @@ export type HistoryScannerErrorCode = export async function scanCanonicalHistoryArtifacts( paths: CanonicalHistoryScannerPaths, + files = createCanonicalHistoryFileAdapter(), ): Promise { - const files = createCanonicalHistoryFileAdapter(); const activeBase = await readArtifact( files.readActiveStrict(paths.activePath), paths.activePath, diff --git a/test/history-processing.test.ts b/test/history-processing.test.ts index f03cd0cd..235112ce 100644 --- a/test/history-processing.test.ts +++ b/test/history-processing.test.ts @@ -917,6 +917,43 @@ describe('owned decoded history rows', () => { }); }); + it('does not let mutations of a returned array poison the cached prefix', async () => { + await withReaders(async (path, reader) => { + await writeFile(path, encode([row('a'), row('b')])); + const first = await reader.readActiveStrict(); + first.pop(); + first[0] = row('replacement'); + expect((await reader.readActiveStrict()).map(value => value.message_id)).toEqual(['msg-a', 'msg-b']); + }); + }); + + it('reparses an unterminated final line and preserves the absolute error line', async () => { + await withReaders(async (path, reader) => { + const text = JSON.stringify(row('a')); + await writeFile(path, text); + await reader.readActiveStrict(); + await writeFile(path, text + 'broken'); + await expect(reader.readActiveStrict()).rejects.toThrow('line 1'); + await writeFile(path, text + '\n' + JSON.stringify(row('b')) + '\n'); + await reader.readActiveStrict(); + await writeFile(path, text + '\n' + JSON.stringify(row('b')) + '\n{broken}\n'); + await expect(reader.readActiveStrict()).rejects.toThrow('line 3'); + }); + }); + + it('keeps a reusable complete prefix when the history exceeds the cache budget', async () => { + await withReaders(async (path, reader) => { + const entries = [row('a', 'a'.repeat(2 * 1024 * 1024)), row('b', 'b'.repeat(2 * 1024 * 1024)), row('c')]; + await writeFile(path, encode(entries)); + const first = await reader.readActiveStrict(); + const next = await reader.readActiveStrict(); + expect(next[0]).toBe(first[0]); + expect(next).toEqual(first); + await writeFile(path, encode([entries[0], entries[1], row('d')])); + expect((await reader.readActiveStrict())[2]!.message_id).toBe('msg-d'); + }); + }); + it('observes same-length edits, truncation, deletion and recreation', async () => { await withReaders(async (path, reader) => { await writeFile(path, encode([row('a', 'first')])); From a1e0521d2ac683b29bcebb0e5659a810d8dccadb Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Mon, 21 Sep 2026 18:49:45 +0800 Subject: [PATCH 05/10] perf: reuse validation and revisions of owned history snapshots --- .../src/infra/file/canonical-history-jsonl.ts | 44 ++++++++++++++----- .../local-runtime-v2/src/infra/file/jsonl.ts | 8 ++-- .../history/canonical-history-provider.ts | 31 +++++++++---- test/history-processing.test.ts | 17 ++++--- 4 files changed, 73 insertions(+), 27 deletions(-) diff --git a/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts b/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts index 271677f9..9e07ebea 100644 --- a/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts +++ b/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts @@ -27,6 +27,12 @@ export type { CanonicalHistoryArtifact } from './canonical-history-artifact.js'; // Only records decoded from file contents and recursively frozen here are trusted. const ownedEnvelopeJson = new WeakMap(); +const ownedRecordArrays = new WeakSet(); +const ownedRevisions = new WeakMap(); +const ownedSequences = new WeakMap< + readonly CanonicalHistoryEnvelope[], + CanonicalHistorySequenceInspection +>(); const ENVELOPE_KEYS = new Set([ 'message_id', @@ -367,6 +373,8 @@ export function assertCanonicalHistorySequence(records: readonly CanonicalHistor export function inspectCanonicalHistorySequence( records: readonly CanonicalHistoryEnvelope[], ): CanonicalHistorySequenceInspection { + const cached = ownedSequences.get(records); + if (cached) return copyInspection(cached); const state: HistorySequenceState = { messageIds: new Set(), toolCallIds: new Set(), @@ -380,14 +388,22 @@ export function inspectCanonicalHistorySequence( validateSequenceRecord(envelope, index, state); } - if (state.pendingToolCallIds && state.pendingToolCallIds.size > 0) { - return { - status: 'pending-tool-results', - settledPrefixLength: state.pendingToolCallStartIndex ?? records.length, - pendingToolCallIds: [...state.pendingToolCallIds], - }; - } - return { status: 'settled' }; + const inspection: CanonicalHistorySequenceInspection = + state.pendingToolCallIds && state.pendingToolCallIds.size > 0 + ? { + status: 'pending-tool-results', + settledPrefixLength: state.pendingToolCallStartIndex ?? records.length, + pendingToolCallIds: [...state.pendingToolCallIds], + } + : { status: 'settled' }; + if (ownedRecordArrays.has(records)) ownedSequences.set(records, copyInspection(inspection)); + return inspection; +} + +function copyInspection(value: CanonicalHistorySequenceInspection): CanonicalHistorySequenceInspection { + return value.status === 'settled' + ? { status: 'settled' } + : { ...value, pendingToolCallIds: [...value.pendingToolCallIds] }; } /** @@ -419,7 +435,10 @@ export class CanonicalHistoryJsonlDataSource { filePath = this.options.activePath, ): Promise { if (!this.options.reuseDecodedRecords) return readStrictEnvelopeFile(filePath); - return readJsonl(filePath, decodeOwnedEnvelope, undefined, this.readCache); + const records = await readJsonl(filePath, decodeOwnedEnvelope, undefined, this.readCache); + Object.freeze(records); + ownedRecordArrays.add(records); + return records; } async readStrict(filePath = this.options.activePath): Promise { @@ -567,6 +586,7 @@ function normalizeActiveRecords( } function decodeRecords(records: readonly CanonicalHistoryEnvelope[]): CanonicalHistoryEnvelope[] { + if (ownedRecordArrays.has(records)) return records as CanonicalHistoryEnvelope[]; const decoded: CanonicalHistoryEnvelope[] = []; for (let index = 0; index < records.length; index += 1) { if (!Object.hasOwn(records, index)) invalidEnvelope(`records[${String(index)}] is sparse`); @@ -576,6 +596,8 @@ function decodeRecords(records: readonly CanonicalHistoryEnvelope[]): CanonicalH } function revisionOfNormalized(records: readonly CanonicalHistoryEnvelope[]): string { + const cached = ownedRevisions.get(records); + if (cached !== undefined) return cached; // Preserve the canonical JSON array bytes without building a sorted copy and // serialized string of the entire history at once. const hash = createHash('sha256').update('['); @@ -589,7 +611,9 @@ function revisionOfNormalized(records: readonly CanonicalHistoryEnvelope[]): str } hash.update(serialized, 'utf8'); } - return `sha256:${hash.update(']').digest('hex')}`; + const revision = `sha256:${hash.update(']').digest('hex')}`; + if (ownedRecordArrays.has(records)) ownedRevisions.set(records, revision); + return revision; } function decodeOwnedEnvelope(value: unknown): CanonicalHistoryEnvelope { diff --git a/packages/local-runtime-v2/src/infra/file/jsonl.ts b/packages/local-runtime-v2/src/infra/file/jsonl.ts index 8a7b2421..18e4d440 100644 --- a/packages/local-runtime-v2/src/infra/file/jsonl.ts +++ b/packages/local-runtime-v2/src/infra/file/jsonl.ts @@ -41,12 +41,13 @@ export async function readJsonl( filePath: string, decode: (value: unknown) => T, onMalformedLine?: (line: JsonlMalformedLine) => void, - /** Only reuse privately owned immutable values from strict reads. */ + /** Strict private reads return immutable arrays when a cache is supplied. */ readCache?: JsonlReadCache, ): Promise { const contents = await readFile(filePath, 'utf-8'); // Tolerant readers must still report every malformed line on every read. const cache = onMalformedLine ? undefined : readCache; + if (cache && contents === cache.text) return cache.records as T[]; const reuse = cache && (contents === cache.text || (cache.text.endsWith('\n') && contents.startsWith(cache.text))); const records: T[] = reuse ? [...cache.records] : []; @@ -70,16 +71,17 @@ export async function readJsonl( const limit = 4 * 1024 * 1024; if (contents.length <= limit) { cache.text = contents; - cache.records = records.slice(); + cache.records = Object.freeze(records); } else { // Keep a bounded complete-line prefix. It remains reusable when the file // grows beyond the budget, without cycling through and evicting every row. const end = contents.lastIndexOf('\n', limit - 1) + 1; const text = Buffer.from(contents.slice(0, end), 'utf8').toString('utf8'); cache.text = text; - cache.records = records.slice(0, text.split('\n').length - 1); + cache.records = Object.freeze(records.slice(0, text.split('\n').length - 1)); } } + if (cache) Object.freeze(records); return records; } diff --git a/packages/local-runtime-v2/src/service/session-system/messages/history/canonical-history-provider.ts b/packages/local-runtime-v2/src/service/session-system/messages/history/canonical-history-provider.ts index f95f241e..8b180162 100644 --- a/packages/local-runtime-v2/src/service/session-system/messages/history/canonical-history-provider.ts +++ b/packages/local-runtime-v2/src/service/session-system/messages/history/canonical-history-provider.ts @@ -213,6 +213,7 @@ export function createSessionSystemCanonicalHistoryProvider( options.locations ?? createSessionHistoryLocationResolver({ dataDir: options.dataDir, sessions: options.sessions }); const indexes = new Map(); + const verifiedRecovery = new WeakMap(); return { read: (sessionId) => inLane(sessionId, () => readSnapshot(sessionId, false)), @@ -323,17 +324,29 @@ export function createSessionSystemCanonicalHistoryProvider( ): Promise { const paths = await ensureInitialized(sessionId); const decoded = await files.readEnvelopesStrict(paths.messages); - const recovery = repairCanonicalHistory(decoded, { allowPendingToolCallTail }); - let records = recovery.records; - if (recovery.issues.length > 0) { - if (allowPendingToolCallTail) { - await files.replaceActive(paths.messages, records); - records = await files.readActiveStrict(paths.messages); + const recoveryMode = allowPendingToolCallTail ? 2 : 1; + const reusable = options.files === undefined; + const verifiedModes = reusable ? (verifiedRecovery.get(decoded) ?? 0) : 0; + let records = decoded; + if ((verifiedModes & recoveryMode) === 0) { + const recovery = repairCanonicalHistory(decoded, { allowPendingToolCallTail }); + if (recovery.issues.length > 0) { + records = recovery.records; + if (allowPendingToolCallTail) { + await files.replaceActive(paths.messages, records); + records = await files.readActiveStrict(paths.messages); + } else { + await files.replace(paths.messages, records); + records = await files.readStrict(paths.messages); + } + options.activity?.notify(sessionId); + } else if (reusable) { + // Only the private reader produces immutable arrays. The two recovery + // modes remain independent, so a pending tail is never treated as settled. + verifiedRecovery.set(decoded, verifiedModes | recoveryMode); } else { - await files.replace(paths.messages, records); - records = await files.readStrict(paths.messages); + records = recovery.records; } - options.activity?.notify(sessionId); } await syncIndex(sessionId, paths, true, records); return historySnapshot(records, allowPendingToolCallTail); diff --git a/test/history-processing.test.ts b/test/history-processing.test.ts index 235112ce..b8d47ef5 100644 --- a/test/history-processing.test.ts +++ b/test/history-processing.test.ts @@ -4,6 +4,7 @@ import { canonicalHistoryRevision, CanonicalHistoryJsonlDataSource, decodeCanonicalHistoryEnvelope, + inspectCanonicalHistorySequence, } from '../packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.js'; import { canonicalJson } from '../packages/local-runtime-v2/src/infra/file/canonical-history-json-value.js'; import { mkdtemp, rm, writeFile } from 'node:fs/promises'; @@ -917,12 +918,12 @@ describe('owned decoded history rows', () => { }); }); - it('does not let mutations of a returned array poison the cached prefix', async () => { + it('keeps private cached arrays immutable without exposing their state', async () => { await withReaders(async (path, reader) => { await writeFile(path, encode([row('a'), row('b')])); const first = await reader.readActiveStrict(); - first.pop(); - first[0] = row('replacement'); + expect(() => first.pop()).toThrow(); + expect(() => { first[0] = row('replacement'); }).toThrow(); expect((await reader.readActiveStrict()).map(value => value.message_id)).toEqual(['msg-a', 'msg-b']); }); }); @@ -1011,6 +1012,11 @@ describe('owned decoded history rows', () => { await writeFile(path, encode([pending])); const records = await reader.readActiveStrict(); canonicalActiveHistoryRevision(records); + const inspection = inspectCanonicalHistorySequence(records); + if (inspection.status === 'pending-tool-results') { + (inspection.pendingToolCallIds as string[]).pop(); + } + expect(inspectCanonicalHistorySequence(records)).toMatchObject({ pendingToolCallIds: ['call-a'] }); expect(() => canonicalHistoryRevision(records)).toThrow('tool results'); await expect(reader.readStrict()).rejects.toThrow('tool results'); expect((await reader.readActiveStrict())[0]).toBe(records[0]); @@ -1022,9 +1028,10 @@ describe('owned decoded history rows', () => { it('does not trust caller-frozen records or leak mutable state from ordinary readers', async () => { await withReaders(async (path) => { const input = Object.freeze(row('a', { nested: ['original'] })); - const before = canonicalHistoryRevision([input]); + const externallyFrozenArray = Object.freeze([input]); + const before = canonicalHistoryRevision(externallyFrozenArray); (input.message.content as { nested: string[] }).nested[0] = 'changed'; - expect(canonicalHistoryRevision([input])).not.toBe(before); + expect(canonicalHistoryRevision(externallyFrozenArray)).not.toBe(before); await writeFile(path, encode([input])); const ordinary = new CanonicalHistoryJsonlDataSource({ activePath: path, From d1e9511b5e3fb6f73a6abe7eea2dbe291e54dae0 Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Mon, 21 Sep 2026 18:53:05 +0800 Subject: [PATCH 06/10] perf: compare history bytes before decoding appended records --- .../src/infra/file/canonical-history-jsonl.ts | 2 +- .../local-runtime-v2/src/infra/file/jsonl.ts | 82 ++++++++++++------- test/history-processing.test.ts | 17 ++++ 3 files changed, 72 insertions(+), 29 deletions(-) diff --git a/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts b/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts index 9e07ebea..7c567fcb 100644 --- a/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts +++ b/packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.ts @@ -414,7 +414,7 @@ function copyInspection(value: CanonicalHistorySequenceInspection): CanonicalHis * not provide a cross-process writer lock. */ export class CanonicalHistoryJsonlDataSource { - private readonly readCache: JsonlReadCache = { text: '', records: [] }; + private readonly readCache: JsonlReadCache = { bytes: Buffer.alloc(0), records: [] }; constructor(private readonly options: CanonicalHistoryJsonlDataSourceOptions) {} diff --git a/packages/local-runtime-v2/src/infra/file/jsonl.ts b/packages/local-runtime-v2/src/infra/file/jsonl.ts index 18e4d440..47ac7974 100644 --- a/packages/local-runtime-v2/src/infra/file/jsonl.ts +++ b/packages/local-runtime-v2/src/infra/file/jsonl.ts @@ -33,7 +33,7 @@ export class JsonlAppendCommitUncertainError extends Error { } export interface JsonlReadCache { - text: string; + bytes: Buffer; records: readonly T[]; } @@ -44,45 +44,71 @@ export async function readJsonl( /** Strict private reads return immutable arrays when a cache is supplied. */ readCache?: JsonlReadCache, ): Promise { - const contents = await readFile(filePath, 'utf-8'); // Tolerant readers must still report every malformed line on every read. - const cache = onMalformedLine ? undefined : readCache; - if (cache && contents === cache.text) return cache.records as T[]; - const reuse = cache && (contents === cache.text || - (cache.text.endsWith('\n') && contents.startsWith(cache.text))); - const records: T[] = reuse ? [...cache.records] : []; - const prefixLength = reuse ? cache.text.length : 0; - const firstLine = records.length; - const lines = contents.slice(prefixLength).split('\n'); + if (readCache && !onMalformedLine) return readCachedJsonl(filePath, decode, readCache); + const contents = await readFile(filePath, 'utf-8'); + const records: T[] = []; + const lines = contents.split('\n'); if (lines.at(-1) === '') lines.pop(); for (const [index, line] of lines.entries()) { try { if (line.trim().length === 0) throw new Error('blank line'); - // Parsed values must not retain a slice of an older whole-file string. - const text = cache ? Buffer.from(line, 'utf8').toString('utf8') : line; - records.push(decode(parseJsonLine(text))); + records.push(decode(parseJsonLine(line))); } catch (error) { - const malformed = { path: filePath, lineNo: firstLine + index + 1, reason: errorReason(error) }; + const malformed = { path: filePath, lineNo: index + 1, reason: errorReason(error) }; if (!onMalformedLine) throw malformedLineError(malformed); onMalformedLine(malformed); } } - if (cache) { - const limit = 4 * 1024 * 1024; - if (contents.length <= limit) { - cache.text = contents; - cache.records = Object.freeze(records); - } else { - // Keep a bounded complete-line prefix. It remains reusable when the file - // grows beyond the budget, without cycling through and evicting every row. - const end = contents.lastIndexOf('\n', limit - 1) + 1; - const text = Buffer.from(contents.slice(0, end), 'utf8').toString('utf8'); - cache.text = text; - cache.records = Object.freeze(records.slice(0, text.split('\n').length - 1)); + return records; +} + +async function readCachedJsonl( + filePath: string, + decode: (value: unknown) => T, + cache: JsonlReadCache, +): Promise { + // Always read fresh bytes: timestamps and file size cannot prove an unchanged + // prefix. Compare before decoding to avoid allocating a whole-history string. + const bytes = await readFile(filePath); + if (bytes.equals(cache.bytes)) return cache.records as T[]; + const reuse = cache.bytes.at(-1) === 10 && + bytes.length >= cache.bytes.length && + bytes.subarray(0, cache.bytes.length).equals(cache.bytes); + const records: T[] = reuse ? [...cache.records] : []; + let offset = reuse ? cache.bytes.length : 0; + const limit = 4 * 1024 * 1024; + let retainedEnd = offset; + let retainedRecords = records.length; + while (offset < bytes.length) { + const newline = bytes.indexOf(10, offset); + const end = newline === -1 ? bytes.length : newline; + try { + // Decode each line separately so parsed values cannot retain a string + // slice of an older whole file. + const line = bytes.toString('utf8', offset, end); + if (line.trim().length === 0) throw new Error('blank line'); + records.push(decode(parseJsonLine(line))); + } catch (error) { + throw malformedLineError({ + path: filePath, lineNo: records.length + 1, reason: errorReason(error), + }); + } + offset = newline === -1 ? bytes.length : newline + 1; + if (newline !== -1 && offset <= limit) { + retainedEnd = offset; + retainedRecords = records.length; } } - if (cache) Object.freeze(records); - return records; + if (bytes.length <= limit) { + cache.bytes = bytes; + cache.records = Object.freeze(records); + } else { + // Copy the bounded prefix so it cannot retain the entire file's buffer. + cache.bytes = Buffer.from(bytes.subarray(0, retainedEnd)); + cache.records = Object.freeze(records.slice(0, retainedRecords)); + } + return Object.freeze(records) as T[]; } function parseJsonLine(line: string): unknown { diff --git a/test/history-processing.test.ts b/test/history-processing.test.ts index b8d47ef5..4a2ccb39 100644 --- a/test/history-processing.test.ts +++ b/test/history-processing.test.ts @@ -918,6 +918,23 @@ describe('owned decoded history rows', () => { }); }); + it('matches uncached UTF-8 replacement and CRLF decoding across appended rows', async () => { + await withReaders(async (path, reader) => { + const first = Buffer.from(JSON.stringify(row('a', 'δΈ­ζ–‡πŸ™‚X')) + '\r\n'); + first[first.indexOf(Buffer.from('X'))] = 0xff; + await writeFile(path, first); + const initial = await reader.readActiveStrict(); + const next = Buffer.concat([first, Buffer.from(JSON.stringify(row('b', 'ε°Ύιƒ¨πŸ™‚')))]); + await writeFile(path, next); + const cached = await reader.readActiveStrict(); + const plain = await new CanonicalHistoryJsonlDataSource({ activePath: path }).readActiveStrict(); + expect(cached[0]).toBe(initial[0]); + expect(cached).toEqual(plain); + expect(canonicalActiveHistoryRevision(cached)).toBe(canonicalActiveHistoryRevision(plain)); + expect(cached[0]!.message.content).toBe('δΈ­ζ–‡πŸ™‚\ufffd'); + }); + }); + it('keeps private cached arrays immutable without exposing their state', async () => { await withReaders(async (path, reader) => { await writeFile(path, encode([row('a'), row('b')])); From 37d6c68c3d606b8d1403d596fcba9bc1d9288b73 Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Mon, 21 Sep 2026 19:03:07 +0800 Subject: [PATCH 07/10] perf: load pi runtime APIs through focused package exports --- .../outbound-message-normalizer.ts | 2 +- .../agent-core/src/pi-turn-runner/tools.ts | 2 +- .../context-manager/src/count-tokens-body.ts | 2 +- .../agent-tools/src/desktop/local-pi-tools.ts | 4 ++-- .../agent-tools/src/shared/read-guards.ts | 2 +- .../src/service/background-bash/executor.ts | 2 +- .../service/model-system/codex-oauth.test.ts | 2 +- .../src/service/model-system/codex-oauth.ts | 2 +- .../src/service/sandbox/deferred-port.ts | 2 +- .../compaction/automatic-context-compactor.ts | 2 +- .../compaction/local-context-compactor.ts | 2 +- .../src/api/host-turn-service-factories.ts | 2 +- .../src/api/hosted-agent-capabilities.ts | 2 +- .../context/messages-count-tokens-messages.ts | 2 +- .../src/context/token-estimator.ts | 2 +- .../test/unit/child-bash-lifecycle.test.ts | 3 ++- packages/tui/src/host/bash-command.ts | 2 +- packages/tui/src/host/image-preview-worker.ts | 2 +- .../packages/coding-agent/package.json | 24 +++++++++++++++++++ tsconfig.standalone.json | 18 ++++++++++++++ 20 files changed, 62 insertions(+), 19 deletions(-) diff --git a/packages/agent-core/src/pi-turn-runner/outbound-message-normalizer.ts b/packages/agent-core/src/pi-turn-runner/outbound-message-normalizer.ts index be83d2d8..ceef328b 100644 --- a/packages/agent-core/src/pi-turn-runner/outbound-message-normalizer.ts +++ b/packages/agent-core/src/pi-turn-runner/outbound-message-normalizer.ts @@ -8,7 +8,7 @@ import type { ThinkingContent, UserMessage, } from '@earendil-works/pi-ai'; -import { convertToLlm } from '@earendil-works/pi-coding-agent'; +import { convertToLlm } from '@earendil-works/pi-coding-agent/messages'; import { imageDimensions } from './image-dimensions.js'; diff --git a/packages/agent-core/src/pi-turn-runner/tools.ts b/packages/agent-core/src/pi-turn-runner/tools.ts index 661f80ff..0ce57348 100644 --- a/packages/agent-core/src/pi-turn-runner/tools.ts +++ b/packages/agent-core/src/pi-turn-runner/tools.ts @@ -11,7 +11,7 @@ import { createEditTool, createReadTool, createWriteTool, -} from '@earendil-works/pi-coding-agent'; +} from '@earendil-works/pi-coding-agent/tools'; import { createBashEnvSpawnHook, resolveBashEnvPolicy } from '../bash-subprocess-env.js'; import type { TSchema } from '@sinclair/typebox'; import type { RuntimeTool, ToolExecutionContext } from '../tools/index.js'; diff --git a/packages/agent-modules/context-manager/src/count-tokens-body.ts b/packages/agent-modules/context-manager/src/count-tokens-body.ts index c3b4c196..2332d759 100644 --- a/packages/agent-modules/context-manager/src/count-tokens-body.ts +++ b/packages/agent-modules/context-manager/src/count-tokens-body.ts @@ -17,7 +17,7 @@ * absorbed by the manager's `safetyMarginTokens` / `reserveTokens` headroom. */ -import { convertToLlm } from '@earendil-works/pi-coding-agent'; +import { convertToLlm } from '@earendil-works/pi-coding-agent/messages'; import type { AgentMessage } from '@earendil-works/pi-agent-core'; import type { Api, Model, Tool } from '@earendil-works/pi-ai'; diff --git a/packages/agent-tools/src/desktop/local-pi-tools.ts b/packages/agent-tools/src/desktop/local-pi-tools.ts index cfc5bc42..81c0c3ed 100644 --- a/packages/agent-tools/src/desktop/local-pi-tools.ts +++ b/packages/agent-tools/src/desktop/local-pi-tools.ts @@ -13,8 +13,8 @@ import { createEditTool, createReadTool, createWriteTool, - getShellConfig, -} from '@earendil-works/pi-coding-agent'; +} from '@earendil-works/pi-coding-agent/tools'; +import { getShellConfig } from '@earendil-works/pi-coding-agent/shell'; import type { AgentTool, AgentToolResult } from '@earendil-works/pi-agent-core'; import { access } from 'node:fs/promises'; import { isAbsolute, resolve as resolvePath } from 'node:path'; diff --git a/packages/agent-tools/src/shared/read-guards.ts b/packages/agent-tools/src/shared/read-guards.ts index 9da6e2da..356e5495 100644 --- a/packages/agent-tools/src/shared/read-guards.ts +++ b/packages/agent-tools/src/shared/read-guards.ts @@ -391,7 +391,7 @@ const PNG_SIGNATURE = [0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]; /** * Replicated from `third_party/pi-mono/packages/coding-agent/src/utils/mime.ts` * `detectSupportedImageMimeType` β€” the function is NOT exported from - * `@earendil-works/pi-coding-agent` (package exposes only the root entry), + * `@earendil-works/pi-coding-agent`, * so we keep a byte-exact copy here. KEEP IN SYNC on pi upstream syncs: * the whole point of this replica is that the exemption face equals pi's * image-branch acceptance face (JPEG minus JPEG-LS, PNG minus APNG, GIF, diff --git a/packages/local-runtime-v2/src/service/background-bash/executor.ts b/packages/local-runtime-v2/src/service/background-bash/executor.ts index b489b5c8..a80dabb6 100644 --- a/packages/local-runtime-v2/src/service/background-bash/executor.ts +++ b/packages/local-runtime-v2/src/service/background-bash/executor.ts @@ -1,6 +1,6 @@ import { StringDecoder } from 'node:string_decoder'; -import { createBashTool, type BashOperations } from '@earendil-works/pi-coding-agent'; +import { createBashTool, type BashOperations } from '@earendil-works/pi-coding-agent/tools'; import { createBashEnvSpawnHook, type BashEnvPolicy, diff --git a/packages/local-runtime-v2/src/service/model-system/codex-oauth.test.ts b/packages/local-runtime-v2/src/service/model-system/codex-oauth.test.ts index 4b570047..44d1c111 100644 --- a/packages/local-runtime-v2/src/service/model-system/codex-oauth.test.ts +++ b/packages/local-runtime-v2/src/service/model-system/codex-oauth.test.ts @@ -1,5 +1,5 @@ import { join } from "node:path"; -import { AuthStorage } from "@earendil-works/pi-coding-agent"; +import { AuthStorage } from '@earendil-works/pi-coding-agent/auth-storage'; import { afterEach, describe, expect, it, vi } from "vitest"; import type { diff --git a/packages/local-runtime-v2/src/service/model-system/codex-oauth.ts b/packages/local-runtime-v2/src/service/model-system/codex-oauth.ts index 118e8898..22bae0ce 100644 --- a/packages/local-runtime-v2/src/service/model-system/codex-oauth.ts +++ b/packages/local-runtime-v2/src/service/model-system/codex-oauth.ts @@ -1,7 +1,7 @@ import { randomUUID } from 'node:crypto'; import { join } from 'node:path'; -import { AuthStorage } from '@earendil-works/pi-coding-agent'; +import { AuthStorage } from '@earendil-works/pi-coding-agent/auth-storage'; import type { LocalByokConfigDraft, diff --git a/packages/local-runtime-v2/src/service/sandbox/deferred-port.ts b/packages/local-runtime-v2/src/service/sandbox/deferred-port.ts index 3cd03567..d0ce856a 100644 --- a/packages/local-runtime-v2/src/service/sandbox/deferred-port.ts +++ b/packages/local-runtime-v2/src/service/sandbox/deferred-port.ts @@ -1,4 +1,4 @@ -import { createLocalBashOperations, type BashOperations } from '@earendil-works/pi-coding-agent'; +import { createLocalBashOperations, type BashOperations } from '@earendil-works/pi-coding-agent/tools'; import type { BashEnvPolicy } from '@mavis/agent-core/bash-subprocess-env'; import type { LocalSandboxBashOperationsFactory } from '@mavis/agent-tools/desktop'; diff --git a/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.ts b/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.ts index 3faa4748..e78182af 100644 --- a/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.ts +++ b/packages/local-runtime-v2/src/service/turn-system/compaction/automatic-context-compactor.ts @@ -1,7 +1,7 @@ import { randomUUID } from 'node:crypto'; import type { AgentMessage } from '@earendil-works/pi-agent-core'; -import { DEFAULT_COMPACTION_SETTINGS } from '@earendil-works/pi-coding-agent'; +import { DEFAULT_COMPACTION_SETTINGS } from '@earendil-works/pi-coding-agent/compaction'; import type { PiBeforeLlmCallHook, PiTurnRunnerLogger } from '@mavis/agent-core/pi-turn-runner'; import { resolveCompactionTokenBudget } from '@mavis/context-manager'; diff --git a/packages/local-runtime-v2/src/service/turn-system/compaction/local-context-compactor.ts b/packages/local-runtime-v2/src/service/turn-system/compaction/local-context-compactor.ts index 1c060ec2..fc1e02fd 100644 --- a/packages/local-runtime-v2/src/service/turn-system/compaction/local-context-compactor.ts +++ b/packages/local-runtime-v2/src/service/turn-system/compaction/local-context-compactor.ts @@ -1,6 +1,6 @@ import { randomUUID } from 'node:crypto'; -import { DEFAULT_COMPACTION_SETTINGS } from '@earendil-works/pi-coding-agent'; +import { DEFAULT_COMPACTION_SETTINGS } from '@earendil-works/pi-coding-agent/compaction'; import type { PiTurnRunnerLogger } from '@mavis/agent-core/pi-turn-runner'; import { resolveCompactionTokenBudget } from '@mavis/context-manager'; diff --git a/packages/local-runtime/src/api/host-turn-service-factories.ts b/packages/local-runtime/src/api/host-turn-service-factories.ts index 84f9a801..7a566184 100644 --- a/packages/local-runtime/src/api/host-turn-service-factories.ts +++ b/packages/local-runtime/src/api/host-turn-service-factories.ts @@ -1,5 +1,5 @@ import { join } from 'node:path'; -import { AuthStorage } from '@earendil-works/pi-coding-agent'; +import { AuthStorage } from '@earendil-works/pi-coding-agent/auth-storage'; import type { LocalRuntimeConfig } from '../config/types.js'; import { LocalModelResolver, diff --git a/packages/local-runtime/src/api/hosted-agent-capabilities.ts b/packages/local-runtime/src/api/hosted-agent-capabilities.ts index 16740f5f..5ebf5382 100644 --- a/packages/local-runtime/src/api/hosted-agent-capabilities.ts +++ b/packages/local-runtime/src/api/hosted-agent-capabilities.ts @@ -1,6 +1,6 @@ import { join } from 'node:path'; -import { AuthStorage } from '@earendil-works/pi-coding-agent'; +import { AuthStorage } from '@earendil-works/pi-coding-agent/auth-storage'; import type { PiLLMRequestFailureHook, PiLLMRequestObserver, diff --git a/packages/local-runtime/src/context/messages-count-tokens-messages.ts b/packages/local-runtime/src/context/messages-count-tokens-messages.ts index efa52aa2..6fb3e72a 100644 --- a/packages/local-runtime/src/context/messages-count-tokens-messages.ts +++ b/packages/local-runtime/src/context/messages-count-tokens-messages.ts @@ -1,5 +1,5 @@ import type { AgentMessage } from '@earendil-works/pi-agent-core'; -import { convertToLlm } from '@earendil-works/pi-coding-agent'; +import { convertToLlm } from '@earendil-works/pi-coding-agent/messages'; import { removeOrphanToolResults } from '@mavis/agent-core/pi-turn-runner'; import type { Api, diff --git a/packages/local-runtime/src/context/token-estimator.ts b/packages/local-runtime/src/context/token-estimator.ts index b1224ae9..7aa26426 100644 --- a/packages/local-runtime/src/context/token-estimator.ts +++ b/packages/local-runtime/src/context/token-estimator.ts @@ -1,6 +1,6 @@ import type { AgentMessage as PiAgentMessage } from '@earendil-works/pi-agent-core'; import type { Api, Model } from '@earendil-works/pi-ai'; -import { calculateContextTokens } from '@earendil-works/pi-coding-agent'; +import { calculateContextTokens } from '@earendil-works/pi-coding-agent/compaction'; import { computeCompactionTriggerAt as computeSharedCompactionTriggerAt, createDefaultTokenEstimator, diff --git a/packages/local-runtime/test/unit/child-bash-lifecycle.test.ts b/packages/local-runtime/test/unit/child-bash-lifecycle.test.ts index 0586d51f..88d2c0c0 100644 --- a/packages/local-runtime/test/unit/child-bash-lifecycle.test.ts +++ b/packages/local-runtime/test/unit/child-bash-lifecycle.test.ts @@ -3,7 +3,8 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { LocalBashTool } from '@mavis/agent-tools/desktop'; -import { createLocalBashOperations, getShellConfig } from '@earendil-works/pi-coding-agent'; +import { createLocalBashOperations } from '@earendil-works/pi-coding-agent/tools'; +import { getShellConfig } from '@earendil-works/pi-coding-agent/shell'; import { createChildBashLifecycle } from '../../src/background-task/child-bash-lifecycle.js'; import { LocalBackgroundTaskService } from '../../src/background-task/service.js'; import { diff --git a/packages/tui/src/host/bash-command.ts b/packages/tui/src/host/bash-command.ts index 6233fd74..b7d1eb1b 100644 --- a/packages/tui/src/host/bash-command.ts +++ b/packages/tui/src/host/bash-command.ts @@ -14,7 +14,7 @@ export type ExecuteTuiBash = (input: { }) => Promise; export const executeTuiBash: ExecuteTuiBash = async (input) => { - const { createLocalBashOperations } = await import('@earendil-works/pi-coding-agent'); + const { createLocalBashOperations } = await import('@earendil-works/pi-coding-agent/tools'); const operations = createLocalBashOperations({ parentDeathGuard: true }); const env = { ...process.env }; stripRuntimeBoundaryKeysFrom(env, 'agent-runtime'); diff --git a/packages/tui/src/host/image-preview-worker.ts b/packages/tui/src/host/image-preview-worker.ts index 16ab1fa3..cd8b14bc 100644 --- a/packages/tui/src/host/image-preview-worker.ts +++ b/packages/tui/src/host/image-preview-worker.ts @@ -1,5 +1,5 @@ import { parentPort, workerData } from 'node:worker_threads'; -import { resizeImage } from '@earendil-works/pi-coding-agent'; +import { resizeImage } from '@earendil-works/pi-coding-agent/image-resize'; const input: unknown = workerData; if ( diff --git a/third_party/pi-mono/packages/coding-agent/package.json b/third_party/pi-mono/packages/coding-agent/package.json index c36126f7..6c97eaf7 100644 --- a/third_party/pi-mono/packages/coding-agent/package.json +++ b/third_party/pi-mono/packages/coding-agent/package.json @@ -12,6 +12,30 @@ ".": { "types": "./dist/index.d.ts", "import": "./dist/index.js" + }, + "./tools": { + "types": "./dist/core/tools/index.d.ts", + "import": "./dist/core/tools/index.js" + }, + "./messages": { + "types": "./dist/core/messages.d.ts", + "import": "./dist/core/messages.js" + }, + "./auth-storage": { + "types": "./dist/core/auth-storage.d.ts", + "import": "./dist/core/auth-storage.js" + }, + "./compaction": { + "types": "./dist/core/compaction/compaction.d.ts", + "import": "./dist/core/compaction/compaction.js" + }, + "./image-resize": { + "types": "./dist/utils/image-resize.d.ts", + "import": "./dist/utils/image-resize.js" + }, + "./shell": { + "types": "./dist/utils/shell.d.ts", + "import": "./dist/utils/shell.js" } }, "files": [ diff --git a/tsconfig.standalone.json b/tsconfig.standalone.json index aeb1e41a..1ad29d2b 100644 --- a/tsconfig.standalone.json +++ b/tsconfig.standalone.json @@ -49,6 +49,24 @@ "@earendil-works/pi-coding-agent": [ "./third_party/pi-mono/packages/coding-agent/src/index.ts" ], + "@earendil-works/pi-coding-agent/auth-storage": [ + "./third_party/pi-mono/packages/coding-agent/src/core/auth-storage.ts" + ], + "@earendil-works/pi-coding-agent/compaction": [ + "./third_party/pi-mono/packages/coding-agent/src/core/compaction/compaction.ts" + ], + "@earendil-works/pi-coding-agent/image-resize": [ + "./third_party/pi-mono/packages/coding-agent/src/utils/image-resize.ts" + ], + "@earendil-works/pi-coding-agent/messages": [ + "./third_party/pi-mono/packages/coding-agent/src/core/messages.ts" + ], + "@earendil-works/pi-coding-agent/shell": [ + "./third_party/pi-mono/packages/coding-agent/src/utils/shell.ts" + ], + "@earendil-works/pi-coding-agent/tools": [ + "./third_party/pi-mono/packages/coding-agent/src/core/tools/index.ts" + ], "@earendil-works/pi-tui": [ "./third_party/pi-mono/packages/tui/src/index.ts" ], From ad0ae3a3324d7cd5cb0e76bfa2133ff082422766 Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Mon, 21 Sep 2026 19:20:36 +0800 Subject: [PATCH 08/10] perf: share unchanged immutable messages between history snapshots --- .../durable-canonical-history-store.ts | 7 +- .../agent-host/history/semantic-identity.ts | 61 ++++++++++++---- test/history-processing.test.ts | 73 +++++++++++++++++++ 3 files changed, 125 insertions(+), 16 deletions(-) diff --git a/packages/local-runtime-v2/src/service/turn-system/agent-host/history/durable-canonical-history-store.ts b/packages/local-runtime-v2/src/service/turn-system/agent-host/history/durable-canonical-history-store.ts index 1298c45f..6b3240d0 100644 --- a/packages/local-runtime-v2/src/service/turn-system/agent-host/history/durable-canonical-history-store.ts +++ b/packages/local-runtime-v2/src/service/turn-system/agent-host/history/durable-canonical-history-store.ts @@ -42,6 +42,7 @@ export interface DurableCanonicalHistoryProvider { */ export class DurableCanonicalHistoryStore implements CanonicalHistoryStore { private readonly lane = new KeyedOperationLane(); + private previousSnapshot?: CanonicalHistorySnapshot; constructor(private readonly provider: DurableCanonicalHistoryProvider) { assertAgentHostCapabilityAvailable( @@ -143,15 +144,17 @@ export class DurableCanonicalHistoryStore implements CanonicalHistoryStore { } private readProviderSnapshot(snapshot: CanonicalHistorySnapshot): CanonicalHistorySnapshot { - const detached = captureSemanticSnapshot(snapshot).value; + const detached = captureSemanticSnapshot(snapshot, this.previousSnapshot).value; validateCanonicalHistorySnapshot(detached); assertCanonicalIdentityVector(detached); - return captureSemanticSnapshot({ + const result = captureSemanticSnapshot({ revision: detached.revision.trim(), // Keep the separately owned arrays reusable at the next snapshot boundary. messages: captureSemanticSnapshot([...detached.messages]).value, identityVector: captureSemanticSnapshot([...detached.identityVector]).value, }).value; + this.previousSnapshot = result; + return result; } } diff --git a/packages/local-runtime-v2/src/service/turn-system/agent-host/history/semantic-identity.ts b/packages/local-runtime-v2/src/service/turn-system/agent-host/history/semantic-identity.ts index ee9f51f4..1ccc478d 100644 --- a/packages/local-runtime-v2/src/service/turn-system/agent-host/history/semantic-identity.ts +++ b/packages/local-runtime-v2/src/service/turn-system/agent-host/history/semantic-identity.ts @@ -13,16 +13,17 @@ const fingerprints = new WeakMap(); * 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. + * encoded payload size. An optional previously owned snapshot lets fresh plain + * data share unchanged immutable descendants while preserving input aliases. */ -export function captureSemanticSnapshot(value: T): SemanticSnapshot { +export function captureSemanticSnapshot(value: T, previous?: T): SemanticSnapshot { // Native cloning preserves external getters and aliases. Data-only wrappers // can instead share descendants already detached and frozen by this module. const snapshot = typeof value === 'object' && value !== null && ownedValues.has(value) ? value : freezeSemanticValue( - cloneOwnedWrapper(value) ?? structuredClone(value), + cloneOwnedWrapper(value, previous) ?? structuredClone(value), new WeakSet(), ); let fingerprint: string | undefined; @@ -46,6 +47,7 @@ export function captureSemanticSnapshot(value: T): SemanticSnapshot { } const NATIVE_CLONE_REQUIRED = Symbol('native-clone-required'); +const PREVIOUS_SHARING_UNAVAILABLE = Symbol('previous-sharing-unavailable'); function plainDataDescriptors(value: object): PropertyDescriptorMap | undefined { // Inspecting a Proxy would invoke traps that native structuredClone rejects. @@ -62,47 +64,78 @@ function plainDataDescriptors(value: object): PropertyDescriptorMap | undefined return descriptors; } -function cloneOwnedWrapper(value: T): T | undefined { +function cloneOwnedWrapper(value: T, previous?: T): T | undefined { if (typeof value !== 'object' || value === null) return undefined; const descriptors = plainDataDescriptors(value); + const reusePrevious = + typeof previous === 'object' && previous !== null && ownedValues.has(previous); if ( !descriptors || - !Object.values(descriptors).some( - (d) => - d.enumerable && typeof d.value === 'object' && d.value !== null && ownedValues.has(d.value), - ) + (!reusePrevious && + !Object.values(descriptors).some( + (d) => + d.enumerable && typeof d.value === 'object' && d.value !== null && ownedValues.has(d.value), + )) ) return undefined; const copies = new WeakMap(); - const clone = (node: unknown): unknown => { + const previousOwners = new WeakMap(); + const clone = (node: unknown, prior?: unknown): unknown => { if (typeof node !== 'object' || node === null) { if (node !== null && !['undefined', 'string', 'boolean', 'number'].includes(typeof node)) { throw NATIVE_CLONE_REQUIRED; } return node; } - if (ownedValues.has(node)) return node; - const previous = copies.get(node); - if (previous) return previous; + if (ownedValues.has(node)) { + // Mixing existing owned nodes with value-based reuse could merge two + // distinct input aliases. Keep the original wrapper path for that case. + if (reusePrevious) throw PREVIOUS_SHARING_UNAVAILABLE; + return node; + } + const existing = copies.get(node); + if (existing) return existing; const fields = node === value ? descriptors : plainDataDescriptors(node); if (!fields) throw NATIVE_CLONE_REQUIRED; const copy = Array.isArray(node) ? new Array(fields.length!.value as number) : {}; copies.set(node, copy); + const candidate = + reusePrevious && + typeof prior === 'object' && prior !== null && ownedValues.has(prior) && + Array.isArray(prior) === Array.isArray(node) && + (!previousOwners.has(prior) || previousOwners.get(prior) === node) + ? prior as Record + : undefined; + const keys = Object.keys(fields).filter((key) => fields[key]!.enumerable); + const priorKeys = candidate ? Object.keys(candidate) : []; + let unchanged = + candidate !== undefined && keys.length === priorKeys.length && + keys.every((key, index) => key === priorKeys[index]) && + (!Array.isArray(node) || node.length === candidate['length']); for (const [key, field] of Object.entries(fields)) { if (!field.enumerable) continue; + const priorChild = candidate && Object.hasOwn(candidate, key) ? candidate[key] : undefined; + const child = clone(field.value, priorChild); + if (unchanged && !Object.is(child, candidate![key])) unchanged = false; Object.defineProperty(copy, key, { - value: clone(field.value), + value: child, enumerable: true, writable: true, configurable: true, }); } + if (unchanged) { + previousOwners.set(candidate!, node); + copies.set(node, candidate!); + return candidate; + } return copy; }; try { - return clone(value) as T; + return clone(value, previous) as T; } catch (error) { + if (error === PREVIOUS_SHARING_UNAVAILABLE) return cloneOwnedWrapper(value); if (error === NATIVE_CLONE_REQUIRED) return undefined; throw error; } diff --git a/test/history-processing.test.ts b/test/history-processing.test.ts index 4a2ccb39..68be2767 100644 --- a/test/history-processing.test.ts +++ b/test/history-processing.test.ts @@ -382,6 +382,57 @@ describe('semantic snapshots', () => { hash.mockRestore(); } }); + it('shares unchanged plain descendants without changing values, fingerprints or byte accounting', () => { + const before = captureSemanticSnapshot({ messages: [{ text: 'old', nested: [-0, undefined] }] }); + const input = { messages: [{ text: 'old', nested: [-0, undefined] }, { text: 'new', nested: [1] }] }; + const shared = captureSemanticSnapshot(input, before.value); + const independent = captureSemanticSnapshot(input); + expect(shared.value).toEqual(independent.value); + expect(shared.fingerprint).toBe(independent.fingerprint); + expect(estimateSemanticValueSize(shared.value)).toBe(estimateSemanticValueSize(independent.value)); + expect(shared.value.messages[0]).toBe(before.value.messages[0]); + expect(shared.value.messages).not.toBe(before.value.messages); + input.messages[0]!.text = 'edited'; + expect(shared.value.messages[0]!.text).toBe('old'); + const edited = captureSemanticSnapshot(input, shared.value); + expect(edited.value.messages[0]).not.toBe(shared.value.messages[0]); + expect(edited.value.messages[1]).toBe(shared.value.messages[1]); + }); + it('preserves split and merged aliases when sharing a previous snapshot', () => { + const alias = { text: 'same' }; + const prior = captureSemanticSnapshot({ a: alias, b: alias }).value; + const split = captureSemanticSnapshot({ a: { text: 'same' }, b: { text: 'same' } }, prior).value; + expect(split.a).not.toBe(split.b); + const merged = captureSemanticSnapshot({ a: alias, b: alias }, split).value; + expect(merged.a).toBe(merged.b); + const mixed = captureSemanticSnapshot({ a: { text: 'same' }, b: prior.a }, prior).value; + expect(mixed.a).not.toBe(mixed.b); + expect(mixed.b).toBe(prior.a); + const mixedFirst = captureSemanticSnapshot({ a: prior.a, b: { text: 'same' } }, prior).value; + expect(mixedFirst.a).not.toBe(mixedFirst.b); + }); + it('preserves key order, sparse arrays, negative zero and native accessor behavior during reuse', () => { + const prior = captureSemanticSnapshot({ a: 1, b: 2 }).value; + const reordered = captureSemanticSnapshot({ b: 2, a: 1 }, prior).value; + expect(Object.keys(reordered)).toEqual(['b', 'a']); + const sparse = new Array(3); + sparse[2] = -0; + const oldArray = captureSemanticSnapshot(sparse).value; + const nextArray = captureSemanticSnapshot([undefined, undefined, 0], oldArray).value; + expect(0 in nextArray).toBe(true); + expect(Object.is(nextArray[2], -0)).toBe(false); + expect(0 in oldArray).toBe(false); + const getter = vi.fn(() => 1); + expect(captureSemanticSnapshot({ get a() { return getter(); }, b: 2 }, prior).value).toEqual(prior); + expect(getter).toHaveBeenCalledTimes(1); + const trap = vi.fn(); + expect(() => captureSemanticSnapshot(new Proxy({}, { ownKeys: trap }), prior)).toThrow(); + expect(trap).not.toHaveBeenCalled(); + const unsafe = Object.freeze({ nested: { text: 'old' } }); + const snapshot = captureSemanticSnapshot({ nested: { text: 'old' } }, unsafe).value; + unsafe.nested.text = 'changed'; + expect(snapshot.nested.text).toBe('old'); + }); it('shares owned history through delivery wrappers and detaches other branches', () => { const history = captureSemanticSnapshot({ messages: [{ text: 'body' }], @@ -844,6 +895,28 @@ describe('committed history read reuse', () => { expect(empty.messages).not.toBe(empty.identityVector); expect(empty.messages).toEqual([]); }); + it('shares immutable messages across fresh provider snapshots while observing edits and replacement', async () => { + let current = { + revision: 'r1', + messages: [{ role: 'user', timestamp: 1, content: 'old' }], + identityVector: ['msg-1'], + }; + const read = async () => structuredClone(current); + const store = new DurableCanonicalHistoryStore({ read, readActive: read, append: read, replace: read }); + const first = await store.read('synthetic-session'); + current = { revision: 'r2', messages: [...current.messages, { role: 'user', timestamp: 2, content: 'new' }], identityVector: ['msg-1', 'msg-2'] }; + const second = await store.read('synthetic-session'); + expect(second.messages[0]).toBe(first.messages[0]); + expect(second.messages).toHaveLength(2); + current.messages[0]!.content = 'edited'; + const third = await store.read('synthetic-session'); + expect(third.messages[0]).not.toBe(second.messages[0]); + expect(third.messages[1]).toBe(second.messages[1]); + expect(second.messages[0]).toMatchObject({ content: 'old' }); + current = { revision: 'r3', messages: [], identityVector: [] }; + expect((await store.read('synthetic-session')).messages).toEqual([]); + expect(first.messages).toHaveLength(1); + }); it('retains legacy rereads and rejects invalid commits or write failures', async () => { const readActive = vi.fn(async () => ({ revision: 'r1', From c689e3f46894e9933a01df8ce539374d88e5720a Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Mon, 21 Sep 2026 19:24:47 +0800 Subject: [PATCH 09/10] perf: compact bundled source to reduce module loading allocations --- scripts/build.mjs | 1 + 1 file changed, 1 insertion(+) diff --git a/scripts/build.mjs b/scripts/build.mjs index cd77d2e6..7bfb42a7 100644 --- a/scripts/build.mjs +++ b/scripts/build.mjs @@ -90,6 +90,7 @@ const result = await build({ format: "esm", platform: "node", minifyIdentifiers: true, + minifyWhitespace: true, target: "node22", chunkNames: "chunks/[name]-[hash]", banner: { js: location.banner }, From b2418c0a4b38c19805bd47244830a2bd20877282 Mon Sep 17 00:00:00 2001 From: saladday <1203511142@qq.com> Date: Mon, 21 Sep 2026 19:59:44 +0800 Subject: [PATCH 10/10] fix: keep precise release audit matches after whitespace minification --- .gitleaks.toml | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/.gitleaks.toml b/.gitleaks.toml index 0d6600dc..d749e8f0 100644 --- a/.gitleaks.toml +++ b/.gitleaks.toml @@ -45,7 +45,21 @@ description = "Bundled TUI shortcut and permission translation key" condition = "AND" paths = ['''(^|/)dist/chunks/(chunk|launcher)-[A-Z0-9]+\.js$'''] regexTarget = "match" -regexes = ['''^defaultKeys: "ctrl\+shift\+down"$''', '''^labelKey: "permission\.scope\.byArgvPrefix2"$'''] +regexes = ['''^defaultKeys: ?"ctrl\+shift\+down"$''', '''^labelKey: ?"permission\.scope\.byArgvPrefix2"$'''] + +[[rules.allowlists]] +description = "Bundled node-forge public PKCS12 algorithm names" +condition = "AND" +paths = ['''(^|/)dist/chunks/chunk-[A-Z0-9]+\.js$'''] +regexTarget = "match" +regexes = ['''^pbeWithSHAAnd3-KeyTripleDES-CBC","pbewithSHAAnd40BitRC2-CBC"$'''] + +[[rules.allowlists]] +description = "Bundled IM binding migration guard, no credential value" +condition = "AND" +paths = ['''(^|/)dist/chunks/chunk-[A-Z0-9]+\.js$'''] +regexTarget = "match" +regexes = ['''^[A-Za-z_$][A-Za-z0-9_$]*\.resolvedProjectKey\|\|[A-Za-z_$][A-Za-z0-9_$]*\.mutationReceipts===void $'''] [[rules.allowlists]] description = "node-forge PKCS12 function alias, no key material"