mirror of
https://github.com/morluto/rea.git
synced 2026-11-10 08:47:00 +00:00
fix(snapshot): replay validated workflow evidence (#984)
* fix(snapshot): bind replay to validated cache entry * fix(snapshot): persist composed workflow results * fix(snapshot): enforce workflow replay identity * fix(snapshot): honor cancellation after route resolution * test(snapshot): make cancellation race deterministic * fix(snapshot): replay all composed workflow evidence * fix(snapshot): index workflow evidence once per snapshot Validate workflow entries against a query and content index instead of scanning the evidence bundle for every entry. * fix(snapshot): reuse previewed routes and bound workflow cache --------- Co-authored-by: Cursor Agent <cursoragent@cursor.com>
This commit is contained in:
1 parent
96a46ada0a
commit
1d62e54e1f
11 files changed
+1462
-80
No files matched your search
@@ -10,15 +10,137 @@ import { createEvidence } from "../../domain/evidence.js";
|
||||
import { createEvidenceBundle } from "../../domain/evidenceBundle.js";
|
||||
import { ok } from "../../domain/result.js";
|
||||
import { createAnalysisExecution } from "../AnalysisProvider.js";
|
||||
import {
|
||||
createAnalysisSnapshotWorkflowEntry,
|
||||
parseAnalysisSnapshot,
|
||||
snapshotBinding,
|
||||
snapshotTarget,
|
||||
} from "../../domain/analysisSnapshot.js";
|
||||
import {
|
||||
REA_WORKFLOW_PROVIDER,
|
||||
workflowAnalysisProfile,
|
||||
} from "../InvestigationProviders.js";
|
||||
import type {
|
||||
AnalysisOperation,
|
||||
CapabilityDescriptor,
|
||||
} from "../AnalysisProvider.js";
|
||||
import {
|
||||
AnalysisSnapshotCache,
|
||||
SNAPSHOT_CACHE_ENTRY_CEILING,
|
||||
isSnapshotCacheable,
|
||||
} from "./AnalysisSnapshotCache.js";
|
||||
|
||||
describe("analysis snapshot cache capacity", () => {
|
||||
it("shares capacity across query kinds while allowing replacements and fresh partitions", () => {
|
||||
const cache = new AnalysisSnapshotCache();
|
||||
const target = ANALYSIS_SNAPSHOT_TARGET;
|
||||
const profile = ANALYSIS_SNAPSHOT_PROFILE;
|
||||
const execution = {
|
||||
...createAnalysisExecution("workflow", REA_WORKFLOW_PROVIDER, {
|
||||
subject: target,
|
||||
}),
|
||||
analysisProfile: workflowAnalysisProfile(profile),
|
||||
};
|
||||
const workflowInput = {
|
||||
target,
|
||||
profile,
|
||||
operation: "trace_feature",
|
||||
parameters: { query: "existing" },
|
||||
execution,
|
||||
};
|
||||
for (let index = 0; index < SNAPSHOT_CACHE_ENTRY_CEILING - 1; index += 1)
|
||||
cache.recordWorkflow({
|
||||
...workflowInput,
|
||||
parameters: { query: String(index) },
|
||||
});
|
||||
// Staging and recording both spend the same budget; an existing query can
|
||||
// still be refreshed when no space remains for a new binding.
|
||||
const directInput = {
|
||||
target,
|
||||
profile,
|
||||
operation: "address_name" as const,
|
||||
parameters: { address: "0x1000", document: "fixture" },
|
||||
execution: createAnalysisExecution(
|
||||
"initial",
|
||||
ANALYSIS_SNAPSHOT_PROVIDER,
|
||||
{ subject: target, analysisProfile: profile },
|
||||
),
|
||||
};
|
||||
cache.record(directInput);
|
||||
cache.record({
|
||||
...directInput,
|
||||
parameters: { address: "0x2000", document: "fixture" },
|
||||
});
|
||||
cache.recordWorkflow(workflowInput);
|
||||
expect(cache.entries()).toHaveLength(1);
|
||||
expect(cache.workflowEntries()).toHaveLength(
|
||||
SNAPSHOT_CACHE_ENTRY_CEILING - 1,
|
||||
);
|
||||
|
||||
cache.record({
|
||||
...directInput,
|
||||
execution: { ...directInput.execution, result: "updated" },
|
||||
});
|
||||
cache.recordWorkflow({
|
||||
...workflowInput,
|
||||
parameters: { query: "0" },
|
||||
execution: { ...execution, result: "updated" },
|
||||
});
|
||||
expect(
|
||||
cache.lookup(target, profile, "address_name", directInput.parameters)
|
||||
?.result,
|
||||
).toBe("updated");
|
||||
expect(
|
||||
cache.workflowEntries().find(({ parameters }) => parameters.query === "0")
|
||||
?.execution.result,
|
||||
).toBe("updated");
|
||||
|
||||
const incoming = createAnalysisSnapshotWorkflowEntry({
|
||||
...workflowInput,
|
||||
target: snapshotTarget(target),
|
||||
binding: snapshotBinding(profile),
|
||||
});
|
||||
const evidence = createEvidence(target, REA_WORKFLOW_PROVIDER, {
|
||||
operation: workflowInput.operation,
|
||||
parameters: workflowInput.parameters,
|
||||
result: execution.result,
|
||||
rawResult: execution.rawResult,
|
||||
analysisProfile: execution.analysisProfile,
|
||||
limitations: execution.limitations,
|
||||
locations: execution.locations,
|
||||
});
|
||||
cache.stage(
|
||||
parseAnalysisSnapshot({
|
||||
target: snapshotTarget(target),
|
||||
binding: snapshotBinding(profile),
|
||||
entries: [],
|
||||
workflow_entries: [incoming],
|
||||
evidence_bundle: createEvidenceBundle([evidence]),
|
||||
}),
|
||||
);
|
||||
expect(cache.workflowEntries()).toHaveLength(
|
||||
SNAPSHOT_CACHE_ENTRY_CEILING - 1,
|
||||
);
|
||||
|
||||
const otherTarget = { ...target, sha256: "b".repeat(64) };
|
||||
cache.recordWorkflow({
|
||||
...workflowInput,
|
||||
target: otherTarget,
|
||||
execution: { ...execution, subject: otherTarget },
|
||||
});
|
||||
expect(cache.entries()).toEqual([]);
|
||||
expect(cache.workflowEntries()).toHaveLength(1);
|
||||
cache.clear();
|
||||
cache.record({
|
||||
...directInput,
|
||||
target: otherTarget,
|
||||
execution: { ...directInput.execution, subject: otherTarget },
|
||||
});
|
||||
expect(cache.entries()).toHaveLength(1);
|
||||
expect(cache.workflowEntries()).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("analysis snapshot cache partitioning", () => {
|
||||
it.each<{
|
||||
operation: Exclude<AnalysisOperation, "health">;
|
||||
|
||||
@@ -9,6 +9,7 @@ import { err, ok, type Result } from "../../domain/result.js";
|
||||
import {
|
||||
analysisQueryId,
|
||||
createAnalysisSnapshotEntry,
|
||||
createAnalysisSnapshotWorkflowEntry,
|
||||
snapshotBinding,
|
||||
snapshotMatchesProfile,
|
||||
snapshotMatchesTarget,
|
||||
@@ -16,6 +17,7 @@ import {
|
||||
parseAnalysisSnapshot,
|
||||
type AnalysisSnapshot,
|
||||
type AnalysisSnapshotEntry,
|
||||
type AnalysisSnapshotWorkflowEntry,
|
||||
type AnalysisSnapshotBinding,
|
||||
type AnalysisSnapshotTarget,
|
||||
} from "../../domain/analysisSnapshot.js";
|
||||
@@ -54,6 +56,9 @@ const DOCUMENT_SCOPED_OPERATIONS: ReadonlySet<string> = new Set(
|
||||
),
|
||||
);
|
||||
|
||||
/** Maximum duplicated result bindings retained across direct and workflow queries. */
|
||||
export const SNAPSHOT_CACHE_ENTRY_CEILING = 10_000;
|
||||
|
||||
/** Whether an operation is immutable and independent of provider UI state. */
|
||||
export const isSnapshotCacheable = (
|
||||
operation: AnalysisOperation,
|
||||
@@ -73,6 +78,7 @@ export const isSnapshotCacheable = (
|
||||
/** Bounded in-memory cache for one immutable binary identity. */
|
||||
export class AnalysisSnapshotCache {
|
||||
readonly #entries = new Map<string, AnalysisSnapshotEntry>();
|
||||
readonly #workflowEntries = new Map<string, AnalysisSnapshotWorkflowEntry>();
|
||||
#target: AnalysisSnapshotTarget | undefined;
|
||||
#binding: AnalysisSnapshotBinding | undefined;
|
||||
|
||||
@@ -93,7 +99,7 @@ export class AnalysisSnapshotCache {
|
||||
|
||||
/** Replace cache state when the target or selected profile changes. */
|
||||
select(target: BinaryTarget, profile: AnalysisProfileCommitment): void {
|
||||
if (!this.matches(target, profile)) this.#entries.clear();
|
||||
if (!this.matches(target, profile)) this.#clearEntries();
|
||||
this.#target = snapshotTarget(target);
|
||||
this.#binding = snapshotBinding(profile);
|
||||
}
|
||||
@@ -109,14 +115,20 @@ export class AnalysisSnapshotCache {
|
||||
snapshot.binding.analysis_profile,
|
||||
))
|
||||
)
|
||||
this.#entries.clear();
|
||||
this.#clearEntries();
|
||||
this.#target = structuredClone(snapshot.target);
|
||||
this.#binding = structuredClone(snapshot.binding);
|
||||
let imported = 0;
|
||||
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, structuredClone(entry));
|
||||
}
|
||||
for (const entry of snapshot.workflow_entries ?? []) {
|
||||
if (!this.#workflowEntries.has(entry.query_id) && !this.#hasCapacity())
|
||||
continue;
|
||||
this.#workflowEntries.set(entry.query_id, structuredClone(entry));
|
||||
}
|
||||
return imported;
|
||||
}
|
||||
|
||||
@@ -137,6 +149,9 @@ export class AnalysisSnapshotCache {
|
||||
target: snapshotTargetIdentity,
|
||||
binding,
|
||||
entries: this.entries(),
|
||||
...(this.workflowEntries().length === 0
|
||||
? {}
|
||||
: { workflow_entries: this.workflowEntries() }),
|
||||
evidence_bundle: retainedEvidence,
|
||||
}),
|
||||
);
|
||||
@@ -193,6 +208,47 @@ export class AnalysisSnapshotCache {
|
||||
.map((entry) => structuredClone(entry));
|
||||
}
|
||||
|
||||
/** Return canonical composed-workflow entries for persistence. */
|
||||
workflowEntries(): AnalysisSnapshotWorkflowEntry[] {
|
||||
return [...this.#workflowEntries.values()]
|
||||
.sort((left, right) => left.query_id.localeCompare(right.query_id))
|
||||
.map((entry) => structuredClone(entry));
|
||||
}
|
||||
|
||||
/** Record one exact derived workflow result unless the shared cache is full. */
|
||||
recordWorkflow(input: {
|
||||
readonly target: BinaryTarget;
|
||||
readonly profile: AnalysisProfileCommitment;
|
||||
readonly operation: string;
|
||||
readonly parameters: Readonly<Record<string, JsonValue>>;
|
||||
readonly execution: {
|
||||
readonly result: JsonValue;
|
||||
readonly rawResult: JsonValue | null;
|
||||
readonly provider: AnalysisExecution["provider"];
|
||||
readonly analysisProfile: AnalysisProfileCommitment;
|
||||
readonly limitations: readonly string[];
|
||||
readonly locations: AnalysisExecution["locations"];
|
||||
readonly subject: AnalysisExecution["subject"];
|
||||
};
|
||||
}): void {
|
||||
this.select(input.target, input.profile);
|
||||
const queryId = analysisQueryId(
|
||||
snapshotTarget(input.target),
|
||||
snapshotBinding(input.profile),
|
||||
input.operation,
|
||||
input.parameters,
|
||||
);
|
||||
if (!this.#workflowEntries.has(queryId) && !this.#hasCapacity()) return;
|
||||
const entry = createAnalysisSnapshotWorkflowEntry({
|
||||
target: snapshotTarget(input.target),
|
||||
binding: snapshotBinding(input.profile),
|
||||
operation: input.operation,
|
||||
parameters: input.parameters,
|
||||
execution: input.execution,
|
||||
});
|
||||
this.#workflowEntries.set(entry.query_id, entry);
|
||||
}
|
||||
|
||||
/** Replay an exact provider-specific query, marking its cached provenance. */
|
||||
lookup(
|
||||
target: BinaryTarget,
|
||||
@@ -246,8 +302,14 @@ export class AnalysisSnapshotCache {
|
||||
readonly execution: AnalysisExecution;
|
||||
}): void {
|
||||
const { target, profile, operation, parameters, execution } = input;
|
||||
if (this.#entries.size >= 10_000) return;
|
||||
this.select(target, profile);
|
||||
const queryId = analysisQueryId(
|
||||
snapshotTarget(target),
|
||||
snapshotBinding(profile),
|
||||
operation,
|
||||
parameters,
|
||||
);
|
||||
if (!this.#entries.has(queryId) && !this.#hasCapacity()) return;
|
||||
const entry = createAnalysisSnapshotEntry({
|
||||
target: snapshotTarget(target),
|
||||
binding: snapshotBinding(profile),
|
||||
@@ -260,8 +322,20 @@ export class AnalysisSnapshotCache {
|
||||
|
||||
/** Forget all target-bound entries. */
|
||||
clear(): void {
|
||||
this.#entries.clear();
|
||||
this.#clearEntries();
|
||||
this.#target = undefined;
|
||||
this.#binding = undefined;
|
||||
}
|
||||
|
||||
#clearEntries(): void {
|
||||
this.#entries.clear();
|
||||
this.#workflowEntries.clear();
|
||||
}
|
||||
|
||||
#hasCapacity(): boolean {
|
||||
return (
|
||||
this.#entries.size + this.#workflowEntries.size <
|
||||
SNAPSHOT_CACHE_ENTRY_CEILING
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -30,7 +30,10 @@ import { BinarySessionRecords } from "./BinarySessionRecords.js";
|
||||
import { binarySessionStatus } from "./BinarySessionStatus.js";
|
||||
import {
|
||||
resolveSessionOpen,
|
||||
resolveSessionTarget,
|
||||
validateResolvedSessionTarget,
|
||||
type BinarySessionOpenOptions,
|
||||
type ResolvedSessionOpen,
|
||||
} from "./BinarySessionOpen.js";
|
||||
import {
|
||||
bindExecutionTarget,
|
||||
@@ -150,6 +153,27 @@ export class BinarySession
|
||||
return () => this.#availabilityListeners.delete(listener);
|
||||
}
|
||||
|
||||
/** Resolve a target and provider profile without creating a provider client. */
|
||||
previewTarget(
|
||||
target: BinaryTarget,
|
||||
options: BinarySessionOpenOptions = {},
|
||||
): Promise<Result<ResolvedSessionOpen, AnalysisError>> {
|
||||
if (isAborted(options.signal))
|
||||
return Promise.resolve(err(new AnalysisCancelledError("open_binary")));
|
||||
return resolveSessionTarget({
|
||||
router: this.#providerRouter,
|
||||
current: this.#active,
|
||||
target,
|
||||
options,
|
||||
stagedSnapshotMatches: (target, profile) =>
|
||||
this.matchesSnapshot(target, profile),
|
||||
}).then((resolved) =>
|
||||
isAborted(options.signal)
|
||||
? err(new AnalysisCancelledError("open_binary"))
|
||||
: resolved,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Open or switch targets after draining calls against the current target.
|
||||
* Returns the switch failure even if best-effort restoration also fails.
|
||||
@@ -157,18 +181,49 @@ export class BinarySession
|
||||
open(
|
||||
path: string,
|
||||
options: BinarySessionOpenOptions = {},
|
||||
): Promise<Result<BinaryTarget, AnalysisError>> {
|
||||
return this.#open(
|
||||
() =>
|
||||
resolveSessionOpen({
|
||||
router: this.#providerRouter,
|
||||
current: this.#active,
|
||||
path,
|
||||
options,
|
||||
stagedSnapshotMatches: (target, profile) =>
|
||||
this.matchesSnapshot(target, profile),
|
||||
}),
|
||||
options,
|
||||
);
|
||||
}
|
||||
|
||||
/** Open a previewed target without repeating parsing or provider profile discovery. */
|
||||
openResolvedTarget(
|
||||
resolved: Pick<ResolvedSessionOpen, "target" | "route">,
|
||||
options: Pick<BinarySessionOpenOptions, "signal" | "snapshot"> = {},
|
||||
): Promise<Result<BinaryTarget, AnalysisError>> {
|
||||
return this.#open(
|
||||
() =>
|
||||
Promise.resolve(
|
||||
validateResolvedSessionTarget({
|
||||
...resolved,
|
||||
current: this.#active,
|
||||
options,
|
||||
stagedSnapshotMatches: (target, profile) =>
|
||||
this.matchesSnapshot(target, profile),
|
||||
}),
|
||||
),
|
||||
options,
|
||||
);
|
||||
}
|
||||
|
||||
#open(
|
||||
resolve: () => Promise<Result<ResolvedSessionOpen, AnalysisError>>,
|
||||
options: Pick<BinarySessionOpenOptions, "signal" | "snapshot">,
|
||||
): Promise<Result<BinaryTarget, AnalysisError>> {
|
||||
return this.#serialize(async () => {
|
||||
if (isAborted(options.signal))
|
||||
return err(new AnalysisCancelledError("open_binary"));
|
||||
const resolved = await resolveSessionOpen({
|
||||
router: this.#providerRouter,
|
||||
current: this.#active,
|
||||
path,
|
||||
options,
|
||||
stagedSnapshotMatches: (target, profile) =>
|
||||
this.matchesSnapshot(target, profile),
|
||||
});
|
||||
const resolved = await resolve();
|
||||
if (!resolved.ok) return resolved;
|
||||
const { target, route, sameTarget } = resolved.value;
|
||||
const { profile, compatibility } = route;
|
||||
|
||||
@@ -46,11 +46,22 @@ interface ResolveSessionOpenInput {
|
||||
) => boolean;
|
||||
}
|
||||
|
||||
interface ResolveSessionTargetInput {
|
||||
readonly router: SessionProviderRouter;
|
||||
readonly current: CurrentOpenBinding | undefined;
|
||||
readonly target: BinaryTarget;
|
||||
readonly options: BinarySessionOpenOptions;
|
||||
readonly stagedSnapshotMatches: (
|
||||
target: BinaryTarget,
|
||||
profile: AnalysisProfileCommitment | null,
|
||||
) => boolean;
|
||||
}
|
||||
|
||||
/** Parse a target, resolve its provider route, and validate snapshot binding. */
|
||||
export const resolveSessionOpen = async (
|
||||
input: ResolveSessionOpenInput,
|
||||
): Promise<Result<ResolvedSessionOpen, AnalysisError>> => {
|
||||
const { router, current, path, options, stagedSnapshotMatches } = input;
|
||||
const { path, options } = input;
|
||||
const parsed = await parseBinaryTarget(
|
||||
path,
|
||||
process.cwd(),
|
||||
@@ -59,7 +70,14 @@ export const resolveSessionOpen = async (
|
||||
options.formatHint,
|
||||
);
|
||||
if (!parsed.ok) return parsed;
|
||||
const target = parsed.value;
|
||||
return resolveSessionTarget({ ...input, target: parsed.value });
|
||||
};
|
||||
|
||||
/** Resolve a known target's provider route and validate snapshot binding. */
|
||||
export const resolveSessionTarget = async (
|
||||
input: ResolveSessionTargetInput,
|
||||
): Promise<Result<ResolvedSessionOpen, AnalysisError>> => {
|
||||
const { router, current, target, options } = input;
|
||||
const sameTarget =
|
||||
current?.target.path === target.path &&
|
||||
snapshotMatchesTarget(snapshotTarget(current.target), target);
|
||||
@@ -68,7 +86,22 @@ export const resolveSessionOpen = async (
|
||||
? ({ ok: true, value: current.route } as const)
|
||||
: await router.resolve(target, options.providerId, options.signal);
|
||||
if (!resolvedRoute.ok) return resolvedRoute;
|
||||
const route = resolvedRoute.value;
|
||||
return validateResolvedSessionTarget({
|
||||
...input,
|
||||
route: resolvedRoute.value,
|
||||
});
|
||||
};
|
||||
|
||||
/** Revalidate a previewed route against the current session and snapshot binding. */
|
||||
export const validateResolvedSessionTarget = (
|
||||
input: Omit<ResolveSessionTargetInput, "router"> & {
|
||||
readonly route: SessionProviderRoute;
|
||||
},
|
||||
): Result<ResolvedSessionOpen, AnalysisError> => {
|
||||
const { current, target, route, options, stagedSnapshotMatches } = input;
|
||||
const sameTarget =
|
||||
current?.target.path === target.path &&
|
||||
snapshotMatchesTarget(snapshotTarget(current.target), target);
|
||||
const snapshotError = validateSnapshot({
|
||||
snapshot: options.snapshot,
|
||||
target,
|
||||
|
||||
@@ -160,6 +160,43 @@ export abstract class BinarySessionRecords {
|
||||
this.#emitSnapshotChanged();
|
||||
}
|
||||
|
||||
/** Retain one derived workflow result alongside its provider cache entries. */
|
||||
recordWorkflowSnapshot(input: {
|
||||
readonly operation: string;
|
||||
readonly parameters: Readonly<
|
||||
Record<string, import("../../domain/jsonValue.js").JsonValue>
|
||||
>;
|
||||
readonly execution: Parameters<
|
||||
AnalysisSnapshotCache["recordWorkflow"]
|
||||
>[0]["execution"];
|
||||
}): Result<null, EvidenceIntegrityError> {
|
||||
const active = this.activeAnalysisBinding();
|
||||
if (active === undefined || active.profile === null)
|
||||
return err(
|
||||
new EvidenceIntegrityError(
|
||||
"Workflow snapshot entries require an active concrete provider profile",
|
||||
),
|
||||
);
|
||||
try {
|
||||
this.#snapshot.recordWorkflow({
|
||||
target: active.target,
|
||||
profile: active.profile,
|
||||
...input,
|
||||
});
|
||||
this.#emitSnapshotChanged();
|
||||
return ok(null);
|
||||
} catch (cause: unknown) {
|
||||
return err(
|
||||
new EvidenceIntegrityError(
|
||||
cause instanceof Error
|
||||
? cause.message
|
||||
: "Workflow snapshot entry validation failed",
|
||||
{ cause },
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
protected invalidateSnapshot(): void {
|
||||
this.#snapshot.clear();
|
||||
this.#snapshotInvalidated = true;
|
||||
|
||||
Reference in new issue
Block a user