diff --git a/.ai/contexts/session-cache.md b/.ai/contexts/session-cache.md index f305b048..9b8689d4 100644 --- a/.ai/contexts/session-cache.md +++ b/.ai/contexts/session-cache.md @@ -343,6 +343,108 @@ untouched. changed file (issue #216's second half) — a changed file discovered this way is still read in full by `readSessionFile`. +### Remote hosts — incremental fetch (issue #257) + +**The fetch itself is now incremental for a growing transcript; the parse +downstream of it is not (still issue #216's second half, unchanged).** +Measured 2026-09-11 (v0.0.76, main.log): 65 poll cycles in 19 min, one live +session's transcript re-`scp`'d whole in 52 of them — the dominant ssh +traffic of the app, since a session that keeps writing never stops being "the +one changed file" for `syncMirror`. + +- **Decision, per file, in `remote-mirror.js`'s `syncMirror`.** For a `rel` + already in the manifest (`previous[rel]` exists) whose remote `size` grew + and whose remote `mtimeMs` did not go backward (`meta.mtimeMs >= + prev.mtimeMs`), and whose **on-disk mirrored file still has exactly + `prev.size` bytes** (`fs.statSync(localPath).size === prev.size` — the + cheap proxy for "the manifest's record of this file is still true"), + `syncMirror` fetches only `[prev.size, meta.size)` and appends it, instead + of re-pulling the whole file. `.meta.json` sidecars are excluded outright + (small, never worth the extra round-trip logic). +- **Invalidation rule — full fetch, never a range, when any of these hold:** + remote size shrank (`meta.size < prev.size` — rotation or truncation); + remote size is unchanged but `mtimeMs` differs (a same-size rewrite, not an + append — nothing to safely append to); `prev` doesn't exist yet (first + pull for this file); the rel is a `.meta.json` sidecar; or the local + mirrored file's on-disk size doesn't match `prev.size` (someone or + something touched the mirror out of band since the manifest was written — + a crash mid-write, a manual edit). **Mtime alone is not trusted as proof of + an untouched prefix** — a rewrite that happens to grow the file could carry + any mtime, and `find -printf %T@`'s resolution/clock skew across hosts is + not something this code verifies further; the *size* check against the + actual on-disk file is what protects the prefix, mtime only screens out the + going-backward case cheaply before bothering to `stat()`. +- **The range fetch itself lives in `remote-transport.js`.** + `createSshTransport().fetchIncremental(alias, requests, destRoot)` takes + `requests: [{ rel, offset }]` and, per file, runs a single `ssh` command — + `` tail -c +${offset + 1} '.claude/projects/' `` (1-indexed: byte + `offset+1` is the first new byte) — capturing stdout as a raw `Buffer` + (`run(..., { binary: true })`), never through the utf8 string path the + inventory/list command uses, so a byte range that happens to split + multi-byte content is preserved exactly. The result is written + copy-then-append-then-rename: `fs.copyFileSync(dest, dest+'.part')`, + `fs.appendFileSync` the new bytes, `fs.renameSync` over `dest` — the + previous mirror is only ever replaced by that final atomic rename, so any + failure before it (ssh exit code, timeout, a size cap on the range output, + a disk error mid-append) leaves `dest` byte-identical to before the call + and removes the `.part`. A transport with no `fetchIncremental` (an older + fake in a test) gets the same files routed through `fetchFiles` instead — + additive, matching the `fileSubsets` precedent above. +- **Cycle bytes are now the transfer size, not the remote file size** — an + incremental candidate counts against `MAX_CYCLE_BYTES` as `meta.size - + offset`, not `meta.size`. Counting the full remote size would silently + undo the point of this feature: a 60 MB transcript that only grew by 4 KB + would otherwise still eat 60 MB of a cycle's 256 MB budget. +- **`cycleFull` is no longer a single sticky flag (issue #257).** The old + loop set one `cycleFull` boolean the first time a file didn't fit either + the file-count or the byte ceiling, and every file listed after it in host + `find` order was deferred too — even a much smaller file that would still + fit the remaining budget. `syncMirror` now (a) sorts every candidate + ascending by transfer size, transcripts before `.meta.json` sidecars (same + priority the sidecar fix already established), and (b) checks each file + independently against the remaining count/byte budget with no flag + latching a permanent "no more this cycle" state — a file that doesn't fit + is deferred on its own, the next (larger-or-equal, after sorting) file is + still evaluated on its own merits. Proven by + `test/remote-mirror.test.js`'s "a single large straggler never defers a + smaller file that still fits the cycle budget" (five 60 MB files plus one + 10 MB file, budget 256 MB — the 10 MB file is always fetched, regardless of + where it sits in host order). +- **A skip/defer that recurs every cycle logs once per doubling of its streak, + not once per cycle.** Same motivation as `onHostFailure`'s throttling in + `remote-index.js` (issue #215) — a transcript permanently over + `MAX_FILE_BYTES`, or a file that keeps losing the cycle-budget race, would + otherwise produce one warning per poll forever. `bumpStreak(manifestPath, + rel)` in `remote-mirror.js` keeps an **in-memory-only** `Map>` (module-level; same durability tradeoff as + `remote-index.js`'s `hostBackoff` — lost on restart, which just logs once + again, cheap) and logs on count 1, 2, 4, 8, 16, ... A file that stops being + skipped/deferred has its streak dropped (`pruneStreaks`), so a later + recurrence logs fresh rather than resuming at its old tier. +- **Mutation-proven**: setting the incremental branch's `offset` to `0` + instead of `localSize` (i.e. breaking the range start so it always starts + from byte 0) reddens `test/remote-mirror.test.js`'s "a transcript that only + grew is fetched incrementally: only the new bytes are requested" — the + fake transport's recorded `bytesRequested` no longer matches the actual + growth (measured 4168 vs the expected 4096 for a 4 KB append with the + mutation live). +- **Still out of scope**: parsing only the appended bytes. The mirrored file + on disk is now correct (fetched incrementally, but byte-identical to a + full fetch), and an incrementally-updated file still reaches + `scanFoldersViaWorker`/`readSessionFile` through the exact same + `fetched` → `changedFolders`/`changedFilesByFolder` → `fileSubsets` path a + fully-fetched file does (`syncMirror`'s return shape is unchanged by fetch + mode) — so the file-level rescan from issue #216 already re-reads only + this file, but still reads *all* of it, not just the new lines. Parse cost + is therefore unchanged by this issue; issue #216's second half remains the + place to fix that. +- **Known gap, not fixed here**: a remote session with a live descriptor + (`~/.claude/sessions/.json`) but no `.jsonl` written yet (a session + that was launched but has not been prompted) is invisible to the + inventory — `LIST_COMMAND`'s `find .claude/projects` only ever sees files + that exist. Observed 2026-09-11. Candidate for a follow-up issue; not + addressed by issue #257. + - **A remote project is never "missing".** `buildProjectsFromCache` sets `missing: false` for any aliased row. Probing the local filesystem for `/srv/supervision` would flag every remote project missing and offer it to the diff --git a/remote-mirror.js b/remote-mirror.js index 2a6cb1cc..fd8d8576 100644 --- a/remote-mirror.js +++ b/remote-mirror.js @@ -11,6 +11,30 @@ const MAX_FILE_BYTES = 64 * 1024 * 1024; const MAX_CYCLE_FILES = 500; const MAX_CYCLE_BYTES = 256 * 1024 * 1024; +// see .ai/contexts/session-cache.md ("Remote hosts — incremental fetch") +const cycleStreaks = new Map(); // manifestPath -> Map + +function isLogTier(count) { + return count === 1 || (count & (count - 1)) === 0; // 1, 2, 4, 8, 16, ... +} + +// Returns { count, shouldLog }. see .ai/contexts/session-cache.md ("Remote hosts — incremental fetch") +function bumpStreak(manifestPath, rel) { + let streaks = cycleStreaks.get(manifestPath); + if (!streaks) { streaks = new Map(); cycleStreaks.set(manifestPath, streaks); } + const count = (streaks.get(rel) || 0) + 1; + streaks.set(rel, count); + return { count, shouldLog: isLogTier(count) }; +} + +function pruneStreaks(manifestPath, seenThisCycle) { + const streaks = cycleStreaks.get(manifestPath); + if (!streaks) return; + for (const rel of [...streaks.keys()]) { + if (!seenThisCycle.has(rel)) streaks.delete(rel); + } +} + function readManifest(manifestPath) { try { const parsed = JSON.parse(fs.readFileSync(manifestPath, 'utf8')); @@ -42,8 +66,12 @@ function pruneEmptyDirs(root, dir) { /** * Bring the local mirror of one host in line with its remote inventory. * Injected transport: - * listFiles(alias) -> Promise<{ files: [{ rel, size, mtimeMs }], sessions: [object] }> - * fetchFiles(alias, rels, destRoot) -> Promise<{ fetched: [], failed: [] }> + * listFiles(alias) -> Promise<{ files: [{ rel, size, mtimeMs }], sessions: [object] }> + * fetchFiles(alias, rels, destRoot) -> Promise<{ fetched: [], failed: [] }> + * fetchIncremental(alias, requests, destRoot) -> Promise<{ fetched: [], failed: [] }> (optional; + * requests is [{ rel, offset }]; a transport without it gets the same + * files routed through fetchFiles instead — see "Remote hosts — + * incremental fetch" in .ai/contexts/session-cache.md) */ async function syncMirror({ alias, transport, projectsDir, manifestPath, log }) { const { files, sessions } = await transport.listFiles(alias); @@ -61,24 +89,21 @@ async function syncMirror({ alias, transport, projectsDir, manifestPath, log }) const previous = readManifest(manifestPath); - const toFetch = []; + // pass 1: full vs incremental per file — see .ai/contexts/session-cache.md ("Remote hosts — incremental fetch") + const seenThisCycle = new Set(); let skippedTooLarge = 0; - let cycleBytes = 0; - let cycleFull = false; - let deferredFiles = 0; - let deferredBytes = 0; - // see .ai/contexts/session-cache.md ("Remote hosts — meta.json sidecars") - const byFetchPriority = [...want.entries()].sort((a, b) => { - const aMeta = a[0].endsWith('.meta.json') ? 1 : 0; - const bMeta = b[0].endsWith('.meta.json') ? 1 : 0; - return aMeta - bMeta; - }); - for (const [rel, meta] of byFetchPriority) { + const candidates = []; + for (const [rel, meta] of want) { // The inventory already carries the size; scp is bounded in time only, so // this is the only place a single oversized transcript can be refused // before it lands. See .ai/contexts/session-cache.md, "Remote hosts". if (meta.size > MAX_FILE_BYTES) { skippedTooLarge++; + seenThisCycle.add(rel); + const { count, shouldLog } = bumpStreak(manifestPath, rel); + if (shouldLog && log && log.warn) { + log.warn(`[remote:${alias}] ${rel} skipped: over ${MAX_FILE_BYTES} bytes (${count}x consecutive)`); + } continue; } const prev = previous[rel]; @@ -86,32 +111,71 @@ async function syncMirror({ alias, transport, projectsDir, manifestPath, log }) if (prev && prev.size === meta.size && prev.mtimeMs === meta.mtimeMs && fs.existsSync(localPath)) { continue; } - if (!cycleFull && toFetch.length < MAX_CYCLE_FILES && cycleBytes + meta.size <= MAX_CYCLE_BYTES) { - toFetch.push(rel); - cycleBytes += meta.size; + + let mode = 'full'; + let offset = 0; + // see .ai/contexts/session-cache.md ("Remote hosts — incremental fetch") for the invalidation rule + if (!rel.endsWith('.meta.json') && prev && meta.size > prev.size && meta.mtimeMs >= prev.mtimeMs) { + let localSize = -1; + try { localSize = fs.statSync(localPath).size; } catch {} + if (localSize === prev.size) { + mode = 'incremental'; + offset = localSize; + } + } + const transferSize = mode === 'incremental' ? (meta.size - offset) : meta.size; + const isMeta = rel.endsWith('.meta.json') ? 1 : 0; + candidates.push({ rel, meta, mode, offset, transferSize, isMeta }); + } + + // pass 2: transcripts first, then transfer size ascending — see .ai/contexts/session-cache.md (cycle ordering) + candidates.sort((a, b) => (a.isMeta - b.isMeta) || (a.transferSize - b.transferSize)); + + const toFetchFull = []; + const toFetchIncremental = []; + let cycleBytes = 0; + let deferredFiles = 0; + let deferredBytes = 0; + for (const c of candidates) { + const wouldExceedCount = (toFetchFull.length + toFetchIncremental.length) >= MAX_CYCLE_FILES; + const wouldExceedBytes = cycleBytes + c.transferSize > MAX_CYCLE_BYTES; + if (!wouldExceedCount && !wouldExceedBytes) { + if (c.mode === 'incremental') toFetchIncremental.push({ rel: c.rel, offset: c.offset }); + else toFetchFull.push(c.rel); + cycleBytes += c.transferSize; } else { - cycleFull = true; deferredFiles++; - deferredBytes += meta.size; + deferredBytes += c.transferSize; + seenThisCycle.add(c.rel); + const { count, shouldLog } = bumpStreak(manifestPath, c.rel); + if (shouldLog && log && log.warn) { + const reason = wouldExceedCount + ? 'deferred to next cycle: over the per-cycle file-count ceiling' + : `deferred to next cycle: ${c.transferSize} bytes over the per-cycle byte ceiling`; + log.warn(`[remote:${alias}] ${c.rel} ${reason} (${count}x consecutive)`); + } } } - if (skippedTooLarge && log && log.warn) { - log.warn(`[remote:${alias}] ${skippedTooLarge} file(s) skipped: over ${MAX_FILE_BYTES} bytes`); - } - if (deferredFiles && log && log.warn) { - log.warn(`[remote:${alias}] ${deferredFiles} file(s) deferred to next cycle: ${deferredBytes} bytes over the per-cycle ceiling`); - } + pruneStreaks(manifestPath, seenThisCycle); let fetched = []; let failed = []; - if (toFetch.length > 0) { - const result = await transport.fetchFiles(alias, toFetch, projectsDir); - fetched = Array.isArray(result?.fetched) ? result.fetched : []; - failed = Array.isArray(result?.failed) ? result.failed : []; + if (toFetchFull.length > 0) { + const result = await transport.fetchFiles(alias, toFetchFull, projectsDir); + fetched.push(...(Array.isArray(result?.fetched) ? result.fetched : [])); + failed.push(...(Array.isArray(result?.failed) ? result.failed : [])); + } + if (toFetchIncremental.length > 0) { + // no fetchIncremental on the transport: route through fetchFiles instead + const result = typeof transport.fetchIncremental === 'function' + ? await transport.fetchIncremental(alias, toFetchIncremental, projectsDir) + : await transport.fetchFiles(alias, toFetchIncremental.map(r => r.rel), projectsDir); + fetched.push(...(Array.isArray(result?.fetched) ? result.fetched : [])); + failed.push(...(Array.isArray(result?.failed) ? result.failed : [])); } const fetchedSet = new Set(fetched); - const attempted = new Set(toFetch); + const attempted = new Set([...toFetchFull, ...toFetchIncremental.map(r => r.rel)]); const nextFiles = {}; for (const [rel, meta] of want) { if (fetchedSet.has(rel)) { nextFiles[rel] = meta; continue; } @@ -171,7 +235,7 @@ async function syncMirror({ alias, transport, projectsDir, manifestPath, log }) total: want.size, fetched: fetched.length, failed: failed.length, - unchanged: want.size - toFetch.length, + unchanged: want.size - (toFetchFull.length + toFetchIncremental.length), removed, changedFolders, changedFilesByFolder, diff --git a/remote-transport.js b/remote-transport.js index 9436655d..8bdf58bd 100644 --- a/remote-transport.js +++ b/remote-transport.js @@ -15,6 +15,8 @@ const DEFAULT_LIST_TIMEOUT_MS = 60_000; const DEFAULT_FETCH_TIMEOUT_MS = 120_000; const MAX_LIST_BYTES = 8 * 1024 * 1024; const DEFAULT_CONCURRENCY = 4; +// see .ai/contexts/session-cache.md ("Remote hosts — incremental fetch") +const MAX_RANGE_FETCH_BYTES = 64 * 1024 * 1024; const SSH_BASE_OPTS = [ '-o', 'BatchMode=yes', @@ -119,22 +121,25 @@ function createSshTransport(opts = {}) { const live = new Set(); let disposed = false; - function run(command, args, { timeoutMs, maxBytes }) { + // `binary: true` captures stdout as a raw Buffer. see .ai/contexts/session-cache.md ("Remote hosts — incremental fetch") + function run(command, args, { timeoutMs, maxBytes, binary }) { return new Promise((resolve) => { if (disposed) { - resolve({ code: -1, stdout: '', stderr: 'transport disposed', timedOut: false }); + resolve({ code: -1, stdout: '', stdoutBuffer: Buffer.alloc(0), stderr: 'transport disposed', timedOut: false }); return; } let child; try { child = spawn(command, args, { windowsHide: true, stdio: ['ignore', 'pipe', 'pipe'] }); } catch (err) { - resolve({ code: -1, stdout: '', stderr: err.message, timedOut: false }); + resolve({ code: -1, stdout: '', stdoutBuffer: Buffer.alloc(0), stderr: err.message, timedOut: false }); return; } live.add(child); let stdout = ''; + const chunks = []; + let stdoutBytes = 0; let stderr = ''; let truncated = false; let timedOut = false; @@ -153,19 +158,38 @@ function createSshTransport(opts = {}) { settled = true; clearTimeout(timer); live.delete(child); - resolve({ code, stdout, stderr: stderr.slice(0, 4096), timedOut, truncated }); + resolve({ + code, + stdout, + stdoutBuffer: binary ? Buffer.concat(chunks) : undefined, + stderr: stderr.slice(0, 4096), + timedOut, + truncated, + }); }; if (child.stdout) { - child.stdout.setEncoding('utf8'); - child.stdout.on('data', (chunk) => { - if (stdout.length + chunk.length > (maxBytes || MAX_LIST_BYTES)) { - truncated = true; - kill(); - return; - } - stdout += chunk; - }); + if (binary) { + child.stdout.on('data', (chunk) => { + stdoutBytes += chunk.length; + if (stdoutBytes > (maxBytes || MAX_RANGE_FETCH_BYTES)) { + truncated = true; + kill(); + return; + } + chunks.push(chunk); + }); + } else { + child.stdout.setEncoding('utf8'); + child.stdout.on('data', (chunk) => { + if (stdout.length + chunk.length > (maxBytes || MAX_LIST_BYTES)) { + truncated = true; + kill(); + return; + } + stdout += chunk; + }); + } } if (child.stderr) { child.stderr.setEncoding('utf8'); @@ -238,6 +262,62 @@ function createSshTransport(opts = {}) { return { fetched, failed }; } + // offset comes from the local mirror's byte count — see .ai/contexts/session-cache.md ("Remote hosts — incremental fetch") + async function fetchOneIncremental(alias, rel, offset, destRoot) { + if (!Number.isInteger(offset) || offset < 0) return false; + const destPath = path.join(destRoot, rel); + if (!fs.existsSync(destPath)) { + log.warn(`[remote:${alias}] range fetch ${rel} skipped — no local file to append to`); + return false; + } + const tmpPath = destPath + '.part'; + const remoteRel = `${REMOTE_PROJECTS_REL}/${rel}`; + // tail -c is 1-indexed: offset+1 is the first new byte. + const rangeCommand = `tail -c +${offset + 1} '${remoteRel}'`; + const res = await run('ssh', [...SSH_BASE_OPTS, '-n', alias, rangeCommand], { + timeoutMs: fetchTimeoutMs, + binary: true, + }); + if (res.code !== 0 || res.timedOut || res.truncated) { + const why = res.timedOut ? 'timed out' : res.truncated ? 'range exceeded the size cap' : `exit ${res.code}: ${res.stderr.trim()}`; + log.warn(`[remote:${alias}] range fetch ${rel} failed — ${why}`); + return false; + } + try { + // Copy-then-append-then-rename. see .ai/contexts/session-cache.md ("Remote hosts — incremental fetch") + fs.copyFileSync(destPath, tmpPath); + fs.appendFileSync(tmpPath, res.stdoutBuffer); + fs.renameSync(tmpPath, destPath); + } catch (err) { + try { fs.rmSync(tmpPath, { force: true }); } catch {} + log.warn(`[remote:${alias}] could not append ${rel}: ${err.message}`); + return false; + } + return true; + } + + // `requests` is [{ rel, offset }]. Same fetched/failed shape as fetchFiles. + async function fetchIncremental(alias, requests, destRoot) { + const fetched = []; + const failed = []; + const queue = (Array.isArray(requests) ? requests : []) + .filter(r => r && isSafeMirrorRelPath(r.rel) && Number.isInteger(r.offset) && r.offset >= 0); + let cursor = 0; + + const workers = Array.from({ length: Math.min(concurrency, queue.length) }, async () => { + while (!disposed) { + const i = cursor++; + if (i >= queue.length) return; + const { rel, offset } = queue[i]; + if (await fetchOneIncremental(alias, rel, offset, destRoot)) fetched.push(rel); + else failed.push(rel); + } + }); + await Promise.all(workers); + for (let i = cursor; i < queue.length; i++) failed.push(queue[i].rel); + return { fetched, failed }; + } + // Kill what is running without ending the transport -- see // .ai/contexts/session-cache.md, "Remote hosts". function cancelInFlight() { @@ -252,7 +332,7 @@ function createSshTransport(opts = {}) { cancelInFlight(); } - return { listFiles, fetchFiles, cancelInFlight, dispose, liveCount: () => live.size }; + return { listFiles, fetchFiles, fetchIncremental, cancelInFlight, dispose, liveCount: () => live.size }; } module.exports = { @@ -267,4 +347,5 @@ module.exports = { SESSIONS_MARKER, MAX_SESSION_DESCRIPTORS, MAX_SESSION_DESCRIPTOR_BYTES, + MAX_RANGE_FETCH_BYTES, }; diff --git a/test/remote-mirror.test.js b/test/remote-mirror.test.js index 41d5173d..2c692dd0 100644 --- a/test/remote-mirror.test.js +++ b/test/remote-mirror.test.js @@ -568,3 +568,191 @@ test('transcripts keep priority over .meta.json sidecars when a cycle is full', assert.deepEqual(asked, [metaRel], 'the deferred sidecar is picked up once the transcripts are unchanged'); } finally { fs.rmSync(dir, { recursive: true, force: true }); } }); + +// issue #257 — incremental fetch by byte range. This fake transport records +// exactly what fetchFiles (full) vs fetchIncremental (range) were asked for, +// so the decision syncMirror makes is visible from the outside, not just its +// end result. +function fakeRangeTransport(files, opts = {}) { + const calls = { list: 0, fetchFiles: [], fetchIncremental: [] }; + return { + calls, + async listFiles() { + calls.list++; + return { + files: Object.entries(files).map(([rel, f]) => ({ + rel, size: f.content.length, mtimeMs: f.mtimeMs, + })), + sessions: [], + }; + }, + async fetchFiles(alias, rels, destRoot) { + calls.fetchFiles.push(...rels); + const fetched = []; + const failed = []; + for (const rel of rels) { + if (opts.failFull && opts.failFull.includes(rel)) { failed.push(rel); continue; } + const dest = path.join(destRoot, rel); + fs.mkdirSync(path.dirname(dest), { recursive: true }); + fs.writeFileSync(dest, files[rel].content, 'utf8'); + fetched.push(rel); + } + return { fetched, failed }; + }, + async fetchIncremental(alias, requests, destRoot) { + const fetched = []; + const failed = []; + for (const { rel, offset } of requests) { + const tail = files[rel].content.slice(offset); + calls.fetchIncremental.push({ rel, offset, bytesRequested: tail.length }); + if (opts.failIncremental && opts.failIncremental.includes(rel)) { failed.push(rel); continue; } + const dest = path.join(destRoot, rel); + fs.appendFileSync(dest, tail, 'utf8'); + fetched.push(rel); + } + return { fetched, failed }; + }, + }; +} + +test('a transcript that only grew is fetched incrementally: only the new bytes are requested', async () => { + const dir = tmp('mirror-incr-grow'); + try { + const projectsDir = path.join(dir, 'projects'); + const manifestPath = path.join(dir, 'inventory.json'); + const files = { '-srv-a/a.jsonl': { content: line('/srv/a'), mtimeMs: 1000 } }; + const t = fakeRangeTransport(files); + await syncMirror({ alias: 'vps', transport: t, projectsDir, manifestPath }); + + t.calls.fetchFiles.length = 0; + const appended = 'z'.repeat(4096); // ~4 KB growth, matching the issue's acceptance case + files['-srv-a/a.jsonl'] = { content: files['-srv-a/a.jsonl'].content + appended, mtimeMs: 2000 }; + const r = await syncMirror({ alias: 'vps', transport: t, projectsDir, manifestPath }); + + assert.equal(t.calls.fetchFiles.length, 0, 'must not fall back to a full fetch'); + assert.equal(t.calls.fetchIncremental.length, 1); + assert.equal(t.calls.fetchIncremental[0].rel, '-srv-a/a.jsonl'); + assert.equal(t.calls.fetchIncremental[0].bytesRequested, appended.length, + 'the transfer is bounded by the growth, not the whole file'); + assert.equal(r.fetched, 1); + assert.equal( + fs.readFileSync(path.join(projectsDir, '-srv-a', 'a.jsonl'), 'utf8'), + files['-srv-a/a.jsonl'].content + ); + } finally { fs.rmSync(dir, { recursive: true, force: true }); } +}); + +// Mutation guard: change the incremental branch's `localSize === prev.size` / +// `offset + 1` computation and this line goes red — see remote-transport- +// incremental.test.js for the byte-count assertion this depends on. +test('a shrunk remote file (rotation/rewrite) falls back to a full fetch, never a byte range', async () => { + const dir = tmp('mirror-incr-shrink'); + try { + const projectsDir = path.join(dir, 'projects'); + const manifestPath = path.join(dir, 'inventory.json'); + const files = { '-srv-a/a.jsonl': { content: line('/srv/a').repeat(50), mtimeMs: 1000 } }; + const t = fakeRangeTransport(files); + await syncMirror({ alias: 'vps', transport: t, projectsDir, manifestPath }); + t.calls.fetchFiles.length = 0; + + files['-srv-a/a.jsonl'] = { content: 'rotated\n', mtimeMs: 2000 }; // much shorter + const r = await syncMirror({ alias: 'vps', transport: t, projectsDir, manifestPath }); + + assert.equal(t.calls.fetchIncremental.length, 0, 'a shrink must never be treated as an append'); + assert.deepEqual(t.calls.fetchFiles, ['-srv-a/a.jsonl']); + assert.equal(r.fetched, 1); + assert.equal(fs.readFileSync(path.join(projectsDir, '-srv-a', 'a.jsonl'), 'utf8'), 'rotated\n'); + } finally { fs.rmSync(dir, { recursive: true, force: true }); } +}); + +test('a local mirror whose on-disk size no longer matches the manifest is never trusted as a prefix', async () => { + const dir = tmp('mirror-incr-stale-prefix'); + try { + const projectsDir = path.join(dir, 'projects'); + const manifestPath = path.join(dir, 'inventory.json'); + const files = { '-srv-a/a.jsonl': { content: line('/srv/a'), mtimeMs: 1000 } }; + const t = fakeRangeTransport(files); + await syncMirror({ alias: 'vps', transport: t, projectsDir, manifestPath }); + t.calls.fetchFiles.length = 0; + + // Something out-of-band touched the local mirror — e.g. a previous crash + // mid-write. Its size no longer agrees with what the manifest recorded. + const localPath = path.join(projectsDir, '-srv-a', 'a.jsonl'); + fs.writeFileSync(localPath, 'tampered-or-torn'); + + files['-srv-a/a.jsonl'] = { content: files['-srv-a/a.jsonl'].content + 'more-bytes-appended', mtimeMs: 2000 }; + const r = await syncMirror({ alias: 'vps', transport: t, projectsDir, manifestPath }); + + assert.equal(t.calls.fetchIncremental.length, 0, 'a size mismatch against the manifest must never be trusted as a prefix'); + assert.deepEqual(t.calls.fetchFiles, ['-srv-a/a.jsonl']); + assert.equal(fs.readFileSync(localPath, 'utf8'), files['-srv-a/a.jsonl'].content); + } finally { fs.rmSync(dir, { recursive: true, force: true }); } +}); + +test('.meta.json sidecars always fetch in full, even when they grow', async () => { + const dir = tmp('mirror-incr-meta-full'); + try { + const projectsDir = path.join(dir, 'projects'); + const manifestPath = path.join(dir, 'inventory.json'); + const metaRel = '-srv-x/parent-uuid/subagents/agent-1.meta.json'; + const files = { [metaRel]: { content: JSON.stringify({ agentType: 'Explore' }), mtimeMs: 1000 } }; + const t = fakeRangeTransport(files); + await syncMirror({ alias: 'vps', transport: t, projectsDir, manifestPath }); + t.calls.fetchFiles.length = 0; + + files[metaRel] = { content: JSON.stringify({ agentType: 'Explore', description: 'grew' }), mtimeMs: 2000 }; + await syncMirror({ alias: 'vps', transport: t, projectsDir, manifestPath }); + + assert.equal(t.calls.fetchIncremental.length, 0); + assert.deepEqual(t.calls.fetchFiles, [metaRel]); + } finally { fs.rmSync(dir, { recursive: true, force: true }); } +}); + +// issue #257: cycleFull used to be a single sticky flag — one file that did +// not fit deferred every smaller file listed after it. Ascending-size +// ordering plus a per-file (not per-cycle) budget check fixes this: the +// smallest file must never be starved just because a straggler ahead of it +// (in host `find` order) didn't fit. +test('a single large straggler never defers a smaller file that still fits the cycle budget', async () => { + const dir = tmp('mirror-cycle-ordering'); + try { + const projectsDir = path.join(dir, 'projects'); + const manifestPath = path.join(dir, 'manifest.json'); + // Each file stays under the 64 MB per-file ceiling (MAX_FILE_BYTES) so + // none is refused outright; their sum (5 x 60 MB = 300 MB) still blows + // the 256 MB per-cycle ceiling (MAX_CYCLE_BYTES). + const size60 = 60 * 1024 * 1024; + const size10 = 10 * 1024 * 1024; + // Host order deliberately lists the small file LAST, behind five 60 MB + // files — the exact shape that starved it under the old sticky + // "cycle full" flag (the first straggler that didn't fit used to defer + // everything listed after it, this file included). + const entries = [ + { rel: '-srv-a/big-0.jsonl', size: size60, mtimeMs: 1 }, + { rel: '-srv-a/big-1.jsonl', size: size60, mtimeMs: 1 }, + { rel: '-srv-a/big-2.jsonl', size: size60, mtimeMs: 1 }, + { rel: '-srv-a/big-3.jsonl', size: size60, mtimeMs: 1 }, + { rel: '-srv-a/big-4.jsonl', size: size60, mtimeMs: 1 }, + { rel: '-srv-a/small.jsonl', size: size10, mtimeMs: 1 }, + ]; + const asked = []; + const transport = { + listFiles: async () => ({ files: entries, sessions: [] }), + fetchFiles: async (_alias, rels, destRoot) => { + asked.push(...rels); + for (const rel of rels) { + const p = path.join(destRoot, rel); + fs.mkdirSync(path.dirname(p), { recursive: true }); + fs.writeFileSync(p, 'x'); + } + return { fetched: rels, failed: [] }; + }, + }; + const r = await syncMirror({ alias: 'vps', transport, projectsDir, manifestPath }); + + assert.ok(asked.includes('-srv-a/small.jsonl'), + 'the 10 MB file must not be starved by the 60 MB files listed ahead of it'); + assert.equal(asked.length, 5, 'small plus four of the five 60 MB files fit the 256 MB budget'); + assert.equal(r.fetched, 5); + } finally { fs.rmSync(dir, { recursive: true, force: true }); } +}); diff --git a/test/remote-transport-incremental.test.js b/test/remote-transport-incremental.test.js new file mode 100644 index 00000000..a5416a24 --- /dev/null +++ b/test/remote-transport-incremental.test.js @@ -0,0 +1,197 @@ +'use strict'; + +// fetchIncremental (byte-range fetch for a transcript that only grew) with +// spawn injected, same style as remote-transport.test.js. See +// .ai/contexts/session-cache.md ("Remote hosts — incremental fetch") and +// issue #257. + +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('fs'); +const os = require('os'); +const path = require('path'); +const { EventEmitter } = require('events'); +const { Readable } = require('stream'); + +const { createSshTransport, REMOTE_PROJECTS_REL } = require('../remote-transport'); + +function fakeChild() { + const child = new EventEmitter(); + child.stdout = new Readable({ read() {} }); + child.stderr = new Readable({ read() {} }); + child.killed = 0; + child.kill = () => { child.killed++; child.emit('close', null); }; + return child; +} + +function spawnRecorder(handler) { + const calls = []; + const spawn = (cmd, args) => { + const child = fakeChild(); + calls.push({ cmd, args, child }); + if (handler) setImmediate(() => handler(child, cmd, args)); + return child; + }; + spawn.calls = calls; + return spawn; +} + +function tmpDir(name) { + return fs.mkdtempSync(path.join(os.tmpdir(), 'switchboard-' + name + '-')); +} + +test('fetchIncremental issues tail -c +N (N = offset+1) as the remote command', async () => { + const dir = tmpDir('incr-cmd'); + try { + const destPath = path.join(dir, '-srv-a', 'a.jsonl'); + fs.mkdirSync(path.dirname(destPath), { recursive: true }); + fs.writeFileSync(destPath, 'x'.repeat(100)); + + const spawn = spawnRecorder((child) => { + child.stdout.push('APPENDED'); + child.stdout.push(null); + child.emit('close', 0); + }); + const t = createSshTransport({ spawn }); + + const r = await t.fetchIncremental('planificator', [{ rel: '-srv-a/a.jsonl', offset: 100 }], dir); + + assert.deepEqual(r.fetched, ['-srv-a/a.jsonl']); + assert.equal(spawn.calls.length, 1); + const { cmd, args } = spawn.calls[0]; + assert.equal(cmd, 'ssh'); + assert.equal(args[args.length - 2], 'planificator'); + const command = args[args.length - 1]; + assert.equal(command, `tail -c +101 '${REMOTE_PROJECTS_REL}/-srv-a/a.jsonl'`); + assert.equal(fs.readFileSync(destPath, 'utf8'), 'x'.repeat(100) + 'APPENDED'); + } finally { fs.rmSync(dir, { recursive: true, force: true }); } +}); + +test('fetchIncremental requests exactly the grown byte count, not the whole file', async () => { + const dir = tmpDir('incr-bytes'); + try { + const destPath = path.join(dir, '-srv-a', 'a.jsonl'); + fs.mkdirSync(path.dirname(destPath), { recursive: true }); + const original = 'y'.repeat(50_000); + fs.writeFileSync(destPath, original); + const grown = 'z'.repeat(4096); // ~4 KB growth, matching the issue's acceptance case + + const spawn = spawnRecorder((child) => { + // Stand in for a real tail: only the requested range is produced. + child.stdout.push(grown); + child.stdout.push(null); + child.emit('close', 0); + }); + const t = createSshTransport({ spawn }); + + await t.fetchIncremental('vps', [{ rel: '-srv-a/a.jsonl', offset: original.length }], dir); + + const finalContent = fs.readFileSync(destPath, 'utf8'); + assert.equal(finalContent.length, original.length + grown.length); + assert.equal(finalContent, original + grown); + } finally { fs.rmSync(dir, { recursive: true, force: true }); } +}); + +test('fetchIncremental preserves bytes exactly (binary capture, no utf8 mangling)', async () => { + const dir = tmpDir('incr-binary'); + try { + const destPath = path.join(dir, '-srv-a', 'a.jsonl'); + fs.mkdirSync(path.dirname(destPath), { recursive: true }); + fs.writeFileSync(destPath, Buffer.from('prefix\n')); + // A byte sequence that is not valid standalone UTF-8 (a lone continuation + // byte) — if the transport decoded stdout as a utf8 string en route, this + // would come back as U+FFFD instead of the original byte. + const raw = Buffer.from([0x41, 0x80, 0x42, 0x0a]); + + const spawn = spawnRecorder((child) => { + child.stdout.push(raw); + child.stdout.push(null); + child.emit('close', 0); + }); + const t = createSshTransport({ spawn }); + + await t.fetchIncremental('vps', [{ rel: '-srv-a/a.jsonl', offset: 7 }], dir); + + const finalBuf = fs.readFileSync(destPath); + assert.deepEqual(finalBuf, Buffer.concat([Buffer.from('prefix\n'), raw])); + } finally { fs.rmSync(dir, { recursive: true, force: true }); } +}); + +test('a failed range fetch leaves the previous mirror byte-identical and no .part behind', async () => { + const dir = tmpDir('incr-fail'); + try { + const destPath = path.join(dir, '-srv-a', 'a.jsonl'); + fs.mkdirSync(path.dirname(destPath), { recursive: true }); + fs.writeFileSync(destPath, 'unchanged-content'); + + const spawn = spawnRecorder((child) => { + child.stderr.push('lost connection'); + child.stderr.push(null); + child.emit('close', 1); + }); + const t = createSshTransport({ spawn }); + + const r = await t.fetchIncremental('vps', [{ rel: '-srv-a/a.jsonl', offset: 18 }], dir); + + assert.deepEqual(r.failed, ['-srv-a/a.jsonl']); + assert.deepEqual(r.fetched, []); + assert.equal(fs.readFileSync(destPath, 'utf8'), 'unchanged-content', 'the previous mirror must be untouched'); + assert.equal(fs.existsSync(destPath + '.part'), false, 'no torn temp file left behind'); + } finally { fs.rmSync(dir, { recursive: true, force: true }); } +}); + +test('a range fetch with no local file to append to fails cleanly instead of creating one from a partial range', async () => { + const dir = tmpDir('incr-nolocal'); + try { + const spawn = spawnRecorder(null); + const t = createSshTransport({ spawn }); + + const r = await t.fetchIncremental('vps', [{ rel: '-srv-a/a.jsonl', offset: 10 }], dir); + + assert.deepEqual(r.failed, ['-srv-a/a.jsonl']); + assert.equal(spawn.calls.length, 0, 'never spawns ssh for a request with nothing to append to'); + } finally { fs.rmSync(dir, { recursive: true, force: true }); } +}); + +test('fetchIncremental writes through a .part file, same torn-write guard as a full fetch', async () => { + const dir = tmpDir('incr-part'); + try { + const destPath = path.join(dir, '-srv-a', 'a.jsonl'); + fs.mkdirSync(path.dirname(destPath), { recursive: true }); + fs.writeFileSync(destPath, 'base'); + let sawPart = false; + + const spawn = spawnRecorder((child) => { + // At the moment ssh "runs", the destination must still be the plain + // file — the .part is a local artifact of the append, never the scp + // destination argv (there is none here; this just documents that the + // range command itself never names a temp path). + child.stdout.push('-more'); + child.stdout.push(null); + child.emit('close', 0); + }); + const t = createSshTransport({ spawn }); + await t.fetchIncremental('vps', [{ rel: '-srv-a/a.jsonl', offset: 4 }], dir); + sawPart = fs.existsSync(destPath + '.part'); + + assert.equal(sawPart, false, 'the .part is renamed away by the time fetchIncremental resolves'); + assert.equal(fs.readFileSync(destPath, 'utf8'), 'base-more'); + } finally { fs.rmSync(dir, { recursive: true, force: true }); } +}); + +test('an unsafe rel path or a negative/non-integer offset never reaches ssh', async () => { + const dir = tmpDir('incr-evil'); + try { + const spawn = spawnRecorder(null); + const t = createSshTransport({ spawn }); + + const r = await t.fetchIncremental('vps', [ + { rel: '../../../etc/shadow.jsonl', offset: 0 }, + { rel: '-srv-a/a.jsonl', offset: -1 }, + { rel: '-srv-a/b.jsonl', offset: 'nope' }, + ], dir); + + assert.equal(spawn.calls.length, 0, 'no process may be spawned for those'); + assert.deepEqual(r.fetched, []); + } finally { fs.rmSync(dir, { recursive: true, force: true }); } +});