mirror of
https://github.com/stablyai/orca.git
synced 2026-09-22 16:02:32 +00:00
fix(chat): enforce legacy import byte budget during reading
This commit is contained in:
@@ -0,0 +1,68 @@
|
||||
# Legacy import source byte budget
|
||||
|
||||
The production importer accepted a transcript larger than its existing 16 MiB
|
||||
quota when the file grew between `stat()` and reading it. The fix applies that
|
||||
same quota to consumed raw bytes and rejects the entire import before replacing
|
||||
the journal. It does not import a truncated prefix.
|
||||
|
||||
## Reproduce
|
||||
|
||||
From the repository root, with dependencies installed:
|
||||
|
||||
```sh
|
||||
ORCA_BACKGROUND_LAUNCH=1 node --max-old-space-size=256 docs/audits/legacy-import-source-budget/reproduce.mjs
|
||||
```
|
||||
|
||||
The script bundles the actual production importer and decoder twice. The baseline
|
||||
restores the importer's UTF-8 stream and omits the new consumed-byte quota in
|
||||
memory; the second run uses the current source. All other dependencies are
|
||||
identical. Explicit file paths bypass discovery; a stub fails if discovery is
|
||||
unexpectedly called. No historical checkout or local-only commit is required.
|
||||
|
||||
Each run creates one real temporary file containing a valid Claude message.
|
||||
The controlled `stat` seam captures its original size, appends spaces until the
|
||||
file is exactly 17 MiB, then returns the original stat. The production importer
|
||||
reads the grown file. A journal spy records replacement attempts and its epoch;
|
||||
no SQLite database or user transcript is touched. Temporary files are removed.
|
||||
|
||||
`results.json` records source hashes, actual consumed bytes, stream closure,
|
||||
and replacement attempts. Before the fix the importer consumes all 17 MiB,
|
||||
returns success, and replaces the journal with the decoded prefix. Afterward it
|
||||
refuses on the first chunk crossing 16 MiB, destroys the stream, and leaves the
|
||||
prior epoch untouched. Read buffering permits one chunk of overshoot.
|
||||
|
||||
## Process and version attribution
|
||||
|
||||
This importer runs in the host runtime through structured-session adoption,
|
||||
handoff, transcript catch-up, and journal recovery. On desktop that runtime is
|
||||
Electron main; a remote runtime imports its own host-local source. The stat-only
|
||||
quota and unrestricted stream already existed in `v1.4.198`, release commit
|
||||
`e0826956fcfc532f5a1e55b5e081f2e57e553c43`.
|
||||
|
||||
The reproduction establishes a quota bypass, not the cause of #19768 or #19831.
|
||||
It requires an import and source growth/replacement after the size check;
|
||||
neither incident provides that evidence. This fix is independent of #20963's
|
||||
resumable JSONL record limit.
|
||||
|
||||
## Other whole-document audit results
|
||||
|
||||
- `ai-vault/session-transcript-reader.ts:152` routes rewritten JSON documents
|
||||
through whole-file `readFile` and `JSON.parse`. Gemini JSON, Hermes, Devin,
|
||||
Cline, Grok metadata, Kimi state, Rovo, and OpenCode file readers lack a source
|
||||
byte quota. These are allocation peaks; the parse cache retains summaries,
|
||||
not the full documents.
|
||||
- The production default, including v1.4.198, places those scans in the forked
|
||||
AI Vault service with a 384 MiB V8 old-space limit. That is not a total RSS
|
||||
limit. `ORCA_AI_VAULT_SERVICE_PROCESS=0` instead puts scans in a worker thread
|
||||
sharing main's PID; its first-prompt fallback runs directly in main.
|
||||
- Existing remote whole-document streaming parsers offer a reuse path, but
|
||||
local conversion must preserve message-sink delivery for search indexing.
|
||||
Streaming also needs an explicit policy for an individually oversized JSON
|
||||
value; arbitrary history truncation was not introduced.
|
||||
- The older `native-chat/transcript-reader.ts` and `transcript-read-cache.ts`
|
||||
have whole-history behavior, including a newest-entry cache exemption, but
|
||||
the current production source has no external call sites for those exports.
|
||||
|
||||
Validation: 50 targeted tests passed, including a regression that failed before
|
||||
the fix, UTF-8 and raw-byte limits, exact-boundary acceptance, and stream cleanup.
|
||||
Node typecheck, targeted lint, formatting, and diff checks passed.
|
||||
@@ -0,0 +1,170 @@
|
||||
import assert from 'node:assert/strict'
|
||||
import { createHash } from 'node:crypto'
|
||||
import * as fs from 'node:fs'
|
||||
import * as fsPromises from 'node:fs/promises'
|
||||
import Module, { createRequire } from 'node:module'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join, relative, resolve } from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { build } from 'esbuild'
|
||||
|
||||
if (process.env.ORCA_BACKGROUND_LAUNCH !== '1') {
|
||||
throw new Error('Run with ORCA_BACKGROUND_LAUNCH=1')
|
||||
}
|
||||
|
||||
const root = fileURLToPath(new URL('../../../', import.meta.url))
|
||||
const require = createRequire(import.meta.url)
|
||||
const sourcePaths = [
|
||||
'src/main/native-chat/agent-session-journal/journal-legacy-import.ts',
|
||||
'src/main/native-chat/transcript-stream-lines.ts'
|
||||
]
|
||||
const limit = 16 * 1024 * 1024
|
||||
const grownBytes = 17 * 1024 * 1024
|
||||
|
||||
function sourceAt(version, path) {
|
||||
const source = fs.readFileSync(resolve(root, path), 'utf8')
|
||||
if (version !== 'before' || path !== sourcePaths[0]) {
|
||||
return source
|
||||
}
|
||||
const stream = 'const stream = createReadStream(input.filePath)'
|
||||
const budget = 'true,\n MAX_LEGACY_IMPORT_SOURCE_BYTES'
|
||||
assert.ok(source.includes(stream) && source.includes(budget), 'Review changed baseline transform')
|
||||
return source
|
||||
.replace(stream, "const stream = createReadStream(input.filePath, { encoding: 'utf-8' })")
|
||||
.replace(budget, 'true')
|
||||
}
|
||||
|
||||
async function run(version) {
|
||||
const sources = new Map(sourcePaths.map((path) => [path, sourceAt(version, path)]))
|
||||
const built = await build({
|
||||
absWorkingDir: root,
|
||||
entryPoints: [sourcePaths[0]],
|
||||
bundle: true,
|
||||
platform: 'node',
|
||||
format: 'cjs',
|
||||
packages: 'external',
|
||||
write: false,
|
||||
plugins: [
|
||||
{
|
||||
name: 'select-production-version',
|
||||
setup(bundler) {
|
||||
bundler.onLoad(
|
||||
{ filter: /(?:journal-legacy-import|transcript-stream-lines)\.ts$/ },
|
||||
(args) => {
|
||||
const path = relative(root, args.path).split('\\').join('/')
|
||||
const contents = sources.get(path)
|
||||
return contents === undefined ? undefined : { contents, loader: 'ts' }
|
||||
}
|
||||
)
|
||||
bundler.onResolve({ filter: /session-file-resolver$/ }, () => ({
|
||||
path: 'unused',
|
||||
namespace: 'probe'
|
||||
}))
|
||||
bundler.onLoad({ filter: /.*/, namespace: 'probe' }, () => ({
|
||||
contents:
|
||||
'export function resolveSessionFilePath() { throw new Error("Unexpected path discovery") }'
|
||||
}))
|
||||
}
|
||||
}
|
||||
]
|
||||
})
|
||||
const directory = await fsPromises.mkdtemp(join(tmpdir(), 'orca-legacy-source-budget-'))
|
||||
const filePath = join(directory, 'growing.jsonl')
|
||||
const initial = `${JSON.stringify({
|
||||
type: 'assistant',
|
||||
uuid: 'valid-prefix',
|
||||
message: {
|
||||
role: 'assistant',
|
||||
content: [{ type: 'text', text: 'Existing journal must survive refusal' }]
|
||||
}
|
||||
})}\n`
|
||||
let consumedBytes = 0
|
||||
let observedStatBytes
|
||||
let stream
|
||||
let replaceCalls = 0
|
||||
let journalEpoch = 'prior-epoch'
|
||||
const compiled = new Module(join(root, 'legacy-import-budget-probe.cjs'))
|
||||
compiled.filename = join(root, 'legacy-import-budget-probe.cjs')
|
||||
compiled.paths = Module._nodeModulePaths(root)
|
||||
compiled.require = (id) => {
|
||||
if (id === 'node:fs/promises') {
|
||||
return {
|
||||
...fsPromises,
|
||||
stat: async (...args) => {
|
||||
const snapshot = await fsPromises.stat(...args)
|
||||
if (args[0] === filePath) {
|
||||
observedStatBytes = snapshot.size
|
||||
await fsPromises.appendFile(filePath, Buffer.alloc(grownBytes - snapshot.size, 0x20))
|
||||
}
|
||||
return snapshot
|
||||
}
|
||||
}
|
||||
}
|
||||
if (id === 'node:fs') {
|
||||
return {
|
||||
...fs,
|
||||
createReadStream: (...args) => {
|
||||
stream = fs.createReadStream(...args)
|
||||
stream.on('data', (chunk) => {
|
||||
consumedBytes += Buffer.isBuffer(chunk)
|
||||
? chunk.byteLength
|
||||
: Buffer.byteLength(chunk, 'utf8')
|
||||
})
|
||||
return stream
|
||||
}
|
||||
}
|
||||
}
|
||||
return require(id)
|
||||
}
|
||||
compiled._compile(built.outputFiles[0].text, compiled.filename)
|
||||
try {
|
||||
await fsPromises.writeFile(filePath, initial)
|
||||
const result = await compiled.exports.importLegacyTranscriptIntoJournal({
|
||||
agent: 'claude',
|
||||
sessionId: 'source-budget-probe',
|
||||
fence: 0,
|
||||
options: { filePath },
|
||||
journal: {
|
||||
cursor: () => ({ epoch: journalEpoch, sequence: 0 }),
|
||||
replaceEpochItems: async () => {
|
||||
replaceCalls++
|
||||
journalEpoch = 'replacement-epoch'
|
||||
return { epoch: journalEpoch, sequence: 1 }
|
||||
}
|
||||
}
|
||||
})
|
||||
assert.equal(observedStatBytes, Buffer.byteLength(initial))
|
||||
assert.equal(stream.destroyed, true)
|
||||
assert.equal(result.ok, version === 'before')
|
||||
assert.equal(replaceCalls, version === 'before' ? 1 : 0)
|
||||
return {
|
||||
source: version === 'before' ? 'working tree without consumed-byte quota' : 'working tree',
|
||||
sourceSha256: Object.fromEntries(
|
||||
[...sources].map(([path, source]) => [
|
||||
path,
|
||||
createHash('sha256').update(source).digest('hex')
|
||||
])
|
||||
),
|
||||
observedStatBytes,
|
||||
sourceBytesAfterGrowth: (await fsPromises.stat(filePath)).size,
|
||||
consumedBytes,
|
||||
limitBytes: limit,
|
||||
streamDestroyed: stream.destroyed,
|
||||
result,
|
||||
journalReplaceCalls: replaceCalls,
|
||||
journalEpoch
|
||||
}
|
||||
} finally {
|
||||
stream?.destroy()
|
||||
await fsPromises.rm(directory, { recursive: true, force: true })
|
||||
}
|
||||
}
|
||||
|
||||
const results = {
|
||||
node: process.version,
|
||||
platform: process.platform,
|
||||
architecture: process.arch,
|
||||
before: await run('before'),
|
||||
after: await run('after')
|
||||
}
|
||||
process.stdout.write(`${JSON.stringify(results, null, 2)}\n`)
|
||||
@@ -0,0 +1,47 @@
|
||||
{
|
||||
"node": "v26.6.0",
|
||||
"platform": "darwin",
|
||||
"architecture": "arm64",
|
||||
"before": {
|
||||
"source": "working tree without consumed-byte quota",
|
||||
"sourceSha256": {
|
||||
"src/main/native-chat/agent-session-journal/journal-legacy-import.ts": "da799f8f34e6c28516b4ac98102b9d1fb76edebefafbdf0ca17d18a06cb79535",
|
||||
"src/main/native-chat/transcript-stream-lines.ts": "5f3a0a1e5b3d003076452fa22234af00502808ed6463672c0e58aa49ae5b6aa1"
|
||||
},
|
||||
"observedStatBytes": 149,
|
||||
"sourceBytesAfterGrowth": 17825792,
|
||||
"consumedBytes": 17825792,
|
||||
"limitBytes": 16777216,
|
||||
"streamDestroyed": true,
|
||||
"result": {
|
||||
"ok": true,
|
||||
"epoch": "replacement-epoch",
|
||||
"cursor": {
|
||||
"epoch": "replacement-epoch",
|
||||
"sequence": 1
|
||||
},
|
||||
"imported": 1,
|
||||
"replaced": true
|
||||
},
|
||||
"journalReplaceCalls": 1,
|
||||
"journalEpoch": "replacement-epoch"
|
||||
},
|
||||
"after": {
|
||||
"source": "working tree",
|
||||
"sourceSha256": {
|
||||
"src/main/native-chat/agent-session-journal/journal-legacy-import.ts": "88c007b4fb83eba6ec5fdf21c50b4c4aff35199cc4eed92abe3a1c29addfc5f3",
|
||||
"src/main/native-chat/transcript-stream-lines.ts": "5f3a0a1e5b3d003076452fa22234af00502808ed6463672c0e58aa49ae5b6aa1"
|
||||
},
|
||||
"observedStatBytes": 149,
|
||||
"sourceBytesAfterGrowth": 17825792,
|
||||
"consumedBytes": 16842752,
|
||||
"limitBytes": 16777216,
|
||||
"streamDestroyed": true,
|
||||
"result": {
|
||||
"ok": false,
|
||||
"error": "Input exceeds 16777216 byte limit (16842752 bytes received)"
|
||||
},
|
||||
"journalReplaceCalls": 0,
|
||||
"journalEpoch": "prior-epoch"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
import { createReadStream } from 'node:fs'
|
||||
import { appendFile, mkdtemp, rm, stat, writeFile } from 'node:fs/promises'
|
||||
import type * as FsPromises from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { prepareLegacyTranscriptImport } from './journal-legacy-import'
|
||||
|
||||
vi.mock(import('node:fs'), async (importOriginal) => {
|
||||
const original = await importOriginal()
|
||||
return { ...original, createReadStream: vi.fn(original.createReadStream) }
|
||||
})
|
||||
|
||||
vi.mock(import('node:fs/promises'), async (importOriginal) => {
|
||||
const original = await importOriginal()
|
||||
return { ...original, stat: vi.fn() }
|
||||
})
|
||||
|
||||
const SOURCE_LIMIT_BYTES = 16 * 1024 * 1024
|
||||
const roots: string[] = []
|
||||
|
||||
afterEach(async () => {
|
||||
vi.clearAllMocks()
|
||||
await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })))
|
||||
})
|
||||
|
||||
describe('legacy import source byte bound', () => {
|
||||
it('refuses a source that grows past the limit after stat and closes the stream', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-legacy-import-bound-'))
|
||||
roots.push(root)
|
||||
const filePath = join(root, 'growing.jsonl')
|
||||
await writeFile(
|
||||
filePath,
|
||||
`${JSON.stringify({
|
||||
type: 'assistant',
|
||||
uuid: 'first-message',
|
||||
message: { role: 'assistant', content: [{ type: 'text', text: 'Keep the prior journal' }] }
|
||||
})}\n`
|
||||
)
|
||||
const actualFs = await vi.importActual<typeof FsPromises>('node:fs/promises')
|
||||
const beforeGrowth = await actualFs.stat(filePath)
|
||||
vi.mocked(stat).mockImplementationOnce(async () => {
|
||||
await appendFile(filePath, Buffer.alloc(SOURCE_LIMIT_BYTES, 0x20))
|
||||
return beforeGrowth
|
||||
})
|
||||
|
||||
const result = await prepareLegacyTranscriptImport({
|
||||
agent: 'claude',
|
||||
sessionId: 'source-bound',
|
||||
options: { filePath }
|
||||
})
|
||||
|
||||
expect(result).toEqual({
|
||||
ok: false,
|
||||
error: expect.stringContaining(`${SOURCE_LIMIT_BYTES} byte limit`)
|
||||
})
|
||||
expect(createReadStream).toHaveBeenCalledOnce()
|
||||
expect(vi.mocked(createReadStream).mock.results[0]?.value.destroyed).toBe(true)
|
||||
})
|
||||
})
|
||||
@@ -194,7 +194,8 @@ async function decodeWithIdentities(input: {
|
||||
const identities: AgentJournalItemIdentity[] = []
|
||||
let lineIndex = 0
|
||||
|
||||
const stream = createReadStream(input.filePath, { encoding: 'utf-8' })
|
||||
// Count raw bytes while reading: the source can grow after the stat check.
|
||||
const stream = createReadStream(input.filePath)
|
||||
const { messages } = await decodeTranscriptStream(
|
||||
stream,
|
||||
input.filePath,
|
||||
@@ -217,7 +218,8 @@ async function decodeWithIdentities(input: {
|
||||
}
|
||||
return message
|
||||
},
|
||||
true
|
||||
true,
|
||||
MAX_LEGACY_IMPORT_SOURCE_BYTES
|
||||
)
|
||||
return { messages, identities }
|
||||
}
|
||||
|
||||
@@ -11,6 +11,35 @@ const decode = (line: string, id: string) => ({
|
||||
})
|
||||
|
||||
describe('decodeTranscriptStream', () => {
|
||||
it('accepts the exact source limit with split UTF-8 bytes and preserves order', async () => {
|
||||
const bytes = Buffer.from('é\n😀\n')
|
||||
const result = await decodeTranscriptStream(
|
||||
Readable.from([bytes.subarray(0, 1), bytes.subarray(1, 5), bytes.subarray(5)]),
|
||||
'/chat.jsonl',
|
||||
0,
|
||||
decode,
|
||||
true,
|
||||
bytes.length
|
||||
)
|
||||
expect(result.messages.map((message) => message.blocks[0])).toEqual([
|
||||
{ type: 'text', text: 'é' },
|
||||
{ type: 'text', text: '😀' }
|
||||
])
|
||||
})
|
||||
|
||||
it.each([Buffer.from('é'), 'é', Buffer.from([0xff, 0xff])])(
|
||||
'counts source bytes before decoding an oversized chunk %j',
|
||||
async (chunk) => {
|
||||
const stream = Readable.from([chunk])
|
||||
const trackedDecode = vi.fn(decode)
|
||||
await expect(
|
||||
decodeTranscriptStream(stream, '/chat.jsonl', 0, trackedDecode, true, 1)
|
||||
).rejects.toThrow('exceeds 1 byte limit')
|
||||
expect(trackedDecode).not.toHaveBeenCalled()
|
||||
expect(stream.destroyed).toBe(true)
|
||||
}
|
||||
)
|
||||
|
||||
it.each([true, false])('preserves chunked record offsets with trailing=%s', async (trailing) => {
|
||||
const first = `${'long record '.repeat(10_000)}😀`
|
||||
const prefix = `\r\n${first}\r\n\n`
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import type { Readable } from 'node:stream'
|
||||
import { StringDecoder } from 'node:string_decoder'
|
||||
import type { NativeChatMessage } from '../../shared/native-chat-types'
|
||||
import { NodeReadableTextTooLargeError } from '../../shared/node-readable-text'
|
||||
import { transcriptFallbackId } from './transcript-fallback-id'
|
||||
|
||||
type TranscriptDecoder = (line: string, fallbackId: string) => NativeChatMessage | null
|
||||
@@ -10,10 +11,12 @@ export async function decodeTranscriptStream(
|
||||
filePath: string,
|
||||
start: number,
|
||||
decode: TranscriptDecoder,
|
||||
includeTrailingLine: boolean
|
||||
includeTrailingLine: boolean,
|
||||
maxSourceBytes = Infinity
|
||||
): Promise<{ messages: NativeChatMessage[]; consumedBytes: number }> {
|
||||
const messages: NativeChatMessage[] = []
|
||||
let consumedBytes = 0
|
||||
let sourceBytes = 0
|
||||
const framer = createTranscriptLineFramer((line, byteLength, terminated) => {
|
||||
if (terminated || includeTrailingLine) {
|
||||
decodeLine(line, consumedBytes)
|
||||
@@ -21,6 +24,12 @@ export async function decodeTranscriptStream(
|
||||
}
|
||||
})
|
||||
for await (const chunk of stream) {
|
||||
if (maxSourceBytes !== Infinity) {
|
||||
sourceBytes += Buffer.isBuffer(chunk) ? chunk.byteLength : Buffer.byteLength(chunk, 'utf8')
|
||||
if (sourceBytes > maxSourceBytes) {
|
||||
throw new NodeReadableTextTooLargeError(sourceBytes, maxSourceBytes)
|
||||
}
|
||||
}
|
||||
framer.write(chunk)
|
||||
}
|
||||
framer.end()
|
||||
|
||||
Reference in New Issue
Block a user