From 0464ced7e9c1ae8e15d2549164aafbc4c60b03cd Mon Sep 17 00:00:00 2001 From: forrestlinfeng Date: Tue, 10 Nov 2026 22:56:46 +0800 Subject: [PATCH 1/3] feat(reuse): link distilled Genes to reused assets, name the host in the ledger, resend source refusals Distill requests carry reused_asset_ids: the sha256 assets the session's injection ledger booked into the turns the range covers (same session epoch), in injection order, at most 20, left out when there are none or the session file cannot be read. Proxy 2.0.50 and 2.0.38 distill the same Gene with or without the field, so no version gate is needed. Local ledger events (value.reuse_hit / value.reuse_outcome) also carry platform, taskId, tier and stage. evolver-core's root event schema keeps the payload open (z.unknown since it was introduced), and 2.0.50's Ingestor and the protected-audit envelope check accept the enriched payload. A report the Hub refuses with validation_error while carrying evidence.source is sent once more without source; that answer settles it and the outbox keeps the report as last sent. Old rejected entries are not re-driven. Tests stop writing into the developer's home: the MiniMax lifecycle fixture points the process home at its fixture (the resident ledger thread reads process.env), with a test that the reuse lands there. The dsh helper stops every loaded plugin before its fixture Proxy and home go away, and the proxyPort test keeps a current Proxy version in settings: a session start raised as a test ended used to find no Proxy, install the real engine from npm and start its Proxy on the default port under the process's own home. Co-Authored-By: Claude Opus 5.5 --- docs/architecture.md | 28 +++++++-- lib/app/deliver-reports.js | 4 +- lib/app/distill-run.js | 15 ++++- lib/distill/request.js | 25 +++++++- lib/local-ledger.js | 13 +++- lib/report-outbox.js | 2 +- lib/reuse.js | 20 +++++- test/app/deliver-reports.test.js | 36 +++++++++++ test/app/distill-session.test.js | 27 +++++++- test/distill-prompt.test.js | 40 +++++++++++- test/helpers/process-home.js | 22 +++++++ test/hosts/dsh/host.js | 13 +++- test/hosts/dsh/plugin.test.js | 6 +- test/hosts/minimax/lifecycle.test.js | 21 ++++++- test/reuse.test.js | 94 +++++++++++++++++++++++++++- 15 files changed, 336 insertions(+), 30 deletions(-) create mode 100644 test/app/deliver-reports.test.js create mode 100644 test/helpers/process-home.js diff --git a/docs/architecture.md b/docs/architecture.md index 3168416d..e13fc8f9 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/lib/app/deliver-reports.js b/lib/app/deliver-reports.js index 8601aa3f..6ad0eead 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 d796b6fe..226ad3f6 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, rangeOf, rangeRouteProblem } 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; @@ -106,8 +107,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 0cc746b7..e3ffa7a0 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 3da020e3..bd97ae50 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 fe978d87..e20c3be2 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 a821bc34..8840f3d1 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/test/app/deliver-reports.test.js b/test/app/deliver-reports.test.js new file mode 100644 index 00000000..209b5a35 --- /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 eaf45dbc..dd3781fd 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 10bd2a1e..f046bc1e 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 00000000..2e4e0b29 --- /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 ede4e1fe..0db03593 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 a7a5e391..c0e061e4 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/minimax/lifecycle.test.js b/test/hosts/minimax/lifecycle.test.js index 2912fe61..88eb1b94 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 b87b3936..7829b211 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, []); +}); From 41fcb115d411f6e38de80d5b4a3f76dcc5275c99 Mon Sep 17 00:00:00 2001 From: "xyanua." Date: Wed, 11 Nov 2026 08:16:16 +0700 Subject: [PATCH 2/3] fix: read Kimi background tasks only for the requested turn --- hosts/kimi/transcript.js | 14 +++++++------- lib/runtime/session-log.js | 4 ++-- test/hosts/kimi/readers.test.js | 21 +++++++++++++++------ test/runtime/session-log.test.js | 31 +++++++++++++++++++++++++++++++ 4 files changed, 55 insertions(+), 15 deletions(-) diff --git a/hosts/kimi/transcript.js b/hosts/kimi/transcript.js index 360e694b..09749932 100644 --- a/hosts/kimi/transcript.js +++ b/hosts/kimi/transcript.js @@ -21,17 +21,17 @@ 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. +// Only the requested turn's tasks can affect its evidence. const MAX_TASKS = 64; /** Only discovers fixed task IDs; the reader chooses a bounded file under the known session root. */ -export function taskIdsOf(records = []) { +export function taskIdsOf(records = [], { turnId } = {}) { 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])); + for (const call of callsOf(records, nativeTurnIdOf(turnId))) { + if (call.name === 'TaskOutput') ids.push(taskId(call.args?.task_id)); + if (call.name === 'Bash') { + const result = resultOf(records, call); + ids.push(taskId(/^task_id:\s*([A-Za-z0-9_-]+)(?:\r?\n|$)/.exec(text(result?.output))?.[1])); } } const newestLast = [...new Set(ids.filter(Boolean).reverse())].reverse(); diff --git a/lib/runtime/session-log.js b/lib/runtime/session-log.js index a3acd48c..21137ad0 100644 --- a/lib/runtime/session-log.js +++ b/lib/runtime/session-log.js @@ -132,7 +132,7 @@ 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 + * 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. */ @@ -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/hosts/kimi/readers.test.js b/test/hosts/kimi/readers.test.js index b2411e8d..61071f34 100644 --- a/test/hosts/kimi/readers.test.js +++ b/test/hosts/kimi/readers.test.js @@ -91,13 +91,22 @@ 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 the requested turn and rejects paths and unrelated output', () => { + const call = (id, task, turnId = '0') => event({ + type: 'tool.call', uuid: `call-${id}`, toolCallId: `tool-${id}`, turnId, name: 'TaskOutput', args: { task_id: task }, + }); + const records = [ + call(0, 'bash-task'), + call(1, '../outside'), + call(2, 'foreign-task', '1'), + 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) => call(index, `task-${index}`)); + const ids = taskIdsOf([...many, call(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/runtime/session-log.test.js b/test/runtime/session-log.test.js index ed983a2c..83826b6a 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,32 @@ 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', (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: '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 }]); +}); From 201aed204c9182a53e77723718d8dae06186ae25 Mon Sep 17 00:00:00 2001 From: forrestlinfeng Date: Wed, 11 Nov 2026 10:12:11 +0800 Subject: [PATCH 3/3] Discover only the background tasks the requested Kimi turn started A TaskOutput call can name another turn's task, which no note of this turn reads; a damaged file of that task still hid the turn reading its output. Co-Authored-By: Claude Opus 5.5 --- hosts/kimi/transcript.js | 19 ++++++++----------- lib/runtime/session-log.js | 4 ++-- test/hosts/kimi/readers.test.js | 20 +++++++++++--------- test/runtime/session-log.test.js | 4 +++- 4 files changed, 24 insertions(+), 23 deletions(-) diff --git a/hosts/kimi/transcript.js b/hosts/kimi/transcript.js index 09749932..cf32183f 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); -// Only the requested turn's tasks can affect its evidence. 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. */ +/** + * 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 = []; - for (const call of callsOf(records, nativeTurnIdOf(turnId))) { - if (call.name === 'TaskOutput') ids.push(taskId(call.args?.task_id)); - if (call.name === 'Bash') { - const result = resultOf(records, call); - ids.push(taskId(/^task_id:\s*([A-Za-z0-9_-]+)(?:\r?\n|$)/.exec(text(result?.output))?.[1])); - } - } + 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/runtime/session-log.js b/lib/runtime/session-log.js index 21137ad0..e79ad75f 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, { 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 + * 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) { diff --git a/test/hosts/kimi/readers.test.js b/test/hosts/kimi/readers.test.js index 61071f34..d042aabe 100644 --- a/test/hosts/kimi/readers.test.js +++ b/test/hosts/kimi/readers.test.js @@ -91,20 +91,22 @@ test('unsupported schemas and foreign agents do not supply evidence', () => { assert.equal(status([meta, { ...prompt(0), origin: { kind: 'cron' } }]), null); }); -test('fixed task discovery reads only the requested turn and rejects paths and unrelated output', () => { - const call = (id, task, turnId = '0') => event({ - type: 'tool.call', uuid: `call-${id}`, toolCallId: `tool-${id}`, turnId, name: 'TaskOutput', args: { task_id: task }, - }); +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 = [ - call(0, 'bash-task'), - call(1, '../outside'), - call(2, 'foreign-task', '1'), + ...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) => call(index, `task-${index}`)); - const ids = taskIdsOf([...many, call(70, 'task-0')], { turnId: '0' }); + 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 named again by a later call is kept'); }); diff --git a/test/runtime/session-log.test.js b/test/runtime/session-log.test.js index 83826b6a..f411fc14 100644 --- a/test/runtime/session-log.test.js +++ b/test/runtime/session-log.test.js @@ -140,7 +140,7 @@ test('a task the log names many times counts once toward the task limit', (t) => 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', (t) => { +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, }); @@ -153,6 +153,8 @@ test('a damaged task of an earlier turn does not hide a later completed turn', ( 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' },