diff --git a/vinci/extensions/lib/task-outcome.ts b/vinci/extensions/lib/task-outcome.ts index 071def20c..54e9b631f 100644 --- a/vinci/extensions/lib/task-outcome.ts +++ b/vinci/extensions/lib/task-outcome.ts @@ -79,6 +79,13 @@ export type VinciTaskUsage = { estimatedCostUsd: number; providers: string[]; models: string[]; + // Same three-way split as the accumulator: `models` stays the collapsed observed-or-resolved + // field other consumers read, while these carry the two facts separately. Kept in lockstep with + // VinciAccumulatedUsage -- receipt-integration asserts the two rollups are deep-equal, and a + // divergence here is a real inconsistency, not a stale expectation. + observedModels: string[]; + resolvedModels: string[]; + observedModelCalls: number; }; export type VinciTaskOutcome = { @@ -285,6 +292,9 @@ function userFacingFailure(error: string): string { function summarizeAssistantUsage(messages: readonly unknown[]): VinciTaskUsage { const providers = new Set(); const models = new Set(); + const observedModels = new Set(); + const resolvedModels = new Set(); + let observedModelCalls = 0; // Assistant-stream and supplemental costs both accumulate as integer micro-USD internally. let estimatedCostMicroUsd = 0; const usage: VinciTaskUsage = { @@ -297,6 +307,9 @@ function summarizeAssistantUsage(messages: readonly unknown[]): VinciTaskUsage { estimatedCostUsd: 0, providers: [], models: [], + observedModels: [], + resolvedModels: [], + observedModelCalls: 0, }; for (const message of assistantMessages(messages)) { usage.modelCalls++; @@ -307,16 +320,25 @@ function summarizeAssistantUsage(messages: readonly unknown[]): VinciTaskUsage { usage.reasoningTokens += finite(message.usage?.reasoning); estimatedCostMicroUsd += usdToMicroUsd(finite(message.usage?.cost?.total)); if (typeof message.provider === "string" && message.provider) providers.add(message.provider); - const model = typeof message.responseModel === "string" && message.responseModel - ? message.responseModel - : typeof message.model === "string" - ? message.model - : ""; + // Machine-observed served id: only ever what the provider reported on the wire. + const observed = + typeof message.responseModel === "string" && message.responseModel ? message.responseModel : ""; + // Resolver's choice: recorded unconditionally, and never used to stand in for `observed`. + const resolved = typeof message.model === "string" && message.model ? message.model : ""; + if (observed) { + observedModels.add(observed); + observedModelCalls += 1; + } + if (resolved) resolvedModels.add(resolved); + const model = observed || resolved; if (model) models.add(model); } usage.estimatedCostUsd = microUsdToUsd(estimatedCostMicroUsd); usage.providers = [...providers].sort(); usage.models = [...models].sort(); + usage.observedModels = [...observedModels].sort(); + usage.resolvedModels = [...resolvedModels].sort(); + usage.observedModelCalls = observedModelCalls; return usage; } diff --git a/vinci/extensions/lib/usage-accumulator.ts b/vinci/extensions/lib/usage-accumulator.ts index 0c0dd1457..1b5b3deed 100644 --- a/vinci/extensions/lib/usage-accumulator.ts +++ b/vinci/extensions/lib/usage-accumulator.ts @@ -13,6 +13,18 @@ export type VinciAccumulatedUsage = { estimatedCostUsd: number; providers: string[]; models: string[]; + // MACHINE-OBSERVED model identity only: the ids the provider itself reported on the wire + // (`responseModel`). NEVER populated from the requested/resolved id, so an empty array means + // "the provider did not tell us what served this call" and never "it served what we asked for". + // `models` above deliberately keeps its existing collapsed meaning; other consumers read it. + observedModels: string[]; + // The id the in-process RESOLVER settled on (`message.model`), recorded unconditionally. This is + // what `buildFallbackModel` relabels, so resolved==requested proves nothing about what ran -- it + // is the middle term, and it is only meaningful next to `observedModels`. + resolvedModels: string[]; + // Count of responses whose model identity was machine-observed. `modelCalls - observedModelCalls` + // is the number of calls whose served identity is unknown. + observedModelCalls: number; }; export type VinciUsageCall = { @@ -105,6 +117,9 @@ export function emptyVinciAccumulatedUsage(): VinciAccumulatedUsage { estimatedCostUsd: 0, providers: [], models: [], + observedModels: [], + resolvedModels: [], + observedModelCalls: 0, }; } @@ -119,6 +134,9 @@ function normalizedUsage(usage: Readonly): VinciAccumulat estimatedCostUsd: finite(usage.estimatedCostUsd), providers: [...new Set(usage.providers.filter(Boolean))].sort(), models: [...new Set(usage.models.filter(Boolean))].sort(), + observedModels: [...new Set((usage.observedModels ?? []).filter(Boolean))].sort(), + resolvedModels: [...new Set((usage.resolvedModels ?? []).filter(Boolean))].sort(), + observedModelCalls: finite(usage.observedModelCalls), }; } @@ -139,6 +157,16 @@ export function addVinciAccumulatedUsage( target.estimatedCostUsd = microUsdToUsd(targetMicroUsd + additionMicroUsd); target.providers = [...new Set([...target.providers, ...addition.providers.filter(Boolean)])].sort(); target.models = [...new Set([...target.models, ...addition.models.filter(Boolean)])].sort(); + target.observedModels = [ + ...new Set([...(target.observedModels ?? []), ...(addition.observedModels ?? []).filter(Boolean)]), + ].sort(); + target.resolvedModels = [ + ...new Set([...(target.resolvedModels ?? []), ...(addition.resolvedModels ?? []).filter(Boolean)]), + ].sort(); + // `finite(target...)` rather than `+=`: this adder is also called with objects assembled + // elsewhere (combinedTaskUsage in task-outcome.ts) which may predate these fields. `undefined += n` + // is NaN, and NaN then travels as a plausible-looking number instead of failing. + target.observedModelCalls = finite(target.observedModelCalls) + finite(addition.observedModelCalls); return target; } @@ -152,14 +180,27 @@ export function vinciResponseKey(response: ModelResponseLike): string | undefine return responseKey(response); } +/** + * The wire boundary, exported for tests. This is where a provider response is split into what the + * provider actually reported (`observedModels`) and what the resolver chose (`resolvedModels`); a + * regression here re-fuses the two and is invisible to every test that starts from the already-split + * persisted shape. + */ +export function vinciUsageFromResponse(response: ModelResponseLike): VinciAccumulatedUsage { + return usageFromResponse(response); +} + function usageFromResponse(response: ModelResponseLike): VinciAccumulatedUsage { const provider = typeof response.provider === "string" && response.provider ? [response.provider] : []; - const model = - typeof response.responseModel === "string" && response.responseModel - ? response.responseModel - : typeof response.model === "string" && response.model - ? response.model - : ""; + // The ONLY machine observation of served identity available on this path. Upstream sets + // `responseModel` exclusively when the wire-reported id DIFFERS from the requested one + // (`packages/ai/src/api/openai-completions.ts:324`, guarded by `chunk.model !== model.id`), so its + // absence conflates "the provider agreed" with "the provider said nothing". We therefore record + // only what was actually observed and let the absence stay an absence. + const observed = + typeof response.responseModel === "string" && response.responseModel ? response.responseModel : ""; + const resolved = typeof response.model === "string" && response.model ? response.model : ""; + const model = observed || resolved; return { modelCalls: 1, inputTokens: finite(response.usage?.input), @@ -170,6 +211,9 @@ function usageFromResponse(response: ModelResponseLike): VinciAccumulatedUsage { estimatedCostUsd: finite(response.usage?.cost?.total), providers: provider, models: model ? [model] : [], + observedModels: observed ? [observed] : [], + resolvedModels: resolved ? [resolved] : [], + observedModelCalls: observed ? 1 : 0, }; } diff --git a/vinci/test/worker-generation-identity-consumer.mjs b/vinci/test/worker-generation-identity-consumer.mjs new file mode 100644 index 000000000..38aad3a66 --- /dev/null +++ b/vinci/test/worker-generation-identity-consumer.mjs @@ -0,0 +1,882 @@ +// S03 generation identity, CONSUMER-TESTED end to end. +// +// The unit controls in worker-generation-identity.mjs stop at buildEconomicsSummary's return value. +// This drives the REAL worker daemon (`worker.mjs start --once`) against a fake bus and fixture +// binaries, and reads the identity chain back out of the artifact a consumer actually reads: +// `economics//economics-summary.json` on disk, whose sha256 is carried on the terminal post. +// +// The whole chain, every link asserted on the persisted artifact: +// +// envelope `model:` header ....... requested -> route.initial_provider / initial_model +// session usage entry ............ resolved -> generation_identity.resolved_model +// session usage entry ............ observed -> generation_identity.observed_model +// terminal bus post .............. used -> economics digest + the usage[] actually billed +// +// The discriminating property: the fixture reports a served model that DIFFERS from the requested +// one, so an implementation that echoes the request back cannot pass. +import assert from "node:assert/strict"; +import { createHash } from "node:crypto"; +import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { spawn, spawnSync, execFileSync } from "node:child_process"; +import { tmpdir } from "node:os"; +import { dirname, join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { WorkerTestFixture } from "./lib/worker-fixture.mjs"; + +const ROOT = join(dirname(fileURLToPath(import.meta.url)), "..", ".."); +const TOOLS = join(dirname(fileURLToPath(import.meta.url)), "fixtures", "worker-test-tools"); + +const REQUESTED_MODEL = "requested/model-A"; +const RESOLVED_MODEL = "resolved/model-B"; +const OBSERVED_MODEL = "observed/model-C"; +const FALLBACK_MODEL = "fallback/provider-default-D"; +// Deliberately NOT the requested provider ("openrouter"): if these matched, substituting one for +// the other would be undetectable and the provider assertions would false-green. +const RUNTIME_PROVIDER = "runtime-gateway-Z"; + +// A session the fixture `vinci` binary appends: one outcome plus one usage entry whose observed id +// is neither the requested id nor the resolved id. +function sessionFixture({ + observed, + resolved = RESOLVED_MODEL, + runtimeProvider = RUNTIME_PROVIDER, + // These two exist so each arm of worker.mjs's `generationOccurred` disjunction can be driven + // ALONE. With both always set together, deleting either arm changed nothing observable. + modelCalls = 1, + responseKey = "openrouter resp-1", +}) { + const usageBlock = (models, observedModels) => ({ + modelCalls, + inputTokens: 10, + outputTokens: 5, + cachedTokens: 0, + cacheWriteTokens: 0, + reasoningTokens: 0, + estimatedCostUsd: 0.01, + providers: runtimeProvider ? [runtimeProvider] : [], + models, + observedModels, + resolvedModels: [resolved], + observedModelCalls: observedModels.length > 0 ? 1 : 0, + }); + const collapsed = [observed ?? resolved]; + const observedModels = observed ? [observed] : []; + const outcome = { + type: "custom", + customType: "vinci-task-outcome", + data: { + schemaVersion: 1, + taskId: "SESSION_ID", + state: "DONE", + reason: "fixture outcome", + changedFiles: [], + verificationStatus: "passed", + verificationCommand: "fixture check", + usage: usageBlock(collapsed, observedModels), + recordedAt: "2026-09-10T00:00:00Z", + }, + }; + // `responseKey` may be a single key, null (no key at all), or an ARRAY -- an array produces one + // usage entry per key, which is how a multi-generation attempt is driven. + const keys = Array.isArray(responseKey) ? responseKey : [responseKey]; + const usageEntries = keys.map((key) => ({ + type: "custom", + customType: "vinci-task-usage", + data: { + ...(key ? { responseKey: key } : {}), + usage: usageBlock(collapsed, observedModels), + }, + })); + return [JSON.stringify(outcome), ...usageEntries.map((e) => JSON.stringify(e))].join("\n") + "\n"; +} + +// Read BOTH result.json and session.jsonl out of the one tarball the worker handed the uploader. +// They must agree: the generations the evidence claims were used are exactly the generations +// present in the session that produced it. Reading them from the same bundle is the point -- a +// claim checked against a different artifact than the one shipped proves nothing. +function uploadedBundle(fixture) { + const awsCalls = join(fixture.tempDir, "aws-calls.txt"); + if (!existsSync(awsCalls)) return { result: null, sessionKeys: null }; + const calls = readFileSync(awsCalls, "utf8") + .split("\n") + .filter((line) => line.startsWith("{")) + .map((line) => JSON.parse(line)); + if (calls.length === 0) return { result: null, sessionKeys: null }; + const out = mkdtempSync(join(tmpdir(), "gi-bundle-")); + try { + const tar = spawnSync("tar", ["xzf", calls[0].argv[3], "-C", out], { encoding: "utf8" }); + if (tar.status !== 0) return { result: null, sessionKeys: null }; + const result = JSON.parse(readFileSync(join(out, "result.json"), "utf8")); + // The immutable identities actually present in the consumed session. + const sessionPath = join(out, "session.jsonl"); + const sessionKeys = existsSync(sessionPath) + ? readFileSync(sessionPath, "utf8") + .split("\n") + .filter(Boolean) + .map((line) => { + try { return JSON.parse(line); } catch { return null; } + }) + .filter((e) => e && e.type === "custom" && e.customType === "vinci-task-usage") + .map((e) => e?.data?.responseKey) + .filter((k) => typeof k === "string" && k) + .sort() + : null; + return { result, sessionKeys }; + } finally { + rmSync(out, { recursive: true, force: true }); + } +} + +function uploadedResultJson(fixture) { + const awsCalls = join(fixture.tempDir, "aws-calls.txt"); + if (!existsSync(awsCalls)) return null; + const calls = readFileSync(awsCalls, "utf8") + .split("\n") + .filter((line) => line.startsWith("{")) + .map((line) => JSON.parse(line)); + if (calls.length === 0) return null; + const tarPath = calls[0].argv[3]; + const out = mkdtempSync(join(tmpdir(), "gi-bundle-")); + try { + const tar = spawnSync("tar", ["xzf", tarPath, "-C", out], { encoding: "utf8" }); + if (tar.status !== 0) return null; + return JSON.parse(readFileSync(join(out, "result.json"), "utf8")); + } finally { + rmSync(out, { recursive: true, force: true }); + } +} + +async function runWorker({ + observed, + resolved, + runtimeProvider, + modelCalls, + responseKey, + taskId, + name, + workerId, + evidence, + env = {}, +}) { + const fixture = new WorkerTestFixture(name); + try { + fixture.createRepo("test", "repo"); + fixture.linkTools(TOOLS); + const sessionPath = join(fixture.tempDir, `session-${taskId}.jsonl`); + writeFileSync(sessionPath, sessionFixture({ observed, resolved, runtimeProvider, modelCalls, responseKey })); + + await fixture.startBus([ + { + message_id: taskId, + kind: "handoff", + to_agent: `worker:${workerId}`, + subject: "generation identity", + // The REQUESTED pair enters the system here and nowhere else. + body: `repo: test/repo\nprovider: openrouter\nmodel: ${REQUESTED_MODEL}\nevidence: ${evidence ?? "none"}\nbudget_usd: 20\nref: job_gi${taskId}\n\nDo the task`, + ts: "2026-09-10T10:00:00Z", + posted_by: "scheduler", + }, + ]); + + const proc = spawn( + "node", + [ + join(ROOT, "vinci/worker/worker.mjs"), + "start", + "--id", + workerId, + "--server", + fixture.busUrl(), + "--once", + "--state-dir", + fixture.tempDir, + ], + { + env: fixture.getEnv({ + FAKE_VINCI_USAGE: "1", + FAKE_VINCI_SESSION_FIXTURE: sessionPath, + // Without a prefix uploadEvidence returns before building a bundle, so result.json -- + // the artifact the downstream consumer reads -- would never exist. + VINCI_EVIDENCE_URI_PREFIX: "s3://bucket/vinci-evidence/", + // The fake `aws` only records when told where to. + FAKE_AWS_RECORD: join(fixture.tempDir, "aws-calls.txt"), + ...env, + }), + stdio: ["ignore", "pipe", "pipe"], + }, + ); + let stderr = ""; + proc.stderr.on("data", (d) => { + stderr += d; + }); + const code = await new Promise((r) => proc.on("close", r)); + assert.equal(code, 0, `worker exited ${code}: ${stderr.slice(-1200)}`); + + // A completed attempt writes economics into the ATTEMPT dir; only early terminals (no + // repository) fall back to /economics/. Discover it rather than guessing. + const found = execFileSync("find", [fixture.tempDir, "-name", "economics-summary.json"], { encoding: "utf8" }) + .split("\n").filter(Boolean); + assert.equal(found.length > 0, true, + `no economics summary anywhere under ${fixture.tempDir}\n${stderr.slice(-1500)}`); + const file = found[0]; + const raw = readFileSync(file, "utf8"); + const taskFile = join(fixture.tempDir, "tasks", `${taskId}.json`); + const task = existsSync(taskFile) ? JSON.parse(readFileSync(taskFile, "utf8")) : null; + // The fake `aws` records `s3 cp --no-progress `. Read result.json back out of + // the tarball the worker ACTUALLY handed the uploader -- that is the artifact a downstream + // consumer of the evidence bundle receives, not a local copy we arranged for the test. + const { result, sessionKeys } = uploadedBundle(fixture); + return { summary: JSON.parse(raw), raw, task, result, sessionKeys, posts: fixture.getPostedMessages() }; + } finally { + fixture.cleanup?.(); + } +} + +const results = []; +const check = async (name, fn) => { + try { + await fn(); + results.push(`PASS ${name}`); + } catch (error) { + results.push(`FAIL ${name}: ${error.message}`); + process.exitCode = 1; + } +}; + +// --------------------------------------------------------------------------------------------- +// The whole chain, through the real worker, with all three identities different. +// --------------------------------------------------------------------------------------------- +await check("the whole chain survives to the on-disk economics artifact", async () => { + const { summary, raw, posts } = await runWorker({ + observed: OBSERVED_MODEL, + taskId: "91", + name: "gen-identity-observed", + workerId: "w6", + }); + + // requested -- from the envelope header, through the daemon, onto the artifact. + assert.equal(summary.route.initial_provider, "openrouter", "requested provider missing from the artifact"); + assert.equal(summary.route.initial_model, REQUESTED_MODEL, "requested model missing from the artifact"); + + const gi = summary.generation_identity; + assert.ok(gi, "generation_identity absent from the persisted artifact"); + assert.equal(gi.resolved_model, RESOLVED_MODEL, "resolved model missing from the artifact"); + assert.equal(gi.observed_model, OBSERVED_MODEL, "observed model missing from the artifact"); + assert.equal(gi.observation, "observed"); + assert.equal(gi.observation_source, "response-stream"); + assert.equal(gi.matches_requested, false, "a served model different from the request must not read as a match"); + + // All three pairwise distinct ON THE ARTIFACT, not merely in memory. + assert.equal( + new Set([summary.route.initial_model, gi.resolved_model, gi.observed_model]).size, + 3, + "the three identities collapsed somewhere on the way to disk", + ); + + // used -- what was actually billed, and the digest a consumer binds the evidence by. + assert.ok(Array.isArray(summary.usage) && summary.usage.length > 0, "no usage rows: nothing was actually consumed"); + assert.equal(gi.observed_model_calls, 1, "the observed call was not counted"); + assert.equal(gi.unobserved_model_calls, 0); + + const digest = createHash("sha256").update(raw).digest("hex"); + const terminal = posts.find((m) => typeof m.body === "string" && m.body.includes(digest)); + assert.ok( + terminal, + `no terminal bus post carries the artifact digest ${digest.slice(0, 12)} -- ` + + "the consumer cannot bind the identity evidence to this attempt", + ); +}); + +// --------------------------------------------------------------------------------------------- +// The refusal, end to end. The provider reported nothing; both wrong answers are present in the +// same artifact and neither may be substituted. +// --------------------------------------------------------------------------------------------- +await check("an unobserved run stays unknown on the artifact", async () => { + const { summary } = await runWorker({ observed: null, taskId: "92", name: "gen-identity-unknown", workerId: "w7" }); + + // Control preconditions: the two values a lazy implementation would copy ARE present here. + assert.equal(summary.route.initial_model, REQUESTED_MODEL, "precondition: requested present and copyable"); + assert.equal( + summary.generation_identity.resolved_model, + RESOLVED_MODEL, + "precondition: resolved present and copyable", + ); + + const gi = summary.generation_identity; + assert.equal(gi.observed_model, null, "unknown served identity was back-filled on the artifact"); + assert.notEqual(gi.observed_model, REQUESTED_MODEL); + assert.notEqual(gi.observed_model, RESOLVED_MODEL); + assert.equal(gi.observation, "unavailable"); + assert.equal(gi.matches_requested, null, "unknown collapsed into a matched/mismatched boolean on the artifact"); + assert.equal(gi.unobserved_model_calls >= 1, true, "an unobserved call was not counted as unobserved"); + + // And the run really happened -- this is a refusal to CLAIM, not an absence of work. + assert.ok( + Array.isArray(summary.usage) && summary.usage.length > 0, + "no usage rows: this would be a vacuous pass, since a run with no calls trivially observes nothing", + ); +}); + +// --------------------------------------------------------------------------------------------- +// DOWNSTREAM CONSUMER. Producing the field is not the deliverable -- something has to READ it and +// carry the distinction into what it emits. Two real downstream surfaces do: +// * the terminal bus post, which a scheduler/human reads without opening the bundle +// * result.json inside the evidence bundle +// Both must show `used_model` as a machine observation or as `unknown`, and must never print the +// requested or resolved string in that slot. +// --------------------------------------------------------------------------------------------- +await check("a downstream consumer preserves the distinction in emitted evidence", async () => { + const { result, posts } = await runWorker({ + observed: OBSERVED_MODEL, + taskId: "93", + name: "gen-identity-downstream", + workerId: "w4", + }); + + // Consumer 1: the evidence bundle's result.json. + assert.ok(result, "no result.json in the evidence bundle"); + const ri = result.generation_identity; + assert.ok(ri, "result.json does not carry the identity interface -- the consumer dropped the field"); + assert.equal(ri.requested_model, REQUESTED_MODEL, "requested lost crossing into result.json"); + assert.equal(ri.resolved_model, RESOLVED_MODEL, "resolved lost crossing into result.json"); + assert.equal(ri.observed_model, OBSERVED_MODEL, "observed lost crossing into result.json"); + assert.equal(new Set([ri.requested_model, ri.resolved_model, ri.observed_model]).size, 3, + "the consumer collapsed the three identities"); + + // STAGE 4, actually_used. The consumer must be able to say WHICH generation it used by an + // immutable event id, not by a copied model string. Assert the id exists, is not any of the + // three model strings, and is the id the observation itself came from. + assert.ok(ri.used_generation_id || ri.used_generation_ids?.length > 0, + "no generation id: the lineage references only a model string"); + for (const modelString of [REQUESTED_MODEL, RESOLVED_MODEL, OBSERVED_MODEL]) { + assert.notEqual(ri.used_generation_id, modelString, + `used_generation_id is a model string (${modelString}), not an event identity`); + } + assert.deepEqual(ri.observed_generation_ids, [ri.used_generation_id], + "the observed identity is not bound to the generation event it came from"); + // All FOUR stages distinct as observable values. + assert.equal( + new Set([ri.requested_model, ri.resolved_model, ri.observed_model, ri.used_generation_id]).size, + 4, + "the four stages are not four distinct observable values", + ); + + // Observation provenance survived downstream. + assert.equal(ri.observation_source, "response-stream", "observation provenance lost crossing into the bundle"); + // Provider is NEVER observed: no adapter reads a served provider off the wire. The field stays + // present and null so a consumer can see that, and what the runtime actually ran on is carried + // separately under its own name. + assert.equal(ri.observed_provider, null, "observed_provider claims provider provenance that does not exist"); + assert.equal(ri.runtime_provider, RUNTIME_PROVIDER, "runtime provider lost crossing into the bundle"); + assert.notEqual(ri.runtime_provider, ri.requested_provider, + "runtime provider was taken from the requested provider"); + assert.equal(ri.requested_provider, "openrouter", "precondition: requested provider is present and copyable"); + + // Consumer 2: the terminal bus post, read as fields. + const terminal = posts.find((m) => typeof m.body === "string" && m.body.includes("used_model=")); + assert.ok(terminal, "no terminal post carries used_model= -- the distinction never reached the bus"); + const fields = Object.fromEntries( + terminal.body.split(/\s+/).filter((t) => t.includes("=")).map((t) => { + const i = t.indexOf("="); + return [t.slice(0, i), t.slice(i + 1)]; + }), + ); + assert.equal(fields.used_model, OBSERVED_MODEL, "the post reports a used_model that was not observed"); + assert.equal(fields.observation, "observed"); + assert.equal(fields.requested_model, REQUESTED_MODEL); + assert.equal(fields.identity_matches_requested, "false"); + assert.equal(fields.observation_source, "response-stream", "observation provenance never reached the bus"); + assert.equal(fields.observed_provider, undefined, "an unobservable provider was posted as observed"); + assert.equal(fields.runtime_provider, RUNTIME_PROVIDER, "runtime provider never reached the bus"); + assert.notEqual(fields.runtime_provider, fields.requested_model, "provider/model confusion on the post"); + // A real generation id is `provider\0responseId` and is not field-safe, so the post carries a + // digest of it rather than truncating the body. The binding must still be checkable: the digest + // has to be the digest OF the id the bundle carries verbatim. + const safe = /^[A-Za-z0-9._:@+-]+$/.test(ri.used_generation_id); + if (safe) { + assert.equal(fields.used_generation_id, ri.used_generation_id, + "the post's lineage id disagrees with the bundle's"); + } else { + const expected = createHash("sha256").update(ri.used_generation_id).digest("hex").slice(0, 12); + assert.equal(fields.used_generation_id_sha256, expected, + "the post's lineage digest does not bind to the generation id in the bundle"); + assert.equal(fields.used_generation_id, undefined, + "an unsafe generation id was printed raw and truncated the field list"); + // And the truncation this guards against did not happen: fields after it survived. + assert.equal(fields.observation, "observed", "the field list was corrupted by the id"); + } +}); + +await check("a downstream consumer reports unknown, not the requested model", async () => { + const { result, posts } = await runWorker({ + observed: null, + taskId: "94", + name: "gen-identity-downstream-unknown", + workerId: "w5", + }); + + const ri = result?.generation_identity; + assert.ok(ri, "result.json does not carry the identity interface"); + // Control precondition: both wrong answers are present in the very object the consumer reads. + assert.equal(ri.requested_model, REQUESTED_MODEL, "precondition: requested present and copyable"); + assert.equal(ri.resolved_model, RESOLVED_MODEL, "precondition: resolved present and copyable"); + assert.equal(ri.observed_model, null, "the consumer back-filled an unobservable identity"); + + const terminal = posts.find((m) => typeof m.body === "string" && m.body.includes("used_model=")); + assert.ok(terminal, "no terminal post carries used_model="); + const fields = Object.fromEntries( + terminal.body.split(/\s+/).filter((t) => t.includes("=")).map((t) => { + const i = t.indexOf("="); + return [t.slice(0, i), t.slice(i + 1)]; + }), + ); + assert.equal(fields.used_model, "unknown", `the post printed used_model=${fields.used_model} for an unobserved run`); + assert.notEqual(fields.used_model, REQUESTED_MODEL, "the requested string was printed as the served model"); + assert.notEqual(fields.used_model, RESOLVED_MODEL, "the resolved string was printed as the served model"); + assert.equal(fields.observation, "unavailable"); + // Not measured, so nothing is claimed either way. + assert.equal(fields.identity_matches_requested, undefined, + "an unmeasured comparison was printed as a match verdict"); +}); + +// --------------------------------------------------------------------------------------------- +// FALLBACK. `buildFallbackModel` returns the provider default's ENTIRE configuration with only +// `id`/`name` overwritten by the requested string -- a different model wearing the right name. The +// interface must show the fallback for what it is and must never relabel it as the requested model. +// --------------------------------------------------------------------------------------------- +await check("a fallback is not relabelled as the requested model", async () => { + // The fallback case as it reaches this layer: the resolver produced a DIFFERENT model + // (FALLBACK_MODEL) while the caller asked for REQUESTED_MODEL, and the provider confirmed the + // fallback on the wire. + const { summary, result, posts } = await runWorker({ + observed: FALLBACK_MODEL, + resolved: FALLBACK_MODEL, + taskId: "95", + name: "gen-identity-fallback", + workerId: "w3", + }); + + const gi = summary.generation_identity; + assert.equal(gi.requested_model, REQUESTED_MODEL, "the request was lost"); + assert.equal(gi.observed_model, FALLBACK_MODEL, "the fallback was not reported as what served the call"); + assert.notEqual(gi.observed_model, REQUESTED_MODEL, "the fallback was relabelled as the requested model"); + assert.equal(gi.matches_requested, false, "a fallback must not read as a match for the request"); + + assert.equal(result.generation_identity.observed_model, FALLBACK_MODEL); + const terminal = posts.find((m) => typeof m.body === "string" && m.body.includes("used_model=")); + assert.ok(terminal); + assert.ok(terminal.body.includes(`used_model=${FALLBACK_MODEL}`), + "the terminal post did not name the fallback as the served model"); + assert.ok(terminal.body.includes("identity_matches_requested=false"), + "a fallback was posted as matching the request"); +}); + +// --------------------------------------------------------------------------------------------- +// The FAITHFUL fallback shape, and the one that matters most. +// +// `buildFallbackModel` returns `{...baseModel, id: modelId, name: modelId}` -- the provider +// default's entire configuration (baseUrl, api, cost, contextWindow, maxTokens, reasoning, compat) +// with only the NAME overwritten by what the caller asked for. So downstream, `message.model` is +// the REQUESTED string: resolved == requested, and every identity keyed on `model.id` reports the +// request back as though it were an observation. The wire report is the only thing that can +// discriminate, and it must not be allowed to agree by default. +// --------------------------------------------------------------------------------------------- +await check("a relabelled fallback is exposed even though resolved == requested", async () => { + const { summary, result, posts } = await runWorker({ + // The relabel: the resolver reports the REQUESTED id back, because that is what it wrote onto + // the object. Only the provider knows a different model actually served the call. + resolved: REQUESTED_MODEL, + observed: FALLBACK_MODEL, + taskId: "96", + name: "gen-identity-relabelled-fallback", + workerId: "w2", + evidence: "none", + }); + + const gi = summary.generation_identity; + // The trap: these two agreeing is exactly what the relabel manufactures, and it must NOT be + // read as confirmation that the request was honoured. + assert.equal(gi.requested_model, REQUESTED_MODEL); + assert.equal(gi.resolved_model, REQUESTED_MODEL, "precondition: the relabel makes resolved == requested"); + + assert.equal(gi.observed_model, FALLBACK_MODEL, "the served model was not exposed"); + assert.notEqual(gi.observed_model, REQUESTED_MODEL, "the fallback was relabelled as the requested model"); + assert.equal(gi.matches_requested, false, + "resolved == requested was taken as agreement while a different model actually served the call"); + + assert.equal(result.generation_identity.observed_model, FALLBACK_MODEL, "the bundle hid the fallback"); + const terminal = posts.find((m) => typeof m.body === "string" && m.body.includes("used_model=")); + assert.ok(terminal.body.includes(`used_model=${FALLBACK_MODEL}`), + "the terminal post named the requested model as the served one"); + assert.ok(terminal.body.includes("identity_matches_requested=false"), + "a relabelled fallback was posted as matching the request"); +}); + +// --------------------------------------------------------------------------------------------- +// POSITIVE CONTROL at the consumer, A -> A -> A. Without this the strictness above is satisfiable +// by a downstream that prints `unknown` unconditionally. The unit-level control proves the +// aggregation can represent a match; this proves the match survives all the way to emitted +// evidence. +// --------------------------------------------------------------------------------------------- +await check("a fully-agreeing run reports a match end to end, not unknown", async () => { + const { summary, result, posts } = await runWorker({ + resolved: REQUESTED_MODEL, + observed: REQUESTED_MODEL, + taskId: "97", + name: "gen-identity-agreement", + workerId: "w1", + }); + + const gi = summary.generation_identity; + assert.equal(gi.observation, "observed", "an observed agreement read as unavailable"); + assert.equal(gi.observed_model, REQUESTED_MODEL); + assert.equal(gi.matches_requested, true, "agreement was not reported as a match"); + assert.equal(gi.observed_model_calls, 1); + assert.equal(gi.unobserved_model_calls, 0); + // Still bound to the event, even when every string agrees -- this is the case where a model + // string is least able to identify anything. + assert.ok(gi.used_generation_id, "no event identity on an agreeing run"); + + assert.equal(result.generation_identity.matches_requested, true, "the bundle lost the match"); + const terminal = posts.find((m) => typeof m.body === "string" && m.body.includes("used_model=")); + assert.ok(terminal.body.includes(`used_model=${REQUESTED_MODEL}`)); + assert.ok(terminal.body.includes("identity_matches_requested=true"), + "an observed agreement was not posted as a match"); + assert.ok(terminal.body.includes("observation=observed")); +}); + +// --------------------------------------------------------------------------------------------- +// THE BINDING CONTROL. A copied model string is not evidence that a particular generation was +// consumed. The consumer must be able to PROVE which generation it used, and this must fail if the +// evidence claims a generation other than the one actually consumed. +// +// Both artifacts come out of the SAME shipped tarball: result.json's claim is checked against +// session.jsonl's immutable response keys -- the generations that actually produced the run. +// --------------------------------------------------------------------------------------------- +await check("evidence claims exactly the generations the run actually consumed", async () => { + const { result, sessionKeys } = await runWorker({ + observed: OBSERVED_MODEL, + taskId: "98", + name: "gen-identity-binding", + workerId: "w8", + }); + + assert.ok(Array.isArray(sessionKeys), "no session.jsonl in the bundle: the claim cannot be checked"); + assert.ok(sessionKeys.length > 0, + "the consumed session carries no generation identities -- every claim below would be vacuous"); + + const claimed = [...(result.generation_identity.used_generation_ids ?? [])].sort(); + assert.deepEqual(claimed, sessionKeys, + `evidence claims generations ${JSON.stringify(claimed)} but the session it shipped ` + + `actually contains ${JSON.stringify(sessionKeys)}`); + + // And the observed identity is attributed to a generation that is really in that set, not to a + // free-floating id. + for (const id of result.generation_identity.observed_generation_ids ?? []) { + assert.ok(sessionKeys.includes(id), + `observed identity attributed to generation ${id}, which the consumed session does not contain`); + } +}); + +// --------------------------------------------------------------------------------------------- +// NON-COMPLETED TERMINALS. Every case above drives the task to DONE/COMPLETED. A run that actually +// ran, actually spent, and actually observed a divergent served model, but then terminates FAILED +// (post-run git/network/evidence errors are routine) must not lose the identity on the way to the +// bus -- that is precisely the run most in need of an accurate trail, and it is the one the +// happy-path tests could never see. +// --------------------------------------------------------------------------------------------- +await check("a FAILED run that actually observed a divergence still reports it on the bus", async () => { + const { summary, posts } = await runWorker({ + observed: OBSERVED_MODEL, + taskId: "99", + name: "gen-identity-failed-terminal", + workerId: "w6", + env: { FAKE_VINCI_EXIT: "1" }, + }); + + // Precondition: the run really happened and the divergence really was computed. Without this the + // assertion below could pass vacuously on a task that never ran. + const gi = summary.generation_identity; + assert.equal(gi.observation, "observed", "precondition: the divergence must have been observed"); + assert.equal(gi.observed_model, OBSERVED_MODEL, "precondition: the served model was identified"); + assert.equal(gi.matches_requested, false, "precondition: this run diverged from the request"); + + const terminal = posts.find((m) => typeof m.body === "string" && m.body.includes("state=FAILED")); + assert.ok(terminal, "no FAILED terminal post"); + assert.ok( + terminal.body.includes("used_model="), + "the FAILED terminal post carries no generation identity at all -- the divergence was computed " + + "and then dropped on the way to the bus, which is the one consumer an operator reads without " + + "opening the bundle", + ); + assert.ok(terminal.body.includes(`used_model=${OBSERVED_MODEL}`), + "the FAILED post does not name the model that actually served the run"); + assert.ok(terminal.body.includes("identity_matches_requested=false"), + "a diverged FAILED run was not posted as diverging"); +}); + +// --------------------------------------------------------------------------------------------- +// UNVERIFIED is where a run that SPENT and DIVERGED most often lands: the session ran, tokens were +// paid for, a different model served the call, and then publication or evidence failed. It is a +// different terminal path from COMPLETED and a different one again from the early aborts, and it +// must not lose the identity. An independent review found this whole class; this is the +// fixture-reachable member of it. +// --------------------------------------------------------------------------------------------- +for (const [label, extraEnv, evidence] of [ + ["gh failure after the run", { FAKE_GH_EXIT: "1" }, "pr"], + ["no commit produced", { FAKE_VINCI_NO_COMMIT: "1" }, "pr"], +]) { + await check(`an UNVERIFIED run (${label}) still reports the served model`, async () => { + const { summary, posts } = await runWorker({ + observed: OBSERVED_MODEL, + taskId: label.startsWith("gh") ? "201" : "203", + name: `gen-identity-unverified-${label.startsWith("gh") ? "gh" : "nocommit"}`, + workerId: label.startsWith("gh") ? "w1" : "w3", + evidence, + env: extraEnv, + }); + + // Preconditions: the run really spent and really diverged, so the assertion cannot pass on a + // task that never reached a provider. + const gi = summary.generation_identity; + assert.equal(gi.observation, "observed", "precondition: the divergence must have been observed"); + assert.equal(gi.matches_requested, false, "precondition: this run diverged from the request"); + assert.ok(summary.usage?.length > 0, "precondition: the run must actually have spent"); + + const terminal = posts.find((m) => typeof m.body === "string" && /state=(UNVERIFIED|FAILED|BLOCKED)/.test(m.body)); + assert.ok(terminal, "no non-COMPLETED terminal post"); + assert.ok( + terminal.body.includes("used_model="), + `the ${(terminal.body.match(/state=(\w+)/) || [])[1]} terminal post carries no generation identity -- ` + + "the divergence was computed and then dropped before the bus", + ); + assert.ok(terminal.body.includes(`used_model=${OBSERVED_MODEL}`), + "the terminal post does not name the model that actually served this run"); + assert.ok(terminal.body.includes("identity_matches_requested=false"), + "a diverged run was not posted as diverging"); + }); +} + +// --------------------------------------------------------------------------------------------- +// PRE-RUN REFUSAL, via the BLOCKER post path. +// +// 🔴 READ THE NAME CAREFULLY: this case does NOT exercise the `generationOccurred` gate, and it +// cannot fail if that gate is removed. A deadline refusal terminates through `blockerPostBody` / +// `terminalPostBody`, which structurally never carry identity fields at all -- so these assertions +// hold whether or not the gate exists. An independent review flagged exactly this, and it is +// correct. +// +// It is kept, and relabelled, rather than deleted: it pins a real and different property -- that +// the blocker path stays free of identity fields as that path evolves -- which nothing else covers. +// The gate itself is proven by "a postFinal terminal with no generation asserts no identity" +// below, which drives a branch-lease refusal through `postFinal` and DOES fail when the gate is +// removed. Do not read this case as the discriminating one. +// --------------------------------------------------------------------------------------------- +await check("the blocker post path carries no identity fields (does NOT test the gate)", async () => { + const fixture = new WorkerTestFixture("gen-identity-prerun"); + try { + fixture.createRepo("test", "repo"); + fixture.linkTools(TOOLS); + await fixture.startBus([ + { + message_id: "205", + kind: "handoff", + to_agent: "worker:w7", + subject: "pre-run refusal", + // A deadline already in the past: refused before anything is spawned. + body: `repo: test/repo\nprovider: openrouter\nmodel: ${REQUESTED_MODEL}\ndeadline: 2020-01-01T00:00:00Z\n\nTask`, + ts: "2026-09-10T10:00:00Z", + posted_by: "scheduler", + }, + ]); + const proc = spawn( + "node", + [join(ROOT, "vinci/worker/worker.mjs"), "start", "--id", "w7", "--server", fixture.busUrl(), + "--once", "--state-dir", fixture.tempDir], + { env: fixture.getEnv(), stdio: ["ignore", "pipe", "pipe"] }, + ); + await new Promise((r) => proc.on("close", r)); + const posts = fixture.getPostedMessages(); + // Reachability control: the task must actually have been claimed and refused, otherwise the + // absence assertions below are vacuous -- a worker that never saw the task trivially asserts + // nothing about it. Pre-run refusals post via blockerPostBody, not postFinal. + const about = posts.filter((m) => m.subject !== undefined && !/online/.test(String(m.subject))); + assert.ok(about.length > 0, + `the task was never claimed or refused, so this proves nothing. posts: ${posts.map((m) => m.subject).join(" | ")}`); + + // Nothing ran, so nothing may be claimed about what served it -- on ANY post, not just one. + for (const m of about) { + const body = String(m.body ?? ""); + assert.ok(!body.includes("used_model="), + `a pre-run refusal asserted a served model: ${body.slice(0, 180)}`); + assert.ok(!body.includes("observation="), + "a pre-run refusal asserted an observation status for a generation that never happened"); + assert.ok(!body.includes("used_generation_id"), + "a pre-run refusal invented a generation identity"); + } + } finally { + await fixture.cleanup(); + } +}); + +// --------------------------------------------------------------------------------------------- +// PROVIDER PROVENANCE ABSENT. With no provider recorded anywhere in the session, neither the +// observed nor the runtime provider may be conjured from the requested one. +// --------------------------------------------------------------------------------------------- +await check("absent provider provenance stays absent, not requested", async () => { + const { summary, posts } = await runWorker({ + observed: OBSERVED_MODEL, + runtimeProvider: null, + taskId: "206", + name: "gen-identity-no-provider", + workerId: "w4", + }); + const gi = summary.generation_identity; + // Control precondition: the requested provider is present and copyable. + assert.equal(gi.requested_provider, "openrouter", "precondition: requested provider present"); + assert.equal(gi.observation, "observed", "precondition: a model WAS observed, so only provider is missing"); + + assert.equal(gi.observed_provider, null, "provider provenance was invented"); + assert.equal(gi.runtime_provider, null, "runtime provider was back-filled from the request"); + assert.deepEqual(gi.runtime_providers, [], "a provider appeared from nowhere"); + + const terminal = posts.find((m) => typeof m.body === "string" && m.body.includes("used_model=")); + assert.ok(terminal, "no terminal post"); + assert.ok(!terminal.body.includes("runtime_provider="), + "the post asserted a runtime provider that the session never recorded"); +}); + +// --------------------------------------------------------------------------------------------- +// A postFinal PATH WHERE NO GENERATION OCCURRED. Pre-run refusals never reach postFinal, so the +// "did a generation happen" gate needs a path that does: a branch-lease refusal terminates through +// postFinal before any session is spawned. It must not assert an observation status. +// --------------------------------------------------------------------------------------------- +await check("a postFinal terminal with no generation asserts no identity", async () => { + const fixture = new WorkerTestFixture("gen-identity-no-generation"); + try { + fixture.createRepo("test", "repo"); + fixture.linkTools(TOOLS); + await fixture.startBus([ + { + message_id: "207", + kind: "handoff", + to_agent: "worker:w2", + subject: "no generation", + body: `repo: test/repo\nprovider: openrouter\nmodel: ${REQUESTED_MODEL}\nevidence: none\nbudget_usd: 20\nref: job_207\n\nTask`, + ts: "2026-09-10T10:00:00Z", + posted_by: "scheduler", + }, + ]); + const proc = spawn( + "node", + [join(ROOT, "vinci/worker/worker.mjs"), "start", "--id", "w2", "--server", fixture.busUrl(), + "--once", "--state-dir", fixture.tempDir], + // Branch leases ON with no governor reachable: refused before anything is spawned. + { env: fixture.getEnv({ VINCI_BRANCH_LEASE: "1" }), stdio: ["ignore", "pipe", "pipe"] }, + ); + await new Promise((r) => proc.on("close", r)); + const posts = fixture.getPostedMessages().filter((m) => !/online/.test(String(m.subject))); + assert.ok(posts.length > 0, "the task was never claimed, so this proves nothing"); + for (const m of posts) { + const body = String(m.body ?? ""); + assert.ok(!body.includes("observation="), + `a terminal with no generation asserted an observation status: ${body.slice(0, 180)}`); + assert.ok(!body.includes("used_model="), + "a terminal with no generation asserted a served model"); + } + } finally { + await fixture.cleanup(); + } +}); + +// --------------------------------------------------------------------------------------------- +// EACH ARM OF THE `generationOccurred` GATE, ALONE (worker.mjs). +// +// The gate is `(model_calls > 0) || (used_generation_ids.length > 0)`. An independent review +// deleted each arm on its own and the whole suite stayed green, because every fixture produced the +// two together -- so neither arm was ever load-bearing in a test. These two cases drive exactly one +// arm each. Both describe real partial-usage records: a response key whose call count was lost, and +// a counted call that carried no key to name it. In both, spend happened, so identity must be +// reported rather than suppressed. +// --------------------------------------------------------------------------------------------- +await check("a generation id with zero counted calls still reports identity", async () => { + const { summary, posts } = await runWorker({ + observed: OBSERVED_MODEL, + modelCalls: 0, + taskId: "208", + name: "gen-identity-arm-ids-only", + workerId: "w6", + }); + const gi = summary.generation_identity; + // Preconditions isolate the arm: no counted calls, but a generation id IS present. + assert.equal(gi.model_calls, 0, "precondition: this case must have NO counted calls"); + assert.ok((gi.used_generation_ids?.length ?? 0) > 0, + "precondition: it must still carry a generation id, else the arm is not isolated"); + + const terminal = posts.find((m) => typeof m.body === "string" && /state=/.test(m.body)); + assert.ok(terminal, "no terminal post"); + assert.ok(terminal.body.includes("used_model="), + "identity was suppressed for a run that produced a generation id -- the used_generation_ids arm " + + "of the gate is not doing its job"); +}); + +await check("counted calls with no generation id still report identity", async () => { + const { summary, posts } = await runWorker({ + observed: OBSERVED_MODEL, + responseKey: null, + taskId: "209", + name: "gen-identity-arm-calls-only", + workerId: "w7", + }); + const gi = summary.generation_identity; + // The mirror precondition: calls counted, but nothing to name them by. + assert.ok((gi.model_calls ?? 0) > 0, "precondition: this case must have counted calls"); + assert.deepEqual(gi.used_generation_ids, [], + "precondition: it must carry NO generation id, else the arm is not isolated"); + + const terminal = posts.find((m) => typeof m.body === "string" && /state=/.test(m.body)); + assert.ok(terminal, "no terminal post"); + assert.ok(terminal.body.includes("used_model="), + "identity was suppressed for a run that actually spent -- the model_calls arm of the gate is " + + "not doing its job"); + // Spend with no id to bind it to is unknown lineage, and must be visible as such rather than omitted. + assert.ok(!terminal.body.includes("used_generation_id="), + "a generation id was asserted for a run that never recorded one"); +}); + +// --------------------------------------------------------------------------------------------- +// MULTI-GENERATION ON THE BUS POST. `generationIdFields`'s `ids.length > 1` branch had zero +// coverage: an independent review deleted it entirely and every test stayed green, because no +// fixture ever drove more than one generation. That branch exists so a multi-generation attempt is +// never summarised by one id standing for all of them -- which is the exact case the lineage was +// built for -- so it must not be the dark corner. +// --------------------------------------------------------------------------------------------- +await check("two generations reach the bus post as a count and a digest, not one id", async () => { + const { summary, posts } = await runWorker({ + observed: OBSERVED_MODEL, + responseKey: ["openrouter resp-1", "openrouter resp-2"], + taskId: "210", + name: "gen-identity-multi-generation", + workerId: "w8", + }); + + // Precondition: two generations actually reached the summary, else the post assertions are vacuous. + const gi = summary.generation_identity; + assert.equal(gi.used_generation_ids.length, 2, + `precondition: expected 2 generations, got ${JSON.stringify(gi.used_generation_ids)}`); + assert.equal(gi.used_generation_id, null, "the singular field must stay null for a 2-generation attempt"); + + const terminal = posts.find((m) => typeof m.body === "string" && m.body.includes("used_model=")); + assert.ok(terminal, "no terminal post carrying identity"); + assert.ok(terminal.body.includes("used_generation_count=2"), + `the post did not report the generation COUNT: ${terminal.body.slice(0, 220)}`); + assert.ok(/used_generation_ids_sha256=[0-9a-f]{12}\b/.test(terminal.body), + "the post did not carry a digest binding the set of generations"); + // 🔴 The substitution this branch exists to prevent: one id standing for several. + assert.ok(!terminal.body.includes("used_generation_id="), + "a single generation id was posted for an attempt that consumed two"); +}); + +console.log(results.join("\n")); +if (process.exitCode === 1) console.error("worker-generation-identity-consumer: FAILURES above"); +else console.log(`worker-generation-identity-consumer: ${results.length} checks passed`); diff --git a/vinci/test/worker-generation-identity.mjs b/vinci/test/worker-generation-identity.mjs new file mode 100644 index 000000000..d97451cf6 --- /dev/null +++ b/vinci/test/worker-generation-identity.mjs @@ -0,0 +1,444 @@ +// S03 generation identity: prove that "what we asked for", "what the resolver chose" and "what the +// provider says actually served the call" survive as THREE separate values all the way to the +// economics summary on disk, and that an unobservable identity is never back-filled from either of +// the other two. +// +// This drives the REAL producer->consumer path for everything downstream of the provider: a real +// session JSONL on disk -> the real `readSessionState` parser -> the real `buildEconomicsSummary`. +// The provider itself is the one thing fixtured, because observing a disagreement requires a +// provider that reports a model different from the one requested. +// +// Filename note: `vinci/test/worker-*.mjs` is glob-discovered by run.sh:283; this name is chosen not +// to collide with `worker-input-artifacts.mjs` (PR #72). +import assert from "node:assert/strict"; +import { mkdtempSync, mkdirSync, writeFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join, dirname, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; +import { createJiti } from "jiti/static"; + +const here = dirname(fileURLToPath(import.meta.url)); +const ROOT = join(here, "..", ".."); +// Same loader the other extension tests use, so these cases execute the REAL TypeScript module +// rather than a re-implementation of it. +const loader = createJiti(import.meta.url, { + alias: { "@earendil-works/pi-agent-core": resolve(here, "../../packages/agent/src/index.ts") }, + moduleCache: false, + tryNative: false, +}); +const taskOutcome = await loader.import(resolve(here, "../extensions/lib/task-outcome.ts"), { default: false }); +const { readSessionState } = await import(join(ROOT, "vinci/worker/session-read.mjs")); +const { buildEconomicsSummary } = await import(join(ROOT, "vinci/worker/economics.mjs")); + +const SESSION_ID = "sess-gen-identity"; + +// One persisted `vinci-task-usage` entry, in exactly the shape the accumulator writes: +// `models` is the COLLAPSED field that already existed (observed-or-resolved), while +// `observedModels` carries only what the provider reported and `resolvedModels` only what the +// resolver chose. +function usageEntry({ responseKey, provider, resolved, observed, modelCalls = 1 }) { + const models = observed ? [observed] : resolved ? [resolved] : []; + return { + type: "custom", + customType: "vinci-task-usage", + data: { + responseKey, + usage: { + modelCalls, + inputTokens: 10, + outputTokens: 5, + cachedTokens: 0, + cacheWriteTokens: 0, + reasoningTokens: 0, + estimatedCostUsd: 0.001, + providers: provider ? [provider] : [], + models, + observedModels: observed ? [observed] : [], + resolvedModels: resolved ? [resolved] : [], + observedModelCalls: observed ? 1 : 0, + }, + }, + }; +} + +function summaryFor(entries, { requestedProvider, requestedModel }) { + const dir = mkdtempSync(join(tmpdir(), "s03-gen-identity-")); + try { + const sessionDir = join(dir, "sessions", "task-1"); + mkdirSync(sessionDir, { recursive: true }); + const lines = [JSON.stringify({ type: "session", id: SESSION_ID }), ...entries.map((e) => JSON.stringify(e))]; + writeFileSync(join(sessionDir, `${SESSION_ID}.jsonl`), lines.join("\n") + "\n"); + + const session = readSessionState(sessionDir, SESSION_ID); + // Reachability: if the parser did not find our entries, every identity assertion below would + // pass vacuously on an empty rollup. Fail loudly here instead. + assert.equal( + session.usageEntries.length, + entries.length, + `parser did not reach the usage entries (got ${session.usageEntries.length} of ${entries.length}) -- ` + + "every assertion after this point would be vacuous", + ); + + return buildEconomicsSummary({ + task: { id: "task-1", envelope: { ref: "task-1" }, attempt: 1 }, + workOrderId: "task-1", + attemptLabel: "task-1/1", + sessionState: session, + sessionId: SESSION_ID, + usageEntries: session.usageEntries, + requestedProvider, + requestedModel, + started: "2026-09-10T00:00:00Z", + finished: "2026-09-10T00:01:00Z", + taskState: "DONE", + }); + } finally { + rmSync(dir, { recursive: true, force: true }); + } +} + +const results = []; +const check = (name, fn) => { + try { + fn(); + results.push(`PASS ${name}`); + } catch (error) { + results.push(`FAIL ${name}: ${error.message}`); + process.exitCode = 1; + } +}; + +// --------------------------------------------------------------------------------------------- +// CASE 1 -- all three disagree. requested=A, resolved=B, observed=C. All three must survive. +// --------------------------------------------------------------------------------------------- +check("three-way disagreement keeps all three identities distinct", () => { + const summary = summaryFor( + [usageEntry({ responseKey: "r1", provider: "openrouter", resolved: "model-B", observed: "model-C" })], + { requestedProvider: "openrouter", requestedModel: "model-A" }, + ); + const gi = summary.generation_identity; + assert.equal(summary.route.initial_model, "model-A", "requested id lost from route.initial_model"); + assert.equal(summary.route.initial_provider, "openrouter"); + assert.equal(gi.resolved_model, "model-B", "resolved id lost"); + assert.equal(gi.observed_model, "model-C", "observed id lost"); + assert.equal(gi.observation, "observed"); + assert.equal(gi.observation_source, "response-stream"); + assert.equal(gi.matches_requested, false, "observed != requested must report a mismatch"); + // The three are pairwise distinct in the artifact, not merely present. + assert.equal(new Set([summary.route.initial_model, gi.resolved_model, gi.observed_model]).size, 3); +}); + +// --------------------------------------------------------------------------------------------- +// CASE 2 -- the load-bearing one. observed is UNKNOWN, and a plausible wrong answer (model-B) is +// sitting right there in the same record. Nothing may convert UNKNOWN into it, or into requested. +// --------------------------------------------------------------------------------------------- +check("unobserved identity is never back-filled from resolved or requested", () => { + const summary = summaryFor( + [usageEntry({ responseKey: "r1", provider: "openrouter", resolved: "model-B", observed: null })], + { requestedProvider: "openrouter", requestedModel: "model-A" }, + ); + const gi = summary.generation_identity; + // Discriminator: the wrong answers were REACHABLE. resolved is present in the same object, and + // requested is present in route. A back-filling implementation passes the null check below only + // by not doing the substitution -- it cannot pass by having nothing to substitute. + assert.equal(gi.resolved_model, "model-B", "control precondition: resolved must be present and copyable"); + assert.equal(summary.route.initial_model, "model-A", "control precondition: requested must be present and copyable"); + + assert.equal(gi.observed_model, null, "UNKNOWN observed identity was back-filled"); + assert.notEqual(gi.observed_model, "model-B", "observed was taken from resolved"); + assert.notEqual(gi.observed_model, "model-A", "observed was taken from requested"); + assert.equal(gi.observation, "unavailable"); + assert.equal(gi.observation_source, null); + // null, not false: we did not measure a mismatch, we failed to measure at all. + assert.equal(gi.matches_requested, null, "unknown collapsed into a matched/mismatched boolean"); + assert.equal(gi.observed_model_calls, 0); + assert.equal(gi.unobserved_model_calls, 1, "a call with no observation must be counted as unobserved"); +}); + +// --------------------------------------------------------------------------------------------- +// CASE 3 -- positive control. Everything agrees and IS observed. Without this, the strictness above +// would be satisfiable by an implementation that reports "unavailable" unconditionally. +// --------------------------------------------------------------------------------------------- +check("agreement that was actually observed reports observed, not unknown", () => { + const summary = summaryFor( + [usageEntry({ responseKey: "r1", provider: "openrouter", resolved: "model-A", observed: "model-A" })], + { requestedProvider: "openrouter", requestedModel: "model-A" }, + ); + const gi = summary.generation_identity; + assert.equal(gi.observed_model, "model-A"); + assert.equal(gi.observation, "observed", "an observed agreement must not read as unavailable"); + assert.equal(gi.matches_requested, true); + assert.equal(gi.observed_model_calls, 1); + assert.equal(gi.unobserved_model_calls, 0); +}); + +// --------------------------------------------------------------------------------------------- +// CASE 4 -- "observed" and "unavailable" must be distinguishable for the SAME final model id. This +// is the pair the old boolean collapsed: both runs end up serving model-A as far as any consumer +// reading `usage[].model` can tell. +// --------------------------------------------------------------------------------------------- +check("observed-agreement and could-not-tell are distinguishable at the same model id", () => { + const observedRun = summaryFor( + [usageEntry({ responseKey: "r1", provider: "openrouter", resolved: "model-A", observed: "model-A" })], + { requestedProvider: "openrouter", requestedModel: "model-A" }, + ); + const unknownRun = summaryFor( + [usageEntry({ responseKey: "r1", provider: "openrouter", resolved: "model-A", observed: null })], + { requestedProvider: "openrouter", requestedModel: "model-A" }, + ); + // The collapsed field cannot tell them apart -- this is the defect, asserted so it stays visible. + assert.deepEqual( + observedRun.usage.map((u) => u.model), + unknownRun.usage.map((u) => u.model), + "precondition: the pre-existing collapsed usage[].model is identical across both runs", + ); + // The new field can. + assert.notEqual( + observedRun.generation_identity.observation, + unknownRun.generation_identity.observation, + "the two runs are indistinguishable -- the whole point of the field is lost", + ); + assert.equal(observedRun.generation_identity.matches_requested, true); + assert.equal(unknownRun.generation_identity.matches_requested, null); +}); + +// --------------------------------------------------------------------------------------------- +// CASE 5 -- two calls reporting DIFFERENT served models inside one attempt is a conflict, not a +// silent pick-the-first. +// --------------------------------------------------------------------------------------------- +check("conflicting observations within one attempt report conflict", () => { + const summary = summaryFor( + [ + usageEntry({ responseKey: "r1", provider: "openrouter", resolved: "model-A", observed: "model-C" }), + usageEntry({ responseKey: "r2", provider: "openrouter", resolved: "model-A", observed: "model-D" }), + ], + { requestedProvider: "openrouter", requestedModel: "model-A" }, + ); + const gi = summary.generation_identity; + assert.equal(gi.observation, "conflict"); + assert.equal(gi.observed_model, null, "a conflict must not resolve to one of the conflicting values"); + assert.deepEqual(gi.observed_models, ["model-C", "model-D"]); + assert.equal(gi.matches_requested, false); +}); + +// --------------------------------------------------------------------------------------------- +// CASE 6 -- the SECOND consumer. `task-outcome.ts` rolls the same messages up independently for the +// receipt, and had its own copy of the `responseModel ?? message.model` collapse. A mutation that +// back-filled observed from resolved HERE survived every case above, so this control exists because +// the mutation battery found the gap, not because the shape looked untested. +// --------------------------------------------------------------------------------------------- +const assistantMessage = ({ model, responseModel, responseId }) => ({ + role: "assistant", + provider: "openrouter", + model, + ...(responseModel ? { responseModel } : {}), + responseId, + stopReason: "stop", + timestamp: 1, + usage: { input: 10, output: 5, cacheRead: 0, cacheWrite: 0, reasoning: 0, cost: { total: 0.001 } }, +}); + +check("task-outcome rollup keeps observed and resolved separate", () => { + const drifted = taskOutcome.summarizeVinciTaskUsage([ + assistantMessage({ model: "model-B", responseModel: "model-C", responseId: "d1" }), + ]); + assert.deepEqual(drifted.observedModels, ["model-C"], "observed id lost in the receipt rollup"); + assert.deepEqual(drifted.resolvedModels, ["model-B"], "resolved id lost in the receipt rollup"); + assert.equal(drifted.observedModelCalls, 1); +}); + +check("task-outcome rollup does not back-fill an unobserved id", () => { + const silent = taskOutcome.summarizeVinciTaskUsage([ + assistantMessage({ model: "model-B", responseModel: null, responseId: "s1" }), + ]); + // Control precondition: the wrong answer is present and copyable in the same rollup. + assert.deepEqual(silent.resolvedModels, ["model-B"], "precondition: resolved must be present"); + assert.deepEqual(silent.models, ["model-B"], "precondition: the collapsed field still shows model-B"); + + assert.deepEqual(silent.observedModels, [], "unobserved id was back-filled in the receipt rollup"); + assert.equal(silent.observedModelCalls, 0, "an unobserved call was counted as observed"); +}); + +check("task-outcome and accumulator agree on a NaN-free count", () => { + // The adder is called with objects assembled elsewhere; `undefined += n` silently yields NaN. + const combined = taskOutcome.summarizeVinciTaskUsage( + [assistantMessage({ model: "model-B", responseModel: "model-C", responseId: "n1" })], + "task-nan-check", + ); + assert.equal(Number.isFinite(combined.observedModelCalls), true, "observedModelCalls is not finite (NaN)"); + assert.equal(Number.isFinite(combined.modelCalls), true); +}); + +// --------------------------------------------------------------------------------------------- +// CASE 7 -- the adder's legacy-object guard. `addVinciAccumulatedUsage` is called with objects +// assembled by other modules; one of them predated these fields, and `undefined += n` yields NaN, +// which then travels as a plausible-looking number rather than failing. +// +// This control exists because a mutation restoring the `+=` form survived every other case: the +// reachable NaN had already been closed by giving VinciTaskUsage the field, leaving the guard +// itself unfalsifiable. Rather than keep an untested guard, exercise the exact shape it defends. +// --------------------------------------------------------------------------------------------- +const usageAccumulator = await loader.import( + resolve(here, "../extensions/lib/usage-accumulator.ts"), + { default: false }, +); + +check("adder tolerates a legacy target with no observed fields", () => { + // A target shaped the way callers built it before these fields existed. + const legacyTarget = { + modelCalls: 1, + inputTokens: 0, + outputTokens: 0, + cachedTokens: 0, + cacheWriteTokens: 0, + reasoningTokens: 0, + estimatedCostUsd: 0, + providers: [], + models: [], + }; + const addition = { + ...usageAccumulator.emptyVinciAccumulatedUsage(), + modelCalls: 1, + observedModels: ["model-C"], + observedModelCalls: 1, + }; + const merged = usageAccumulator.addVinciAccumulatedUsage(legacyTarget, addition); + assert.equal( + Number.isFinite(merged.observedModelCalls), + true, + `observedModelCalls is ${merged.observedModelCalls} -- a legacy target produced a non-finite count`, + ); + assert.equal(merged.observedModelCalls, 1); + assert.deepEqual(merged.observedModels, ["model-C"]); +}); + +// --------------------------------------------------------------------------------------------- +// CASE 8 (F3) -- ONE persisted entry carrying TWO distinct observed ids. Crew/helper rollups go +// through the same recordVinciTaskUsage path and legitimately produce this. Taking [0] reported a +// confident `observed` for what is actually a disagreement, and credited both calls to one id. +// Found by an independent reviewer against unmutated code, not by a mutation. +// --------------------------------------------------------------------------------------------- +check("two observed ids inside ONE entry is a conflict, not a silent pick-first", () => { + const entry = usageEntry({ responseKey: "agg1", provider: "openrouter", resolved: "model-B", observed: "model-C" }); + // Exactly the aggregate shape: one entry, two sub-calls, two different served ids. + entry.data.usage.observedModels = ["model-C", "model-D"]; + entry.data.usage.observedModelCalls = 2; + entry.data.usage.modelCalls = 2; + const summary = summaryFor([entry], { requestedProvider: "openrouter", requestedModel: "model-A" }); + const gi = summary.generation_identity; + assert.equal(gi.observation, "conflict", `two served ids in one entry reported as ${gi.observation}`); + assert.equal(gi.observed_model, null, "a conflict resolved to one of the conflicting values"); + assert.deepEqual(gi.observed_models, ["model-C", "model-D"], "the second observed id was dropped"); +}); + +// --------------------------------------------------------------------------------------------- +// CASE 9 (F2) -- the WIRE BOUNDARY itself. Every case above hand-builds the already-split +// observedModels/resolvedModels fields, so none of them reaches `usageFromResponse`, which is the +// function that actually performs the split. A mutation reverting it to `responseModel || model` +// survived the whole suite. Drive the real function with a real response object instead. +// --------------------------------------------------------------------------------------------- +check("the wire boundary splits observed from resolved (usageFromResponse)", () => { + const drift = usageAccumulator.vinciUsageFromResponse({ + provider: "openrouter", + model: "model-B", + responseModel: "model-C", + responseId: "w1", + usage: { input: 1, output: 1, cacheRead: 0, cacheWrite: 0, reasoning: 0, cost: { total: 0.001 } }, + }); + assert.deepEqual(drift.observedModels, ["model-C"], "wire boundary lost the observed id"); + assert.deepEqual(drift.resolvedModels, ["model-B"], "wire boundary lost the resolved id"); + assert.equal(drift.observedModelCalls, 1); + + const silent = usageAccumulator.vinciUsageFromResponse({ + provider: "openrouter", + model: "model-B", + // no responseModel: the provider reported nothing + responseId: "w2", + usage: { input: 1, output: 1, cacheRead: 0, cacheWrite: 0, reasoning: 0, cost: { total: 0.001 } }, + }); + // Control precondition: the wrong answer is present and copyable at this exact boundary. + assert.deepEqual(silent.resolvedModels, ["model-B"], "precondition: resolved present at the boundary"); + assert.deepEqual(silent.models, ["model-B"], "precondition: the collapsed field still shows model-B"); + + assert.deepEqual(silent.observedModels, [], "wire boundary back-filled observed from resolved"); + assert.equal(silent.observedModelCalls, 0, "an unobserved call was counted as observed at the boundary"); +}); + +// --------------------------------------------------------------------------------------------- +// CASE 10 (F3) -- MORE THAN ONE generation in one attempt. An independent review found this branch +// had zero coverage: the whole `ids.length > 1` arm could be deleted and every test stayed green, +// because every fixture drove exactly one responseKey. That arm exists so a multi-generation +// attempt is never summarised by a single id standing for all of them -- the case the lineage was +// built for -- so it is the last place that should be dark. +// --------------------------------------------------------------------------------------------- +check("two generations in one attempt are both carried, never collapsed to one", () => { + const summary = summaryFor( + [ + usageEntry({ responseKey: "openrouterresp-1", provider: "openrouter", resolved: "model-B", observed: "model-C" }), + usageEntry({ responseKey: "openrouterresp-2", provider: "openrouter", resolved: "model-B", observed: "model-C" }), + ], + { requestedProvider: "openrouter", requestedModel: "model-A" }, + ); + const gi = summary.generation_identity; + assert.equal(gi.used_generation_ids.length, 2, "a second generation was dropped"); + assert.deepEqual(gi.used_generation_ids, ["openrouterresp-1", "openrouterresp-2"]); + // 🔴 The load-bearing assertion: with more than one generation, the singular field must be null. + // A non-null value here is one id silently standing for both. + assert.equal(gi.used_generation_id, null, + `used_generation_id is ${JSON.stringify(gi.used_generation_id)} for a 2-generation attempt -- ` + + "one id is standing in for both"); + assert.equal(gi.observed_generation_ids.length, 2, "an observation lost its generation binding"); + // Both generations agreed on the served model, so this is still a clean observation, not a conflict. + assert.equal(gi.observation, "observed"); + assert.equal(gi.observed_model, "model-C"); +}); + +// --------------------------------------------------------------------------------------------- +// CASE 11 (F2) -- Economics-level fixtures that isolate the two partial-usage shapes independently. +// +// The gate is `(model_calls > 0) || (used_generation_ids.length > 0)`. A review deleted each arm +// independently and the suite stayed green both times, because every fixture produced the two +// together. These two cases separate them at the economics level, so a silent break in either arm +// is visible downstream. +// +// Note: this test file proves behavior at the fixture level (the two partial-usage shapes are +// distinguishable) and at the economics level (buildEconomicsSummary handles them correctly). The +// `generationOccurred` branch itself lives in worker.mjs and is not directly exercised here — the +// fixture exercises the implications of it at the economics layer and downstream consumers make +// the resulting behavior observable. This case is not a direct unit test of the gate. +// --------------------------------------------------------------------------------------------- +check("a generation id with no counted calls still counts as a generation", () => { + // A partial/malformed usage record: the response key survived, the call count did not. + const summary = summaryFor( + [usageEntry({ responseKey: "openrouterresp-9", provider: "openrouter", resolved: "model-B", observed: null, modelCalls: 0 })], + { requestedProvider: "openrouter", requestedModel: "model-A" }, + ); + const gi = summary.generation_identity; + assert.equal(gi.model_calls, 0, "precondition: this fixture must have NO counted calls"); + assert.deepEqual(gi.used_generation_ids, ["openrouterresp-9"], + "precondition: but it DOES carry a generation id -- otherwise this arm is not isolated"); + // A generation happened; the count is simply missing. Identity must still be reportable. + assert.equal(gi.used_generation_id, "openrouterresp-9"); +}); + +check("counted calls with no generation id still count as a generation", () => { + // The mirror: the call was counted but carried no response key to name it. + const summary = summaryFor( + [usageEntry({ responseKey: undefined, provider: "openrouter", resolved: "model-B", observed: null })], + { requestedProvider: "openrouter", requestedModel: "model-A" }, + ); + const gi = summary.generation_identity; + assert.equal(gi.model_calls, 1, "precondition: this fixture must have a counted call"); + assert.deepEqual(gi.used_generation_ids, [], + "precondition: and NO generation id -- otherwise this arm is not isolated"); + // Spend happened with no id to bind it to: that is unknown identity, not absence of a generation. + assert.equal(gi.used_generation_id, null); + assert.equal(gi.observation, "unavailable"); +}); + +console.log(results.join("\n")); +if (process.exitCode === 1) { + console.error("worker-generation-identity: FAILURES above"); +} else { + console.log(`worker-generation-identity: ${results.length} checks passed`); +} diff --git a/vinci/worker/economics.mjs b/vinci/worker/economics.mjs index fb98d210d..0d8b2b640 100644 --- a/vinci/worker/economics.mjs +++ b/vinci/worker/economics.mjs @@ -50,6 +50,126 @@ function str(value) { return typeof value === "string" && value.length <= 512 ? value : null; } +// Machine-observed generation identity for this attempt. +// +// The contract, and the reason this is not a boolean: a run whose served model is UNKNOWN must stay +// distinguishable from one where the served model was observed to equal what we asked for. Three +// distinct states, never two: +// +// observation: "unavailable" no call reported a served id -> `observed_model` is null and +// `matches_requested` is null. NOT evidence of agreement. +// observation: "observed" >=1 call reported a served id and all reports agree. +// observation: "conflict" calls reported DIFFERENT served ids within one attempt. +// +// `observed_model` is only ever a value the provider itself put on the wire. It is never derived +// from the requested or resolved id -- that substitution is the defect this exists to make visible +// (`buildFallbackModel` returns another model's entire configuration relabelled with the requested +// id, so every identity keyed on `model.id` reports the request back as though it were an +// observation). +function observeGeneration(entries, requestedProvider, requestedModel) { + const observedModels = new Set(); + const resolvedModels = new Set(); + // The provider the RUNTIME actually used. Every adapter sets a response's `provider` from + // `model.provider` (packages/ai/src/api/*.ts) -- it is configuration carried alongside the call, + // and it is never read back off the wire. It is therefore routing/runtime state, NOT an + // observation, and it is named accordingly. + const runtimeProviders = new Set(); + const seenResponseIds = new Set(); + // Lineage: the actual generation events this attempt consumed. A model STRING cannot identify + // what ran -- two different generations can carry the same string, and a relabelled fallback + // carries a string that was never served. The response id names the event itself. + const usedGenerationIds = new Set(); + const observedGenerationIds = new Set(); + let totalCalls = 0; + let observedCalls = 0; + + for (const entry of entries) { + if (!entry || typeof entry !== "object") continue; + // Same dedup rule as the cost rollup: one provider response counted once. + if (typeof entry.responseId === "string" && entry.responseId) { + if (seenResponseIds.has(entry.responseId)) continue; + seenResponseIds.add(entry.responseId); + } + if (typeof entry.model_calls === "number" && entry.model_calls > 0) totalCalls += entry.model_calls; + const runtimeProvider = str(entry.provider); + if (runtimeProvider) runtimeProviders.add(runtimeProvider); + const generationId = str(entry.responseId); + if (generationId) usedGenerationIds.add(generationId); + // Prefer the full arrays; a single entry may carry more than one of either. `observed_model` + // / `resolved_model` remain as the singular convenience fields and are only consulted when the + // arrays are absent (legacy entries written before this shape existed). + const resolvedList = Array.isArray(entry.resolved_models) ? entry.resolved_models : []; + if (resolvedList.length > 0) for (const m of resolvedList) { if (str(m)) resolvedModels.add(m); } + else { const resolved = str(entry.resolved_model); if (resolved) resolvedModels.add(resolved); } + + const observedList = Array.isArray(entry.observed_models) ? entry.observed_models : []; + const observedHere = []; + if (observedList.length > 0) { for (const m of observedList) { if (str(m)) observedHere.push(m); } } + else { const observed = str(entry.observed_model); if (observed) observedHere.push(observed); } + if (observedHere.length > 0) { + for (const m of observedHere) observedModels.add(m); + if (generationId) observedGenerationIds.add(generationId); + const n = typeof entry.observed_model_calls === "number" ? entry.observed_model_calls : 0; + observedCalls += n > 0 ? n : observedHere.length; + } + } + + const sorted = [...observedModels].sort(); + const resolvedSorted = [...resolvedModels].sort(); + let observation = "unavailable"; + if (sorted.length === 1) observation = "observed"; + else if (sorted.length > 1) observation = "conflict"; + + const observedModel = sorted.length === 1 ? sorted[0] : null; + // null, not false: with no observation there is nothing to compare, and reporting `false` here + // would assert a mismatch we never measured. + let matchesRequested = null; + if (observation === "observed" && requestedModel !== null) matchesRequested = observedModel === requestedModel; + else if (observation === "conflict") matchesRequested = false; + + const runtimeSorted = [...runtimeProviders].sort(); + const usedIds = [...usedGenerationIds].sort(); + const observedIds = [...observedGenerationIds].sort(); + return { + observation, + // Carried here as well as in `route.initial_*` so a consumer reads ONE object and gets the + // whole chain; a consumer that has to join two places to tell requested from served is a + // consumer that will eventually print the wrong one. + requested_provider: requestedProvider, + requested_model: requestedModel, + // 🔴 ALWAYS null on every current path, and that is the correct answer rather than an omission. + // No adapter reads a served provider off the wire, so there is no provider evidence at + // observation strength anywhere in this system. This field previously echoed + // `requestedProvider` whenever any MODEL was observed, which asserted provider provenance the + // data never had -- exactly the substitution the rest of this interface exists to prevent. + // It stays present, and null, so a consumer can see that provider is never observed. + observed_provider: null, + // Routing/runtime state, taken from the entries themselves. One value when the whole attempt + // ran on one provider, null when it did not -- never collapsed to a first element. + runtime_provider: runtimeSorted.length === 1 ? runtimeSorted[0] : null, + runtime_providers: runtimeSorted, + // What was actually consumed. `used_generation_id` is filled only when the attempt consumed + // exactly one generation; otherwise the caller must read the list rather than be handed a + // single id that silently stands for several. + used_generation_id: usedIds.length === 1 ? usedIds[0] : null, + used_generation_ids: usedIds, + // The subset whose served identity was machine-observed. `used_model` is the served model of + // record: it is `observed_model` or nothing, never the requested or resolved string. + observed_generation_ids: observedIds, + // The middle term. Present whenever any call ran, and deliberately NOT compared against + // `observed_model` to produce a verdict here -- a consumer that wants drift reads all three. + resolved_model: resolvedSorted.length === 1 ? resolvedSorted[0] : null, + resolved_models: resolvedSorted, + observed_model: observedModel, + observed_models: sorted, + observation_source: observation === "unavailable" ? null : "response-stream", + model_calls: totalCalls, + observed_model_calls: observedCalls, + unobserved_model_calls: Math.max(0, totalCalls - observedCalls), + matches_requested: matchesRequested, + }; +} + function rollupUsage(entries, flags) { const rollup = new Map(); // One provider response is one response regardless of which (provider, model) row it lands in. @@ -254,7 +374,18 @@ export function buildEconomicsSummary(input = {}) { summary.finished_at = finishedAt; if (work !== null) summary.work = work; if (usage.length > 0) summary.usage = usage; - summary.route = { policy_id: "none", initial_provider: null, initial_model: null, escalations: [] }; + // Milestone 2: the REQUESTED pair. `usage[].provider/model` stays the (collapsed) observed-or- + // requested pair it already was; these two slots were in the schema and null on every path, so + // requested-vs-observed was not computable from this artifact at all. + const requestedProvider = str(input.requestedProvider); + const requestedModel = str(input.requestedModel); + summary.route = { + policy_id: "none", + initial_provider: requestedProvider, + initial_model: requestedModel, + escalations: [], + }; + summary.generation_identity = observeGeneration(usageArray, requestedProvider, requestedModel); summary.assets_consumed = []; summary.compactions = 0; summary.human_interventions = []; @@ -285,6 +416,28 @@ export function buildEconomicsSummary(input = {}) { started_at: null, finished_at: null, route: { policy_id: "none", initial_provider: null, initial_model: null, escalations: [] }, + // The builder threw. Nothing here was observed, and the field is emitted rather than omitted + // so a consumer never has to infer meaning from its absence. + generation_identity: { + observation: "unavailable", + observed_model: null, + observed_models: [], + resolved_model: null, + resolved_models: [], + requested_provider: null, + requested_model: null, + observed_provider: null, + runtime_provider: null, + runtime_providers: [], + used_generation_id: null, + used_generation_ids: [], + observed_generation_ids: [], + observation_source: null, + model_calls: 0, + observed_model_calls: 0, + unobserved_model_calls: 0, + matches_requested: null, + }, assets_consumed: [], compactions: 0, human_interventions: [], diff --git a/vinci/worker/session-read.mjs b/vinci/worker/session-read.mjs index d8f615166..c5b9f54eb 100644 --- a/vinci/worker/session-read.mjs +++ b/vinci/worker/session-read.mjs @@ -129,11 +129,32 @@ function usageEntryToRecord(entry) { const modelCalls = numberOrZero(usage.modelCalls); const providers = Array.isArray(usage.providers) ? usage.providers.filter((p) => typeof p === "string" && p) : []; const models = Array.isArray(usage.models) ? usage.models.filter((m) => typeof m === "string" && m) : []; + // Machine-observed served identity, carried separately from `models` because `models` falls back + // to the REQUESTED id when the provider reported nothing. Absent/legacy entries yield [], which + // reads as "unknown", never as agreement. + const observedModels = Array.isArray(usage.observedModels) + ? usage.observedModels.filter((m) => typeof m === "string" && m) + : []; + const observedModelCalls = numberOrZero(usage.observedModelCalls); + const resolvedModels = Array.isArray(usage.resolvedModels) + ? usage.resolvedModels.filter((m) => typeof m === "string" && m) + : []; const costUsd = numberOrZero(usage.estimatedCostUsd); const responseId = typeof entry?.data?.responseKey === "string" && entry.data.responseKey ? entry.data.responseKey : null; return { provider: providers[0] ?? null, model: models[0] ?? null, + // null means the provider did not report a served model for this call. It must never be + // back-filled from `model` above -- that is the substitution this field exists to prevent. + // The FULL sets, not [0]. One persisted entry can aggregate sub-calls served by different + // models (crew/helper rollups go through the same recordVinciTaskUsage path), and collapsing + // to the first silently credits every call in the entry to one identity and reports a + // confident "observed" for what is actually a disagreement. + observed_model: observedModels.length === 1 ? observedModels[0] : null, + observed_models: observedModels, + resolved_model: resolvedModels.length === 1 ? resolvedModels[0] : null, + resolved_models: resolvedModels, + observed_model_calls: observedModelCalls, model_calls: modelCalls, input_tokens: numberOrZero(usage.inputTokens), cached_read_tokens: numberOrZero(usage.cachedTokens), diff --git a/vinci/worker/worker.mjs b/vinci/worker/worker.mjs index 813c1b283..ee0bcd839 100644 --- a/vinci/worker/worker.mjs +++ b/vinci/worker/worker.mjs @@ -9,6 +9,7 @@ import { unlinkSync, writeFileSync, } from "node:fs"; +import { createHash } from "node:crypto"; import { join, resolve } from "node:path"; import { replayPending } from "./outbox.mjs"; import { seedProviderDefinitions } from "./provider-definitions.mjs"; @@ -483,6 +484,11 @@ async function emitEconomics({ task: { id: taskId, envelope: { ref: envelopeToUse.ref }, attempt: attempt.attempt || attempt }, workOrderId: contractFields?.work_order_id ?? envelopeToUse?.ref ?? null, attemptLabel: `${taskId}/${attempt.attempt || attempt}`, + // The REQUESTED pair, straight off the envelope the daemon spawned `vinci -p` with. Requested + // only -- what actually served the call is observed separately and must never be back-filled + // from these two. + requestedProvider: envelopeToUse?.provider ?? null, + requestedModel: envelopeToUse?.model ?? null, lease: lease || null, sessionState: session, sessionId, @@ -534,6 +540,29 @@ async function emitEconomics({ work_order_id: contractFields?.work_order_id ?? envelopeToUse?.ref ?? null, attempt_label: `${taskId}/${attempt?.attempt ?? attempt ?? 0}`, route: { policy_id: "none", initial_provider: null, initial_model: null, escalations: [] }, + // Emitted, not omitted. This is the outer crash path -- if the builder or the canonicaliser + // throws, the artifact still carries the field, so a consumer never has to read meaning into + // its absence. Nothing was observed here, and that is exactly what it says. + generation_identity: { + observation: "unavailable", + resolved_model: null, + resolved_models: [], + requested_provider: null, + requested_model: null, + used_generation_id: null, + used_generation_ids: [], + observed_generation_ids: [], + observed_provider: null, + runtime_provider: null, + runtime_providers: [], + observed_model: null, + observed_models: [], + observation_source: null, + model_calls: 0, + observed_model_calls: 0, + unobserved_model_calls: 0, + matches_requested: null, + }, assets_consumed: [], compactions: 0, human_interventions: [], @@ -556,6 +585,26 @@ async function emitEconomics({ } } +// A generation id is only printable as a whitespace-delimited field when it contains no +// whitespace, NUL or `=`. Anything else is emitted as a digest so the body stays parseable while +// the reference remains checkable against result.json, which carries the id verbatim. +const FIELD_SAFE_ID = /^[A-Za-z0-9._:@+-]+$/; +function generationIdFields(gi) { + const ids = Array.isArray(gi.used_generation_ids) ? gi.used_generation_ids : []; + if (ids.length === 0) return []; + const single = ids.length === 1 ? ids[0] : null; + if (single) { + return FIELD_SAFE_ID.test(single) + ? [`used_generation_id=${single}`] + : [`used_generation_id_sha256=${createHash("sha256").update(single).digest("hex").slice(0, 12)}`]; + } + // More than one generation: never print a single id that would stand for all of them. + return [ + `used_generation_count=${ids.length}`, + `used_generation_ids_sha256=${createHash("sha256").update(ids.join("\u0000")).digest("hex").slice(0, 12)}`, + ]; +} + function terminalPostBody(details) { return `${details} worker_build=${formatWorkerBuild(workerBuild)} vinci_binary=${formatVinciBinary(vinciBinary)}`; } @@ -571,7 +620,13 @@ function blockerPostBody(record, details, fallback = null) { return terminalPostBody(tag ? `${tag} ${details}` : details); } -async function postFinal(bus, message, envelope, state, evidence, economicsSha = null) { +// `economics` is the emitEconomics result -- `{ summary, sha256 }` -- passed WHOLE and on purpose. +// The digest and the identity are two projections of ONE summary, so the post cannot carry a SHA +// from one summary and an identity from another. Identity is evidence about the execution, not +// task state: it must not travel via the lifecycle, whose terminal state is correctly immutable +// (`record()` throws once terminal, so every post-terminal path could never have used it anyway). +async function postFinal(bus, message, envelope, state, evidence, economics = null) { + const economicsSha = typeof economics?.sha256 === "string" ? economics.sha256 : null; const subject = `task ${message.message_id} ${state.state.toLowerCase()}`; // uri/sha256 are advertised only when the bundle actually reached S3 (`uploaded === true`, // set by uploadEvidence solely after a successful `aws s3 cp`); a failed upload also carries @@ -608,6 +663,44 @@ async function postFinal(bus, message, envelope, state, evidence, economicsSha = policy.blocked > 0 ? `policy_blocked_sites=${policy.sites.blocked.join(",")}` : undefined, ] : []; + // The identity distinction, carried as FIELDS on the terminal post so a downstream reader gets + // it without opening the bundle. Never collapsed: `observation=unavailable` and an observed + // match are different facts, and `used_model` is emitted ONLY from a machine observation, so a + // reader can never mistake the requested string for what served the call. Same rule as W2's + // three profile outcomes directly above -- do not fold these into one field. + const gi = economics?.summary?.generation_identity ?? null; + // Emit identity ONLY where a generation could actually have occurred. A pre-run refusal (bad + // bounds, past deadline, provider not allowed, branch-lease refusal) has no calls, and printing + // `observation=unavailable` there would assert an absence of evidence about a generation that + // never happened. A run that DID spend always prints, even when the served identity is unknown -- + // that is the case most in need of the trail, not the one to hide. + const generationOccurred = (gi?.model_calls ?? 0) > 0 || (gi?.used_generation_ids?.length ?? 0) > 0; + const identityDetails = gi && generationOccurred + ? [ + `observation=${gi.observation}`, + // Provenance of the observation itself: WHICH channel established identity. Without it a + // reader cannot tell a gateway-attested identity from one scraped off a response stream, + // and `unavailable` cannot be distinguished from "nobody looked". + gi.observation_source ? `observation_source=${gi.observation_source}` : undefined, + // `observed_provider` is never emitted: no adapter reads a served provider off the wire, + // so there is nothing to report at observation strength. What the runtime actually ran on + // is reported under its own name instead. + gi.runtime_provider ? `runtime_provider=${gi.runtime_provider}` : undefined, + gi.requested_model ? `requested_model=${gi.requested_model}` : undefined, + gi.resolved_model ? `resolved_model=${gi.resolved_model}` : undefined, + // `unknown`, never the requested or resolved string: an unobservable identity must not be + // reported as a served one. + `used_model=${gi.observed_model ?? "unknown"}`, + // The body is whitespace-delimited fields. A generation id is provider-shaped + // (`provider\0responseId`) and can contain whitespace or NUL, which would truncate the + // field and silently corrupt every token after it. Print it only when it is field-safe, + // otherwise print a digest -- lineage stays referenceable against the bundle, which + // carries the id verbatim, and the body stays parseable. + ...generationIdFields(gi), + // null means "not measured"; only an actual comparison prints true/false. + gi.matches_requested === null ? undefined : `identity_matches_requested=${gi.matches_requested}`, + ] + : []; const details = [ `state=${state.state}`, `exit_code=${state.exit_code}`, @@ -616,6 +709,7 @@ async function postFinal(bus, message, envelope, state, evidence, economicsSha = state.head ? `head=${state.head}` : undefined, state.pr ? `pr=${state.pr}` : undefined, ...policyDetails, + ...identityDetails, contractTag(state), ...economicsDetails, ...evidenceDetails, @@ -1256,7 +1350,7 @@ async function processHandoff( lifecycle.transition("BLOCKED", { outcome: { reason }, publish: "skipped", pr: null, fenced_out: reason }); await releaseLease("BLOCKED"); const econBranch = await emitEconomics({ taskId, attempt: lifecycle.snapshot().attempt ?? 0, stateDir, envelopeToUse, lease, lifecycle, contractFields, sessionId: attempt?.sessionId ?? null }); - await postFinal(bus, message, envelopeToUse, lifecycle.snapshot(), null, econBranch.sha256); + await postFinal(bus, message, envelopeToUse, lifecycle.snapshot(), null, econBranch); return true; } branchLease = acquired.lease; @@ -1276,7 +1370,7 @@ async function processHandoff( lifecycle.transition("BLOCKED", { outcome: { reason: authorityLost }, publish: "skipped", pr: null, fenced_out: authorityLost, lease: { ...lifecycle.snapshot().lease, ...lease } }); await releaseLease("BLOCKED"); const econLost = await emitEconomics({ taskId, attempt: lifecycle.snapshot().attempt ?? 0, stateDir, envelopeToUse, lease, lifecycle, contractFields, sessionId: attempt?.sessionId ?? null }); - await postFinal(bus, message, envelopeToUse, lifecycle.snapshot(), null, econLost.sha256); + await postFinal(bus, message, envelopeToUse, lifecycle.snapshot(), null, econLost); return true; } // #18: probe the binary IMMEDIATELY before the spawn — after the Governor lease and the clone, @@ -1420,6 +1514,8 @@ async function processHandoff( // A governed handoff has no envelope.ref; its id is the contract's work_order_id. workOrderId: contractFields?.work_order_id ?? envelopeToUse.ref ?? null, attemptLabel: `${taskId}/${attempt.attempt}`, + requestedProvider: envelopeToUse?.provider ?? null, + requestedModel: envelopeToUse?.model ?? null, lease: lease || null, sessionState: session, usageEntries: session.usageEntries || [], @@ -1453,6 +1549,9 @@ async function processHandoff( const economicsSha = economicsSha256(economicsCanonical); extraFiles["economics-summary.json"] = economicsCanonical; resultJson.economics_sha256 = economicsSha; + // Downstream consumers of the evidence bundle read result.json, not the economics summary. + // Carry the identity interface across that boundary rather than making them join two files. + resultJson.generation_identity = economicsSummary.generation_identity ?? null; // Local copy beside the attempt: a box without VINCI_EVIDENCE_URI_PREFIX uploads nothing, and // the runs that actually spent must not be the only ones that leave no file behind. try { @@ -1511,7 +1610,7 @@ async function processHandoff( // L4: release with the committed state's outcome, BEFORE the final post so the lease is not // held across a bus retry. A release failure is logged; the state above is already final. await releaseLease(state); - await postFinal(bus, message, envelopeToUse, lifecycle.snapshot(), evidenceResult, economicsSha); + await postFinal(bus, message, envelopeToUse, lifecycle.snapshot(), evidenceResult, { summary: economicsSummary, sha256: economicsSha }); } catch (error) { // A terminal state is immutable: if the failure happened after it was committed (e.g. the // final bus post), surface the error to the daemon loop instead of rewriting the record. @@ -1535,7 +1634,7 @@ async function processHandoff( await releaseLease("FAILED"); // A session may already have run and spent here (exception after runVinci): read it. const econFailed = await emitEconomics({ taskId, attempt: lifecycle.snapshot().attempt ?? 0, stateDir, envelopeToUse: envelope, lease: lease ?? null, lifecycle, contractFields, sessionId: lifecycle.snapshot().session_id ?? null }); - await postFinal(bus, message, envelope, lifecycle.snapshot(), null, econFailed.sha256); + await postFinal(bus, message, envelope, lifecycle.snapshot(), null, econFailed); } return true; }