fix: revalidate restored snapshot sources lazily

This commit is contained in:
morluto committed 2026-11-04 14:06:59 +08:00
1 parent 75c7187290
commit 89d5bb7056
4 files changed
+149 -11

No files matched your search

@@ -1307,8 +1307,53 @@ it("replays source-bearing entries only when their complete retained closure exi
});
const incomplete = new AnalysisSnapshotCache();
incomplete.stage(pruned);
const partial = new Map(
[parent, movedSource].map((record) => [record.evidence_id, record]),
);
expect(
incomplete.lookup(target, profile, "procedure_info", parameters),
incomplete.lookup(target, profile, "procedure_info", parameters, (id) =>
partial.get(id),
),
).toBeUndefined();
partial.set(leaf.evidence_id, leaf);
expect(
incomplete
.lookup(target, profile, "procedure_info", parameters, (id) =>
partial.get(id),
)
?.sourceEvidence?.map((record) => record.evidence_id),
).toEqual([source.evidence_id]);
const otherSource = immutableEvidence(
createEvidence(target, ANALYSIS_SNAPSHOT_PROVIDER, {
operation: "other_source",
parameters: {},
result: "different source",
}),
);
const otherParent = immutableEvidence(
createAnalysisExecutionEvidence(
"procedure_info",
parameters,
{ ...execution, sourceEvidence: [otherSource] },
"observed",
),
);
const records = [parent, movedSource, leaf, otherParent, otherSource];
const ambiguous = new AnalysisSnapshotCache();
ambiguous.stage(
parseAnalysisSnapshot({
...exported.value,
evidence_bundle: createEvidenceBundle(records),
}),
);
const allRecords = new Map(
records.map((record) => [record.evidence_id, record]),
);
expect(
ambiguous.lookup(target, profile, "procedure_info", parameters, (id) =>
allRecords.get(id),
),
).toBeUndefined();
});
@@ -147,9 +147,12 @@ export const SNAPSHOT_REPLAY_LIMITATION =
/** Bounded in-memory cache for one immutable binary identity. */
export class AnalysisSnapshotCache {
readonly #entries = new Map<string, AnalysisSnapshotEntry>();
// Null means imported provenance cannot be faithfully replayed. It is distinct
// from an entry recorded before an Evidence ID was assigned.
readonly #replayBindings = new Map<string, SnapshotReplayBinding | null>();
// Null is ambiguous provenance; a pending ID is an exact parent whose source
// closure was missing and can be checked again against the owning ledger.
readonly #replayBindings = new Map<
string,
SnapshotReplayBinding | { readonly pendingEvidenceId: string } | null
>();
readonly #workflowEntries = new Map<string, AnalysisSnapshotWorkflowEntry>();
#target: AnalysisSnapshotTarget | undefined;
#binding: AnalysisSnapshotBinding | undefined;
@@ -242,11 +245,15 @@ export class AnalysisSnapshotCache {
else evidenceByQuery.set(queryId, evidence);
}
for (const [queryId, evidence] of evidenceByQuery) {
if (ambiguous.has(queryId)) {
this.#replayBindings.set(queryId, null);
continue;
}
this.#replayBindings.set(
queryId,
ambiguous.has(queryId)
? null
: snapshotReplayBinding(evidence, (id) => records.get(id)),
snapshotReplayBinding(evidence, (id) => records.get(id)) ?? {
pendingEvidenceId: evidence.evidence_id,
},
);
}
for (const entry of snapshot.workflow_entries) {
@@ -272,7 +279,12 @@ export class AnalysisSnapshotCache {
replacements.map(({ current }) => [current.evidence_id, current]),
);
for (const [queryId, binding] of this.#replayBindings) {
if (binding === null || binding.sourceEvidence === undefined) continue;
if (
binding === null ||
"pendingEvidenceId" in binding ||
binding.sourceEvidence === undefined
)
continue;
const sources = binding.sourceEvidence;
if (!sources.some((source) => currentRecords.has(source.evidence_id)))
continue;
@@ -475,6 +487,7 @@ export class AnalysisSnapshotCache {
profile: AnalysisProfileCommitment,
operation: AnalysisOperation,
parameters: Readonly<Record<string, JsonValue>>,
lookup?: (id: string) => Evidence | undefined,
): AnalysisExecution | undefined {
const queryId = analysisQueryId(
snapshotTarget(target),
@@ -484,10 +497,18 @@ export class AnalysisSnapshotCache {
);
const cached = this.#entries.get(queryId);
if (cached === undefined) return undefined;
const binding = this.#replayBindings.get(queryId);
let binding = this.#replayBindings.get(queryId);
// Do not turn a refused provenance selection into a fresh, unlinked record.
// The provider must produce a new observation before this query can replay.
if (binding === null) return undefined;
if (binding !== undefined && "pendingEvidenceId" in binding) {
const parent = lookup?.(binding.pendingEvidenceId);
if (parent === undefined || lookup === undefined) return undefined;
const restored = snapshotReplayBinding(parent, lookup);
if (restored === null) return undefined;
this.#replayBindings.set(queryId, restored);
binding = restored;
}
const cachedSubject = cached.execution.subject;
const subject =
cachedSubject == null
@@ -323,7 +323,9 @@ export abstract class BinarySessionRecords {
>,
): AnalysisExecution | undefined {
if (this.#snapshotInvalidated) return undefined;
return this.#snapshot.lookup(target, profile, operation, parameters);
return this.#snapshot.lookup(target, profile, operation, parameters, (id) =>
this.evidenceForAnalysis(id),
);
}
protected recordSnapshot(
@@ -25,7 +25,10 @@ import { createAnalysisProfile } from "../../../src/domain/analysisProfile.js";
import { createEvidence } from "../../../src/domain/evidence.js";
import { readAnalysisSnapshot } from "../../../src/application/binary/AnalysisSnapshotFiles.js";
import { ok } from "../../../src/domain/result.js";
import { evidenceBundleSchema } from "../../../src/domain/evidenceBundle.js";
import {
createEvidenceBundle,
evidenceBundleSchema,
} from "../../../src/domain/evidenceBundle.js";
import { silentLogger } from "../../../src/logger.js";
import { createServer } from "../../../src/server/createServer.js";
import { toolContract } from "../../../src/contracts/toolContracts.js";
@@ -691,8 +694,75 @@ it("replays a source-bearing observation through an exported MCP snapshot", asyn
},
],
});
await expectRestoredSourceReplay(provider, targetPath, snapshotPath, calls);
});
const expectRestoredSourceReplay = async (
provider: AnalysisProvider,
targetPath: string,
snapshotPath: string,
calls: string[],
) => {
const saved = await readAnalysisSnapshot(snapshotPath);
if (!saved.ok) throw saved.error;
const source = saved.value.evidence_bundle.records.find(
(record) => record.operation === "health",
);
if (source === undefined) throw new Error("Expected historical source");
const directory = dirname(snapshotPath);
const prunedPath = join(directory, "missing-source.json");
const sourcePath = join(directory, "source-bundle.json");
await writeFile(
prunedPath,
JSON.stringify({
...saved.value,
evidence_bundle: createEvidenceBundle(
saved.value.evidence_bundle.records.filter(
(record) => record.evidence_id !== source.evidence_id,
),
),
}),
);
await writeFile(sourcePath, JSON.stringify(createEvidenceBundle([source])));
const { mcp } = await connectWorkflowMcp(
createWorkflowSession(provider),
"source-restored-after-open",
);
expect(
(
await mcp.callTool({
name: "open_binary",
arguments: { path: targetPath, snapshot_path: prunedPath },
})
).isError,
).not.toBe(true);
expect(
(
await mcp.callTool({
name: "import_evidence_bundle",
arguments: { path: sourcePath },
})
).isError,
).not.toBe(true);
const replayed = toolContract("list_strings").outputSchema.parse(
(
await mcp.callTool({
name: "list_strings",
arguments: { document: "fixture" },
})
).structuredContent,
);
expect(replayed.evidence_links).toEqual([source.evidence_id]);
expect(replayed.limitations).toEqual(
expect.arrayContaining([
expect.stringContaining("did not re-run the provider"),
]),
);
expect(
calls.filter((operation) => operation === "list_strings"),
).toHaveLength(1);
};
const createWorkflowFiles = async (
prefix: string,
filenames: readonly string[],