diff --git a/docs/architecture.md b/docs/architecture.md index 57d0426..964c2c2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -96,7 +96,8 @@ pure rules and never Node; `lib/runtime/` implements the ports and is the only s reads files or branches on the platform; `hosts//` holds data and pure functions, no IO; `bin/` and every `hosts//dist/` are entries that reach the plugin only through a channel; no shared layer names a host. Every session id the use cases see is qualified with its host when it leaves (`claude-code::`), so task ids -and ledger cycles say which agent a report came from. +say which agent a report came from; a local ledger record's cycle is the host's bare session id, and its `platform` +names the host. **Profiles.** `defineProfile` completes a host's profile with its channel's defaults and fails on load if a difference point is missing. The points are the channel; the event map onto eight canonical events @@ -149,7 +150,13 @@ Each IR that keeps the IR's shape (fields, counts, `scope.signals` of at most 64 record) goes to `POST /conversation/distill` as its own request: `ir_version`, `ir`, the same `record`, `user_prompt`, `final_answer`, `execution.trace` as `{ command, exit, source }` per check (`ok`, what its output said, in place of `exit` when the host gives no exit code, which is never made up), `project` (`{ root, head, diff }` -with the range's starting commit as `head`, or only `{ diff_omitted }`), `project_dir` and the trigger. The record is +with the range's starting commit as `head`, or only `{ diff_omitted }`), `project_dir`, the trigger and +`reused_asset_ids`: the strategies the session's injection ledger booked into the range's turns (a turn of an +earlier session file with the same number is another turn), `sha256:` ids only, in the order they were injected, at +most 20, left out when there are none or the session's file cannot be read. A Proxy that knows the field keeps it as +the Gene's `generation_meta.reused_asset_ids`, so a Gene names the assets the work it was distilled from reused. The +distiller reads only the body fields it knows, so a Proxy from before the field (2.0.50 and earlier) distills the same +Gene with or without it and the request never fails over it. The record is fitted to the Proxy's 256 KiB before the model reads it (the diff is left out as `too_large` first, then the diff lines, then the oldest turns), so the lines an IR cites are the lines sent. The Proxy's answer is `stored`, a draft, or refused with a typed `quality.reason` and the `terms` it found; every outcome, the model's skip and a failure are @@ -316,14 +323,25 @@ scripts/ pack.mjs, publish-tag.sh distillation in those projects. An unknown check is a plain command in the distillation record and provides no execution trace row; a check whose result is known without an exit code is a check line with no `exit` (and `failed: true` when it failed). A native exit code of zero still passes without requiring output. - A tier `none` report counts as exposure only on the Hub and is not written to the local ledger. An older Proxy - drops `source` and the plugin runs unchanged. + A tier `none` report counts as exposure only on the Hub and is not written to the local ledger. Any other report the + Hub took is also appended to evolver-core's root events, `value.reuse_hit` for a success and `value.reuse_outcome` + (with its `outcome`) otherwise, with `assetId`, `cycleId` (the host's session id), `platform` (the host), + `taskId`, `tier` and `stage`, so a reader of the ledger can tell which agent reused what on which evidence. + evolver-core's event schema leaves the payload open (2.0.50 and every release before it), so a core that reads + none of the four keeps them and records the event as before. + + The Proxy forwards `evidence` whole, and a Hub that does not know `source` refuses the whole report as + `validation_error`, which the Proxy passes on as a terminal rejection: on one machine from 5 to 8 November 2026, + all 175 Claude Code reports a 2.0.49 Proxy relayed with `source` were refused that way. Such a report goes once + more at once, without `source`; that answer settles it, and the outbox keeps the report as last sent, so a retry + never carries `source` again. - **Report outbox.** Reports are keyed by task id and asset, queued once and retried for 8 days, long enough for a retention follow-up a week later. A report counts as delivered once the Hub answers it at all: `recorded`, `duplicate_report`, or `eligible: false` (`daily_cap`, `same_owner`, `no_delivery_proof`) are final, so only an unreachable Proxy or an unrecorded answer is retried. The Hub counts one turn-end report per node, asset and UTC day. A request the Hub refuses as invalid can never succeed: the Proxy marks it `terminal: true` (an older one - says `hub 400` or `hub 422`), and the report is settled as rejected and kept out of the local ledger. `hub 404` + says `hub 400` or `hub 422`), and the report is settled as rejected and kept out of the local ledger (a + `validation_error` of a report with a `source` only once it was refused without it too, above). `hub 404` (the Hub may not serve the route yet), `hub 5xx`, 401, 403, 429 and timeouts are retried until the report expires. - **Turn numbers** come from a per-session counter, one per turn whatever started it, kept past the session's end: a resumed session continues under the same id, and a repeated task id is one the Hub drops as a duplicate. diff --git a/hosts/kimi/transcript.js b/hosts/kimi/transcript.js index 360e694..cf32183 100644 --- a/hosts/kimi/transcript.js +++ b/hosts/kimi/transcript.js @@ -21,19 +21,16 @@ const validResult = (result) => result && typeof result === 'object' && typeof r const pathOf = (input, cwd) => text(input?.path) && (isAbsolute(input.path) ? input.path : text(cwd) && isAbsolute(cwd) ? resolve(cwd, input.path) : null); -// The newest turns' tasks are the ones a turn still waiting for its end can name; older ones are past their end. const MAX_TASKS = 64; +const backgroundTaskOf = (result) => taskId(/^task_id:\s*([A-Za-z0-9_-]+)(?:\r?\n|$)/.exec(text(result?.output))?.[1]); -/** Only discovers fixed task IDs; the reader chooses a bounded file under the known session root. */ -export function taskIdsOf(records = []) { - const ids = []; - for (const record of records.filter(main)) { - const event = record.type === 'context.append_loop_event' ? record.event : null; - if (event?.type === 'tool.call' && event.name === 'TaskOutput') ids.push(taskId(event.args?.task_id)); - if (event?.type === 'tool.result') { - ids.push(taskId(/^task_id:\s*([A-Za-z0-9_-]+)(?:\r?\n|$)/.exec(text(event.result?.output))?.[1])); - } - } +/** + * Only discovers fixed task IDs, those the requested turn's Bash calls started: no other task can settle its + * evidence. The reader chooses a bounded file under the known session root. + */ +export function taskIdsOf(records = [], { turnId } = {}) { + const ids = callsOf(records, nativeTurnIdOf(turnId)).filter((call) => call.name === 'Bash') + .map((call) => backgroundTaskOf(resultOf(records, call))); const newestLast = [...new Set(ids.filter(Boolean).reverse())].reverse(); return newestLast.slice(-MAX_TASKS); } @@ -63,7 +60,7 @@ function notesOf(records, tasks, turnId, cwd) { const result = resultOf(records, call); const input = call.args && typeof call.args === 'object' ? call.args : {}; if (call.name === 'Bash' && text(input.command)) { - const expectedTask = taskId(/^task_id:\s*([A-Za-z0-9_-]+)(?:\r?\n|$)/.exec(text(result?.output))?.[1]); + const expectedTask = backgroundTaskOf(result); const matching = tasks.filter((task) => expectedTask && task?.kind === 'process' && task.parentToolCallId === call.toolCallId && task.command === input.command && task.taskId === expectedTask); const task = matching.length === 1 ? matching[0] : null; diff --git a/lib/app/deliver-reports.js b/lib/app/deliver-reports.js index 8601aa3..6ad0eea 100644 --- a/lib/app/deliver-reports.js +++ b/lib/app/deliver-reports.js @@ -3,9 +3,9 @@ import { deliverPending, isLocalRecordDue, postPending } from '../reuse.js'; -const ledgersOf = ({ proxy, ledger }) => ({ +const ledgersOf = ({ identity, proxy, ledger }) => ({ proxyFetch: (method, path, body) => proxy.request(method, path, body), - recordLocally: (entry) => ledger.record(entry), + recordLocally: (entry) => ledger.record({ ...entry, platform: identity.agent }), onRejected: async () => {}, }); diff --git a/lib/app/distill-run.js b/lib/app/distill-run.js index a0a346f..f1d6fab 100644 --- a/lib/app/distill-run.js +++ b/lib/app/distill-run.js @@ -10,9 +10,10 @@ import { defaultKindOf } from '../distill/kinds.js'; import { withOutcome } from '../distill/outcomes.js'; import { isCurrentRange } from '../distill/range.js'; import { sessionRecordOf } from '../distill/record.js'; -import { distillBodiesOf } from '../distill/request.js'; +import { distillBodiesOf, reusedAssetIdsOf } from '../distill/request.js'; import { blockedModelReason, distillResultOf, isDiscarded, postedOutcomeOf } from '../distill/result.js'; import { withoutSecrets } from '../redact.js'; +import { sessionOf } from '../session/state.js'; import { taskSessionId } from './turn-reports.js'; export const DISTILL_TIMEOUT_MS = 120_000; @@ -94,8 +95,18 @@ async function keptResult(ports, claimed, result) { return result.draft ? savedDraft(ports, claimed, result) : result; } +// A session file that cannot be read costs the distillation its lineage, never the distillation itself. +async function reusedAssetIdsIn(ports, { sessionId, range }) { + try { + return reusedAssetIdsOf(sessionOf(await ports.sessions.read(sessionId)), range); + } catch { + return []; + } +} + async function posted(ports, claimed, session, { irs, usage, signal }) { - const bodies = distillBodiesOf({ platform: ports.identity.agent, sessionId: taskSessionId(ports.identity, claimed.sessionId), trigger: claimed.trigger, session, irs, usage }); + const reusedAssetIds = await reusedAssetIdsIn(ports, claimed); + const bodies = distillBodiesOf({ platform: ports.identity.agent, sessionId: taskSessionId(ports.identity, claimed.sessionId), trigger: claimed.trigger, session, irs, usage, reusedAssetIds }); const results = []; for (const body of bodies) { // The range watch polls; a correction since the last look (while a refused IR was logged, or during the previous diff --git a/lib/distill/request.js b/lib/distill/request.js index 0cc746b..e3ffa7a 100644 --- a/lib/distill/request.js +++ b/lib/distill/request.js @@ -7,6 +7,25 @@ import { PROXY_BODY_MAX_BYTES } from './record.js'; const bytesOf = (value) => new TextEncoder().encode(JSON.stringify(value)).length; +const ASSET_ID = /^sha256:[0-9a-f]{64}$/; +const MAX_REUSED_ASSET_IDS = 20; + +/** + * The assets the session's injection ledger booked into the turns `range` covers, in the order they were injected, at + * most 20, for the Proxy to record what the distilled work reused. A turn of another session epoch (a session file + * started over) is another turn, whatever its number. + */ +export function reusedAssetIdsOf({ injected = {}, sessionEpoch }, range) { + const epochOf = (turn) => turn.sessionEpoch ?? range.sessionEpoch; + const isSessionTurn = (turn) => sessionEpoch === undefined || epochOf(turn) === undefined || epochOf(turn) === sessionEpoch; + const turns = new Set(range.turns.filter(isSessionTurn).map(({ turn }) => turn)); + return Object.entries(injected) + .filter(([assetId, entry]) => ASSET_ID.test(assetId) && turns.has(entry?.turn)) + .sort(([, left], [, right]) => left.turn - right.turn || (left.injectedAt ?? 0) - (right.injectedAt ?? 0)) + .slice(0, MAX_REUSED_ASSET_IDS) + .map(([assetId]) => assetId); +} + // The record is fitted before the model reads it; an IR long enough to push the body over still gets through by // losing the diff, then the diff lines, which come last so no other line's number changes. function withinBodyLimit(body) { @@ -18,9 +37,10 @@ function withinBodyLimit(body) { /** * One `POST /conversation/distill` body per IR, all on the same record (see lib/distill/record.js). The model's usage - * rides the first only, since one call wrote them all. + * rides the first only, since one call wrote them all. `reusedAssetIds` rides every one; a Proxy that does not know + * the field ignores it, as it ignores any field it does not read. */ -export function distillBodiesOf({ platform, sessionId, trigger, session, irs, usage = null, persist = true }) { +export function distillBodiesOf({ platform, sessionId, trigger, session, irs, usage = null, persist = true, reusedAssetIds = [] }) { return irs.map((ir, index) => withinBodyLimit({ platform, session_id: sessionId, @@ -33,6 +53,7 @@ export function distillBodiesOf({ platform, sessionId, trigger, session, irs, us execution: { trace: session.trace }, ...(session.project && { project: session.project }), ...(session.projectDir && { project_dir: session.projectDir }), + ...(reusedAssetIds.length > 0 && { reused_asset_ids: reusedAssetIds }), persist, publish: false, ...(usage && index === 0 && { distill_usage: usage }), diff --git a/lib/local-ledger.js b/lib/local-ledger.js index 3da020e..bd97ae5 100644 --- a/lib/local-ledger.js +++ b/lib/local-ledger.js @@ -70,6 +70,13 @@ function cycleIdOf(sessionId) { return sessionId || 'unknown'; } +const REUSE_DETAILS = ['platform', 'taskId', 'tier', 'stage']; + +// Which agent reused the asset, under which task, on what evidence: the cycle alone is a host's bare session id. +function reuseDetailsOf(record) { + return Object.fromEntries(REUSE_DETAILS.filter((key) => typeof record[key] === 'string' && record[key]).map((key) => [key, record[key]])); +} + // The verdict has to become a root_event to be worth anything: the actuator // that re-orders candidates reads root_events, not the Hub, and a Hub without // the route records nothing at all. Success and the negative verdicts ride @@ -80,7 +87,9 @@ export async function loadReuseLedger({ enginePrefix } = {}) { await loadIngestor(enginePrefix); } -export async function recordReuseLocally({ assetId, outcome, sessionId }, { enginePrefix } = {}) { +/** Appends one sent report to the local ledger: `{ assetId, outcome, sessionId, platform?, taskId?, tier?, stage? }`. */ +export async function recordReuseLocally(record, { enginePrefix } = {}) { + const { assetId, outcome, sessionId } = record; if (!assetId || !outcome) return false; const ingestor = await loadIngestor(enginePrefix); if (!ingestor) return false; @@ -101,7 +110,7 @@ export async function recordReuseLocally({ assetId, outcome, sessionId }, { engi title: `Evolver reuse ${outcome}`.slice(0, MAX_TITLE_CHARS), detail: `cycle ${cycleId}`, }, - payload: success ? { assetId, cycleId } : { assetId, cycleId, outcome }, + payload: { assetId, cycleId, ...(!success && { outcome }), ...reuseDetailsOf(record) }, }); } catch (error) { // Core acquires the lock before appending. Other errors can follow a write. diff --git a/lib/report-outbox.js b/lib/report-outbox.js index fe978d8..e20c3be 100644 --- a/lib/report-outbox.js +++ b/lib/report-outbox.js @@ -44,6 +44,6 @@ function localRecordOf({ local }, entry, now) { export function withAttempt(outbox, key, attempt, now) { const entry = outbox[key]; if (!entry) return outbox; - const attempted = { ...entry, attempts: entry.attempts + 1, ...localRecordOf(attempt, entry, now) }; + const attempted = { ...entry, ...(attempt.report && { report: attempt.report }), attempts: entry.attempts + 1, ...localRecordOf(attempt, entry, now) }; return { ...outbox, [key]: { ...attempted, ...settlementOf(attempt, now) } }; } diff --git a/lib/reuse.js b/lib/reuse.js index a821bc3..8840f3d 100644 --- a/lib/reuse.js +++ b/lib/reuse.js @@ -136,17 +136,31 @@ export const isLocalRecordDue = (entry) => Boolean(entry.sentAt) && isLocalRecor async function recordedLocally(recordLocally, { report, sessionId }) { try { - const recorded = await recordLocally({ assetId: report.asset_id, outcome: report.outcome, sessionId }); + const { tier, stage } = report.evidence; + const recorded = await recordLocally({ assetId: report.asset_id, outcome: report.outcome, sessionId, taskId: report.task_id, tier, stage }); return { local: recorded ? RECORDED : NOT_RECORDED }; } catch { return { local: UNCERTAIN }; } } +const VALIDATION_ERROR = 'validation_error'; + +const isRefusedForSource = (delivery, report) => delivery.verdict === REJECTED && delivery.reason === VALIDATION_ERROR && report.evidence?.source !== undefined; + +function withoutSource(report) { + const { source, ...evidence } = report.evidence; + return { ...report, evidence }; +} + +// A Hub that does not know `evidence.source` refuses the whole report as invalid, though the rest of it is evidence it +// takes: the report goes once more without it, and that answer settles it. The outbox keeps the report as last sent. async function postedOf(ledgers, entry) { - const delivery = await postToHub(ledgers.proxyFetch, entry); + const first = await postToHub(ledgers.proxyFetch, entry); + const resent = isRefusedForSource(first, entry.report) ? withoutSource(entry.report) : null; + const delivery = resent ? await postToHub(ledgers.proxyFetch, { ...entry, report: resent }) : first; if (delivery.verdict === REJECTED) await ledgers.onRejected({ taskId: entry.report.task_id, assetId: entry.report.asset_id, ...delivery }); - return delivery; + return resent ? { ...delivery, report: resent } : delivery; } async function attemptOf(ledgers, entry) { diff --git a/lib/runtime/session-log.js b/lib/runtime/session-log.js index a3acd48..e79ad75 100644 --- a/lib/runtime/session-log.js +++ b/lib/runtime/session-log.js @@ -132,8 +132,8 @@ function sessionDirOf(declaration, home, nativeId) { * names the log home (`root(env)`), its workspace index (`workspaceIndex`, listed as `[id, projectDir]` by * `workspacesOf`), a session's folder in a workspace (`sessionDir(workspaceId, nativeId)`) and its log (`logFile`); * `nativeSessionOf(sessionId, scope)` takes the plugin's session id back to the host's. `decode(records, { turnId, - * tasks, projectDir, unread })` answers the row, where `unread` is the byte span of a long log left unread; `taskIds(records)` and `tasksDir(sessionDir)` name the background task files - * it also reads. `scope` is a pseudonym of the log home, for session ids, so no path reaches a Hub task id. Anything + * tasks, projectDir, unread })` answers the row, where `unread` is the byte span of a long log left unread; + * `taskIds(records, { turnId })` and `tasksDir(sessionDir)` name the background task files it also reads. `scope` is a pseudonym of the log home, for session ids, so no path reaches a Hub task id. Anything * unreadable answers null, never an empty valid turn. */ export function sessionLogReaderOf(declaration, env = process.env) { @@ -160,7 +160,7 @@ export function sessionLogReaderOf(declaration, env = process.env) { const key = JSON.stringify([session.dir, stat.dev, stat.ino, stat.size, stat.mtimeMs]); if (parsed?.key !== key) parsed = { key, ...recordsOf(fd, stat.size) }; const { records, unread } = parsed; - const tasks = tasksOf(root, declaration.tasksDir?.(session.dir), declaration.taskIds?.(records) ?? []); + const tasks = tasksOf(root, declaration.tasksDir?.(session.dir), declaration.taskIds?.(records, { turnId }) ?? []); if (!tasks) return null; return declaration.decode(records, { turnId, tasks, projectDir: session.projectDir, unread }) ?? null; } catch { diff --git a/test/app/deliver-reports.test.js b/test/app/deliver-reports.test.js new file mode 100644 index 0000000..209b5a3 --- /dev/null +++ b/test/app/deliver-reports.test.js @@ -0,0 +1,36 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; + +import { recordReports } from '../../lib/app/deliver-reports.js'; +import { withQueued } from '../../lib/report-outbox.js'; +import { immediateReportOf } from '../../lib/reuse.js'; + +const NOW = 1_800_000_000_000; + +function portsHolding(outbox) { + const records = []; + let held = outbox; + return { + records, + outbox: () => held, + ports: { + identity: { agent: 'codex', label: 'Codex' }, + proxy: { request: async () => ({ ok: true, data: { recorded: true } }) }, + ledger: { load: async () => {}, record: async (entry) => { records.push(entry); return true; } }, + outbox: { + some: async (predicate) => Object.values(held).some(predicate), + drain: async (deliver) => { held = await deliver(held, () => {}); return true; }, + }, + clock: { now: () => NOW }, + }, + }; +} + +test('a delivered report is recorded in the local ledger under the host that reused the asset', async () => { + const report = immediateReportOf({ asset: { assetId: 'sha256:a' }, sessionId: 'codex:s1', turn: 2, turnReason: 'answer', checks: [{ command: 'npm test', failed: false }], applied: 'used' }); + const [[key, entry]] = Object.entries(withQueued({}, [{ report, sessionId: 's1' }], NOW)); + const held = portsHolding({ [key]: { ...entry, sentAt: NOW } }); + await recordReports(held.ports); + assert.deepEqual(held.records, [{ assetId: 'sha256:a', outcome: 'success', sessionId: 's1', taskId: 'codex:s1:2', tier: 'L1', stage: 'immediate', platform: 'codex' }]); + assert.equal(held.outbox()[key].localAt, NOW); +}); diff --git a/test/app/distill-session.test.js b/test/app/distill-session.test.js index 2804758..fbe8d38 100644 --- a/test/app/distill-session.test.js +++ b/test/app/distill-session.test.js @@ -11,7 +11,7 @@ import { TASKS } from '../../lib/app/ports.js'; import { draftSourcesOf } from '../../lib/distill.js'; import { turnEnded } from '../../lib/app/turn-end.js'; import { BLOCKED_ROUTE_TTL_MS, EPISODE_QUIET_MS, IDLE_DISTILL_MS, RETRY_BASE_MS, UNDISTILLED_LEFTOVER_TTL_MS, nextDueAt } from '../../lib/distill/range.js'; -import { memoryPorts } from './memory-ports.js'; +import { ASSET, memoryPorts } from './memory-ports.js'; import { DRAFTED, IR, changeTurn, fixTurn, modelAnswering, turnOf } from './work-turns.js'; const PROJECT = { @@ -76,8 +76,9 @@ test('a session-end distillation posts each IR in the contract\'s shape, keeps t assert.equal((await made.ports.candidates.read()).s1, undefined); assert.equal((await made.ports.drafts.read())[0].title, IR.title); const [{ body }] = distills(made.requests); - assert.deepEqual(Object.keys(body), ['platform', 'session_id', 'trigger', 'ir_version', 'ir', 'record', 'user_prompt', 'final_answer', 'execution', 'project', 'project_dir', 'persist', 'publish', 'distill_usage']); + assert.deepEqual(Object.keys(body), ['platform', 'session_id', 'trigger', 'ir_version', 'ir', 'record', 'user_prompt', 'final_answer', 'execution', 'project', 'project_dir', 'reused_asset_ids', 'persist', 'publish', 'distill_usage']); assert.deepEqual([body.platform, body.session_id, body.trigger, body.ir_version, body.persist, body.publish, body.project_dir], ['testagent', 'testagent:s1', 'session_end', 'session-gene-ir-v1', true, false, '/proj']); + assert.deepEqual(body.reused_asset_ids, [ASSET.asset_id], 'the strategy the turn was recalled'); assert.deepEqual(body.ir, IR); assert.deepEqual(body.record.slice(0, 9), [ { kind: 'turn', text: 'Turn 1' }, @@ -95,6 +96,28 @@ test('a session-end distillation posts each IR in the contract\'s shape, keeps t assert.deepEqual([kept.gate.passed, kept.kind, kept.record], [true, 'fix', body.record]); }); +test('a distillation names only the assets injected into the turns it read', async () => { + const offTopic = { ...ASSET, asset_id: 'sha256:' + '1'.repeat(64), short_title: 'Weather lookups', signals_match: ['weather'] }; + const made = memoryPorts({ assets: [offTopic, ASSET], distillAnswer: DRAFTED, project: PROJECT, model: modelAnswering(IR) }); + await turnOf(made.ports, 's1', { prompt: 'what is the weather like', answer: 'Sunny.' }); + await fixTurn(made.ports, 's1'); + const { injected } = await made.ports.sessions.read('s1'); + assert.deepEqual([injected[offTopic.asset_id].turn, injected[ASSET.asset_id].turn], [1, 2], 'the answered question was recalled for too'); + await endedAndDistilled(made); + const [{ body }] = distills(made.requests); + assert.deepEqual(body.reused_asset_ids, [ASSET.asset_id]); +}); + +test('a session ledger that cannot be read leaves out the reused assets, never the distillation', async () => { + const made = await fixedSession({ model: modelAnswering(IR) }); + await sessionEnded(made.ports, { sessionId: 's1', projectDir: '/proj' }); + made.ports.sessions.read = async () => { throw Object.assign(new Error('Invalid JSON in the session file'), { code: 'EINVALIDSTATE' }); }; + const [outcome] = await distillDue(made.ports); + assert.equal(outcome.status, 'drafted'); + const [{ body }] = distills(made.requests); + assert.equal('reused_asset_ids' in body, false); +}); + test('the record holds every work turn in order: a failed check\'s redacted output tail, the correction, and a diff cut to fit a prompt', async () => { const output = [...Array.from({ length: 30 }, (_, index) => `line ${index}`), 'Authorization: Bearer abcdefghijklmnopqrstuvwxyz0123'].join('\n'); const files = Array.from({ length: 40 }, (_, index) => `diff --git a/src/f${index}.js b/src/f${index}.js\n--- a/src/f${index}.js\n+++ b/src/f${index}.js\n@@ -0,0 +1,40 @@\n${'+export const value = 1;\n'.repeat(39)}+++count;\n`); diff --git a/test/distill-prompt.test.js b/test/distill-prompt.test.js index 10bd2a1..f046bc1 100644 --- a/test/distill-prompt.test.js +++ b/test/distill-prompt.test.js @@ -4,7 +4,7 @@ import { test } from 'node:test'; import { irPromptOf } from '../lib/distill/ir-prompt.js'; import { defaultKindOf } from '../lib/distill/kinds.js'; import { sessionRecordOf } from '../lib/distill/record.js'; -import { distillBodiesOf } from '../lib/distill/request.js'; +import { distillBodiesOf, reusedAssetIdsOf } from '../lib/distill/request.js'; import { workTurnOf } from '../lib/distill/work-turn.js'; const rangeOf = (records, source = 'observed') => ({ @@ -77,3 +77,41 @@ test('the Proxy body keeps a clean saved snapshot as it was saved: an empty diff assert.deepEqual(bodyOf(range).project, range.project, diff); } }); + +const assetId = (digit) => `sha256:${String(digit).repeat(64)}`; +const injectedAt = (turn, at = turn) => ({ turn, injectedAt: at, strategyText: '' }); + +test('the reused assets of a range are those injected into its turns, valid ids only, in injection order, at most 20', () => { + const range = { sessionEpoch: 7, turns: [{ turn: 2 }, { turn: 3, sessionEpoch: 7 }] }; + const injected = { + [assetId(3)]: injectedAt(3, 30), + [assetId(1)]: injectedAt(1), + [assetId(2)]: injectedAt(2), + [assetId(4)]: injectedAt(3, 20), + 'sha256:step1': injectedAt(2), + [`sha256:${'A'.repeat(64)}`]: injectedAt(2), + [assetId(5)]: { turn: '2', injectedAt: 1 }, + }; + assert.deepEqual(reusedAssetIdsOf({ injected, sessionEpoch: 7 }, range), [assetId(2), assetId(4), assetId(3)]); + + const many = Object.fromEntries(Array.from({ length: 25 }, (_, index) => [`sha256:${index.toString(16).padStart(64, '0')}`, injectedAt(2, 100 - index)])); + const kept = reusedAssetIdsOf({ injected: many, sessionEpoch: 7 }, range); + assert.equal(kept.length, 20); + assert.equal(kept[0], `sha256:${(24).toString(16).padStart(64, '0')}`, 'the earliest injections are kept'); +}); + +test('a turn of another session epoch reused nothing this session file booked', () => { + const injected = { [assetId(1)]: injectedAt(1) }; + assert.deepEqual(reusedAssetIdsOf({ injected, sessionEpoch: 9 }, { sessionEpoch: 7, turns: [{ turn: 1 }] }), []); + assert.deepEqual(reusedAssetIdsOf({ injected, sessionEpoch: 9 }, { sessionEpoch: 7, turns: [{ turn: 1, sessionEpoch: 9 }] }), [assetId(1)]); + assert.deepEqual(reusedAssetIdsOf({ injected }, { sessionEpoch: 7, turns: [{ turn: 1 }] }), [assetId(1)], 'a ledger from before epochs'); + assert.deepEqual(reusedAssetIdsOf({}, { turns: [{ turn: 1 }] }), []); +}); + +test('a body carries reused_asset_ids only when there are some, on every IR', () => { + const session = sessionRecordOf(rangeOf([{ checks: [{ command: 'npm test', exit: 0, failed: false }] }])); + const bodies = (reusedAssetIds) => distillBodiesOf({ platform: 'codex', sessionId: 's1', trigger: 'idle', session, irs: [{ title: 'A' }, { title: 'B' }], reusedAssetIds }); + assert.deepEqual(bodies([assetId(1)]).map((body) => body.reused_asset_ids), [[assetId(1)], [assetId(1)]]); + assert.equal('reused_asset_ids' in bodies([])[0], false); + assert.equal('reused_asset_ids' in bodies(undefined)[0], false); +}); diff --git a/test/helpers/process-home.js b/test/helpers/process-home.js new file mode 100644 index 0000000..2e4e0b2 --- /dev/null +++ b/test/helpers/process-home.js @@ -0,0 +1,22 @@ +import { join } from 'node:path'; + +const KEYS = ['HOME', 'USERPROFILE', 'EVOLVER_HOME', 'EVOMAP_HOME']; + +/** + * Points this process's home and Evolver home at `home` until `restore()`. A resident host's ledger thread and its + * evolver-core read the process environment, never the `env` its ports were built with, so a fixture home passed only + * as `env` leaves the reuse records in the developer's own `~/.evomap`. `env` carries the same values for the ports. + */ +export function useProcessHome(home) { + const previous = Object.fromEntries(KEYS.map((key) => [key, process.env[key]])); + const env = { HOME: home, USERPROFILE: home, EVOLVER_HOME: join(home, '.evomap') }; + delete process.env.EVOMAP_HOME; + Object.assign(process.env, env); + return { + env, + rootEventsPath: join(env.EVOLVER_HOME, 'evolution', 'root_events.jsonl'), + restore: () => { + for (const [key, value] of Object.entries(previous)) if (value === undefined) delete process.env[key]; else process.env[key] = value; + }, + }; +} diff --git a/test/hosts/dsh/host.js b/test/hosts/dsh/host.js index ede4e1f..0db0359 100644 --- a/test/hosts/dsh/host.js +++ b/test/hosts/dsh/host.js @@ -29,6 +29,10 @@ after(() => scratch.forEach((dir) => rmSync(dir, { recursive: true, force: true let profileDir = null; +// A session start raised while a test ends checks the engine after its Proxy and home are gone; only a stopped plugin +// keeps that check from installing the real engine and starting its Proxy under the process's own home. +const loaded = new Set(); + /** A dsh profile with the packed folder linked in, the way `dsh plugin --profile

add -w ` leaves it. */ export function packedProfile() { if (profileDir) return profileDir; @@ -68,7 +72,8 @@ export async function startProxy(t, options = {}) { Object.assign(process.env, { HOME: proxy.env.HOME, USERPROFILE: proxy.env.HOME }); delete process.env.EVOMAP_PROXY_PORT; delete process.env.EVOLVER_RECALL; - t.after(() => { + t.after(async () => { + await Promise.all([...loaded].map((dsh) => dsh.dispose())); for (const [key, value] of Object.entries(previous)) if (value === undefined) delete process.env[key]; else process.env[key] = value; proxy.close(); }); @@ -194,7 +199,10 @@ export async function startDsh({ config = {}, llm = fakeLlm() } = {}) { event(agent, type, data) { emit('session/event', agent.session, { type, data }); }, - dispose: async () => Promise.all(disposers.map((dispose) => dispose?.())), + dispose: async () => { + if (!loaded.delete(dsh)) return; + await Promise.all(disposers.map((dispose) => dispose?.())); + }, /** Resolves once every host event raised so far for the agent's session was handled: they are handled in order. */ settled: (agent) => dsh.tool(agent, { name: 'todo_write', arguments: { todos: [] }, result: { isError: false, content: [] } }), disposeSession(agent) { @@ -221,6 +229,7 @@ export async function startDsh({ config = {}, llm = fakeLlm() } = {}) { return steps; }, }; + loaded.add(dsh); return dsh; } diff --git a/test/hosts/dsh/plugin.test.js b/test/hosts/dsh/plugin.test.js index a7a5e39..c0e061e 100644 --- a/test/hosts/dsh/plugin.test.js +++ b/test/hosts/dsh/plugin.test.js @@ -260,8 +260,10 @@ test('a turn no person\'s message opened is recalled for nothing and records not test('options: proxyPort reaches the Proxy when settings name none, and assetPrimeEnabled: false turns recall off', async (t) => { const proxy = await startProxy(t); - const { rmSync } = await import('node:fs'); - rmSync(`${proxy.home}/.evolver/settings.json`); + const { writeFileSync } = await import('node:fs'); + // Still a current Proxy's version: with none, a session start would install the real engine from npm and start + // its Proxy on the default port, under whatever home the process has by then. + writeFileSync(`${proxy.home}/.evolver/settings.json`, JSON.stringify({ proxy: { version: '2.0.50' } })); const dsh = await startDsh({ config: { proxyPort: proxy.port } }); const agent = await dsh.createAgent(); const decision = await dsh.preStep(agent, { turn: 1, step: 1, messages: [{ ...DSH.preStep.prompt.messages[0], content: [{ type: 'text', text: PROMPT }] }] }); diff --git a/test/hosts/kimi/readers.test.js b/test/hosts/kimi/readers.test.js index b2411e8..d042aab 100644 --- a/test/hosts/kimi/readers.test.js +++ b/test/hosts/kimi/readers.test.js @@ -91,13 +91,24 @@ test('unsupported schemas and foreign agents do not supply evidence', () => { assert.equal(status([meta, { ...prompt(0), origin: { kind: 'cron' } }]), null); }); -test('fixed task discovery rejects paths and does not inspect arbitrary output paths', () => { - const records = [event({ type: 'tool.call', name: 'TaskOutput', args: { task_id: 'bash-task' } }), event({ type: 'tool.result', result: { output: 'task_id: bash-task\nstatus: running' } }), event({ type: 'tool.call', name: 'TaskOutput', args: { task_id: '../outside' } })]; - assert.deepEqual(taskIdsOf(records), ['bash-task']); - const many = Array.from({ length: 70 }, (_, index) => event({ type: 'tool.call', name: 'TaskOutput', args: { task_id: `task-${index}` } })); - const ids = taskIdsOf([...many, many[0]]); +test('fixed task discovery reads only background tasks the requested turn started and rejects paths', () => { + const bash = (id, task, turnId = '0') => [ + event({ type: 'tool.call', uuid: `call-${id}`, toolCallId: `tool-${id}`, turnId, name: 'Bash', args: { command: 'npm test' } }), + event({ type: 'tool.result', parentUuid: `call-${id}`, toolCallId: `tool-${id}`, result: { output: `task_id: ${task}\nstatus: running`, durationMs: 1 } }), + ]; + const records = [ + ...bash(0, 'bash-task'), + ...bash(1, '../outside'), + ...bash(2, 'foreign-task', '1'), + event({ type: 'tool.call', uuid: 'read-uuid', toolCallId: 'read-call', turnId: '0', name: 'TaskOutput', args: { task_id: 'read-task' } }), + event({ type: 'tool.result', result: { output: 'task_id: unrelated-task\nstatus: running' } }), + ]; + assert.deepEqual(taskIdsOf(records, { turnId: '0' }), ['bash-task']); + assert.deepEqual(taskIdsOf(records, { turnId: '1' }), ['foreign-task']); + const many = Array.from({ length: 70 }, (_, index) => bash(index, `task-${index}`)).flat(); + const ids = taskIdsOf([...many, ...bash(70, 'task-0')], { turnId: '0' }); assert.equal(ids.length, 64); - assert.equal(ids.at(-1), 'task-0', 'a task the newest turn names again is kept'); + assert.equal(ids.at(-1), 'task-0', 'a task named again by a later call is kept'); }); test('duplicate native events do not duplicate checks and mismatched results remain unknown', () => { diff --git a/test/hosts/minimax/lifecycle.test.js b/test/hosts/minimax/lifecycle.test.js index 2912fe6..88eb1b9 100644 --- a/test/hosts/minimax/lifecycle.test.js +++ b/test/hosts/minimax/lifecycle.test.js @@ -1,7 +1,7 @@ import assert from 'node:assert/strict'; import { spawn } from 'node:child_process'; import { once } from 'node:events'; -import { mkdirSync, writeFileSync } from 'node:fs'; +import { mkdirSync, readFileSync, writeFileSync } from 'node:fs'; import { join } from 'node:path'; import { test } from 'node:test'; @@ -15,18 +15,20 @@ import { transcripts } from '../../../lib/runtime/transcripts.js'; import { PROMPT } from '../../app/memory-ports.js'; import { IR } from '../../app/work-turns.js'; import { fakeProxy } from '../../helpers/fake-proxy.js'; +import { useProcessHome } from '../../helpers/process-home.js'; const DRAFT = { ok: true, status: 'draft', distill_id: 'minimax-draft', quality: { ok: true, score: 6 }, gene: { type: 'Gene', asset_id: `sha256:${'1'.repeat(64)}` }, capsule: { type: 'Capsule', asset_id: `sha256:${'2'.repeat(64)}` } }; async function fixture({ failFirst = false } = {}) { const proxy = await fakeProxy({ distillAnswer: DRAFT }); + const home = useProcessHome(proxy.env.HOME); const project = join(proxy.env.HOME, 'project'); mkdirSync(project); const file = join(project, 'fixture.txt'); writeFileSync(file, 'before'); const host = spawn(process.execPath, ['-e', 'setInterval(()=>{},60000)'], { windowsHide: true, stdio: 'ignore' }); await once(host, 'spawn'); let owner = host.pid, calls = 0, skew = 0; const nativeData = join(proxy.env.HOME, 'native'); - const env = { ...proxy.env, MINIMAX_DATA_DIR: nativeData, PLUGIN_DATA: join(nativeData, 'v2', 'plugin-data', 'hooks', 'evolver-plugin') }; + const env = { ...proxy.env, ...home.env, MINIMAX_DATA_DIR: nativeData, PLUGIN_DATA: join(nativeData, 'v2', 'plugin-data', 'hooks', 'evolver-plugin') }; const { ports } = environmentFor({ profile, env, lifetime: RESIDENT, runBackground: runBackgroundTasks, overrides: { owner: { current: () => owner, self: () => process.pid, isGone: ownerIsGone }, clock: { now: () => Date.now() + skew, sleep: (ms) => new Promise((resolve) => setTimeout(resolve, ms)) }, @@ -54,12 +56,25 @@ async function fixture({ failFirst = false } = {}) { const close = async () => { await idle(); ports.scheduler.stop(); if (host.exitCode === null && host.signalCode === null) { const ended = once(host, 'close'); host.kill(); await ended; } + home.restore(); proxy.close(); }; const advance = (ms) => { skew += ms; }; - return { proxy, ports, hook, idle, fixedError, killHost, close, advance, calls: () => calls }; + const ledgerEvents = () => readFileSync(home.rootEventsPath, 'utf8').trim().split('\n').map((line) => JSON.parse(line)); + return { proxy, ports, hook, idle, fixedError, killHost, close, advance, ledgerEvents, calls: () => calls }; } +test('the reuse a verified turn reports reaches the ledger under the fixture home, naming its host and evidence', async () => { + const f = await fixture(); + try { + await f.fixedError(); + const reuses = f.ledgerEvents().filter(({ type }) => type === 'value.reuse_hit'); + assert.deepEqual(reuses.map(({ payload }) => payload), [ + { assetId: `sha256:${'a'.repeat(64)}`, cycleId: 's1', platform: 'minimax', taskId: 'minimax:s1:1', tier: 'L1', stage: 'immediate' }, + ]); + } finally { await f.close(); } +}); + test('a dead native owner is recovered without SessionEnd, distilled once and kept as an unpublished draft', async () => { const f = await fixture(); try { diff --git a/test/reuse.test.js b/test/reuse.test.js index b87b393..7829b21 100644 --- a/test/reuse.test.js +++ b/test/reuse.test.js @@ -12,6 +12,7 @@ import { correctionReportOf, deliverPending, deliveryOf, + postPending, immediateReportOf, reportedReuseOf, retentionReportOf, @@ -142,7 +143,7 @@ test('evidence-bearing reports are recorded locally once; tier none and a trial const control = { report: { asset_id: 'sha256:a', withheld_id: 'wh-1', task_id: 'claude-code:s1:3', passed: true, platform: 'claude-code' }, sessionId: 'claude-code:s1', path: '/asset/trial-control' }; const outbox = withQueued({}, [{ report: verified, sessionId: 'claude-code:s1' }, { report: immediate({ turn: 5 }), sessionId: 'claude-code:s1' }, control], NOW); await deliverPending(sinks, await deliverPending(sinks, outbox, NOW), NOW); - assert.deepEqual(sinks.locals, [{ assetId: 'sha256:a', outcome: 'failed', sessionId: 'claude-code:s1' }]); + assert.deepEqual(sinks.locals, [{ assetId: 'sha256:a', outcome: 'failed', sessionId: 'claude-code:s1', taskId: 'claude-code:s1:4', tier: 'L1', stage: 'immediate' }]); assert.deepEqual(sinks.posts.map(({ path }) => path), ['/asset/reuse-result', '/asset/reuse-result', '/asset/trial-control', '/asset/reuse-result', '/asset/reuse-result', '/asset/trial-control']); assert.deepEqual(sinks.posts[2].body, control.report); }); @@ -157,7 +158,7 @@ test('a sent report retries a pre-write local failure without posting to the Hub const up = ledgers(); const recorded = await deliverPending(up, sent, NOW + 1); assert.equal(up.posts.length, 0); - assert.deepEqual(up.locals, [{ assetId: 'sha256:a', outcome: 'success', sessionId: 'claude-code:s1' }]); + assert.deepEqual(up.locals, [{ assetId: 'sha256:a', outcome: 'success', sessionId: 'claude-code:s1', taskId: 'claude-code:s1:4', tier: 'L1', stage: 'immediate' }]); assert.equal(Object.values(recorded)[0].sentAt, NOW, 'the original Hub delivery time stays intact'); assert.equal(Object.values(recorded)[0].localAt, NOW + 1); await deliverPending(up, recorded, NOW + 2); @@ -235,7 +236,19 @@ test('a ledger path blocked by a directory retries locally after the path is rep assert.equal(sinks.posts.length, 1); const entries = new events.Ingestor({ path }).readAllStrict(); assert.deepEqual(entries.map(({ type, payload }) => ({ type, payload })), [ - { type: 'value.reuse_hit', payload: { assetId: ASSET.assetId, cycleId: 'claude-code:s1' } }, + { type: 'value.reuse_hit', payload: { assetId: ASSET.assetId, cycleId: 'claude-code:s1', taskId: 'claude-code:s1:4', tier: 'L1', stage: 'immediate' } }, + ]); +}); + +test('a ledger record names the host, task, tier and stage of the reuse, and a negative verdict its outcome', async (t) => { + useTemporaryLedger(t); + const record = { assetId: ASSET.assetId, sessionId: 's1', platform: 'claude-code', taskId: 'claude-code:s1:4:retention', tier: 'L2', stage: 'retention' }; + assert.equal(await recordReuseLocally({ ...record, outcome: 'failed' }), true); + assert.equal(await recordReuseLocally({ ...record, outcome: 'success', platform: '', tier: undefined }), true); + const entries = new events.Ingestor({ path: events.rootEventsPath() }).readAllStrict(); + assert.deepEqual(entries.map(({ type, payload }) => ({ type, payload })), [ + { type: 'value.reuse_outcome', payload: { assetId: ASSET.assetId, cycleId: 's1', outcome: 'failed', platform: 'claude-code', taskId: 'claude-code:s1:4:retention', tier: 'L2', stage: 'retention' } }, + { type: 'value.reuse_hit', payload: { assetId: ASSET.assetId, cycleId: 's1', taskId: 'claude-code:s1:4:retention', stage: 'retention' } }, ]); }); @@ -352,3 +365,78 @@ test('a reuse that cannot be judged is exposure only, without the unused penalty assert.match(report.reason, /applied: unknown/); assert.equal(immediate({ applied: 'unknown', checks: [{ command: 'npm test', failed: false }] }).evidence.tier, 'none'); }); + +const withSource = () => verified(); +const refusedForSource = { ok: true, data: { recorded: false, terminal: true, reason: 'validation_error', status: 400 } }; + +/** A Hub that answers each post in turn with the next of `answers`, the last one from then on. */ +function answering(answers) { + const sinks = ledgers(); + let index = 0; + return { + ...sinks, + proxyFetch: async (method, path, body) => { + sinks.posts.push({ method, path, body: structuredClone(body) }); + const answer = answers[Math.min(index++, answers.length - 1)]; + if (answer instanceof Error) throw answer; + return answer; + }, + }; +} + +test('a report the Hub refuses as invalid for its evidence source goes once more without it, and that answer settles it', async () => { + const sinks = answering([refusedForSource, { ok: true, data: { recorded: true } }]); + const outbox = await deliverPending(sinks, queuedOne(withSource()), NOW); + const [entry] = Object.values(outbox); + assert.equal(sinks.posts.length, 2); + assert.equal(sinks.posts[0].body.evidence.source, 'observed'); + const { source, ...evidence } = withSource().evidence; + assert.deepEqual(sinks.posts[1].body, { ...withSource(), evidence }); + assert.equal(entry.sentAt, NOW); + assert.equal(entry.rejectedAt, undefined); + assert.equal(entry.report.evidence.source, undefined, 'the outbox keeps the report as last sent'); + assert.deepEqual(sinks.rejections, []); + assert.deepEqual(sinks.locals, [{ assetId: 'sha256:a', outcome: 'success', sessionId: 'claude-code:s1', taskId: 'claude-code:s1:4', tier: 'L1', stage: 'immediate' }]); +}); + +test('a report refused again without its source is settled as rejected and never sent a third time', async () => { + const sinks = answering([refusedForSource]); + const outbox = await deliverPending(sinks, queuedOne(withSource()), NOW); + const [entry] = Object.values(outbox); + assert.equal(entry.rejectedAt, NOW); + assert.deepEqual(entry.rejection, { reason: 'validation_error', status: 400 }); + assert.deepEqual(sinks.rejections, [{ taskId: 'claude-code:s1:4', assetId: 'sha256:a', verdict: 'rejected', reason: 'validation_error', status: 400 }]); + await deliverPending(sinks, outbox, NOW + 1); + assert.equal(sinks.posts.length, 2); + assert.deepEqual(sinks.locals, []); +}); + +test('a resend without source that cannot reach the Hub is retried without source, as one post', async () => { + const sinks = answering([refusedForSource, { ok: false, error: 'timed out' }, { ok: true, data: { recorded: true } }]); + const retried = await deliverPending(sinks, queuedOne(withSource()), NOW); + assert.equal(Object.values(retried)[0].sentAt, undefined); + const sent = await deliverPending(sinks, retried, NOW + 1); + assert.equal(sinks.posts.length, 3); + assert.equal(sinks.posts[2].body.evidence.source, undefined); + assert.equal(Object.values(sent)[0].sentAt, NOW + 1); +}); + +test('only a validation_error of a report that carries a source is resent', async () => { + const unsourced = correctionReportOf(reportedReuseOf(verified()), 'correction'); + const plain = answering([refusedForSource]); + await deliverPending(plain, queuedOne(unsourced), NOW); + assert.equal(plain.posts.length, 1); + + const other = answering([{ ok: true, data: { recorded: false, terminal: true, reason: 'invalid_evidence', status: 400 } }]); + await deliverPending(other, queuedOne(withSource()), NOW); + assert.equal(other.posts.length, 1); +}); + +test('the Hub delivery alone (no local record) resends a source refusal too', async () => { + const sinks = answering([refusedForSource, { ok: true, data: { recorded: true } }]); + const outbox = await postPending(sinks, queuedOne(withSource()), NOW); + assert.equal(sinks.posts.length, 2); + assert.equal(Object.values(outbox)[0].sentAt, NOW); + assert.equal(Object.values(outbox)[0].report.evidence.source, undefined); + assert.deepEqual(sinks.locals, []); +}); diff --git a/test/runtime/session-log.test.js b/test/runtime/session-log.test.js index ed983a2..f411fc1 100644 --- a/test/runtime/session-log.test.js +++ b/test/runtime/session-log.test.js @@ -4,6 +4,8 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { test } from 'node:test'; +import { taskIdsOf, turnStatusOf } from '../../hosts/kimi/transcript.js'; + import { recordReaderOf } from '../../lib/runtime/records.js'; function fixture(t, overrides = {}) { @@ -137,3 +139,34 @@ test('a task the log names many times counts once toward the task limit', (t) => assert.ok(f.reader(f.sessionId, '0')); assert.deepEqual(f.decoded.at(-1).context.tasks, [{ taskId: 'task', status: 'running' }]); }); + +test('a damaged task of an earlier turn does not hide a later completed turn, even one that reads its output', (t) => { + const f = fixture(t, { + taskIds: taskIdsOf, tasksDir: (dir) => join(dir, 'agents', 'main', 'tasks'), decode: turnStatusOf, + }); + const prompt = (turnId) => ({ type: 'turn.prompt', agentId: 'main', turnId, promptId: `prompt-${turnId}`, origin: { kind: 'user' } }); + const event = (value) => ({ type: 'context.append_loop_event', agentId: 'main', event: value }); + const records = [ + { type: 'metadata', protocol_version: '1.5' }, + prompt(0), + event({ type: 'tool.call', uuid: 'old-call', turnId: '0', toolCallId: 'old-check', name: 'Bash', args: { command: 'npm test', run_in_background: true } }), + event({ type: 'tool.result', parentUuid: 'old-call', toolCallId: 'old-check', result: { output: 'task_id: old-task\nstatus: running', durationMs: 1 } }), + { type: 'turn.ended', agentId: 'main', turnId: 0, reason: 'completed' }, + prompt(1), + event({ type: 'tool.call', uuid: 'output-call', turnId: '1', toolCallId: 'old-output', name: 'TaskOutput', args: { task_id: 'old-task' } }), + event({ type: 'tool.result', parentUuid: 'output-call', toolCallId: 'old-output', result: { output: 'status: completed', durationMs: 1 } }), + event({ type: 'tool.call', uuid: 'current-call', turnId: '1', toolCallId: 'current-check', name: 'Bash', args: { command: 'npm test' } }), + event({ type: 'tool.result', parentUuid: 'current-call', toolCallId: 'current-check', result: { output: '# pass 1\n# fail 0', durationMs: 1 } }), + { type: 'turn.ended', agentId: 'main', turnId: 1, reason: 'completed' }, + ]; + writeFileSync(f.wire, records.map(line).join('')); + mkdirSync(join(f.agent, 'tasks')); + const taskFile = join(f.agent, 'tasks', 'old-task.json'); + writeFileSync(taskFile, JSON.stringify({ taskId: 'old-task', kind: 'process', parentToolCallId: 'old-check', command: 'npm test', status: 'completed', exitCode: 0 })); + assert.equal(f.reader(f.sessionId, '1').status, 'completed'); + writeFileSync(taskFile, '{'); + assert.equal(f.reader(f.sessionId, '0'), null, 'the requested turn still requires its own task to be readable'); + const current = f.reader(f.sessionId, '1'); + assert.equal(current?.status, 'completed'); + assert.deepEqual(current.record.checks, [{ command: 'npm test', exit: null, failed: false }]); +});