Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 23 additions & 5 deletions docs/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -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/<host>/` holds data and pure functions, no IO; `bin/` and every
`hosts/<host>/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:<session>:<turn>`), 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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand Down
21 changes: 9 additions & 12 deletions hosts/kimi/transcript.js
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down Expand Up @@ -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;
Expand Down
4 changes: 2 additions & 2 deletions lib/app/deliver-reports.js
Original file line number Diff line number Diff line change
Expand Up @@ -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 () => {},
});

Expand Down
15 changes: 13 additions & 2 deletions lib/app/distill-run.js
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down
25 changes: 23 additions & 2 deletions lib/distill/request.js
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand All @@ -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,
Expand All @@ -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 }),
Expand Down
13 changes: 11 additions & 2 deletions lib/local-ledger.js
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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;
Expand All @@ -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.
Expand Down
2 changes: 1 addition & 1 deletion lib/report-outbox.js
Original file line number Diff line number Diff line change
Expand Up @@ -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) } };
}
20 changes: 17 additions & 3 deletions lib/reuse.js
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
6 changes: 3 additions & 3 deletions lib/runtime/session-log.js
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand All @@ -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 {
Expand Down
36 changes: 36 additions & 0 deletions test/app/deliver-reports.test.js
Original file line number Diff line number Diff line change
@@ -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);
});
Loading
Loading