fix(evidence): link composed workflows to consumed observations (#7028)

This commit is contained in:
Wenyu Chiou
2026-10-24 14:11:32 -04:00
committed by GitHub
parent ed79bf9361
commit 1ca7d38764
9 changed files with 558 additions and 3 deletions
+2
View File
@@ -51,6 +51,8 @@ export type AnalysisOperation =
/** Atomic successful provider observation with exact producing provenance. */
export interface AnalysisExecution extends AnalysisObservation {
readonly provider: ProviderIdentity;
/** Session-issued identity of the committed record consumed by this execution. */
readonly evidenceId?: string;
/** Result-specific veto of snapshot replay; Evidence remains retainable. */
readonly cachePolicy?: "live";
/**
+3
View File
@@ -589,6 +589,9 @@ const executeAnalysisTool = async (input: {
result: result.value.result,
upstreamProfile: evidenceProfile,
upstreamLimitations: result.value.limitations,
...(result.value.evidenceLinks === undefined
? {}
: { upstreamEvidenceLinks: result.value.evidenceLinks }),
});
return ok({
evidence,
+12
View File
@@ -1,3 +1,4 @@
import { compareUnicodeCodePoints } from "../domain/unicodeCodePointOrder.js";
import { jsonStringByteLength } from "../domain/jsonStringByteLength.js";
import { REA_WORKFLOW_PROVIDER } from "../domain/investigationProviders.js";
import { workflowAnalysisProfile } from "./WorkflowAnalysisProfile.js";
@@ -74,11 +75,13 @@ export const executeComposedWorkflow = async (
readonly result: JsonValue;
readonly limitations: readonly string[];
readonly cachePolicy?: "live";
readonly evidenceLinks?: readonly string[];
},
AnalysisError
>
> => {
const limitations = new Map<string, SourceLimitation>();
const evidenceLinks = new Set<string>();
const perItemGroups = new Set<string>();
let perItemBytes = 0;
let omittedObservations = 0;
@@ -87,6 +90,8 @@ export const executeComposedWorkflow = async (
resourceScope: analysis.resourceScope,
execute: async (operation, parameters, options) => {
const execution = await analysis.execute(operation, parameters, options);
if (execution.ok && execution.value.evidenceId !== undefined)
evidenceLinks.add(execution.value.evidenceId);
if (execution.ok && execution.value.cachePolicy === "live")
cachePolicy = "live";
if (execution.ok)
@@ -140,6 +145,9 @@ export const executeComposedWorkflow = async (
]),
],
...(cachePolicy === undefined ? {} : { cachePolicy }),
...(evidenceLinks.size === 0
? {}
: { evidenceLinks: [...evidenceLinks].sort(compareUnicodeCodePoints) }),
});
};
@@ -151,6 +159,7 @@ export const createWorkflowEvidence = (input: {
readonly result: JsonValue;
readonly upstreamProfile: AnalysisProfileCommitment | undefined;
readonly upstreamLimitations: readonly string[];
readonly upstreamEvidenceLinks?: readonly string[];
}): Evidence =>
createEvidence(input.target, REA_WORKFLOW_PROVIDER, {
operation: input.operation,
@@ -165,6 +174,9 @@ export const createWorkflowEvidence = (input: {
),
}),
confidence: "derived",
...(input.upstreamEvidenceLinks === undefined
? {}
: { evidenceLinks: input.upstreamEvidenceLinks }),
limitations: [
"Derived by an REA composed workflow.",
...input.upstreamLimitations,
@@ -96,6 +96,7 @@ export const requiresNativeSourceAdmissionBeforeReplay = (
/** Bounded in-memory cache for one immutable binary identity. */
export class AnalysisSnapshotCache {
readonly #entries = new Map<string, AnalysisSnapshotEntry>();
readonly #evidenceIds = new Map<string, string>();
readonly #workflowEntries = new Map<string, AnalysisSnapshotWorkflowEntry>();
#target: AnalysisSnapshotTarget | undefined;
#binding: AnalysisSnapshotBinding | undefined;
@@ -138,10 +139,40 @@ export class AnalysisSnapshotCache {
this.#target = structuredClone(snapshot.target);
this.#binding = structuredClone(snapshot.binding);
let imported = 0;
const stagedQueryIds = new Set<string>();
for (const entry of snapshot.entries) {
if (!this.#entries.has(entry.query_id) && !this.#hasCapacity()) continue;
if (!this.#entries.has(entry.query_id)) imported += 1;
this.#entries.set(entry.query_id, entry);
stagedQueryIds.add(entry.query_id);
this.#evidenceIds.delete(entry.query_id);
}
// Snapshot validation authenticated these records. Recover the original
// identity before lookup adds replay caveats; do not mint a new record.
const evidenceIds = new Map<string, string>();
const ambiguous = new Set<string>();
for (const evidence of snapshot.evidence_bundle.records) {
const queryId = analysisQueryId(
snapshot.target,
snapshot.binding,
evidence.operation,
evidence.parameters,
);
if (!stagedQueryIds.has(queryId)) continue;
const entry = this.#entries.get(queryId);
if (
entry === undefined ||
!evidenceMatchesEntry(evidence, entry, snapshot)
)
continue;
const previous = evidenceIds.get(queryId);
if (previous !== undefined && previous !== evidence.evidence_id)
ambiguous.add(queryId);
else evidenceIds.set(queryId, evidence.evidence_id);
}
for (const [queryId, evidenceId] of evidenceIds) {
if (ambiguous.has(queryId)) this.#evidenceIds.delete(queryId);
else this.#evidenceIds.set(queryId, evidenceId);
}
for (const entry of snapshot.workflow_entries) {
if (!this.#workflowEntries.has(entry.query_id) && !this.#hasCapacity())
@@ -363,7 +394,9 @@ export class AnalysisSnapshotCache {
},
target,
);
const evidenceId = this.#evidenceIds.get(queryId);
return structuredClone({
...(evidenceId === undefined ? {} : { evidenceId }),
result: cached.execution.result,
rawResult: cached.execution.raw_result,
provider: cached.execution.provider,
@@ -431,6 +464,9 @@ export class AnalysisSnapshotCache {
},
});
this.#entries.set(entry.query_id, entry);
if (execution.evidenceId === undefined)
this.#evidenceIds.delete(entry.query_id);
else this.#evidenceIds.set(entry.query_id, execution.evidenceId);
}
/** Forget all target-bound entries. */
@@ -442,6 +478,7 @@ export class AnalysisSnapshotCache {
#clearEntries(): void {
this.#entries.clear();
this.#evidenceIds.clear();
this.#workflowEntries.clear();
}
+17 -3
View File
@@ -136,6 +136,16 @@ const trackCleanupFailures = (
};
};
/** Provider metadata cannot confer authority to an uncommitted Evidence ID. */
const unrecordedExecution = (
result: Result<AnalysisExecution, AnalysisError>,
): Result<AnalysisExecution, AnalysisError> => {
if (!result.ok || !("evidenceId" in result.value)) return result;
const observation = { ...result.value };
delete observation.evidenceId;
return ok(observation);
};
/**
* Owns the single active target shared by CLI and MCP adapters.
*
@@ -1180,7 +1190,9 @@ export class BinarySession
"This operation requires an executable analysis target; the active session contains only app resource access.",
),
);
const result = await resources.client.execute(name, arguments_, options);
const result = unrecordedExecution(
await resources.client.execute(name, arguments_, options),
);
if (!result.ok) return result;
const execution = { ...result.value, subject: null };
return ok(execution);
@@ -1207,7 +1219,7 @@ export class BinarySession
// Stop replay before dispatch, including reads overlapping the write.
if (capability.effects.mutatesArtifact) this.invalidateSnapshot();
const call = active.client.execute(name, arguments_, options);
const result = await call;
const result = unrecordedExecution(await call);
const profiled = bindExecutionTarget(
commitExecutionProfile(name, result, profile),
name,
@@ -1266,14 +1278,16 @@ export class BinarySession
);
const recorded = this.recordEvidence(evidence);
if (!recorded.ok) return recorded;
const retained = { ...execution.value, evidenceId: evidence.evidence_id };
if (execution.value.cachePolicy !== "live")
this.recordSnapshot({
target: active.target,
profile,
operation: name,
parameters: arguments_,
execution: execution.value,
execution: retained,
});
return ok(retained);
}
return execution;
}
+3
View File
@@ -275,6 +275,9 @@ const executeEnhancedTool = async (
result: result.value.result,
upstreamProfile: registration.analysisProfile?.(name),
upstreamLimitations: result.value.limitations,
...(result.value.evidenceLinks === undefined
? {}
: { upstreamEvidenceLinks: result.value.evidenceLinks }),
});
const recorded = registration.recordEvidence?.(evidence);
if (recorded !== undefined && !recorded.ok)
@@ -0,0 +1,293 @@
import { executeComposedWorkflow } from "../../../src/application/WorkflowEvidence.js";
import { parseAnalysisSnapshot } from "../../../src/domain/analysisSnapshot.js";
import { ok } from "../../../src/domain/result.js";
import { createDeferred } from "../../fixtures/binarySession.js";
import { expect, it } from "vitest";
import { runDirectAnalysis } from "../../../src/application/DirectAnalysis.js";
import { createEvidence, parseEvidence } from "../../../src/domain/evidence.js";
import { composedWorkflowFixture } from "../../fixtures/composedWorkflowEvidence.js";
it.each([false, true])(
"links a profiled binary overview only to the atomic record it consumed (cached=%s)",
async (warmCache) => {
const fixture = await composedWorkflowFixture();
const { path, summary, calls, session } = fixture;
const createSession = async () => {
const opened = await session.open(path);
if (!opened.ok) throw opened.error;
const unrelated = await session.execute("address_name", {
document: "fixture",
address: "0x2000",
});
if (!unrelated.ok) throw unrelated.error;
if (warmCache) {
const warmed = await session.execute(
"read_binary_inventory_summary",
{},
);
if (!warmed.ok) throw warmed.error;
}
return session;
};
const dependencies = {
...fixture.dependencies,
createBinarySession: createSession,
};
try {
const evidence = parseEvidence(
await runDirectAnalysis(dependencies, path, "binary_overview", {}),
);
expect(evidence.normalized_result).toEqual(summary);
expect(evidence.confidence).toBe("derived");
const records = fixture.closedRecords();
const atomic = records.filter(
(record) => record.operation === "read_binary_inventory_summary",
);
const unrelated = records.filter(
(record) => record.operation === "address_name",
);
expect(atomic).toHaveLength(1);
expect(unrelated).toHaveLength(1);
expect(atomic[0]?.normalized_result).toEqual(summary);
expect(
calls.filter(
(operation) => operation === "read_binary_inventory_summary",
),
).toHaveLength(1);
// These are real, already committed ledger IDs, not hashes fabricated by
// the workflow and not a before/after scan used to infer consumption.
expect(evidence.evidence_links).toEqual(
atomic.map((record) => record.evidence_id),
);
expect(evidence.evidence_links).not.toContain(unrelated[0]?.evidence_id);
} finally {
expect((await session.close()).ok).toBe(true);
}
},
);
it("deduplicates repeated and nested consumed records without linking unrelated concurrent work", async () => {
const entered = createDeferred<void>();
const release = createDeferred<void>();
const fixture = await composedWorkflowFixture({
beforeSummary: async () => {
entered.resolve();
await release.promise;
},
});
await fixture.open();
const overview = executeComposedWorkflow(
fixture.session,
async (observed) => {
const first = await observed.execute("read_binary_inventory_summary", {});
if (!first.ok) return first;
const nested = await executeComposedWorkflow(observed, async (inner) => {
const repeated = await inner.execute(
"read_binary_inventory_summary",
{},
);
return repeated.ok ? ok(repeated.value.result) : repeated;
});
return nested.ok ? ok(nested.value.result) : nested;
},
);
await entered.promise;
try {
const other = await executeComposedWorkflow(
fixture.session,
async (observed) => {
const execution = await observed.execute("address_name", {
address: "0x2000",
document: "fixture",
});
return execution.ok ? ok(execution.value.result) : execution;
},
);
if (!other.ok) throw other.error;
release.resolve();
const result = await overview;
if (!result.ok) throw result.error;
const atomic = fixture.session
.exportEvidenceBundle()
.records.find(
(record) => record.operation === "read_binary_inventory_summary",
);
const unrelated = fixture.session
.exportEvidenceBundle()
.records.find((record) => record.operation === "address_name");
expect(atomic).toBeDefined();
expect(unrelated).toBeDefined();
expect(result.value.evidenceLinks).toEqual([atomic?.evidence_id]);
expect(other.value.evidenceLinks).toEqual([unrelated?.evidence_id]);
expect(
fixture.calls.filter(
(operation) => operation === "read_binary_inventory_summary",
),
).toHaveLength(1);
} finally {
release.resolve();
await overview;
}
});
it("rehydrates the original atomic identity from an authenticated snapshot without rerunning the provider", async () => {
const original = await composedWorkflowFixture();
await original.open();
const executed = await original.session.execute(
"read_binary_inventory_summary",
{},
);
if (!executed.ok) throw executed.error;
const exported = original.session.exportAnalysisSnapshot();
if (!exported.ok) throw exported.error;
const snapshot = parseAnalysisSnapshot(
JSON.parse(JSON.stringify(exported.value)),
);
const atomic = snapshot.evidence_bundle.records.find(
(record) => record.operation === "read_binary_inventory_summary",
);
expect(executed.value.evidenceId).toBe(atomic?.evidence_id);
const replay = await composedWorkflowFixture();
expect(replay.session.importAnalysisSnapshot(snapshot).ok).toBe(true);
await replay.open();
const composed = await executeComposedWorkflow(
replay.session,
async (observed) => {
const execution = await observed.execute(
"read_binary_inventory_summary",
{},
);
if (!execution.ok) return execution;
expect(execution.value.limitations).toContain(
"Loaded from a local REA analysis snapshot; this call did not re-run the provider.",
);
return ok(execution.value.result);
},
);
if (!composed.ok) throw composed.error;
expect(composed.value.evidenceLinks).toEqual([atomic?.evidence_id]);
expect(replay.calls).not.toContain("read_binary_inventory_summary");
expect(
replay.session.evidenceById(atomic!.evidence_id)?.normalized_result,
).toEqual(original.summary);
});
it.each(["profileless", "uncacheable"] as const)(
"does not trust provider-supplied Evidence IDs on %s executions",
async (mode) => {
const spoofedEvidenceId = `ev_${"f".repeat(64)}`;
const fixture = await composedWorkflowFixture({
profiled: mode !== "profileless",
live: mode === "uncacheable",
spoofedEvidenceId,
});
await fixture.open();
const composed = await executeComposedWorkflow(
fixture.session,
async (observed) => {
const execution = await observed.execute(
"read_binary_inventory_summary",
{},
);
if (!execution.ok) return execution;
expect(execution.value.evidenceId).toBeUndefined();
return ok(execution.value.result);
},
);
if (!composed.ok) throw composed.error;
expect(composed.value.evidenceLinks).toBeUndefined();
expect(fixture.session.exportEvidenceBundle().records).toEqual([]);
},
);
it("does not manufacture links for empty or failed composed calls", async () => {
const fixture = await composedWorkflowFixture();
await fixture.open();
const empty = await executeComposedWorkflow(fixture.session, async () =>
ok(null),
);
expect(empty).toEqual(ok({ result: null, limitations: [] }));
const failed = await executeComposedWorkflow(
fixture.session,
async (observed) => {
const result = await observed.execute("list_strings", {});
return result.ok ? ok(result.value.result) : result;
},
);
expect(failed.ok).toBe(false);
expect(fixture.session.exportEvidenceBundle().records).toEqual([]);
});
it("drops an old cached ID rather than guessing after an ambiguous snapshot replacement", async () => {
const fixture = await composedWorkflowFixture();
const target = await fixture.open();
const initial = await fixture.session.execute(
"read_binary_inventory_summary",
{},
);
if (!initial.ok) throw initial.error;
const snapshot = fixture.session.exportAnalysisSnapshot();
if (!snapshot.ok) throw snapshot.error;
const unrelated = await fixture.session.execute("address_name", {
address: "0x2000",
document: "fixture",
});
if (!unrelated.ok || unrelated.value.evidenceId === undefined)
throw new Error("Expected recorded dependency");
const alternative = createEvidence(target, fixture.profile.provider, {
operation: "read_binary_inventory_summary",
parameters: {},
result: fixture.summary,
analysisProfile: fixture.profile,
evidenceLinks: [unrelated.value.evidenceId],
});
expect(fixture.session.recordEvidence(alternative).ok).toBe(true);
const ambiguous = fixture.session.exportAnalysisSnapshot();
if (!ambiguous.ok) throw ambiguous.error;
const replay = await composedWorkflowFixture();
expect(replay.session.importAnalysisSnapshot(snapshot.value).ok).toBe(true);
await replay.open();
const before = await replay.session.execute(
"read_binary_inventory_summary",
{},
);
if (!before.ok) throw before.error;
expect(before.value.evidenceId).toBe(initial.value.evidenceId);
expect(replay.session.importAnalysisSnapshot(ambiguous.value).ok).toBe(true);
const after = await replay.session.execute(
"read_binary_inventory_summary",
{},
);
if (!after.ok) throw after.error;
expect(after.value.result).toEqual(fixture.summary);
expect(after.value.evidenceId).toBeUndefined();
expect(replay.calls).not.toContain("read_binary_inventory_summary");
});
it("replaces provider-supplied identity with the record actually committed by the session", async () => {
const spoofedEvidenceId = `ev_${"f".repeat(64)}`;
const fixture = await composedWorkflowFixture({ spoofedEvidenceId });
await fixture.open();
const executed = await fixture.session.execute(
"read_binary_inventory_summary",
{},
);
if (!executed.ok) throw executed.error;
const records = fixture.session.exportEvidenceBundle().records;
expect(records).toHaveLength(1);
expect(executed.value.evidenceId).toBe(records[0]?.evidence_id);
expect(executed.value.evidenceId).not.toBe(spoofedEvidenceId);
expect((await fixture.session.close()).ok).toBe(true);
await fixture.open();
const refreshed = await fixture.session.execute(
"read_binary_inventory_summary",
{},
);
if (!refreshed.ok) throw refreshed.error;
expect(
fixture.calls.filter(
(operation) => operation === "read_binary_inventory_summary",
),
).toHaveLength(2);
expect(fixture.session.hasEvidence(refreshed.value.evidenceId!)).toBe(true);
});
@@ -0,0 +1,67 @@
import { Client, InMemoryTransport } from "@modelcontextprotocol/client";
import { expect, it } from "vitest";
import { createServer } from "../../../src/server/createServer.js";
import { silentLogger } from "../../../src/logger.js";
import { parseEvidence } from "../../../src/domain/evidence.js";
import { runDirectAnalysis } from "../../../src/application/DirectAnalysis.js";
import { composedWorkflowFixture } from "../../fixtures/composedWorkflowEvidence.js";
it("returns the same consumed atomic provenance through CLI application and MCP boundaries", async () => {
const cli = await composedWorkflowFixture();
const direct = parseEvidence(
await runDirectAnalysis(cli.dependencies, cli.path, "binary_overview", {}),
);
const fixture = await composedWorkflowFixture();
const server = createServer(
{ kind: "session", session: fixture.session },
{ logger: silentLogger },
);
const client = new Client({
name: "workflow-evidence-links",
version: "1.0.0",
});
const [clientTransport, serverTransport] =
InMemoryTransport.createLinkedPair();
try {
await server.connect(serverTransport);
await client.connect(clientTransport);
const opened = await client.callTool({
name: "open_binary",
arguments: { path: fixture.path },
});
expect(opened.isError, JSON.stringify(opened.content)).not.toBe(true);
const response = await client.callTool({
name: "binary_overview",
arguments: {},
});
expect(response.isError, JSON.stringify(response.content)).not.toBe(true);
const mcp = parseEvidence(response.structuredContent);
const atomic = fixture.session
.exportEvidenceBundle()
.records.filter(
(record) => record.operation === "read_binary_inventory_summary",
);
expect(atomic).toHaveLength(1);
expect(mcp.evidence_links).toEqual(
atomic.map((record) => record.evidence_id),
);
expect(mcp.evidence_links).toEqual(direct.evidence_links);
expect(mcp.normalized_result).toEqual(direct.normalized_result);
const again = await client.callTool({
name: "binary_overview",
arguments: {},
});
expect(again.isError).not.toBe(true);
expect(parseEvidence(again.structuredContent).evidence_links).toEqual(
mcp.evidence_links,
);
expect(
fixture.calls.filter(
(operation) => operation === "read_binary_inventory_summary",
),
).toHaveLength(1);
} finally {
await client.close();
await server.close();
}
});
+124
View File
@@ -0,0 +1,124 @@
import { writeFile } from "node:fs/promises";
import { join } from "node:path";
import { onTestFinished } from "vitest";
import { parseConfig } from "../../src/config/parseConfig.js";
import {
createAnalysisExecution,
type AnalysisProvider,
} from "../../src/application/AnalysisProvider.js";
import type { DirectAnalysisDependencies } from "../../src/application/DirectAnalysisDependencies.js";
import { createAnalysisProfile } from "../../src/domain/analysisProfile.js";
import type { Evidence } from "../../src/domain/evidence.js";
import { ok } from "../../src/domain/result.js";
import { createTestBinarySession } from "./binarySession.js";
import { createTestTempDirectory } from "./temporaryDirectory.js";
/** Real session/ledger with a deterministic provider that never launches a target. */
export const composedWorkflowFixture = async (
options: {
readonly profiled?: boolean;
readonly live?: boolean;
readonly spoofedEvidenceId?: string;
readonly beforeSummary?: () => Promise<void>;
} = {},
) => {
const directory = await createTestTempDirectory("rea-workflow-links-");
const path = join(directory, "fixture.hop");
await writeFile(path, "synthetic analysis database");
const identity = { id: "fixture", name: "Fixture", version: "1" };
const profile = createAnalysisProfile(identity, { fixture: true });
const summary = {
document: "fixture",
procedure_count: 12,
string_count: 34,
};
const calls: string[] = [];
let closedRecords: readonly Evidence[] = [];
const provider: AnalysisProvider = {
identity: () => identity,
capabilities: () =>
(["read_binary_inventory_summary", "address_name"] as const).map(
(operation) => ({
provider: identity,
operation,
available: true,
reason: null,
cachePolicy: options.live === true ? "live" : "snapshot",
limitations: [],
effects: {
mutatesArtifact: false,
launchesProcess: false,
mayShowUi: false,
mayAccessNetwork: false,
mayWriteFilesystem: false,
changesPermissions: false,
requiresRoot: false,
},
}),
),
...(options.profiled === false
? {}
: { resolveAnalysisProfile: async () => ok({ profile }) }),
createClient: () => ({
execute: async (operation) => {
calls.push(operation);
if (operation === "read_binary_inventory_summary")
await options.beforeSummary?.();
if (
!["health", "address_name", "read_binary_inventory_summary"].includes(
operation,
)
)
throw new Error(`Unexpected operation: ${operation}`);
const observation = createAnalysisExecution(
operation === "health"
? null
: operation === "address_name"
? "unrelated history"
: summary,
identity,
);
return ok({
...observation,
...(options.spoofedEvidenceId === undefined
? {}
: { evidenceId: options.spoofedEvidenceId }),
});
},
close: async () => {
// DirectAnalysis clears the ledger only after this legitimate close.
closedRecords = session.exportEvidenceBundle().records;
return ok(null);
},
}),
};
const session = createTestBinarySession(provider);
onTestFinished(async () => {
const closed = await session.close();
if (!closed.ok) throw closed.error;
});
const createSession = () => session;
const dependencies: DirectAnalysisDependencies = {
readConfiguration: () => parseConfig({}),
createBinarySession: createSession,
createNativeDispatchBinarySession: createSession,
createArtifactBinarySession: createSession,
createNativeBinarySession: createSession,
createManagedBinarySession: createSession,
};
const open = async () => {
const opened = await session.open(path);
if (!opened.ok) throw opened.error;
return opened.value;
};
return {
path,
profile,
summary,
calls,
session,
dependencies,
open,
closedRecords: () => closedRecords,
};
};