diff --git a/docs/first-agent-history-import.md b/docs/first-agent-history-import.md index 00691ca..12b1a38 100644 --- a/docs/first-agent-history-import.md +++ b/docs/first-agent-history-import.md @@ -36,6 +36,11 @@ imported; the user is asked to run a fresh discovery and approve its current inventory. The CLI then imports one discovered history at a time through the authenticated sync client. Each source fingerprint receives a deterministic immutable raw archive; a complete archive is hash-verified and reused on retry. +Codex rollout archives store fixed-size content-addressed chunks, so subsequent +snapshots of a growing transcript reuse the unchanged prefix. Background sync +also waits five minutes for active files to settle and refreshes previously +imported conversations no more than once every six hours. Explicit +`ae sync once` and approved first imports are not delayed. The content cursor advances only after the synchronous import result is durable. The page can request cancellation at the next item boundary. Closing the page diff --git a/packages/cli/README.md b/packages/cli/README.md index 4f35d23..977e20a 100644 --- a/packages/cli/README.md +++ b/packages/cli/README.md @@ -78,8 +78,14 @@ into `config.yaml`, verifies the full bundle fingerprint again before reading, and records a resumable reconciled inventory. Changed bundles require a fresh preview and approval; inaccessible sources receive safe permission guidance. Supported history sources are Claude Code, Codex, Cowork, and local directories. -The sync cursor and raw source archive remain local under `AE_HOME`. Raw archives -are content-addressed and reused when an import is retried. Writes fail closed +The sync cursor and raw source archive remain local under `AE_HOME`. Codex's +append-heavy JSONL transcripts use fixed-size, content-addressed chunks, so a +growing session writes only its changed tail instead of another full copy. +Background sync waits for a file to settle and refreshes an already-imported +conversation at most once every six hours; `ae sync once` remains immediate. +The daemon scans every five minutes by default and reports deferred files plus +new and reused archive bytes. Other raw archives are content-addressed and +reused when an import is retried. Writes fail closed before exceeding a 256 MiB bundle limit, a 10 GiB total archive limit, or a 10 GiB free-space reserve. Override those byte counts with `AE_RAW_ARCHIVE_MAX_BUNDLE_BYTES`, `AE_RAW_ARCHIVE_MAX_TOTAL_BYTES`, and diff --git a/packages/cli/src/__tests__/codex-normalize.test.ts b/packages/cli/src/__tests__/codex-normalize.test.ts index e192009..3e4458c 100644 --- a/packages/cli/src/__tests__/codex-normalize.test.ts +++ b/packages/cli/src/__tests__/codex-normalize.test.ts @@ -7,6 +7,7 @@ import { readFileSync, readdirSync, rmSync, + utimesSync, writeFileSync, } from 'node:fs'; import { tmpdir } from 'node:os'; @@ -256,21 +257,30 @@ describe('Codex conversation normalization', () => { expect(searchText).not.toContain('future_value'); const archiveRoot = join(codexHome, '..', 'ae-home', 'raw-archive'); - const [archiveName] = readdirSync(archiveRoot); + const [archiveName] = readdirSync(archiveRoot).filter((name) => !name.startsWith('.')); const manifest = JSON.parse( readFileSync(join(archiveRoot, archiveName, 'manifest.json'), 'utf8'), ) as { + version: number; adapter_name: string; - files: Array<{ path: string; archive_path: string; sha256: string }>; + files: Array<{ + path: string; + sha256: string; + chunks: Array<{ archive_path: string }>; + }>; }; const expectedHash = createHash('sha256').update(readFileSync(rolloutPath)).digest('hex'); expect(manifest).toMatchObject({ + version: 2, adapter_name: 'codex-history', files: [{ path: rolloutPath, sha256: expectedHash }], }); expect(conversation.source_sha256).toBe(expectedHash); expect(result.sourceFingerprint).toBe(expectedHash); - expect(readFileSync(join(archiveRoot, archiveName, manifest.files[0].archive_path))) + const archivedBytes = Buffer.concat( + manifest.files[0].chunks.map((chunk) => readFileSync(join(archiveRoot, chunk.archive_path))), + ); + expect(archivedBytes) .toEqual(readFileSync(rolloutPath)); }); @@ -437,6 +447,46 @@ describe('Codex conversation normalization', () => { source_identifier: 'openai_codex:codex:codex-session-915', 'metadata.sync.adapter_name': 'codex-history', }); - expect(readdirSync(join(codexHome, '..', 'ae-home', 'raw-archive'))).toHaveLength(1); + const archiveRoot = join(codexHome, '..', 'ae-home', 'raw-archive'); + expect(readdirSync(archiveRoot).filter((name) => !name.startsWith('.'))).toHaveLength(1); + expect(readdirSync(archiveRoot)).toContain('.chunks'); + }); + + it('throttles changed conversation snapshots only in the background daemon', async () => { + const { codexHome, rolloutPath, database } = makeCodexFixture(); + const cursorFile = join(codexHome, 'cursor.json'); + const client = { + submitSyncImport: vi.fn((request: ImportRequest) => Promise.resolve({ + success: true as const, + data: { + totalItems: request.items.length, + completedItems: request.items.length, + failedItems: 0, + contentIds: ['content-codex'], + failures: [], + }, + })), + }; + const now = Date.now(); + const stableTime = new Date(now - 10 * 60 * 1000); + utimesSync(rolloutPath, stableTime, stableTime); + const options = { + sourceId: 'codex' as const, + paths: [rolloutPath], cursorFile, batchSize: 1, client, + }; + + const first = await runSyncOnce({ ...options, enforceArchiveThrottle: true, nowMs: now }); + writeFileSync(rolloutPath, `${readFileSync(rolloutPath, 'utf8')}\n`); + utimesSync(rolloutPath, stableTime, stableTime); + const deferred = await runSyncOnce({ + ...options, enforceArchiveThrottle: true, nowMs: now + 20 * 60 * 1000, + }); + const manual = await runSyncOnce(options); + closeTestDatabase(database); + + expect(first.turnsImported).toBe(1); + expect(deferred).toMatchObject({ turnsImported: 0, deferredFiles: 1 }); + expect(manual.turnsImported).toBe(1); + expect(client.submitSyncImport).toHaveBeenCalledTimes(2); }); }); diff --git a/packages/cli/src/__tests__/raw-archive.test.ts b/packages/cli/src/__tests__/raw-archive.test.ts index 19adf53..ad92ed2 100644 --- a/packages/cli/src/__tests__/raw-archive.test.ts +++ b/packages/cli/src/__tests__/raw-archive.test.ts @@ -1,7 +1,7 @@ import { createHash } from 'node:crypto'; -import { existsSync, mkdirSync, mkdtempSync, readFileSync, readdirSync, rmSync, writeFileSync } from 'node:fs'; +import { appendFileSync, existsSync, mkdirSync, mkdtempSync, readFileSync, readdirSync, rmSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; -import { join } from 'node:path'; +import { dirname, join } from 'node:path'; import { afterEach, describe, expect, it } from 'vitest'; import { RawArchiveCapacityError, @@ -9,6 +9,8 @@ import { applyRawArchiveRetention, inspectRawArchive, planRawArchiveRetention, + readRawArchiveFile, + writeChunkedRawArchive, writeRawArchive, } from '../sync/raw-archive.js'; @@ -30,6 +32,48 @@ afterEach(() => { }); describe('writeRawArchive', () => { + it('deduplicates unchanged chunks when an append-only transcript grows', async () => { + const root = makeTempDir(); + process.env.AE_HOME = join(root, 'ae-home'); + const sourcePath = join(root, 'large.jsonl'); + const testChunkBytes = 64 * 1024; + const initial = Buffer.alloc(testChunkBytes + 1024, 97); + writeFileSync(sourcePath, initial); + + const first = await writeChunkedRawArchive(sourcePath, { + adapterName: 'append-only-test', adapterVersion: '1.0.0', + chunkBytes: testChunkBytes, + }); + appendFileSync(sourcePath, Buffer.alloc(128, 98)); + const second = await writeChunkedRawArchive(sourcePath, { + adapterName: 'append-only-test', adapterVersion: '1.0.0', + chunkBytes: testChunkBytes, + }); + + expect(first.manifest.version).toBe(2); + expect(first.newlyArchivedBytes).toBe(initial.length); + expect(second.newlyArchivedBytes).toBeLessThanOrEqual(testChunkBytes + 128); + expect(second.reusedBytes).toBeGreaterThanOrEqual(testChunkBytes); + expect(await readRawArchiveFile(second, sourcePath)).toEqual(readFileSync(sourcePath)); + expect(readdirSync(second.archiveDir)).toEqual(['manifest.json']); + }); + + it('rejects a tampered shared chunk before returning archived bytes', async () => { + const root = makeTempDir(); + process.env.AE_HOME = join(root, 'ae-home'); + const sourcePath = join(root, 'source.jsonl'); + writeFileSync(sourcePath, '{"message":"integrity"}\n'); + const archive = await writeChunkedRawArchive(sourcePath, { + adapterName: 'append-only-test', adapterVersion: '1.0.0', + }); + const entry = archive.manifest.files[0]; + const chunk = entry?.chunks[0]; + expect(chunk).toBeDefined(); + writeFileSync(join(dirname(archive.archiveDir), chunk!.archive_path), 'tampered'); + + await expect(readRawArchiveFile(archive, sourcePath)).rejects.toThrow(/integrity/i); + }); + it('requires adapter versioning in the manifest contract', () => { expect(RawArchiveManifestSchema.safeParse({ version: 1, diff --git a/packages/cli/src/commands/sync.ts b/packages/cli/src/commands/sync.ts index 5a50f92..acc505a 100644 --- a/packages/cli/src/commands/sync.ts +++ b/packages/cli/src/commands/sync.ts @@ -54,7 +54,7 @@ import { } from '../sync/raw-archive.js'; const DEFAULT_SYNC_BATCH_SIZE = 25; -const DEFAULT_POLL_INTERVAL_SECONDS = 15; +const DEFAULT_POLL_INTERVAL_SECONDS = 300; interface SyncCommandOptions { source?: string; @@ -137,7 +137,9 @@ function printLoopSummary(summary: SyncRunSummary): void { console.log( `[${timestamp}] source=${summary.sourceId} files=${summary.filesScanned} ` + `found=${summary.turnsFound} imported=${summary.turnsImported} ` + - `failed=${summary.failedItems} parseErrors=${summary.parseErrors}` + `deferred=${summary.deferredFiles} failed=${summary.failedItems} ` + + `parseErrors=${summary.parseErrors} archiveWritten=${summary.archiveBytesWritten} ` + + `archiveReused=${summary.archiveBytesReused}` ); } diff --git a/packages/cli/src/sync/raw-archive.ts b/packages/cli/src/sync/raw-archive.ts index 5e1626e..cde3be9 100644 --- a/packages/cli/src/sync/raw-archive.ts +++ b/packages/cli/src/sync/raw-archive.ts @@ -6,6 +6,8 @@ import { z } from 'zod'; import { logsDir, rawArchiveDir } from '../home.js'; export const RAW_ARCHIVE_MANIFEST_VERSION = 1 as const; +export const RAW_ARCHIVE_CHUNKED_MANIFEST_VERSION = 2 as const; +export const RAW_ARCHIVE_CHUNK_BYTES = 4 * 1024 * 1024; const NonEmptyStringSchema = z.string().trim().min(1); const Sha256Schema = z.string().regex(/^[a-f0-9]{64}$/i, 'must be a SHA-256 hex digest'); @@ -13,6 +15,10 @@ const ArchivePathSchema = z.string().regex( /^files\/[a-zA-Z0-9._-]+$/, 'must remain inside the archive files directory', ); +const ChunkArchivePathSchema = z.string().regex( + /^\.chunks\/sha256-[a-f0-9]{64}$/, + 'must be a content-addressed file inside the shared chunk directory', +); export const RawFileManifestEntrySchema = z.object({ path: NonEmptyStringSchema, @@ -24,7 +30,7 @@ export const RawFileManifestEntrySchema = z.object({ adapter_version: NonEmptyStringSchema, }).strict(); -export const RawArchiveManifestSchema = z.object({ +export const RawArchiveManifestV1Schema = z.object({ version: z.literal(RAW_ARCHIVE_MANIFEST_VERSION), created_at: z.string().datetime(), adapter_name: NonEmptyStringSchema, @@ -33,8 +39,40 @@ export const RawArchiveManifestSchema = z.object({ files: z.array(RawFileManifestEntrySchema), }).strict(); +export const RawChunkManifestEntrySchema = z.object({ + archive_path: ChunkArchivePathSchema, + size: z.number().int().nonnegative(), + sha256: Sha256Schema, +}).strict(); + +export const RawChunkedFileManifestEntrySchema = z.object({ + path: NonEmptyStringSchema, + size: z.number().int().nonnegative(), + mtime: z.string().datetime(), + sha256: Sha256Schema, + adapter_name: NonEmptyStringSchema, + adapter_version: NonEmptyStringSchema, + chunks: z.array(RawChunkManifestEntrySchema), +}).strict(); + +export const RawArchiveManifestV2Schema = z.object({ + version: z.literal(RAW_ARCHIVE_CHUNKED_MANIFEST_VERSION), + created_at: z.string().datetime(), + adapter_name: NonEmptyStringSchema, + adapter_version: NonEmptyStringSchema, + source_fingerprint: Sha256Schema, + files: z.array(RawChunkedFileManifestEntrySchema), +}).strict(); + +export const RawArchiveManifestSchema = z.discriminatedUnion('version', [ + RawArchiveManifestV1Schema, + RawArchiveManifestV2Schema, +]); + export type RawFileManifestEntry = z.infer; export type RawArchiveManifest = z.infer; +export type RawArchiveManifestV1 = z.infer; +export type RawArchiveManifestV2 = z.infer; export const DEFAULT_RAW_ARCHIVE_MAX_BUNDLE_BYTES = 256 * 1024 * 1024; export const DEFAULT_RAW_ARCHIVE_MAX_TOTAL_BYTES = 10 * 1024 * 1024 * 1024; @@ -61,10 +99,22 @@ export interface WriteRawArchiveOptions { limits?: Partial; } +export interface WriteChunkedRawArchiveOptions extends WriteRawArchiveOptions { + chunkBytes?: number; +} + export interface WriteRawArchiveResult { archiveDir: string; manifestPath: string; - manifest: RawArchiveManifest; + manifest: RawArchiveManifestV1; +} + +export interface WriteChunkedRawArchiveResult { + archiveDir: string; + manifestPath: string; + manifest: RawArchiveManifestV2; + newlyArchivedBytes: number; + reusedBytes: number; } export interface RawArchiveInventoryEntry { @@ -101,7 +151,7 @@ export interface RawArchiveRetentionResult { export function attachRawArchiveManifest< T extends { provider_metadata_json: Record }, ->(conversations: readonly T[], archive: WriteRawArchiveResult): T[] { +>(conversations: readonly T[], archive: WriteRawArchiveResult | WriteChunkedRawArchiveResult): T[] { const rawArchiveManifest = { manifest_path: archive.manifestPath, ...archive.manifest, @@ -285,7 +335,7 @@ async function validateExistingArchive( adapterVersion: string, ): Promise { const manifestPath = join(archiveDir, 'manifest.json'); - const manifest = RawArchiveManifestSchema.parse(JSON.parse(await readFile(manifestPath, 'utf8'))); + const manifest = RawArchiveManifestV1Schema.parse(JSON.parse(await readFile(manifestPath, 'utf8'))); const fingerprint = bundleFingerprint(snapshots, adapterName, adapterVersion); if (manifest.adapter_name !== adapterName || manifest.adapter_version !== adapterVersion || manifest.files.length !== snapshots.length @@ -573,7 +623,7 @@ export async function writeRawArchive( adapter_version: adapter.adapter_version, }); } - const manifest = RawArchiveManifestSchema.parse({ + const manifest = RawArchiveManifestV1Schema.parse({ version: RAW_ARCHIVE_MANIFEST_VERSION, created_at: adapter.created_at, adapter_name: adapter.adapter_name, @@ -605,3 +655,193 @@ export async function writeRawArchive( } }); } + +function sha256Bytes(bytes: Uint8Array): string { + return createHash('sha256').update(bytes).digest('hex'); +} + +async function validateChunk(path: string, size: number, sha256: string): Promise { + const fileStat = await lstat(path); + if (!fileStat.isFile() || fileStat.isSymbolicLink()) { + throw new Error(`Raw archive chunk failed integrity validation: ${path}`); + } + const bytes = await readFile(path); + if (bytes.length !== size || sha256Bytes(bytes) !== sha256) { + throw new Error(`Raw archive chunk failed integrity validation: ${path}`); + } +} + +export async function readRawArchiveFile( + archive: WriteRawArchiveResult | WriteChunkedRawArchiveResult, + sourcePath: string, +): Promise { + let bytes: Buffer; + if (archive.manifest.version === RAW_ARCHIVE_MANIFEST_VERSION) { + const entry = archive.manifest.files.find((candidate) => candidate.path === sourcePath); + if (!entry) throw new Error(`Raw archive manifest omitted ${sourcePath}`); + const archivedPath = join(archive.archiveDir, entry.archive_path); + if (!isWithin(archive.archiveDir, archivedPath)) { + throw new Error(`Raw archive file path failed integrity validation: ${entry.archive_path}`); + } + bytes = await readFile(archivedPath); + } else { + const entry = archive.manifest.files.find((candidate) => candidate.path === sourcePath); + if (!entry) throw new Error(`Raw archive manifest omitted ${sourcePath}`); + const archiveRoot = dirname(archive.archiveDir); + const parts: Buffer[] = []; + for (const chunk of entry.chunks) { + const chunkPath = join(archiveRoot, chunk.archive_path); + if (!isWithin(join(archiveRoot, '.chunks'), chunkPath)) { + throw new Error(`Raw archive chunk path failed integrity validation: ${chunk.archive_path}`); + } + await validateChunk(chunkPath, chunk.size, chunk.sha256); + parts.push(await readFile(chunkPath)); + } + bytes = Buffer.concat(parts); + } + + const expected = archive.manifest.files.find((candidate) => candidate.path === sourcePath); + if (!expected || bytes.length !== expected.size || sha256Bytes(bytes) !== expected.sha256) { + throw new Error(`Raw archive file failed integrity validation: ${sourcePath}`); + } + return bytes; +} + +export async function writeChunkedRawArchive( + sourcePath: string, + options: WriteChunkedRawArchiveOptions, +): Promise { + return serializeArchiveWrite(async () => { + const createdAt = options.createdAt ?? new Date().toISOString(); + const adapter = z.object({ + adapter_name: NonEmptyStringSchema, + adapter_version: NonEmptyStringSchema, + created_at: z.string().datetime(), + }).parse({ + adapter_name: options.adapterName, + adapter_version: options.adapterVersion, + created_at: createdAt, + }); + const limits = archiveLimits(options.limits); + const chunkBytes = options.chunkBytes ?? RAW_ARCHIVE_CHUNK_BYTES; + if (!Number.isSafeInteger(chunkBytes) || chunkBytes <= 0 || chunkBytes > RAW_ARCHIVE_CHUNK_BYTES) { + throw new RawArchiveCapacityError( + `Raw archive chunks must be between 1 and ${RAW_ARCHIVE_CHUNK_BYTES} bytes`, + ); + } + const archiveRoot = options.archiveDir ? dirname(resolve(options.archiveDir)) : rawArchiveDir(); + if (isWithin(rawArchiveDir(), sourcePath)) { + throw new Error(`Raw archive cannot archive its own contents: ${sourcePath}`); + } + await mkdir(archiveRoot, { recursive: true, mode: 0o700 }); + const releaseLock = await acquireArchiveLock(archiveRoot); + try { + const snapshot = await snapshotSource(sourcePath); + if (snapshot.size > limits.maxBundleBytes) { + throw new RawArchiveCapacityError( + `Raw archive bundle exceeds ${limits.maxBundleBytes} byte per-bundle limit`, + ); + } + const sourceBytes = await readFile(sourcePath); + if (sourceBytes.length !== snapshot.size || sha256Bytes(sourceBytes) !== snapshot.sha256) { + throw new Error(`Raw archive source changed while being chunked: ${sourcePath}`); + } + + const fingerprint = bundleFingerprint([snapshot], adapter.adapter_name, adapter.adapter_version); + const archiveDir = options.archiveDir + ? resolve(options.archiveDir) + : join(archiveRoot, `${safeBaseName(adapter.adapter_name)}-${fingerprint}`); + const manifestPath = join(archiveDir, 'manifest.json'); + if (!isWithin(archiveRoot, archiveDir) || resolve(archiveDir) === resolve(archiveRoot)) { + throw new Error('Raw archive target must remain inside its archive root'); + } + + try { + await lstat(archiveDir); + if (options.archiveDir) throw new Error(`Raw archive already exists: ${archiveDir}`); + const manifest = RawArchiveManifestV2Schema.parse(JSON.parse(await readFile(manifestPath, 'utf8'))); + const result = { archiveDir, manifestPath, manifest, newlyArchivedBytes: 0, reusedBytes: snapshot.size }; + await readRawArchiveFile(result, sourcePath); + return result; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; + } + + const chunks = []; + let newlyArchivedBytes = 0; + let reusedBytes = 0; + const chunkRoot = join(archiveRoot, '.chunks'); + await mkdir(chunkRoot, { recursive: true, mode: 0o700 }); + for (let offset = 0; offset < sourceBytes.length; offset += chunkBytes) { + const bytes = sourceBytes.subarray(offset, Math.min(offset + chunkBytes, sourceBytes.length)); + const sha256 = sha256Bytes(bytes); + const archivePath = `.chunks/sha256-${sha256}`; + const chunkPath = join(archiveRoot, archivePath); + try { + await lstat(chunkPath); + await validateChunk(chunkPath, bytes.length, sha256); + reusedBytes += bytes.length; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; + newlyArchivedBytes += bytes.length; + } + chunks.push({ archive_path: archivePath, size: bytes.length, sha256, bytes }); + } + + const existingBytes = await directoryBytes(archiveRoot, limits.maxTotalBytes); + const requiredBytes = newlyArchivedBytes + 64 * 1024; + if (existingBytes + requiredBytes > limits.maxTotalBytes) { + throw new RawArchiveCapacityError( + `Raw archive total limit would be exceeded (${existingBytes} existing + ${requiredBytes} required > ${limits.maxTotalBytes})`, + ); + } + const freeBytes = await availableBytes(archiveRoot); + if (freeBytes - requiredBytes < limits.minFreeBytes) { + throw new RawArchiveCapacityError( + `Raw archive write would breach the ${limits.minFreeBytes} byte free-space reserve`, + ); + } + + for (const chunk of chunks) { + const chunkPath = join(archiveRoot, chunk.archive_path); + try { + await writeFile(chunkPath, chunk.bytes, { mode: 0o600, flag: 'wx' }); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'EEXIST') throw error; + await validateChunk(chunkPath, chunk.size, chunk.sha256); + } + } + + const manifest = RawArchiveManifestV2Schema.parse({ + version: RAW_ARCHIVE_CHUNKED_MANIFEST_VERSION, + created_at: adapter.created_at, + adapter_name: adapter.adapter_name, + adapter_version: adapter.adapter_version, + source_fingerprint: fingerprint, + files: [{ + path: snapshot.path, + size: snapshot.size, + mtime: snapshot.mtime, + sha256: snapshot.sha256, + adapter_name: adapter.adapter_name, + adapter_version: adapter.adapter_version, + chunks: chunks.map(({ bytes: _bytes, ...chunk }) => chunk), + }], + }); + const stagingDir = join(archiveRoot, `.staging-${randomUUID()}`); + try { + await mkdir(stagingDir, { mode: 0o700 }); + await writeFile(join(stagingDir, 'manifest.json'), `${JSON.stringify(manifest, null, 2)}\n`, { + encoding: 'utf8', mode: 0o600, flag: 'wx', + }); + await rename(stagingDir, archiveDir); + } catch (error) { + await rm(stagingDir, { recursive: true, force: true }); + throw error; + } + return { archiveDir, manifestPath, manifest, newlyArchivedBytes, reusedBytes }; + } finally { + await releaseLock(); + } + }); +} diff --git a/packages/cli/src/sync/sources/codex.ts b/packages/cli/src/sync/sources/codex.ts index 0dfe84e..0e25740 100644 --- a/packages/cli/src/sync/sources/codex.ts +++ b/packages/cli/src/sync/sources/codex.ts @@ -2,7 +2,11 @@ import { createHash } from 'node:crypto'; import { readdir, readFile, stat } from 'node:fs/promises'; import { homedir } from 'node:os'; import { basename, dirname, join, resolve, sep } from 'node:path'; -import { attachRawArchiveManifest, writeRawArchive } from '../raw-archive.js'; +import { + attachRawArchiveManifest, + readRawArchiveFile, + writeChunkedRawArchive, +} from '../raw-archive.js'; import type { ConversationReadResult, TranscriptDiscoverOptions, @@ -156,11 +160,7 @@ function isRecord(value: unknown): value is Record { return value !== null && typeof value === 'object' && !Array.isArray(value); } -async function parseJsonlFile( - readPath: string, - filePath: string = readPath, -): Promise { - const buffer = await readFile(readPath); +function parseJsonlBuffer(buffer: Buffer, filePath: string): ParsedJsonlFile { const endsWithNewline = buffer.at(-1) === 10; const lines = buffer.toString('utf8').split('\n'); if (endsWithNewline) lines.pop(); @@ -242,6 +242,10 @@ function findThreadMetadata( export const codexSource: TranscriptSource = { id: SOURCE_ID, label: SOURCE_NAME, + archiveThrottle: { + minStableMs: 5 * 60 * 1000, + minRefreshMs: 6 * 60 * 60 * 1000, + }, async discover(options: TranscriptDiscoverOptions = {}): Promise { const home = codexHome(); @@ -266,16 +270,13 @@ export const codexSource: TranscriptSource = { }, async readConversations(file: TranscriptFile): Promise { - const archive = await writeRawArchive([file.path], { + const archive = await writeChunkedRawArchive(file.path, { adapterName: CODEX_ADAPTER_NAME, adapterVersion: CODEX_ADAPTER_VERSION, }); const manifest = archive.manifest.files[0]; if (!manifest) throw new Error(`Raw archive manifest omitted ${file.path}`); - const parsed = await parseJsonlFile( - join(archive.archiveDir, manifest.archive_path), - file.path, - ); + const parsed = parseJsonlBuffer(await readRawArchiveFile(archive, file.path), file.path); const metadata = await cachedThreadMetadata(join(codexHome(), 'state_5.sqlite')); const threadMeta = findThreadMetadata(metadata, file.path, parsed.records); const conversations = normalizeCodexSession({ @@ -291,6 +292,8 @@ export const codexSource: TranscriptSource = { errors: parsed.errors, processedLines: parsed.processedLines, sourceFingerprint: manifest.sha256, + archiveBytesWritten: archive.newlyArchivedBytes, + archiveBytesReused: archive.reusedBytes, }; }, }; diff --git a/packages/cli/src/sync/sync-daemon.ts b/packages/cli/src/sync/sync-daemon.ts index ea5cb40..de3f41e 100644 --- a/packages/cli/src/sync/sync-daemon.ts +++ b/packages/cli/src/sync/sync-daemon.ts @@ -40,6 +40,8 @@ export interface SyncRunOptions { client: SyncClient; onWarning?: (message: string) => void; inventoryOnly?: boolean; + enforceArchiveThrottle?: boolean; + nowMs?: number; } export interface SyncRunSummary { @@ -55,6 +57,9 @@ export interface SyncRunSummary { parseErrors: number; contentIds: string[]; archiveManifestPaths: string[]; + deferredFiles: number; + archiveBytesWritten: number; + archiveBytesReused: number; } export interface SyncLoopOptions extends SyncRunOptions { @@ -179,6 +184,9 @@ async function runLocalDirSyncOnce( parseErrors: 0, contentIds: [], archiveManifestPaths: [], + deferredFiles: 0, + archiveBytesWritten: 0, + archiveBytesReused: 0, }; const discoveredPaths = new Set(files.map((file) => file.path)); @@ -335,6 +343,9 @@ export async function runSyncOnce(options: SyncRunOptions): Promise { const cursor = await cursorStore.get(source.id, file.path); if (source.readConversations) { + if (options.enforceArchiveThrottle && source.archiveThrottle) { + const nowMs = options.nowMs ?? Date.now(); + const lastRefreshMs = cursor.updatedAt ? Date.parse(cursor.updatedAt) : Number.NaN; + const sourceIsSettled = nowMs - file.mtimeMs >= source.archiveThrottle.minStableMs; + const refreshIsDue = !Number.isFinite(lastRefreshMs) + || nowMs - lastRefreshMs >= source.archiveThrottle.minRefreshMs; + if (!sourceIsSettled || (cursor.sourceSha256 !== undefined && !refreshIsDue)) { + summary.deferredFiles += 1; + return; + } + } const sourceSha256 = source.fingerprint ? await source.fingerprint(file) : await fingerprintFile(file.path); @@ -353,7 +375,7 @@ export async function runSyncOnce(options: SyncRunOptions): Promise; readNewTurns?(file: TranscriptFile, cursor: FileCursor): Promise; fingerprint?(file: TranscriptFile): Promise; diff --git a/scripts/verify-real-history.ts b/scripts/verify-real-history.ts index 72468c8..045b4a6 100644 --- a/scripts/verify-real-history.ts +++ b/scripts/verify-real-history.ts @@ -7,6 +7,8 @@ import { fileURLToPath } from 'node:url'; import { z } from 'zod'; import { RawArchiveManifestSchema, + RawArchiveManifestV1Schema, + RawArchiveManifestV2Schema, } from '../packages/cli/src/sync/raw-archive.js'; import { syncCursorFilePath } from '../packages/cli/src/home.js'; import { createDatabasePool } from './database.js'; @@ -56,9 +58,10 @@ const SyncSummaryDocumentSchema = z.preprocess((input) => { return input; }, SyncRunSummarySchema); -const StoredRawArchiveManifestSchema = RawArchiveManifestSchema.extend({ - manifest_path: NonEmptyStringSchema, -}); +const StoredRawArchiveManifestSchema = z.discriminatedUnion('version', [ + RawArchiveManifestV1Schema.extend({ manifest_path: NonEmptyStringSchema }), + RawArchiveManifestV2Schema.extend({ manifest_path: NonEmptyStringSchema }), +]); const RealHistoryRowSchema = z.object({ id: NonEmptyStringSchema, @@ -424,18 +427,38 @@ async function verifySampledManifest(row: ValidatedHistoryRow): Promise } for (const entry of diskManifest.files) { - const bytes = await readBytes( - archiveFilePath(manifestPath, entry.archive_path), - `${row.source} sampled archive file`, - ); - const actualSha256 = createHash('sha256').update(bytes).digest('hex'); - if (actualSha256 !== entry.sha256.toLowerCase()) { - throw new RealHistoryVerificationError( - `${row.source} sampled archive SHA-256 mismatch`, + if (diskManifest.version === 1) { + const bytes = await readBytes( + archiveFilePath(manifestPath, entry.archive_path), + `${row.source} sampled archive file`, ); + const actualSha256 = createHash('sha256').update(bytes).digest('hex'); + if (actualSha256 !== entry.sha256.toLowerCase()) { + throw new RealHistoryVerificationError( + `${row.source} sampled archive SHA-256 mismatch`, + ); + } + if (bytes.byteLength !== entry.size) { + throw new RealHistoryVerificationError(`${row.source} sampled archive size mismatch`); + } + continue; + } + + const archiveRoot = dirname(dirname(manifestPath)); + const digest = createHash('sha256'); + let totalBytes = 0; + for (const chunk of entry.chunks) { + const chunkPath = archiveFilePath(join(archiveRoot, 'manifest.json'), chunk.archive_path); + const bytes = await readBytes(chunkPath, `${row.source} sampled archive chunk`); + if (bytes.byteLength !== chunk.size + || createHash('sha256').update(bytes).digest('hex') !== chunk.sha256.toLowerCase()) { + throw new RealHistoryVerificationError(`${row.source} sampled archive chunk integrity mismatch`); + } + digest.update(bytes); + totalBytes += bytes.byteLength; } - if (bytes.byteLength !== entry.size) { - throw new RealHistoryVerificationError(`${row.source} sampled archive size mismatch`); + if (totalBytes !== entry.size || digest.digest('hex') !== entry.sha256.toLowerCase()) { + throw new RealHistoryVerificationError(`${row.source} sampled archive integrity mismatch`); } } return diskManifest.files.length;