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" 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/config/src/config.ts b/packages/config/src/config.ts index db9f7109..75bccea1 100644 --- a/packages/config/src/config.ts +++ b/packages/config/src/config.ts @@ -1649,11 +1649,28 @@ function syncManagedPresetBaseUrl(configPath: string): void { ) return; + const originalContent = fs.readFileSync(configPath); (options as Record).baseURL = presetBaseURL; - writePrivateConfigFileSync( - configPath, - yaml.dump(raw, { indent: 2, lineWidth: -1, noRefs: true }), - ); + try { + writePrivateConfigFileSync( + configPath, + yaml.dump(raw, { indent: 2, lineWidth: -1, noRefs: true }), + ); + } catch (error) { + // This on-disk sync is optional, but a failure after truncation is not safe + // to hide. Only continue when the original document is still intact. + let unchanged = false; + try { + unchanged = fs.readFileSync(configPath).equals(originalContent); + } catch { + // Preserve the original write error if integrity cannot be verified. + } + if (!unchanged) throw error; + // Do not include the error message: config errors may contain credentials. + console.warn( + "[config] managed preset baseURL sync skipped; config file unchanged, using runtime provider settings", + ); + } } function buildPresetEntry(key: PresetKey) { diff --git a/packages/config/src/private-config-file.ts b/packages/config/src/private-config-file.ts index 82e4f090..9112c419 100644 --- a/packages/config/src/private-config-file.ts +++ b/packages/config/src/private-config-file.ts @@ -18,11 +18,13 @@ export function writePrivateConfigFileSync( content: string | Buffer, exclusive = false, ): void { - const fd = fs.openSync( - filePath, - exclusive ? "wx" : "a", - PRIVATE_CONFIG_FILE_MODE, - ); + // Append handles cannot be truncated on Windows. Defer truncation until + // permissions have been restricted, including for an existing POSIX file. + const flags = + fs.constants.O_WRONLY | + fs.constants.O_CREAT | + (exclusive ? fs.constants.O_EXCL : 0); + const fd = fs.openSync(filePath, flags, PRIVATE_CONFIG_FILE_MODE); try { fs.fchmodSync(fd, PRIVATE_CONFIG_FILE_MODE); fs.ftruncateSync(fd, 0); diff --git a/packages/config/test/managed-preset-sync.test.ts b/packages/config/test/managed-preset-sync.test.ts new file mode 100644 index 00000000..dff3082a --- /dev/null +++ b/packages/config/test/managed-preset-sync.test.ts @@ -0,0 +1,168 @@ +import fs from "node:fs"; +import os from "node:os"; +import { join } from "node:path"; +import yaml from "js-yaml"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + getConfig, + resetConfig, + setLegacyByokProviderMigrationEnabled, + setManagedPresetBaseUrlSyncEnabled, +} from "../src/config.js"; +import { loadConfigFromFile } from "../src/file-loader.js"; + +const secret = "synthetic-preset-sync-secret"; +const oldBaseURL = "https://agent.minimax.io/mavis/api/v1/llm/v1"; +const presetBaseURL = + "https://matrix-overseas-pre.example.invalid/mavis/api/v1/llm/v1"; +const original = yaml.dump({ + logLevel: "debug", + provider: { minimax: { options: { baseURL: oldBaseURL, apiKey: secret } } }, + custom_provider: { example: { options: { apiKey: secret }, models: {} } }, +}); +let root: string; +let file: string; + +beforeEach(() => { + root = fs.mkdtempSync(join(os.tmpdir(), "managed-preset-sync-")); + file = join(root, "config.yaml"); + vi.stubEnv("__MAVIS_RUNTIME_MANAGED", "1"); + vi.stubEnv("__MAVIS_RUNTIME_DATA_DIR", root); + vi.stubEnv("MINIMAX_DATA_DIR", root); + vi.stubEnv("MAVIS_REGION", "en"); + vi.stubEnv("MAVIS_BUILD_ENV", "staging"); + setLegacyByokProviderMigrationEnabled(false); + setManagedPresetBaseUrlSyncEnabled(true); + resetConfig(); + fs.writeFileSync(file, original, { mode: 0o600 }); +}); + +afterEach(() => { + vi.restoreAllMocks(); + vi.unstubAllEnvs(); + resetConfig(); + setLegacyByokProviderMigrationEnabled(true); + setManagedPresetBaseUrlSyncEnabled(true); + fs.rmSync(root, { recursive: true, force: true }); +}); + +describe.each([ + ["default config", () => getConfig()], + ["explicit config", () => loadConfigFromFile(file, { dataDir: root })], +] as const)("%s preset synchronization", (_name, load) => { + it("persists the current preset and preserves user settings", () => { + const config = load(); + expect(config.logLevel).toBe("debug"); + expect(config.provider.minimax?.options).toMatchObject({ + baseURL: presetBaseURL, + apiKey: secret, + }); + const persisted = yaml.load(fs.readFileSync(file, "utf8")); + expect(persisted).toMatchObject({ + custom_provider: { example: { options: { apiKey: secret } } }, + provider: { + minimax: { options: { baseURL: presetBaseURL, apiKey: secret } }, + }, + }); + }); + + it.each(["openSync", "fchmodSync", "ftruncateSync"] as const)( + "loads the effective preset when %s fails without changing the file", + (operation) => { + const failure = Object.assign(new Error(secret), { code: "EPERM" }); + if (operation === "openSync") { + const open = fs.openSync; + vi.spyOn(fs, "openSync").mockImplementation((path, flags, mode) => { + if ( + flags === "a" || + (typeof flags === "number" && flags & fs.constants.O_WRONLY) + ) { + throw failure; + } + return open(path, flags, mode); + }); + } else { + vi.spyOn(fs, operation).mockImplementation(() => { + throw failure; + }); + } + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + const config = load(); + expect(config.logLevel).toBe("debug"); + expect(config.provider.minimax?.options).toMatchObject({ + baseURL: presetBaseURL, + apiKey: secret, + }); + expect(fs.readFileSync(file, "utf8")).toBe(original); + expect(warn).toHaveBeenCalledTimes(1); + expect(warn).toHaveBeenCalledWith( + expect.stringContaining("preset baseURL sync skipped"), + ); + expect(JSON.stringify(warn.mock.calls)).not.toContain(secret); + }, + ); + + it("does not hide a failure after the original file has been truncated", () => { + const failure = new Error("synthetic write failure"); + vi.spyOn(fs, "writeFileSync").mockImplementation(() => { + throw failure; + }); + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + expect(() => load()).toThrow(failure); + expect(warn).not.toHaveBeenCalled(); + expect(fs.readFileSync(file, "utf8")).toBe(""); + }); + + it("preserves the write error when the original file cannot be verified", () => { + const failure = new Error("synthetic truncate failure"); + let failed = false; + vi.spyOn(fs, "ftruncateSync").mockImplementation(() => { + failed = true; + throw failure; + }); + const read = fs.readFileSync; + vi.spyOn(fs, "readFileSync").mockImplementation((...args) => { + if (failed && args[0] === file) throw new Error("synthetic read failure"); + return read(...args); + }); + expect(() => load()).toThrow(failure); + }); +}); + +it("honors the disabled preset synchronization policy", () => { + setManagedPresetBaseUrlSyncEnabled(false); + const truncate = vi.spyOn(fs, "ftruncateSync"); + expect(getConfig().provider.minimax?.options?.baseURL).toBe(presetBaseURL); + expect(truncate).not.toHaveBeenCalled(); + expect(fs.readFileSync(file, "utf8")).toBe(original); +}); + +it.each([ + ["prod", oldBaseURL], + ["test", presetBaseURL], +] as const)( + "preserves %s provider policy after a safe sync failure", + (buildEnv, expectedBaseURL) => { + vi.stubEnv("MAVIS_BUILD_ENV", buildEnv); + // A staging endpoint needs syncing in both prod and test builds. + fs.writeFileSync(file, original.replace(oldBaseURL, presetBaseURL)); + vi.spyOn(fs, "ftruncateSync").mockImplementation(() => { + throw new Error("synthetic failure"); + }); + vi.spyOn(console, "warn").mockImplementation(() => {}); + const config = getConfig(); + expect(config.provider.minimax?.options?.baseURL).toBe(expectedBaseURL); + expect(config.provider.minimax?.options?.apiKey).toBe(secret); + }, +); + +it("still reports failure when creating the required initial config", () => { + fs.unlinkSync(file); + const failure = Object.assign(new Error("synthetic create failure"), { + code: "EPERM", + }); + vi.spyOn(fs, "openSync").mockImplementation(() => { + throw failure; + }); + expect(() => getConfig()).toThrow(failure); +}); diff --git a/packages/config/test/private-config-file.test.ts b/packages/config/test/private-config-file.test.ts new file mode 100644 index 00000000..6de21633 --- /dev/null +++ b/packages/config/test/private-config-file.test.ts @@ -0,0 +1,85 @@ +import fs from "node:fs"; +import os from "node:os"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { writePrivateConfigFileSync } from "../src/private-config-file.js"; + +let root: string; +let file: string; + +beforeEach(() => { + root = fs.mkdtempSync(join(os.tmpdir(), "private-config-write-")); + file = join(root, "config.yaml"); +}); + +afterEach(() => { + vi.restoreAllMocks(); + fs.rmSync(root, { recursive: true, force: true }); +}); + +// These use the native filesystem on every OS, including Windows. +describe("private config writes", () => { + it.each([false, true])( + "creates a missing file (exclusive=%s)", + (exclusive) => { + writePrivateConfigFileSync(file, "logLevel: info\n", exclusive); + expect(fs.readFileSync(file, "utf8")).toBe("logLevel: info\n"); + if (process.platform !== "win32") { + expect(fs.statSync(file).mode & 0o777).toBe(0o600); + } + }, + ); + + it.each(["logLevel: info\n", Buffer.from("logLevel: info\n")])( + "replaces longer content without appending or retaining a suffix", + (content) => { + fs.writeFileSync( + file, + "logLevel: debug\n# old credentials and trailing data\n", + ); + writePrivateConfigFileSync(file, content); + expect(fs.readFileSync(file, "utf8")).toBe("logLevel: info\n"); + }, + ); + + it("does not overwrite an existing file in exclusive mode", () => { + fs.writeFileSync(file, "original"); + expect(() => + writePrivateConfigFileSync(file, "replacement", true), + ).toThrow(); + expect(fs.readFileSync(file, "utf8")).toBe("original"); + }); + + it.each(["fchmodSync", "ftruncateSync"] as const)( + "closes the descriptor and stops writing when %s fails", + (operation) => { + fs.writeFileSync(file, "original"); + const failure = Object.assign(new Error("synthetic failure"), { + code: "EPERM", + }); + vi.spyOn(fs, operation).mockImplementation(() => { + throw failure; + }); + const write = vi.spyOn(fs, "writeFileSync"); + const close = vi.spyOn(fs, "closeSync"); + expect(() => writePrivateConfigFileSync(file, "replacement")).toThrow( + failure, + ); + expect(write).not.toHaveBeenCalled(); + expect(close).toHaveBeenCalledTimes(1); + expect(fs.readFileSync(file, "utf8")).toBe("original"); + }, + ); + + it("propagates a write failure and closes the descriptor", () => { + const failure = new Error("synthetic write failure"); + vi.spyOn(fs, "writeFileSync").mockImplementation(() => { + throw failure; + }); + const close = vi.spyOn(fs, "closeSync"); + expect(() => writePrivateConfigFileSync(file, "replacement")).toThrow( + failure, + ); + expect(close).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/local-runtime-v2/src/infra/db/write-transaction.ts b/packages/local-runtime-v2/src/infra/db/write-transaction.ts new file mode 100644 index 00000000..5146ca9b --- /dev/null +++ b/packages/local-runtime-v2/src/infra/db/write-transaction.ts @@ -0,0 +1,86 @@ +import { setTimeout as delay } from 'node:timers/promises'; +import { sql } from 'drizzle-orm'; +import type { AppDb } from './client.js'; + +const WRITE_LOCK_BUDGET_MS = 10_000; +const WRITE_LOCK_ATTEMPT_MS = 50; + +/** Cancellation before the mutation callback starts; no write needs to be replayed. */ +export class WriteLockWaitAbortedError extends Error { + override readonly name = 'WriteLockWaitAbortedError'; + + constructor(readonly signal: AbortSignal) { + super('SQLite write lock wait was cancelled', { cause: signal.reason }); + } +} + +/** + * Retry only transaction admission: a callback that has started is never replayed. + * The signal cancels contention waits, not an immediately available write. This + * lets post-cancellation tool completion and cleanup messages remain durable. + */ +export async function runWithWriteLock( + db: AppDb, + mutation: (tx: AppDb) => T, + options: { readonly signal?: AbortSignal; readonly timeoutMs?: number } = {}, +): Promise { + const budget = options.timeoutMs ?? WRITE_LOCK_BUDGET_MS; + if (!Number.isFinite(budget) || budget <= 0) + throw new RangeError('Invalid write lock budget'); + const deadline = performance.now() + budget; + let attempt = 0; + let hasContended = false; + let lastBusy: unknown = new Error('SQLite write lock wait exceeded its deadline'); + for (;;) { + if (hasContended) throwIfWaitAborted(options.signal); + const remaining = deadline - performance.now(); + if (remaining <= 0) throw lastBusy; + const previous = db.get<{ timeout: number }>(sql`PRAGMA busy_timeout`).timeout; + let entered = false; + try { + const nativeWaitMs = options.signal?.aborted + ? 0 + : Math.ceil(Math.min(WRITE_LOCK_ATTEMPT_MS, remaining)); + db.run(sql.raw(`PRAGMA busy_timeout = ${nativeWaitMs}`)); + return db.transaction( + (tx) => { + entered = true; + // Only lock acquisition gets a short timeout. Restore the connection's + // policy before callbacks (including nested transactions) can use it. + db.run(sql.raw(`PRAGMA busy_timeout = ${previous}`)); + return mutation(tx); + }, + { behavior: 'immediate' }, + ); + } catch (error) { + if (entered || !isBusy(error)) throw error; + lastBusy = error; + } finally { + // No await occurs while the shared connection has a temporary timeout. + db.run(sql.raw(`PRAGMA busy_timeout = ${previous}`)); + } + hasContended = true; + throwIfWaitAborted(options.signal); + const wait = Math.min( + deadline - performance.now(), + 25 * 2 ** Math.min(attempt++, 3) + Math.random() * 25, + ); + if (wait <= 0) throw lastBusy; + try { + await delay(wait, undefined, { signal: options.signal }); + } catch (error) { + if (options.signal?.aborted && error instanceof Error && error.name === 'AbortError') { + throw new WriteLockWaitAbortedError(options.signal); + } + throw error; + } + } +} + +function throwIfWaitAborted(signal: AbortSignal | undefined): void { + if (signal?.aborted) throw new WriteLockWaitAbortedError(signal); +} + +function isBusy(error: unknown): boolean { + return error instanceof Error && Reflect.get(error, 'code') === 'SQLITE_BUSY'; +} 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 8b042504..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 @@ -6,6 +6,7 @@ import { readJsonl, writeJsonlAtomically, type JsonlMalformedLine, + type JsonlReadCache, } from './jsonl.js'; import { decodeCanonicalHistoryArtifact, @@ -24,6 +25,15 @@ 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 ownedRecordArrays = new WeakSet(); +const ownedRevisions = new WeakMap(); +const ownedSequences = new WeakMap< + readonly CanonicalHistoryEnvelope[], + CanonicalHistorySequenceInspection +>(); + const ENVELOPE_KEYS = new Set([ 'message_id', 'turn_id', @@ -268,6 +278,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 +306,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'); @@ -358,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(), @@ -371,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] }; } /** @@ -389,6 +414,8 @@ export function inspectCanonicalHistorySequence( * not provide a cross-process writer lock. */ export class CanonicalHistoryJsonlDataSource { + private readonly readCache: JsonlReadCache = { bytes: Buffer.alloc(0), records: [] }; + constructor(private readonly options: CanonicalHistoryJsonlDataSourceOptions) {} async readActive(): Promise { @@ -398,7 +425,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 +434,11 @@ export class CanonicalHistoryJsonlDataSource { async readEnvelopesStrict( filePath = this.options.activePath, ): Promise { - return readStrictEnvelopeFile(filePath); + if (!this.options.reuseDecodedRecords) return readStrictEnvelopeFile(filePath); + const records = await readJsonl(filePath, decodeOwnedEnvelope, undefined, this.readCache); + Object.freeze(records); + ownedRecordArrays.add(records); + return records; } async readStrict(filePath = this.options.activePath): Promise { @@ -555,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`); @@ -564,7 +596,37 @@ function decodeRecords(records: readonly CanonicalHistoryEnvelope[]): CanonicalH } function revisionOfNormalized(records: readonly CanonicalHistoryEnvelope[]): string { - return `sha256:${createHash('sha256').update(canonicalJson(records), 'utf8').digest('hex')}`; + 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('['); + for (let index = 0; index < records.length; index += 1) { + if (index > 0) hash.update(','); + 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'); + } + const revision = `sha256:${hash.update(']').digest('hex')}`; + if (ownedRecordArrays.has(records)) ownedRevisions.set(records, revision); + return revision; +} + +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 { diff --git a/packages/local-runtime-v2/src/infra/file/jsonl.ts b/packages/local-runtime-v2/src/infra/file/jsonl.ts index 5f0bdb52..47ac7974 100644 --- a/packages/local-runtime-v2/src/infra/file/jsonl.ts +++ b/packages/local-runtime-v2/src/infra/file/jsonl.ts @@ -32,27 +32,30 @@ export class JsonlAppendCommitUncertainError extends Error { } } +export interface JsonlReadCache { + bytes: Buffer; + records: readonly T[]; +} + export async function readJsonl( filePath: string, decode: (value: unknown) => T, onMalformedLine?: (line: JsonlMalformedLine) => void, + /** Strict private reads return immutable arrays when a cache is supplied. */ + readCache?: JsonlReadCache, ): Promise { + // Tolerant readers must still report every malformed line on every read. + if (readCache && !onMalformedLine) return readCachedJsonl(filePath, decode, readCache); const contents = await readFile(filePath, 'utf-8'); - if (contents.length === 0) return []; - + const records: T[] = []; const lines = contents.split('\n'); if (lines.at(-1) === '') lines.pop(); - const records: T[] = []; for (const [index, line] of lines.entries()) { try { if (line.trim().length === 0) throw new Error('blank line'); records.push(decode(parseJsonLine(line))); } catch (error) { - const malformed = { - path: filePath, - lineNo: index + 1, - reason: errorReason(error), - }; + const malformed = { path: filePath, lineNo: index + 1, reason: errorReason(error) }; if (!onMalformedLine) throw malformedLineError(malformed); onMalformedLine(malformed); } @@ -60,6 +63,54 @@ export async function readJsonl( 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 (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 { try { return JSON.parse(line) as unknown; 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/session-system/agent-projection.ts b/packages/local-runtime-v2/src/service/session-system/agent-projection.ts index 3b8c37a1..3d8f61d2 100644 --- a/packages/local-runtime-v2/src/service/session-system/agent-projection.ts +++ b/packages/local-runtime-v2/src/service/session-system/agent-projection.ts @@ -37,6 +37,7 @@ export interface SessionAgentEventContext { export interface SessionAgentProjectionInput { readonly context: SessionAgentEventContext; readonly event: RuntimeEvent; + readonly signal?: AbortSignal; } export interface SessionSystemAgentProjectionOptions { @@ -76,19 +77,25 @@ export function createSessionSystemAgentProjection(options: SessionSystemAgentPr const message = displayMessage(input.event); if (message && !isTerminalAssistantDisplayError(message)) { const queryKey = await queryKeyForTurn(options, input.context); - await options.messages.upsert({ - sessionId: input.context.sessionId, - turnId: input.context.turnId, - message: { - ...message, - turn_id: input.context.turnId, - ...(queryKey ? { query_key: queryKey } : {}), + await options.messages.upsert( + { + sessionId: input.context.sessionId, + turnId: input.context.turnId, + message: { + ...message, + turn_id: input.context.turnId, + ...(queryKey ? { query_key: queryKey } : {}), + }, + source: input.context.provenance?.source ?? 'agent', + ...(input.context.provenance?.sourceContext + ? { sourceContext: input.context.provenance.sourceContext } + : {}), }, - source: input.context.provenance?.source ?? 'agent', - ...(input.context.provenance?.sourceContext - ? { sourceContext: input.context.provenance.sourceContext } - : {}), - }); + // Complete tool messages describe work already executed, including + // abort cleanup. Persist these facts within the write-lock budget + // even when the lease is cancelled; text-only waits may stop early. + message.tool_calls?.length ? undefined : { signal: input.signal }, + ); } await projectQueryCollapse(() => options.queryCollapse?.projectRuntimeEvent(input)); }, 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..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 @@ -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)); @@ -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); @@ -384,7 +397,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/packages/local-runtime-v2/src/service/session-system/messages/repo/contract.ts b/packages/local-runtime-v2/src/service/session-system/messages/repo/contract.ts index 69dee5ef..197c3f35 100644 --- a/packages/local-runtime-v2/src/service/session-system/messages/repo/contract.ts +++ b/packages/local-runtime-v2/src/service/session-system/messages/repo/contract.ts @@ -37,6 +37,10 @@ export interface MessageUpsertInput { readonly source?: string; readonly sourceContext?: Record; } +export interface MessageWriteOptions { + /** Cancels lock contention waits; immediately available cleanup writes still commit. */ + readonly signal?: AbortSignal; +} export interface UserMessageCommitInput extends MessageUpsertInput { readonly unconsumedFromTurnIds?: readonly string[]; /** Trusted Queue startup lineage; the initial Host never read these rows. */ @@ -97,8 +101,11 @@ export interface MessageRepository { }, ): Promise; commitUserMessage(input: UserMessageCommitInput): Promise; - upsert(input: MessageUpsertInput): Promise; - upsertMany(inputs: readonly MessageUpsertInput[]): Promise; + upsert(input: MessageUpsertInput, options?: MessageWriteOptions): Promise; + upsertMany( + inputs: readonly MessageUpsertInput[], + options?: MessageWriteOptions, + ): Promise; replace(input: MessageReplaceInput): Promise; replaceStream(input: MessageReplaceStreamInput): Promise; rewindInclusive(input: MessageRewindInclusiveInput): Promise; 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 44591897..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,6 +1,7 @@ -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'; import { legacyMessages, messageRowMigrations, @@ -22,6 +23,7 @@ import type { MessageRewindInclusiveInput, MessageRewindInclusiveResult, MessageUpsertInput, + MessageWriteOptions, DisplayMessageRecord, NormalizedDisplayMessage, UserMessageCommitInput, @@ -35,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) { @@ -43,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; } @@ -95,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( @@ -205,14 +239,18 @@ class DrizzleMessageRepository implements MessageRepository { ); } - async upsert(input: MessageUpsertInput): Promise { - const [result] = await this.upsertMany([input]); + async upsert( + input: MessageUpsertInput, + options?: MessageWriteOptions, + ): Promise { + const [result] = await this.upsertMany([input], options); if (!result) throw new Error('Message upsert returned no row'); return result; } async upsertMany( inputs: readonly MessageUpsertInput[], + options?: MessageWriteOptions, ): Promise { const sessionId = inputs[0]?.sessionId; if (!sessionId) return []; @@ -229,11 +267,16 @@ class DrizzleMessageRepository implements MessageRepository { }), replaceProvenance: hasMessageProvenance(input), })); - this.mutationTransaction(sessionId, (tx) => { - normalized.forEach(({ message, replaceProvenance }) => - this.write(tx, sessionId, message, replaceProvenance), - ); - }); + await runWithWriteLock( + this.options.db, + (tx) => { + this.ensureReadyInTransaction(tx, sessionId); + normalized.forEach(({ message, replaceProvenance }) => + this.write(tx, sessionId, message, replaceProvenance), + ); + }, + options, + ); return normalized.map(({ message }) => message); } diff --git a/packages/local-runtime-v2/src/service/session-system/query-collapse-state.ts b/packages/local-runtime-v2/src/service/session-system/query-collapse-state.ts index 4d7b044f..825a0db5 100644 --- a/packages/local-runtime-v2/src/service/session-system/query-collapse-state.ts +++ b/packages/local-runtime-v2/src/service/session-system/query-collapse-state.ts @@ -217,19 +217,30 @@ function readByCurrentTurn(db: AppDb, sessionId: string, currentTurnId: string) .get(); } +const processingQueries = new WeakMap>(); + function readProcessingByCurrentTurn(db: AppDb, sessionId: string, currentTurnId: string) { + let query = processingQueries.get(db); + if (!query) { + query = prepareProcessingQuery(db); + processingQueries.set(db, query); + } + return query.get({ sessionId, currentTurnId }); +} + +function prepareProcessingQuery(db: AppDb) { return db .select() .from(queryCollapseViewStates) .where( and( - eq(queryCollapseViewStates.sessionId, sessionId), - eq(queryCollapseViewStates.currentTurnId, currentTurnId), + eq(queryCollapseViewStates.sessionId, sql.placeholder('sessionId')), + eq(queryCollapseViewStates.currentTurnId, sql.placeholder('currentTurnId')), isNull(queryCollapseViewStates.processingFinishedAtMs), ), ) .orderBy(desc(queryCollapseViewStates.updatedAtMs), desc(queryCollapseViewStates.queryKey)) - .get(); + .prepare(); } function readByKey(db: AppDb, sessionId: string, queryKey: string) { 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/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/packages/local-runtime-v2/src/service/turn-system/agent-host/events/contracts.ts b/packages/local-runtime-v2/src/service/turn-system/agent-host/events/contracts.ts index 65d6e2f8..f2008e06 100644 --- a/packages/local-runtime-v2/src/service/turn-system/agent-host/events/contracts.ts +++ b/packages/local-runtime-v2/src/service/turn-system/agent-host/events/contracts.ts @@ -30,6 +30,10 @@ export type AgentEventResult = }; export interface AgentEventDelivery { - handleRuntimeEvent(context: AgentEventContext, event: RuntimeEvent): Promise; + handleRuntimeEvent( + context: AgentEventContext, + event: RuntimeEvent, + signal?: AbortSignal, + ): Promise; handleHistoryCommitted(context: AgentEventContext, change: CommittedHistoryChange): Promise; } diff --git a/packages/local-runtime-v2/src/service/turn-system/agent-host/events/required-agent-event-delivery.ts b/packages/local-runtime-v2/src/service/turn-system/agent-host/events/required-agent-event-delivery.ts index eb3d0be0..c43e0558 100644 --- a/packages/local-runtime-v2/src/service/turn-system/agent-host/events/required-agent-event-delivery.ts +++ b/packages/local-runtime-v2/src/service/turn-system/agent-host/events/required-agent-event-delivery.ts @@ -23,6 +23,8 @@ type ObservationStage = 'runtime-event' | 'history-committed' | 'history-failure interface RuntimeProjectionInput { readonly context: AgentEventContext; readonly event: RuntimeEvent; + /** Process-local control; excluded from semantic snapshots and replay identities. */ + readonly signal?: AbortSignal; } interface HistoryProjectionInput { @@ -156,7 +158,11 @@ export class RequiredAgentEventDelivery implements AgentEventDelivery, AgentHost this.historyReplays = new SemanticReplayRegistry(maximum); } - handleRuntimeEvent(context: AgentEventContext, event: RuntimeEvent): Promise { + handleRuntimeEvent( + context: AgentEventContext, + event: RuntimeEvent, + signal?: AbortSignal, + ): Promise { try { const snapshot = captureSemanticSnapshot({ context, event }); validateRuntimeInput(snapshot.value.context, snapshot.value.event); @@ -167,7 +173,7 @@ export class RequiredAgentEventDelivery implements AgentEventDelivery, AgentHost conflict: () => new AgentEventIdentityConflictError('runtime-event', identity), execute: () => this.lane.run(snapshot.value.context.sessionId, () => - this.projectRuntime(snapshot.value.context, snapshot.value.event), + this.projectRuntime(snapshot.value.context, snapshot.value.event, signal), ), }); } catch (error) { @@ -223,6 +229,7 @@ export class RequiredAgentEventDelivery implements AgentEventDelivery, AgentHost private async projectRuntime( context: AgentEventContext, event: RuntimeEvent, + signal?: AbortSignal, ): Promise { const runtimeSequence = this.validateSequence(context, event); const authoritative = await this.options.projectors.session.projectRuntimeEvent({ @@ -233,7 +240,11 @@ export class RequiredAgentEventDelivery implements AgentEventDelivery, AgentHost throw new AgentEventAcknowledgementError(terminalOutcome(event) ?? 'non-terminal', 'missing'); } validateAcknowledgement(event, authoritative); - await this.options.projectors.messages.projectRuntimeEvent({ context, event }); + await this.options.projectors.messages.projectRuntimeEvent({ + context, + event, + ...(signal ? { signal } : {}), + }); await this.options.projectors.stream.projectRuntimeEvent({ context, event }); await this.options.projectors.turnFacts.projectRuntimeEvent({ context, event }); this.commitSequence(context, runtimeSequence); diff --git a/packages/local-runtime-v2/src/service/turn-system/agent-host/events/turn-commit-pipeline.ts b/packages/local-runtime-v2/src/service/turn-system/agent-host/events/turn-commit-pipeline.ts index 5ec1d43b..bc11dd9b 100644 --- a/packages/local-runtime-v2/src/service/turn-system/agent-host/events/turn-commit-pipeline.ts +++ b/packages/local-runtime-v2/src/service/turn-system/agent-host/events/turn-commit-pipeline.ts @@ -34,6 +34,7 @@ import { TurnCommittedHistoryState } from '../history/turn-committed-history-sta import { CanonicalUserMessageIdentityLane } from '../history/canonical-user-message-identities.js'; import { AgentEventAssociationError } from '../preparation/turn-preflight.js'; import type { UserMessageId } from '../../../session-system/index.js'; +import { WriteLockWaitAbortedError } from '../../../../infra/db/write-transaction.js'; export class AgentTerminalConfirmationError extends Error { override readonly name = 'AgentTerminalConfirmationError'; @@ -147,11 +148,26 @@ export class TurnCommitPipeline { } await this.lane.enqueue(async () => { validateRuntimeAssociation(this.dependencies.context, event); - const result = await this.dependencies.events.handleRuntimeEvent( - this.dependencies.context, - event, - ); - new TerminalConfirmation().observe(event, result); + const signal = this.dependencies.lease.signal; + try { + const result = await this.dependencies.events.handleRuntimeEvent( + this.dependencies.context, + event, + signal, + ); + new TerminalConfirmation().observe(event, result); + } catch (error) { + if ( + error instanceof WriteLockWaitAbortedError && + error.signal === signal && + signal.aborted + ) { + // This lease cancelled a projection before its write began. Keep the + // lane available for the runner's abort reconciliation and terminal. + return; + } + throw error; + } }); }; 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 e55a311e..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,14 +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 Object.freeze({ + const result = captureSemanticSnapshot({ revision: detached.revision.trim(), - messages: Object.freeze([...detached.messages]), - identityVector: Object.freeze([...detached.identityVector]), - }); + // 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/incremental-sha256.ts b/packages/local-runtime-v2/src/service/turn-system/agent-host/history/incremental-sha256.ts index 0ae6c201..f2df698c 100644 --- a/packages/local-runtime-v2/src/service/turn-system/agent-host/history/incremental-sha256.ts +++ b/packages/local-runtime-v2/src/service/turn-system/agent-host/history/incremental-sha256.ts @@ -2,26 +2,42 @@ import { createHash } from 'node:crypto'; const UTF8_CHUNK_CODE_UNITS = 8 * 1_024; -/** Native SHA-256 with bounded UTF-8 chunks and the existing update boundaries. */ +/** Native SHA-256 with bounded batching and independent UTF-8 update boundaries. */ export class IncrementalSha256 { private readonly hash = createHash('sha256'); private finalized = false; + private pending = ''; 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.hash.update(value.slice(offset, end), 'utf8'); + let chunk = value.slice(offset, end); + // Each update encodes separately. A trailing high surrogate must not pair + // with a low surrogate supplied by a later update when we batch strings. + if (isHighSurrogate(chunk.charCodeAt(chunk.length - 1))) { + chunk = chunk.slice(0, -1) + '\ufffd'; + } + if (this.pending.length + chunk.length > UTF8_CHUNK_CODE_UNITS) this.flush(); + this.pending += chunk; + if (this.pending.length === UTF8_CHUNK_CODE_UNITS) this.flush(); offset = end; } } digestHex(): string { if (this.finalized) throw new Error('SHA-256 digest is already finalized.'); + this.flush(); this.finalized = true; return this.hash.digest('hex'); } + + private flush(): void { + if (this.pending.length === 0) return; + this.hash.update(this.pending, 'utf8'); + this.pending = ''; + } } function safeUtf8ChunkEnd(value: string, offset: number): number { 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..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; } @@ -118,27 +151,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/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-v2/src/service/turn-system/persistence/turn.repository.ts b/packages/local-runtime-v2/src/service/turn-system/persistence/turn.repository.ts index e310741f..ab24d34d 100644 --- a/packages/local-runtime-v2/src/service/turn-system/persistence/turn.repository.ts +++ b/packages/local-runtime-v2/src/service/turn-system/persistence/turn.repository.ts @@ -1,6 +1,6 @@ import { randomUUID } from 'node:crypto'; -import { and, desc, eq, gt, inArray } from 'drizzle-orm'; +import { and, desc, eq, gt, inArray, sql } from 'drizzle-orm'; import { sessionLocks, @@ -491,11 +491,29 @@ async function completeTurnSessionDeletion( ); } +const receiptQueries = new WeakMap< + TurnRepositoryOptions['db'], + ReturnType +>(); + +function prepareReceiptQuery(db: TurnRepositoryOptions['db']) { + return db + .select() + .from(turnIngress) + .where(eq(turnIngress.turnId, sql.placeholder('turnId'))) + .prepare(); +} + async function findTurnReceipt( options: TurnRepositoryOptions, turnId: string, ): ReturnType { - const receipt = options.db.select().from(turnIngress).where(eq(turnIngress.turnId, turnId)).get(); + let query = receiptQueries.get(options.db); + if (!query) { + query = prepareReceiptQuery(options.db); + receiptQueries.set(options.db, query); + } + const receipt = query.get({ turnId }); if (!receipt) return undefined; if ( !storedBusyReason(receipt.busyReason) || 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..87097d83 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 { @@ -80,6 +81,34 @@ async function fixture() { } const poll = { wait: false, readTaskIds: new Set() }; +// Force Windows PowerShell 5.1 even when pwsh 7 is installed on the host. +// These are real shell checks, not a simulation of PowerShell exit semantics. +describe.skipIf(process.platform !== 'win32')('Windows PowerShell 5.1 exit codes', () => { + const shellPath = join(process.env.SystemRoot ?? 'C:\\Windows', 'System32', 'WindowsPowerShell', 'v1.0', 'powershell.exe'); + const quote = (value: string) => `'${value.replaceAll("'", "''")}'`; + const native = (code: number) => `& ${quote(process.execPath)} -e ${quote(`console.error('native-evidence');process.exit(${code})`)}`; + + describe.each([false, true])('parentDeathGuard=%s', (parentDeathGuard) => { + it.each([ + ['native failure', native(7), 7, 'native-evidence'], + ['native failure followed by PowerShell output', `${native(7)}; Write-Output 'tail'`, 7, 'tail'], + ['native success', native(0), 0, 'native-evidence'], + ['last native command succeeds', `${native(7)}; ${native(0)}`, 0, 'native-evidence'], + ['pure PowerShell success', "Write-Output 'success-evidence'", 0, 'success-evidence'], + ['terminating PowerShell error', "throw 'ps-error'", 1, 'ps-error'], + ['explicit exit', 'exit 9', 9, ''], + ] as const)('%s', async (_name, command, exitCode, evidence) => { + let output = ''; + const result = await createLocalBashOperations({ shellPath, parentDeathGuard }).exec(command, tmpdir(), { + onData: (chunk) => { output += chunk.toString(); }, + timeout: 10, + }); + expect(output).toContain(evidence); + expect(result.exitCode).toBe(exitCode); + }, 15_000); + }); +}); + describe('child Bash lifecycle', () => { it.each(['succeeded', 'failed', 'canceled', 'lost'] as const)( 'notifies %s exactly once without acknowledging consumption', diff --git a/packages/tui/src/headless/invocation.ts b/packages/tui/src/headless/invocation.ts index 8317fda8..b78aa98c 100644 --- a/packages/tui/src/headless/invocation.ts +++ b/packages/tui/src/headless/invocation.ts @@ -10,6 +10,7 @@ import { MINIMAX_CODE_MAX_ATTACHMENT_COUNT, } from '../application/attachment-policy.js'; import { inferTuiNativeVideoMimeType } from '../application/video-mime.js'; +import { resolveWslPath } from '../host/wsl-path.js'; import { TuiExecError } from './exit-policy.js'; import type { TuiExecFormat } from './output.js'; @@ -295,7 +296,12 @@ async function resolveHeadlessAttachment( signal?: AbortSignal, ): Promise { throwIfAborted(signal); - const requestedPath = expandPath(reference, workspaceDir); + const localReference = await resolveWslPath(reference.trim(), signal).catch((error: unknown) => { + throwIfAborted(signal); + throw invocationError(`Cannot attach ${reference}: ${errorMessage(error)}`); + }); + throwIfAborted(signal); + const requestedPath = expandPath(localReference, workspaceDir); const filePath = await realpath(requestedPath).catch((error: unknown) => { throw invocationError(`Cannot attach ${reference}: ${errorMessage(error)}`); }); 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/packages/tui/src/host/wsl-path.ts b/packages/tui/src/host/wsl-path.ts new file mode 100644 index 00000000..ba87c2ed --- /dev/null +++ b/packages/tui/src/host/wsl-path.ts @@ -0,0 +1,46 @@ +import { execFile } from "node:child_process"; +import { platform, release } from "node:os"; +import { posix } from "node:path"; +import { promisify } from "node:util"; + +const executeFile = promisify(execFile); + +/** Translate Windows absolute paths before the host's POSIX resolver sees them. */ +export async function resolveWslPath( + reference: string, + signal?: AbortSignal, +): Promise { + if ( + !/^(?:[a-z]:[\\/]|\\\\)/iu.test(reference) || + platform() !== "linux" || + !( + process.env.WSL_DISTRO_NAME || + process.env.WSL_INTEROP || + process.env.WSLENV || + /microsoft/iu.test(release()) + ) + ) { + return reference; + } + + // Use the distro's converter so custom automount roots and mounted drives work. + // execFile keeps spaces, backslashes and shell metacharacters in one literal argument. + try { + const { stdout } = await executeFile("wslpath", ["-a", "-u", reference], { + encoding: "utf8", + timeout: 1_000, + maxBuffer: 64 * 1024, + ...(signal ? { signal } : {}), + }); + const mapped = stdout.replace(/\r?\n$/u, ""); + if (!posix.isAbsolute(mapped) || /[\r\n\0]/u.test(mapped)) { + throw new Error("wslpath did not return an absolute Linux path."); + } + return mapped; + } catch (error) { + throw new Error( + "Could not convert the Windows path with wslpath. Use an accessible Linux path instead.", + { cause: error }, + ); + } +} diff --git a/packages/tui/src/tui/features/composer/attachments.ts b/packages/tui/src/tui/features/composer/attachments.ts index 6464bf4a..d953115c 100644 --- a/packages/tui/src/tui/features/composer/attachments.ts +++ b/packages/tui/src/tui/features/composer/attachments.ts @@ -3,6 +3,7 @@ import { homedir } from 'node:os'; import { basename, extname, isAbsolute, resolve } from 'node:path'; import type { TuiAttachment } from '../../../application/invocation.js'; import { inferTuiNativeVideoMimeType } from '../../../application/video-mime.js'; +import { resolveWslPath } from '../../../host/wsl-path.js'; import { sanitizeTerminalText } from '../../rendering/terminal-text.js'; import type { TranscriptAttachment } from '../../transcript/model.js'; import { getTuiTerminalImagePasteFallbackPath } from './terminal-image-paste.js'; @@ -54,12 +55,13 @@ export async function resolveTuiAttachment( const normalizedReference = stripMatchingQuotes(reference.trim()); if (!normalizedReference) throw new Error('Attachment path is required.'); const userHome = options.homeDir ?? homedir(); - const expanded = + const expanded = await resolveWslPath( normalizedReference === '~' ? userHome : normalizedReference.startsWith('~/') || normalizedReference.startsWith('~\\') ? resolve(userHome, normalizedReference.slice(2)) - : normalizedReference; + : normalizedReference, + ); let filePath = isAbsolute(expanded) ? resolve(expanded) : resolve(options.workspaceDir, expanded); const info = await stat(filePath) .catch((error: unknown) => { diff --git a/packages/tui/test/unit/wsl-attachment-paths.test.ts b/packages/tui/test/unit/wsl-attachment-paths.test.ts new file mode 100644 index 00000000..48d2d20c --- /dev/null +++ b/packages/tui/test/unit/wsl-attachment-paths.test.ts @@ -0,0 +1,253 @@ +import { mkdtemp, realpath, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { resolveTuiExecInvocation } from "../../src/headless/invocation.js"; +import { resolveWslPath } from "../../src/host/wsl-path.js"; +import { resolveTuiAttachment } from "../../src/tui/features/composer/attachments.js"; +import { TuiComposerDraft } from "../../src/tui/features/composer/draft.js"; +import { isTuiTerminalImagePaste } from "../../src/tui/features/composer/terminal-image-paste.js"; + +const host = vi.hoisted(() => ({ + platform: vi.fn(() => "linux"), + release: vi.fn(() => "6.18.33.2-microsoft-standard-WSL2"), + executeFile: vi.fn(), +})); + +vi.mock("node:os", async (importOriginal) => ({ + ...(await importOriginal()), + platform: host.platform, + release: host.release, +})); +vi.mock("node:child_process", async (importOriginal) => ({ + ...(await importOriginal()), + execFile: Object.assign(vi.fn(), { + [Symbol.for("nodejs.util.promisify.custom")]: host.executeFile, + }), +})); + +let workspaceDir: string; +let imagePath: string; +const windowsPath = String.raw`D:\Users\demo\Documents\Screen shots\ζˆͺε›Ύ.png`; + +beforeEach(async () => { + host.platform.mockReturnValue("linux"); + host.release.mockReturnValue("6.18.33.2-microsoft-standard-WSL2"); + host.executeFile.mockReset(); + vi.stubEnv("WSL_DISTRO_NAME", ""); + vi.stubEnv("WSL_INTEROP", ""); + vi.stubEnv("WSLENV", ""); + workspaceDir = await mkdtemp(join(tmpdir(), "mcode-wsl-path-")); + imagePath = join(workspaceDir, "ζˆͺε›Ύ.png"); + await writeFile(imagePath, Buffer.from([0x89, 0x50, 0x4e, 0x47])); + host.executeFile.mockResolvedValue({ stdout: `${imagePath}\n`, stderr: "" }); +}); + +afterEach(async () => { + vi.unstubAllEnvs(); + await rm(workspaceDir, { recursive: true, force: true }); +}); + +describe("WSL attachment paths", () => { + it.each([ + windowsPath, + `"${windowsPath}"`, + `'${windowsPath}'`, + windowsPath.replaceAll("\\", "/"), + ])( + "queues a Windows image path as an image placeholder: %s", + async (reference) => { + const placeholders = vi.fn(); + const draft = new TuiComposerDraft({ + workspaceDir, + resolveAttachment: resolveTuiAttachment, + append: vi.fn(), + onChanged: vi.fn(), + onAttachmentPlaceholdersChanged: placeholders, + }); + expect(isTuiTerminalImagePaste(reference)).toBe(true); + await draft.queueAttachment(reference, { source: "terminal-paste" }); + expect(draft.snapshot().attachments).toEqual([ + { + type: "image", + filePath: imagePath, + fileName: "ζˆͺε›Ύ.png", + mimeType: "image/png", + sizeBytes: 4, + }, + ]); + expect(placeholders).toHaveBeenLastCalledWith([ + { id: imagePath, label: "[Image #1]" }, + ]); + expect(host.executeFile).toHaveBeenCalledWith( + "wslpath", + ["-a", "-u", reference.replace(/^['"]|['"]$/gu, "")], + expect.objectContaining({ + encoding: "utf8", + timeout: 1_000, + maxBuffer: 64 * 1024, + }), + ); + }, + ); + + it("resolves headless --file through the same conversion before realpath", async () => { + const invocation = await resolveTuiExecInvocation( + "describe", + { cwd: workspaceDir, file: [windowsPath] }, + async () => "", + ); + expect(invocation.attachments).toEqual([ + { + type: "image", + filePath: await realpath(imagePath), + fileName: "ζˆͺε›Ύ.png", + mimeType: "image/png", + sizeBytes: 4, + }, + ]); + expect(host.executeFile).toHaveBeenCalledOnce(); + }); + + it.each(["WSL_DISTRO_NAME", "WSL_INTEROP", "WSLENV"])( + "detects WSL via %s even without a Microsoft kernel name", + async (name) => { + host.release.mockReturnValue("custom-kernel"); + vi.stubEnv(name, "synthetic-wsl"); + await expect(resolveWslPath(windowsPath)).resolves.toBe(imagePath); + }, + ); + + it.each(["darwin", "win32", "linux"])( + "preserves Windows path syntax on a non-WSL %s host", + async (platform) => { + host.platform.mockReturnValue(platform); + host.release.mockReturnValue("generic-kernel"); + await expect(resolveWslPath(windowsPath)).resolves.toBe(windowsPath); + expect(host.executeFile).not.toHaveBeenCalled(); + }, + ); + + it.each([ + "/mnt/d/Screen shots/ζˆͺε›Ύ.png", + "/tmp/literal\\name.png", + "//tmp/image.png", + "image.png", + "~/image.png", + "D:relative.png", + ])( + "leaves native and relative paths unchanged in WSL: %s", + async (reference) => { + await expect(resolveWslPath(reference)).resolves.toBe(reference); + expect(host.executeFile).not.toHaveBeenCalled(); + }, + ); + + it("keeps native attachment resolution working in WSL without a subprocess", async () => { + await expect( + resolveTuiAttachment("ζˆͺε›Ύ.png", { workspaceDir }), + ).resolves.toMatchObject({ filePath: imagePath }); + await expect( + resolveTuiAttachment("~/ζˆͺε›Ύ.png", { + workspaceDir, + homeDir: workspaceDir, + }), + ).resolves.toMatchObject({ filePath: imagePath }); + expect(host.executeFile).not.toHaveBeenCalled(); + }); + + it.each([ + String.raw`d:\Screen shots\$(touch marker);'ζˆͺε›Ύ'.png`, + String.raw`\\server\share\ζˆͺε›Ύ.png`, + ])("passes Windows paths as a literal argument: %s", async (reference) => { + await expect(resolveWslPath(reference)).resolves.toBe(imagePath); + expect(host.executeFile.mock.calls[0]?.slice(0, 2)).toEqual([ + "wslpath", + ["-a", "-u", reference], + ]); + expect(host.executeFile.mock.calls[0]?.[2]).not.toHaveProperty("shell"); + }); + + it("honors the converter output for custom mount roots and preserves trailing spaces", async () => { + host.executeFile.mockResolvedValue({ + stdout: "/custom/windows/d/Screen shots/image.png \n", + }); + await expect(resolveWslPath(windowsPath)).resolves.toBe( + "/custom/windows/d/Screen shots/image.png ", + ); + }); + + it.each([ + "", + "\n", + "relative/image.png\n", + "D:\\image.png\n", + "/tmp/image.png\nextra\n", + "/tmp/image\0.png\n", + ])( + "rejects invalid converter output instead of treating it as a local path: %j", + async (stdout) => { + host.executeFile.mockResolvedValue({ stdout }); + await expect( + resolveTuiAttachment(windowsPath, { workspaceDir }), + ).rejects.toThrow("Could not convert the Windows path with wslpath"); + }, + ); + + it.each(["ENOENT", "EACCES", "ETIMEDOUT"])( + "reports converter failure %s without falling back to a workspace filename", + async (code) => { + if (process.platform !== "win32") + await writeFile(join(workspaceDir, windowsPath), "wrong image"); + host.executeFile.mockRejectedValue( + Object.assign(new Error("converter failed"), { code }), + ); + await expect( + resolveTuiAttachment(windowsPath, { + workspaceDir, + source: "terminal-paste", + }), + ).rejects.toThrow("Use an accessible Linux path instead"); + await expect( + resolveTuiExecInvocation( + "describe", + { cwd: workspaceDir, file: [windowsPath] }, + async () => "", + ), + ).rejects.toMatchObject({ + kind: "invocation", + message: expect.stringContaining("wslpath"), + }); + }, + ); + + it("still rejects missing files and directories after conversion", async () => { + host.executeFile.mockResolvedValue({ + stdout: `${join(workspaceDir, "missing.png")}\n`, + }); + await expect( + resolveTuiAttachment(windowsPath, { workspaceDir }), + ).rejects.toThrow("ENOENT"); + host.executeFile.mockResolvedValue({ stdout: `${workspaceDir}\n` }); + await expect( + resolveTuiAttachment(windowsPath, { workspaceDir }), + ).rejects.toThrow("not a file"); + }); + + it("preserves headless cancellation while converting a path", async () => { + const controller = new AbortController(); + host.executeFile.mockImplementation(async (_file, _args, options) => { + expect(options.signal).toBe(controller.signal); + controller.abort(); + throw new Error("aborted"); + }); + await expect( + resolveTuiExecInvocation( + "describe", + { cwd: workspaceDir, file: [windowsPath] }, + async () => "", + controller.signal, + ), + ).rejects.toMatchObject({ kind: "cancelled" }); + }); +}); diff --git a/release/public-source.json b/release/public-source.json index 8870c53f..b7597ec2 100644 --- a/release/public-source.json +++ b/release/public-source.json @@ -475,6 +475,8 @@ "packages/config/src/tui-status-line-write.ts", "packages/config/test/config-file-permissions.test.ts", "packages/config/test/config-update-preservation.test.ts", + "packages/config/test/managed-preset-sync.test.ts", + "packages/config/test/private-config-file.test.ts", "packages/local-runtime-v2/assets/agents/_default/prompt-base-all.md", "packages/local-runtime-v2/assets/agents/_default/prompt-base-all.md.hbs", "packages/local-runtime-v2/assets/agents/_default/prompt-base-windows.md", @@ -721,6 +723,7 @@ "packages/local-runtime-v2/src/infra/db/schema/turn.ts", "packages/local-runtime-v2/src/infra/db/schema/usage.ts", "packages/local-runtime-v2/src/infra/db/sql-contract.ts", + "packages/local-runtime-v2/src/infra/db/write-transaction.ts", "packages/local-runtime-v2/src/infra/event-bus/index.ts", "packages/local-runtime-v2/src/infra/file/canonical-history-artifact.ts", "packages/local-runtime-v2/src/infra/file/canonical-history-json-value.ts", @@ -2891,6 +2894,7 @@ "packages/tui/src/host/transcript-export.ts", "packages/tui/src/host/tui-keybindings.ts", "packages/tui/src/host/tui-settings.ts", + "packages/tui/src/host/wsl-path.ts", "packages/tui/src/index.ts", "packages/tui/src/observability/incident-reporter.ts", "packages/tui/src/observability/index.ts", @@ -3296,6 +3300,7 @@ "packages/tui/test/unit/tui/widgets/editor/editor-behavior.test.ts", "packages/tui/test/unit/update-application.test.ts", "packages/tui/test/unit/update-service.test.ts", + "packages/tui/test/unit/wsl-attachment-paths.test.ts", "packages/webui/.cve-ignore.json", "packages/webui/README.md", "packages/webui/README.zh-CN.md", @@ -3569,6 +3574,7 @@ "test/public-artifact.test.mjs", "test/smoke.test.mjs", "test/source-sync.test.mjs", + "test/sqlite-message-contention.test.ts", "test/vitest-suites.json", "third_party/pi-mono/.minimax-vendor.json", "third_party/pi-mono/LICENSE", diff --git a/scripts/build.mjs b/scripts/build.mjs index 22f2ca39..5b0ed0f4 100644 --- a/scripts/build.mjs +++ b/scripts/build.mjs @@ -91,6 +91,7 @@ const result = await build({ format: "esm", platform: "node", minifyIdentifiers: true, + minifyWhitespace: true, target: "node22", chunkNames: "chunks/[name]-[hash]", banner: { js: location.banner }, diff --git a/test/byok.test.mjs b/test/byok.test.mjs index fcbc4e29..15feaf54 100644 --- a/test/byok.test.mjs +++ b/test/byok.test.mjs @@ -2,12 +2,15 @@ import test from "node:test"; import assert from "node:assert/strict"; import { createServer } from "node:http"; import { spawn } from "node:child_process"; +import { once } from "node:events"; +import { createRequire } from "node:module"; import { mkdtempSync, mkdirSync, rmSync, existsSync, readFileSync, + readdirSync, writeFileSync, } from "node:fs"; import { tmpdir } from "node:os"; @@ -524,3 +527,304 @@ test( } }, ); + +test( + "cancelling a contended message write persists an aborted turn rather than a failure", + // Node terminates children on Windows SIGINT instead of invoking their handler. + { timeout: 45000, skip: process.platform === "win32" }, + cancellationTest("lock"), +); + +test( + "cancelling a running tool preserves its completed display message", + { timeout: 45000, skip: process.platform === "win32" }, + cancellationTest("tool"), +); + +function cancellationTest(cancellation) { + return async (t) => { + const fixtureDir = mkdtempSync(path.join(tmpdir(), "minimax-code-cancel-write-")); + const dataDir = path.join(fixtureDir, "data"); + const workspaceDir = path.join(fixtureDir, "workspace"); + const homeDir = path.join(fixtureDir, "home"); + for (const dir of [dataDir, workspaceDir, homeDir]) mkdirSync(dir); + const dbPath = path.join(dataDir, "v2", "sqlite", "runtime-state.sqlite"); + const networkAudit = path.join(fixtureDir, "network-audit.log"); + let child, holder, holderClosed, cancelTimer, deadline; + let serverError; + let cancelledWhileLocked = false; + let cancelledDuringTool = false; + const marker = path.join(workspaceDir, "cancel-tool.marker"); + let holderReleased = false; + const server = createServer(async (req, res) => { + try { + let raw = ""; + for await (const chunk of req) raw += chunk; + const body = JSON.parse(raw); + if (req.url?.endsWith("/responses/input_tokens")) { + res + .writeHead(200, { "content-type": "application/json" }) + .end(JSON.stringify({ input_tokens: 1 })); + return; + } + const content = "SYNTHETIC_CANCELLED_ANSWER"; + if (!body.stream) { + res.writeHead(200, { "content-type": "application/json" }).end( + JSON.stringify({ + id: "fixture", + object: "chat.completion", + model: "fixture", + choices: [ + { index: 0, message: { role: "assistant", content }, finish_reason: "stop" }, + ], + }), + ); + return; + } + if (cancellation === "lock") { + // A separate process releases the lock even when the CLI blocks its event loop. + holder = spawn( + process.execPath, + [ + "-e", + ` + const Database = require(process.argv[1]); + const db = new Database(process.argv[2]); + db.exec('BEGIN IMMEDIATE'); + process.send('locked'); + setTimeout(() => { + db.exec('COMMIT'); db.close(); process.disconnect(); + }, 1500); + `, + createRequire(import.meta.url).resolve("better-sqlite3"), + dbPath, + ], + { + stdio: ["ignore", "ignore", "inherit", "ipc"], + }, + ); + holderClosed = once(holder, "close").then(() => { + holderReleased = true; + }); + await Promise.race([ + once(holder, "message"), + holderClosed.then(() => { + throw new Error("Writer exited before acquiring the lock"); + }), + ]); + } + const delta = + cancellation === "tool" + ? { + role: "assistant", + tool_calls: [ + { + index: 0, + id: "cancel-tool-call", + type: "function", + function: { + name: "bash", + arguments: JSON.stringify({ + command: "printf started > cancel-tool.marker; sleep 30", + }), + }, + }, + ], + } + : { role: "assistant", content }; + res.writeHead(200, { "content-type": "text/event-stream" }); + for (const chunk of [ + { + choices: [{ index: 0, delta, finish_reason: null }], + }, + { + choices: [ + { + index: 0, + delta: {}, + finish_reason: cancellation === "tool" ? "tool_calls" : "stop", + }, + ], + usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 }, + }, + ]) + res.write( + `data: ${JSON.stringify({ + id: "fixture", + object: "chat.completion.chunk", + created: 1, + model: "fixture", + ...chunk, + })}\n\n`, + ); + res.end("data: [DONE]\n\n"); + if (cancellation === "tool") { + cancelTimer = setInterval(() => { + if (!existsSync(marker)) return; + clearInterval(cancelTimer); + cancelledDuringTool = true; + child.kill("SIGINT"); + }, 20); + } else { + cancelTimer = setTimeout(() => { + cancelledWhileLocked = !holderReleased; + child.kill("SIGINT"); + }, 200); + } + } catch (error) { + serverError = error; + res.writeHead(500).end(); + } + }); + t.after(async () => { + clearTimeout(cancelTimer); + clearTimeout(deadline); + for (const childProcess of [child, holder]) { + if ( + childProcess && + childProcess.exitCode === null && + childProcess.signalCode === null + ) { + const closed = once(childProcess, "close"); + childProcess.kill("SIGKILL"); + await closed; + } + } + server.closeAllConnections(); + await new Promise((resolve) => server.close(resolve)); + rmSync(fixtureDir, { recursive: true, force: true }); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const origin = `http://127.0.0.1:${server.address().port}`; + writeFileSync( + path.join(dataDir, "config.yaml"), + stringifyYaml({ + custom_provider: { + fixture: { + name: "fixture", + kind: "custom", + enabled: true, + api: "openai-completions", + options: { + apiKey: "synthetic-key", + baseURL: `${origin}/v1`, + authMode: "api-key", + }, + models: { fixture: { limit: { context: 32768, output: 4096 } } }, + }, + }, + }), + ); + child = spawn( + process.execPath, + [ + cli, + "exec", + "Reply with the synthetic fixture response.", + "--model", + "custom_provider:fixture/fixture", + "--permission", + "off", + "--cwd", + workspaceDir, + "--timeout", + "30s", + "--max-steps", + "1", + ], + { + cwd: workspaceDir, + env: { + ...withoutProxyEnvironment(process.env), + HOME: homeDir, + XDG_CONFIG_HOME: path.join(homeDir, "config"), + XDG_DATA_HOME: path.join(homeDir, "data"), + MINIMAX_DATA_DIR: dataDir, + MAVIS_DATA_DIR: dataDir, + MCODE_TEST_ALLOWED_ORIGIN: origin, + MCODE_TEST_NETWORK_AUDIT: networkAudit, + MCODE_TEST_MANAGED_OFFLINE: "1", + NODE_OPTIONS: `--import=${new URL("./network-deny.mjs", import.meta.url).href}`, + }, + stdio: ["ignore", "pipe", "pipe"], + }, + ); + let stdout = "", + stderr = ""; + child.stdout.on("data", (chunk) => { + stdout += chunk; + }); + child.stderr.on("data", (chunk) => { + stderr += chunk; + }); + deadline = setTimeout(() => child.kill("SIGKILL"), 35000); + const [code] = await once(child, "close"); + clearTimeout(deadline); + if (holderClosed) await holderClosed; + assert.equal(serverError, undefined); + if (cancellation === "tool") { + assert.equal(cancelledDuringTool, true, "SIGINT must follow actual Bash execution"); + assert.equal(holder, undefined, "Tool cancellation must not involve a foreign writer"); + } else { + assert.equal( + cancelledWhileLocked, + true, + "SIGINT must arrive while the foreign writer holds its lock", + ); + } + assert.equal(code, 130, `${stdout}\n${stderr}`); + assert.equal(existsSync(networkAudit), false, "No external requests are allowed"); + const db = new Database(dbPath, { readonly: true }); + try { + assert.deepEqual(db.prepare("SELECT status FROM local_runtime_turn_ingress").all(), [ + { status: "aborted" }, + ]); + assert.deepEqual( + db.prepare("SELECT status, error_message FROM local_runtime_sessions").all(), + [{ status: "aborted", error_message: null }], + ); + assert.deepEqual( + db.prepare("SELECT terminal_outcome FROM local_runtime_session_agent_state").all(), + [{ terminal_outcome: "aborted" }], + ); + if (cancellation === "tool") { + const display = db + .prepare( + "SELECT data_json FROM local_runtime_message_rows WHERE role = 'assistant'", + ) + .all() + .map((row) => JSON.parse(row.data_json)); + assert.equal( + display.length, + 1, + "Tool completion must survive reopening display history", + ); + const calls = display[0].tool_calls ?? []; + assert.equal(calls.length, 1); + assert.equal(calls[0].tool_call_id, "cancel-tool-call"); + assert.equal(calls[0].tool_name, "bash"); + assert.ok( + calls[0].tool_call_result_data, + "Persist the completed tool result as well as its call", + ); + } + } finally { + db.close(); + } + const sessionsDir = path.join(dataDir, "v2", "sessions"); + const histories = readdirSync(sessionsDir, { recursive: true }).filter((file) => + file.endsWith("messages.jsonl"), + ); + assert.equal(histories.length, 1); + const roles = readFileSync(path.join(sessionsDir, histories[0]), "utf8") + .trim() + .split("\n") + .filter(Boolean) + .map((line) => JSON.parse(line).message?.role); + assert.deepEqual( + roles, + cancellation === "tool" ? ["user", "assistant", "toolResult"] : ["user"], + "Abort reconciliation must preserve executed tools without retaining cancelled text output", + ); + }; +} diff --git a/test/history-processing.test.ts b/test/history-processing.test.ts index 634e5d96..68be2767 100644 --- a/test/history-processing.test.ts +++ b/test/history-processing.test.ts @@ -1,11 +1,33 @@ import { createHash } from 'node:crypto'; -import { CanonicalHistoryJsonlDataSource } from '../packages/local-runtime-v2/src/infra/file/canonical-history-jsonl.js'; +import { + canonicalActiveHistoryRevision, + 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'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { describe, expect, it, vi } from 'vitest'; +import { eq } from 'drizzle-orm'; +import { DatabaseClient } from '../packages/local-runtime-v2/src/infra/db/client.js'; +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'; @@ -18,6 +40,205 @@ import { } from '../packages/local-runtime-v2/src/service/session-system/messages/history/session-history-paths.js'; import type { SessionRecord } from '../packages/local-runtime-v2/src/service/session-system/sessions/repo/contract.js'; +describe('prepared runtime reads', () => { + async function withDatabase( + run: (client: DatabaseClient, writer: DatabaseClient) => Promise, + ) { + const dataDir = await mkdtemp(join(tmpdir(), 'mcode-prepared-reads-')); + const client = new DatabaseClient({ dataDir }); + const writer = new DatabaseClient({ dataDir }); + try { + await initializeDatabase({ database: client, dataDir }); + await run(client, writer); + } finally { + writer.close(); + client.close(); + await rm(dataDir, { recursive: true, force: true }); + } + } + + 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 }); + expect(await state.findProcessingByCurrentTurn('s1', 't1')).toBeUndefined(); + await state.start({ sessionId: 's1', currentTurnId: 't1', queryKey: 'a' }); + await state.start({ sessionId: 's1', currentTurnId: 't1', queryKey: 'b' }); + await state.start({ sessionId: 's2', currentTurnId: 't1', queryKey: 'other' }); + expect((await state.findProcessingByCurrentTurn('s1', 't1'))?.queryKey).toBe('b'); + expect((await state.findProcessingByCurrentTurn('s2', 't1'))?.queryKey).toBe('other'); + expect(await state.findProcessingByCurrentTurn('s1', 'missing')).toBeUndefined(); + await state.finish({ + sessionId: 's1', + currentTurnId: 't1', + queryKey: 'b', + forceExpanded: false, + }); + expect((await state.findProcessingByCurrentTurn('s1', 't1'))?.queryKey).toBe('a'); + writer.db + .update(queryCollapseViewStates) + .set({ processingFinishedAtMs: 2 }) + .where(eq(queryCollapseViewStates.sessionId, 's1')) + .run(); + expect(await state.findProcessingByCurrentTurn('s1', 't1')).toBeUndefined(); + expect((await state.findProcessingByCurrentTurn('s2', 't1'))?.queryKey).toBe('other'); + client.close(); + const reopened = createQueryCollapseState({ db: client.db }); + expect((await reopened.findProcessingByCurrentTurn('s2', 't1'))?.queryKey).toBe('other'); + }); + }); + + it('rereads receipts and validates corruption after a previous successful lookup', async () => { + await withDatabase(async (client, writer) => { + const repository = createTurnRepository({ + db: client.db, + priorityFence: createQueueTurnAdmissionPriorityFence(), + sessionAdmission: { rejectionInTransaction: () => undefined }, + }); + expect(await repository.findReceipt('turn-1')).toBeUndefined(); + for (let i = 1; i <= 2; i += 1) { + writer.db + .insert(turnIngress) + .values({ + turnId: `turn-${i}`, + sessionId: `s${i}`, + busyReason: 'turn', + inputJson: '{}', + status: 'accepted', + acceptedAtMs: 1, + acceptedSequence: i, + inputDigest: `digest-${i}`, + }) + .run(); + } + expect(await repository.findReceipt('turn-1')).toMatchObject({ + sessionId: 's1', + acceptedSequence: 1, + }); + expect(await repository.findReceipt('turn-2')).toMatchObject({ + sessionId: 's2', + acceptedSequence: 2, + }); + writer.db + .update(turnIngress) + .set({ inputDigest: '' }) + .where(eq(turnIngress.turnId, 'turn-1')) + .run(); + await expect(repository.findReceipt('turn-1')).rejects.toThrow('malformed'); + writer.db + .update(turnIngress) + .set({ inputDigest: 'edited', acceptedSequence: 3 }) + .where(eq(turnIngress.turnId, 'turn-1')) + .run(); + expect(await repository.findReceipt('turn-1')).toMatchObject({ + inputDigest: 'edited', + acceptedSequence: 3, + }); + writer.db.delete(turnIngress).where(eq(turnIngress.turnId, 'turn-1')).run(); + expect(await repository.findReceipt('turn-1')).toBeUndefined(); + expect(await repository.findReceipt("turn-2' OR 1=1 --")).toBeUndefined(); + }); + }); +}); + describe('native incremental semantic hashing', () => { const inputs = ['', 'abc', 'δΈ­ζ–‡πŸ™‚', '\ud800', '\udc00', 'a'.repeat(8191) + 'πŸ™‚tail']; for (const length of [55, 56, 63, 64, 65, 8191, 8192, 8193, 32769]) { @@ -38,9 +259,103 @@ describe('native incremental semantic hashing', () => { createHash('sha256').update('\ud83d').update('\ude42').digest('hex'), ); }); + it('matches native updates across repeated batches and split surrogate pairs', () => { + const actual = new IncrementalSha256(); + const expected = createHash('sha256'); + const parts = ['key', ':', '', '\ud83d', '\ude42', 'δΈ­ζ–‡πŸ™‚', 'x'.repeat(8191), 'πŸ™‚tail']; + for (let index = 0; index < 257; index += 1) { + for (const part of parts) { + actual.update(part); + expected.update(part, 'utf8'); + } + } + expect(actual.digestHex()).toBe(expected.digest('hex')); + }); +}); + +describe('streamed canonical history revisions', () => { + it.each([0, 1, 100])('preserves the canonical JSON digest for %i records', (length) => { + const records = Array.from({ length }, (_, index) => ({ + message_id: `msg-${index}`, + turn_id: `turn-${index}`, + message: { + role: 'user', + timestamp: index, + content: 'δΈ­ζ–‡πŸ™‚\ud800'.repeat(2048), + metadata: { z: [null, true, -0], '10': 'ten', '2': 'two', a: { b: '"\\\n' } }, + }, + })); + const expected = `sha256:${createHash('sha256') + .update(canonicalJson(records.map(decodeCanonicalHistoryEnvelope)), 'utf8') + .digest('hex')}`; + expect(canonicalHistoryRevision(records)).toBe(expected); + expect(canonicalActiveHistoryRevision(records)).toBe(expected); + if (records.length > 0) { + records[0]!.message.content = 'edited'; + expect(canonicalHistoryRevision(records)).not.toBe(expected); + } + }); + it('keeps active and settled sequence validation distinct', () => { + const pending = [ + { + message_id: 'msg-assistant', + turn_id: 'turn-1', + message: { + role: 'assistant', + timestamp: 1, + content: [ + { type: 'toolCall', id: 'call-1', name: 'bash', arguments: { command: 'pwd' } }, + ], + }, + }, + ]; + const expected = `sha256:${createHash('sha256') + .update(canonicalJson(pending.map(decodeCanonicalHistoryEnvelope)), 'utf8') + .digest('hex')}`; + expect(canonicalActiveHistoryRevision(pending)).toBe(expected); + expect(() => canonicalHistoryRevision(pending)).toThrow('tool results'); + expect(() => canonicalActiveHistoryRevision([pending[0]!, pending[0]!])).toThrow(); + }); }); 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 { @@ -67,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' }], @@ -486,6 +852,71 @@ describe('committed history read reuse', () => { expect(readActive).not.toHaveBeenCalled(); expect((await store.read('synthetic-session')).revision).toBe('r2'); }); + it('preserves owned history through the store and the next commit snapshot', async () => { + const original = { + revision: ' r1 ', + messages: [{ role: 'user', timestamp: 1, content: 'original'.repeat(4096) }], + identityVector: ['msg-1'], + }; + const store = new DurableCanonicalHistoryStore({ + read: async () => original, + readActive: async () => original, + append: async () => original, + replace: async () => original, + }); + const committed = await store.append(change); + const expectedContent = original.messages[0]!.content; + original.messages[0]!.content = 'changed'; + original.identityVector[0] = 'changed'; + expect(committed.revision).toBe('r1'); + expect(committed.messages).toEqual([{ role: 'user', timestamp: 1, content: expectedContent }]); + expect(committed.identityVector).toEqual(['msg-1']); + expect(Object.isFrozen(committed.messages[0])).toBe(true); + expect(Object.isFrozen(committed.messages)).toBe(true); + expect(Object.isFrozen(committed.identityVector)).toBe(true); + const clone = vi.spyOn(globalThis, 'structuredClone'); + try { + expect(captureSemanticSnapshot(committed).value).toBe(committed); + expect( + captureSemanticSnapshot({ committedMessages: committed.messages }).value.committedMessages, + ).toBe(committed.messages); + expect(clone).not.toHaveBeenCalled(); + } finally { + clone.mockRestore(); + } + const sharedEmpty: never[] = []; + const emptyStore = new DurableCanonicalHistoryStore({ + read: async () => ({ revision: 'r2', messages: sharedEmpty, identityVector: sharedEmpty }), + readActive: async () => original, + append: async () => original, + replace: async () => original, + }); + const empty = await emptyStore.read('synthetic-session'); + 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', @@ -514,3 +945,241 @@ 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('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')])); + const first = await reader.readActiveStrict(); + expect(() => first.pop()).toThrow(); + expect(() => { first[0] = row('replacement'); }).toThrow(); + 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')])); + 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); + 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]); + 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 externallyFrozenArray = Object.freeze([input]); + const before = canonicalHistoryRevision(externallyFrozenArray); + (input.message.content as { nested: string[] }).nested[0] = 'changed'; + expect(canonicalHistoryRevision(externallyFrozenArray)).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 }); + } + }); +}); diff --git a/test/sqlite-message-contention.test.ts b/test/sqlite-message-contention.test.ts new file mode 100644 index 00000000..8f523b9e --- /dev/null +++ b/test/sqlite-message-contention.test.ts @@ -0,0 +1,317 @@ +import { fork } from 'node:child_process'; +import { once } from 'node:events'; +import { mkdtemp, rm, writeFile } from 'node:fs/promises'; +import { createRequire } from 'node:module'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { sql } from 'drizzle-orm'; +import { RUNTIME_EVENT_SCHEMA, RuntimeEventType } from '@mavis/agent-core/protocol'; +import { RespDataType, type AgentMessage } from '@mavis/agent-core/protocol/agent-message'; +import { afterEach, expect, it, vi } from 'vitest'; +import { + DatabaseClient, + type AppDb, +} from '../packages/local-runtime-v2/src/infra/db/client.js'; +import { initializeDatabase } from '../packages/local-runtime-v2/src/infra/db/initialize.js'; +import { + runWithWriteLock, + WriteLockWaitAbortedError, +} from '../packages/local-runtime-v2/src/infra/db/write-transaction.js'; +import { createSessionSystemAgentProjection } from '../packages/local-runtime-v2/src/service/session-system/agent-projection.js'; +import { createMessageRepository } from '../packages/local-runtime-v2/src/service/session-system/messages/repo/drizzle.js'; +import { RequiredAgentEventDelivery } from '../packages/local-runtime-v2/src/service/turn-system/agent-host/events/required-agent-event-delivery.js'; +import { TurnCommitPipeline } from '../packages/local-runtime-v2/src/service/turn-system/agent-host/events/turn-commit-pipeline.js'; +import { AgentHostCommittedHistoryWriter } from '../packages/local-runtime-v2/src/service/turn-system/agent-host/history/committed-history-writer.js'; + +const cleanup: Array<() => Promise | void> = []; +afterEach(async () => { + for (const dispose of cleanup.splice(0).reverse()) await dispose(); +}); + +async function fixture() { + const dataDir = await mkdtemp(join(tmpdir(), 'mcode-message-contention-')); + cleanup.push(() => rm(dataDir, { recursive: true, force: true })); + const client = new DatabaseClient({ dataDir }); + cleanup.push(() => client.close()); + await initializeDatabase({ database: client, dataDir }); + const messages = createMessageRepository({ db: client.db, sourceProjectionEnabled: false }); + await messages.upsert({ sessionId: 'synthetic', message: { msg_id: 'seed', role: 'user' } }); + return { client, messages, dataDir }; +} + +async function holdWriter(dataDir: string, durationMs: number) { + const file = join(dataDir, 'lock-holder.cjs'); + await writeFile( + file, + ` + const Database = require(process.argv[2]); + const db = new Database(process.argv[3]); + db.exec('BEGIN IMMEDIATE'); + process.send('locked'); + setTimeout(() => { + db.exec('COMMIT'); + db.close(); + process.disconnect(); + }, Number(process.argv[4])); + `, + ); + const child = fork( + file, + [ + createRequire(import.meta.url).resolve('better-sqlite3'), + join(dataDir, 'v2', 'sqlite', 'runtime-state.sqlite'), + String(durationMs), + ], + { execArgv: [], stdio: ['ignore', 'ignore', 'pipe', 'ipc'] }, + ); + const exited = once(child, 'exit'); + cleanup.push(async () => { + if (child.exitCode === null && child.signalCode === null) child.kill(); + await exited; + }); + await Promise.race([ + once(child, 'message'), + exited.then(() => { + throw new Error('Lock holder exited before acquiring the lock'); + }), + ]); + return { exited }; +} + +it('commits one message after a foreign writer outlasts the native busy timeout', async () => { + const { dataDir, messages } = await fixture(); + await holdWriter(dataDir, 6_500); + let ticks = 0; + const timer = setInterval(() => { + ticks += 1; + }, 20); + try { + await messages.upsert({ + sessionId: 'synthetic', + message: { msg_id: 'answer', role: 'assistant' }, + }); + } finally { + clearInterval(timer); + } + expect((await messages.list('synthetic')).messages.map((message) => message.msg_id)).toEqual([ + 'seed', + 'answer', + ]); + expect(ticks).toBeGreaterThan(20); +}, 20_000); + +it('bounds admission retries and restores the shared connection timeout before yielding', async () => { + const { client, dataDir } = await fixture(); + client.db.run(sql`PRAGMA busy_timeout = 1234`); + await holdWriter(dataDir, 2_000); + const mutation = vi.fn(); + const started = performance.now(); + const pending = runWithWriteLock(client.db, mutation, { timeoutMs: 200 }); + expect(client.db.get(sql`PRAGMA busy_timeout`)).toEqual({ timeout: 1234 }); + await expect(pending).rejects.toMatchObject({ code: 'SQLITE_BUSY' }); + expect(performance.now() - started).toBeLessThan(1_000); + expect(mutation).not.toHaveBeenCalled(); + expect(client.db.get(sql`PRAGMA busy_timeout`)).toEqual({ timeout: 1234 }); +}); + +it.each(['SQLITE_BUSY', 'SQLITE_CONSTRAINT_UNIQUE'])( + 'rolls back without replaying a callback that throws %s', + async (code) => { + const { client } = await fixture(); + client.db.run(sql`CREATE TABLE contention_probe (value INTEGER)`); + const error = Object.assign(new Error('Synthetic callback failure'), { code }); + const mutation = vi.fn((tx: AppDb) => { + expect(client.db.get(sql`PRAGMA busy_timeout`)).toEqual({ timeout: 5000 }); + tx.run(sql`INSERT INTO contention_probe VALUES (1)`); + throw error; + }); + await expect(runWithWriteLock(client.db, mutation)).rejects.toBe(error); + expect(mutation).toHaveBeenCalledTimes(1); + expect(client.db.all(sql`SELECT * FROM contention_probe`)).toEqual([]); + expect(client.db.get(sql`PRAGMA busy_timeout`)).toEqual({ timeout: 5000 }); + }, +); + +it('persists uncontended cleanup messages after the lease is cancelled', async () => { + const { messages } = await fixture(); + const controller = new AbortController(); + controller.abort(new Error('Synthetic cancellation')); + await messages.upsert( + { sessionId: 'synthetic', message: { msg_id: 'tool-completion', role: 'assistant' } }, + { signal: controller.signal }, + ); + expect((await messages.list('synthetic')).messages.map((message) => message.msg_id)).toEqual([ + 'seed', + 'tool-completion', + ]); +}); + +it('does not wait for a contended cleanup write when its lease is already cancelled', async () => { + const { client, dataDir, messages } = await fixture(); + const controller = new AbortController(); + controller.abort(new Error('Synthetic cancellation')); + const { exited } = await holdWriter(dataDir, 1_500); + const started = performance.now(); + await expect( + messages.upsert( + { sessionId: 'synthetic', message: { msg_id: 'cancelled', role: 'assistant' } }, + { signal: controller.signal }, + ), + ).rejects.toMatchObject({ + name: 'WriteLockWaitAbortedError', + signal: controller.signal, + cause: controller.signal.reason, + }); + expect(performance.now() - started).toBeLessThan(500); + expect(client.db.get(sql`PRAGMA busy_timeout`)).toEqual({ timeout: 5000 }); + await exited; + expect((await messages.list('synthetic')).messages.map((message) => message.msg_id)).toEqual([ + 'seed', + ]); +}); + +async function pipelineFixture() { + const storage = await fixture(); + const { messages } = storage; + const controller = new AbortController(); + const context = { sessionId: 'synthetic', turnId: 'turn', turnSequence: 1 }; + const projection = createSessionSystemAgentProjection({ + messages, + sessions: { update: vi.fn() }, + state: { markStarted: vi.fn(), markIdle: vi.fn(), markTerminal: vi.fn() }, + stream: { write: vi.fn() }, + conversationFacts: { handle: vi.fn() }, + }); + const stream = { projectRuntimeEvent: vi.fn() }; + const events = new RequiredAgentEventDelivery({ + projectors: { + session: projection.session, + messages: projection.messages, + stream, + turnFacts: { projectRuntimeEvent: vi.fn(), projectHistoryCommitted: vi.fn() }, + }, + }); + const pipeline = new TurnCommitPipeline({ + context, + lease: { + ...context, + leaseId: 'lease', + acceptedSequence: 1, + acceptedAtMs: 1, + busyReason: 'turn', + signal: controller.signal, + }, + initialHistory: { revision: 'initial', messages: [] }, + events, + committedHistory: new AgentHostCommittedHistoryWriter({ + events, + history: { read: vi.fn(), append: vi.fn(), replace: vi.fn() }, + }), + isRuntimeErrorRetryable: () => false, + }); + return { ...storage, controller, pipeline, stream }; +} + +it('cancels a blocked write through the turn event pipeline without persisting it later', async () => { + const { client, dataDir, messages, controller, pipeline, stream } = await pipelineFixture(); + const { exited } = await holdWriter(dataDir, 1_500); + const started = performance.now(); + const timer = setTimeout(() => controller.abort(), 100); + try { + await expect(pipeline.onRuntimeEvent(messageEvent())).resolves.toBeUndefined(); + expect(performance.now() - started).toBeLessThan(1_000); + } finally { + clearTimeout(timer); + } + expect(client.db.get(sql`PRAGMA busy_timeout`)).toEqual({ timeout: 5000 }); + expect(stream.projectRuntimeEvent).not.toHaveBeenCalled(); + await exited; + expect((await messages.list('synthetic')).messages.map((message) => message.msg_id)).toEqual([ + 'seed', + ]); + await expect(pipeline.drain()).resolves.toBeUndefined(); +}); + +it.each(['before', 'during'])( + 'preserves completed tool facts when the lease is cancelled %s contention', + async (timing) => { + const { dataDir, messages, controller, pipeline, stream } = await pipelineFixture(); + const completedTool: AgentMessage = { + msg_id: 'completed-tool', + role: 'assistant', + tool_calls: [ + { + tool_call_id: 'executed-tool', + tool_name: 'bash', + tool_call_status: 2, + tool_call_args: '{"command":"echo synthetic"}', + tool_call_result_data: '{"output":"synthetic","exitCode":0}', + }, + ], + }; + const { exited } = await holdWriter(dataDir, 1_500); + if (timing === 'before') controller.abort(); + const timer = timing === 'during' ? setTimeout(() => controller.abort(), 100) : undefined; + try { + await pipeline.onRuntimeEvent(messageEvent(completedTool)); + } finally { + clearTimeout(timer); + } + await exited; + expect(controller.signal.aborted).toBe(true); + const persisted = (await messages.list('synthetic')).messages; + expect( + persisted.find((message) => message.msg_id === 'completed-tool')?.tool_calls, + ).toEqual(completedTool.tool_calls); + expect(stream.projectRuntimeEvent).toHaveBeenCalledTimes(1); + await expect(pipeline.drain()).resolves.toBeUndefined(); + }, +); + +function messageEvent(message: AgentMessage = { msg_id: 'cancelled', role: 'assistant' }) { + return { + schema: RUNTIME_EVENT_SCHEMA, + event_id: 'blocked-message', + session_id: 'synthetic', + turn_id: 'turn', + runtime_seq: 1, + type: RuntimeEventType.STREAM_RESP, + payload: { + stream_resp: JSON.stringify({ + type: RespDataType.AgentMessage, + agent_message: message, + }), + }, + }; +} + +it.each(['unrelated-abort', 'different-lease', 'mutation-failure'])( + 'preserves fail-closed delivery for %s even when the current lease is cancelled', + async (kind) => { + const { client, messages, controller, pipeline } = await pipelineFixture(); + const other = new AbortController(); + other.abort(); + const error = + kind === 'different-lease' + ? new WriteLockWaitAbortedError(other.signal) + : new DOMException('Synthetic unrelated failure', 'AbortError'); + vi.spyOn(messages, 'upsert').mockImplementation(async () => { + if (kind === 'mutation-failure') { + // Cancellation during a callback must not reclassify its error as an + // abandoned admission: the transaction started and must fail closed. + return runWithWriteLock( + client.db, + () => { + controller.abort(error); + throw error; + }, + { signal: controller.signal }, + ); + } + controller.abort(error); + throw error; + }); + await expect(pipeline.onRuntimeEvent(messageEvent())).rejects.toBe(error); + await expect(pipeline.drain()).rejects.toBe(error); + }, +); diff --git a/test/vitest-suites.json b/test/vitest-suites.json index 24e41b65..3dbe6c1d 100644 --- a/test/vitest-suites.json +++ b/test/vitest-suites.json @@ -20,6 +20,8 @@ "packages/agent-tools/src/shared/replace-all-edit.test.ts", "packages/config/test/config-file-permissions.test.ts", "packages/config/test/config-update-preservation.test.ts", + "packages/config/test/managed-preset-sync.test.ts", + "packages/config/test/private-config-file.test.ts", "packages/local-runtime-v2/src/application/session/process-local-application.test.ts", "packages/local-runtime-v2/src/application/session/session-title-policy.test.ts", "packages/local-runtime-v2/src/compat/v1/agent-host.test.ts", @@ -159,7 +161,9 @@ "packages/tui/test/unit/tui/widgets/editor/editor-behavior.test.ts", "packages/tui/test/unit/update-application.test.ts", "packages/tui/test/unit/update-service.test.ts", - "test/history-processing.test.ts" + "packages/tui/test/unit/wsl-attachment-paths.test.ts", + "test/history-processing.test.ts", + "test/sqlite-message-contention.test.ts" ], "status-contract": [ "packages/tui/test/unit/tui-build-mode-contract.test.ts" diff --git a/third_party/pi-mono/MINIMAX_CHANGES.md b/third_party/pi-mono/MINIMAX_CHANGES.md index c9057b54..fb2a7d86 100644 --- a/third_party/pi-mono/MINIMAX_CHANGES.md +++ b/third_party/pi-mono/MINIMAX_CHANGES.md @@ -330,3 +330,12 @@ For future changes, add one entry per MiniMax patch with: - Reason: immediately submitted local user batches must be consumed in one provider hop while retaining each message identity. - Affected package: `Agent.steerBatch` added to `@earendil-works/pi-agent-core`; individual `steer` / `followUp` and default modes remain compatible. - Validation: `pnpm --filter @earendil-works/pi-agent-core build`; `node scripts/test/focused-vitest.mjs --package @earendil-works/pi-agent-core --skip-workspace-build third_party/pi-mono/packages/agent/test/agent.test.ts` (19 tests); agent-core focused `pi-turn-runner.test.ts` covers local batches and machine-only individual consumption. + +## 2026-09-21: Propagate native exit codes through Windows PowerShell 5.1 wrappers + +- Reason: `wrapWindowsPowerShellStdinCommand` ends with `& ([ScriptBlock]::Create($source))` and `wrapConstrainedWindowsPowerShellCommand` ends with `Invoke-Expression $source`. On Windows PowerShell 5.1, a `-Command` session whose final statement is a scriptblock invocation exits 0 regardless of `$LASTEXITCODE` set by native commands inside it, so every failing native command was reported as success (foreground tool result and background task status) on hosts without pwsh 7. Reproduced before the fix: a `node -e "…;process.exit(7)"` command produced its stderr yet the shell process exited 0. +- Affected package: `@earendil-works/pi-coding-agent` local bash operations, PowerShell 5.1 transport only (`src/core/tools/bash.ts`). pwsh 7 (native `-Command` path) and POSIX shells are untouched. PowerShell-internal terminating errors still exit non-zero via `throw` before the appended statement. +- Type: generic, upstreamable Windows fix using the same `exit $LASTEXITCODE` idiom already used by the first-party `packages/tui/src/update/versioned-prefix.ts` launchers. +- Change: append `exit $LASTEXITCODE` after the scriptblock invocation (stdin transport) and after `Invoke-Expression` (ConstrainedLanguage transport). +- Upstream PR: not opened. +- Validation (Windows 11 x64 build 26200, PowerShell 5.1 default, Node 24.18.0): `packages/local-runtime/test/unit/child-bash-lifecycle.test.ts` 'failure' mode fails before the fix (`expected 'succeeded' to be 'failed'`) and passes after; 'timeout' and 'cancel' modes unaffected. Focused re-run of the affected suites and full `pnpm test:capabilities` show no new failures. Not run: pwsh 7 host validation, ConstrainedLanguage host validation (launcher covered by structure assertions only), macOS/Linux regression runs. 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/third_party/pi-mono/packages/coding-agent/src/core/tools/bash.ts b/third_party/pi-mono/packages/coding-agent/src/core/tools/bash.ts index fab0b193..bca3b8e7 100644 --- a/third_party/pi-mono/packages/coding-agent/src/core/tools/bash.ts +++ b/third_party/pi-mono/packages/coding-agent/src/core/tools/bash.ts @@ -413,7 +413,9 @@ function wrapConstrainedWindowsPowerShellCommand(): { const source = `$__mavis${nonce}Source`; return { environmentVariable, - launcher: `${source} = $env:${environmentVariable}; Remove-Item -LiteralPath 'Env:${environmentVariable}'; Invoke-Expression ${source}`, + // Windows PowerShell 5.1 exits 0 after Invoke-Expression regardless of the + // native command's exit code; propagate it explicitly or failures report success. + launcher: `${source} = $env:${environmentVariable}; Remove-Item -LiteralPath 'Env:${environmentVariable}'; Invoke-Expression ${source}; exit $LASTEXITCODE`, }; } @@ -458,6 +460,9 @@ function wrapWindowsPowerShellStdinCommand(): string { " }", "}", `& ([ScriptBlock]::Create(${source}))`, + // Windows PowerShell 5.1 exits 0 after a scriptblock invocation regardless of + // the native command's exit code; propagate it explicitly or failures report success. + "exit $LASTEXITCODE", ].join("; "); } 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" ],