mirror of
https://github.com/morluto/rea.git
synced 2026-11-15 11:13:21 +00:00
fix(session): clear provider-inferred unavailability after recovery (#8518)
* fix(session): clear provider-inferred unavailability after recovery A provider adapter or process failure marked every routed operation of that provider unavailable, but a later success cleared only the operation that succeeded. Sibling operations, and composed tools built on them, kept advertising the stale failure until each was retried individually. Move runtime availability into one owner that records whether each entry was observed for the operation or inferred from another operation's provider-wide failure. A success clears its own failure and that provider's inferred entries while keeping failures observed on other operations or providers and declared refusals. Outcomes are ordered by dispatch to the provider client, so an older in-flight success no longer erases a newer failure while a request queued behind a failure still clears it. Fixes #8404 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(session): keep newer successes when older failures complete last Runtime availability kept one merged entry per operation and forgot successful dispatches. An older provider-wide failure that completed after a newer success re-inferred unavailability for siblings the newer success had already proved working, and a superseded refusal could survive its own later success. Record only the newest failure and newest success dispatch per operation and per provider, and project unavailability from them: an operation is unavailable while its own newest failure is newer than its own newest success, or, as inferred, while its provider's newest provider-wide failure is newer than any success from that provider. The projection no longer depends on reply completion order. The recovery tests now bound held-reply acknowledgements, release and join every held call in finally, and compare remediation with the owning formatters instead of copying their wording. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
35ae7232bf
commit
3f61303939
4 files changed
+744
-58
No files matched your search
@@ -49,6 +49,7 @@ import { InvestigationRecords } from "../investigation/InvestigationRecords.js";
|
||||
import { recordEvidenceSources } from "../investigation/recordEvidenceSources.js";
|
||||
import { BinarySessionRecords } from "./BinarySessionRecords.js";
|
||||
import { binarySessionStatus } from "./BinarySessionStatus.js";
|
||||
import { SessionRuntimeAvailability } from "./SessionRuntimeAvailability.js";
|
||||
import { isSourceEvidenceRetainable } from "./AnalysisSnapshotCache.js";
|
||||
import type { BinarySessionDetail } from "../../contracts/sessionStatusContract.js";
|
||||
import {
|
||||
@@ -197,10 +198,7 @@ export class BinarySession
|
||||
readonly #bundleResourceFactory: BundleResourceClientFactory | undefined;
|
||||
#activeBundleResources: ActiveBundleResourceSession | undefined;
|
||||
#pendingBundleResourceCleanup: ActiveBundleResourceSession[] = [];
|
||||
readonly #runtimeUnavailability = new Map<
|
||||
string,
|
||||
{ readonly reason: string }
|
||||
>();
|
||||
readonly #runtimeAvailability = new SessionRuntimeAvailability();
|
||||
readonly #availabilityListeners = new Set<() => void | Promise<void>>();
|
||||
|
||||
constructor(
|
||||
@@ -1252,7 +1250,7 @@ export class BinarySession
|
||||
}),
|
||||
route: this.#currentRoute(),
|
||||
router: this.#providerRouter,
|
||||
runtimeUnavailability: this.#runtimeUnavailability,
|
||||
runtimeUnavailability: this.#runtimeAvailability.entries,
|
||||
runId: run?.runId,
|
||||
runtimeLineageSnapshots: client?.runtimeLineageSnapshots?.() ?? [],
|
||||
requestActivitySnapshots: client?.requestActivitySnapshots?.() ?? [],
|
||||
@@ -1414,6 +1412,7 @@ export class BinarySession
|
||||
// A failed reply cannot prove that a dispatched metadata write rolled back.
|
||||
// Stop replay before dispatch, including reads overlapping the write.
|
||||
if (capability.effects.mutatesArtifact) this.invalidateSnapshot();
|
||||
const dispatch = this.#runtimeAvailability.dispatch();
|
||||
const call = active.client.execute(name, arguments_, options);
|
||||
const result = unrecordedExecution(await call);
|
||||
const profiled = bindExecutionTarget(
|
||||
@@ -1442,7 +1441,15 @@ export class BinarySession
|
||||
this.#active === active
|
||||
)
|
||||
this.#active.nativeSourceAdmitted = true;
|
||||
this.#observeRuntimeAvailability(name, execution);
|
||||
if (
|
||||
this.#runtimeAvailability.observe(
|
||||
dispatch,
|
||||
capability,
|
||||
active.route.capabilities.values(),
|
||||
execution,
|
||||
)
|
||||
)
|
||||
this.#emitAvailabilityChanged();
|
||||
// Source facts belong to the observation, even when this query cannot be
|
||||
// replayed (for example a Ghidra procedure read without a document selector).
|
||||
if (execution.ok && execution.value.sourceEvidence !== undefined) {
|
||||
@@ -1486,56 +1493,8 @@ export class BinarySession
|
||||
return execution;
|
||||
}
|
||||
|
||||
#observeRuntimeAvailability(
|
||||
operation: AnalysisOperation,
|
||||
result: Result<AnalysisExecution, AnalysisError>,
|
||||
): void {
|
||||
if (result.ok) {
|
||||
if (this.#markRuntimeAvailable(operation))
|
||||
this.#emitAvailabilityChanged();
|
||||
return;
|
||||
}
|
||||
if (result.error._tag === "AnalysisCapabilityUnavailableError") {
|
||||
if (this.#markRuntimeUnavailable(operation, result.error.message))
|
||||
this.#emitAvailabilityChanged();
|
||||
return;
|
||||
}
|
||||
if (
|
||||
[
|
||||
"ProviderAdapterError",
|
||||
"HopperProcessError",
|
||||
"HopperStartError",
|
||||
].includes(result.error._tag)
|
||||
) {
|
||||
const capabilities = this.#active?.route.capabilities;
|
||||
const providerId = capabilities?.get(operation)?.provider.id;
|
||||
let changed = false;
|
||||
for (const descriptor of capabilities?.values() ?? [])
|
||||
if (providerId !== undefined && descriptor.provider.id === providerId)
|
||||
changed =
|
||||
this.#markRuntimeUnavailable(
|
||||
descriptor.operation,
|
||||
`${result.error._tag}: ${result.error.message}`,
|
||||
) || changed;
|
||||
if (changed) this.#emitAvailabilityChanged();
|
||||
}
|
||||
}
|
||||
|
||||
#markRuntimeAvailable(operation: string): boolean {
|
||||
return this.#runtimeUnavailability.delete(operation);
|
||||
}
|
||||
|
||||
#markRuntimeUnavailable(operation: string, reason: string): boolean {
|
||||
const current = this.#runtimeUnavailability.get(operation);
|
||||
if (current?.reason === reason) return false;
|
||||
this.#runtimeUnavailability.set(operation, { reason });
|
||||
return true;
|
||||
}
|
||||
|
||||
#clearRuntimeAvailability(): void {
|
||||
if (this.#runtimeUnavailability.size === 0) return;
|
||||
this.#runtimeUnavailability.clear();
|
||||
this.#emitAvailabilityChanged();
|
||||
if (this.#runtimeAvailability.clear()) this.#emitAvailabilityChanged();
|
||||
}
|
||||
|
||||
#emitAvailabilityChanged(): void {
|
||||
|
||||
@@ -0,0 +1,169 @@
|
||||
import type { AnalysisError } from "../../domain/analysisErrorBase.js";
|
||||
import type { Result } from "../../domain/result.js";
|
||||
import type { CapabilityDescriptor } from "../AnalysisProvider.js";
|
||||
|
||||
/** Failures of the provider adapter or process rather than of one request. */
|
||||
const PROVIDER_WIDE_FAILURES: ReadonlySet<string> = new Set([
|
||||
"ProviderAdapterError",
|
||||
"HopperProcessError",
|
||||
"HopperStartError",
|
||||
]);
|
||||
|
||||
interface Failure {
|
||||
readonly dispatch: number;
|
||||
readonly reason: string;
|
||||
}
|
||||
|
||||
interface ProviderFailure extends Failure {
|
||||
readonly operation: string;
|
||||
}
|
||||
|
||||
/** Newest failure and newest success, both by dispatch, for one owner. */
|
||||
interface Outcomes<F extends Failure> {
|
||||
failure: F | undefined;
|
||||
succeeded: number;
|
||||
}
|
||||
|
||||
type RoutedOperation = Pick<
|
||||
CapabilityDescriptor,
|
||||
"operation" | "provider" | "available"
|
||||
>;
|
||||
|
||||
/** Reason recorded for an operation's own refusal or provider-wide failure. */
|
||||
export const runtimeFailureReason = (
|
||||
failure: AnalysisError,
|
||||
): string | undefined => {
|
||||
if (failure._tag === "AnalysisCapabilityUnavailableError")
|
||||
return failure.message;
|
||||
return PROVIDER_WIDE_FAILURES.has(failure._tag)
|
||||
? `${failure._tag}: ${failure.message}`
|
||||
: undefined;
|
||||
};
|
||||
|
||||
/** Explain unavailability inferred for an operation no request failed for. */
|
||||
export const inferredUnavailabilityReason = (
|
||||
providerId: string,
|
||||
failure: Pick<ProviderFailure, "operation" | "reason">,
|
||||
): string =>
|
||||
`Inferred from the ${failure.operation} failure (${failure.reason}). Retry after repairing the provider; a later successful ${providerId} request clears this.`;
|
||||
|
||||
/**
|
||||
* Own the runtime unavailability overlaid on the active route's declared
|
||||
* capabilities.
|
||||
*
|
||||
* An operation is unavailable while its newest refusal or provider-wide
|
||||
* failure was dispatched after its newest success. A declared-available
|
||||
* operation is also unavailable, as inferred, while its provider's newest
|
||||
* provider-wide failure was dispatched after any success from that provider.
|
||||
* One success therefore returns inferred siblings to their declared
|
||||
* availability without proving each is healthy, and keeps failures observed
|
||||
* for other operations.
|
||||
*
|
||||
* Only the newest dispatch of each outcome matters, so the projection does not
|
||||
* depend on the order in which overlapping replies complete. Dispatch is the
|
||||
* latest order this session owns; it does not establish how a provider
|
||||
* interleaves overlapping requests.
|
||||
*/
|
||||
export class SessionRuntimeAvailability {
|
||||
readonly #operations = new Map<string, Outcomes<Failure>>();
|
||||
readonly #providers = new Map<string, Outcomes<ProviderFailure>>();
|
||||
#projected: ReadonlyMap<string, { readonly reason: string }> = new Map();
|
||||
#dispatched = 0;
|
||||
|
||||
/** Current runtime unavailability keyed by operation. */
|
||||
get entries(): ReadonlyMap<string, { readonly reason: string }> {
|
||||
return this.#projected;
|
||||
}
|
||||
|
||||
/** Order one request immediately before it reaches the provider client. */
|
||||
dispatch(): number {
|
||||
this.#dispatched += 1;
|
||||
return this.#dispatched;
|
||||
}
|
||||
|
||||
/**
|
||||
* Record one dispatched request's outcome against the route it ran on and
|
||||
* report whether visible availability changed.
|
||||
*/
|
||||
observe(
|
||||
dispatch: number,
|
||||
capability: RoutedOperation,
|
||||
route: Iterable<RoutedOperation>,
|
||||
result: Result<unknown, AnalysisError>,
|
||||
): boolean {
|
||||
const operation = outcomes(this.#operations, capability.operation);
|
||||
const provider = outcomes(this.#providers, capability.provider.id);
|
||||
if (result.ok) {
|
||||
operation.succeeded = Math.max(operation.succeeded, dispatch);
|
||||
provider.succeeded = Math.max(provider.succeeded, dispatch);
|
||||
// A success can only remove unavailability.
|
||||
if (this.#projected.size === 0) return false;
|
||||
} else {
|
||||
const reason = runtimeFailureReason(result.error);
|
||||
if (reason === undefined) return false;
|
||||
operation.failure = newer(operation.failure, { dispatch, reason });
|
||||
if (PROVIDER_WIDE_FAILURES.has(result.error._tag))
|
||||
provider.failure = newer(provider.failure, {
|
||||
dispatch,
|
||||
reason,
|
||||
operation: capability.operation,
|
||||
});
|
||||
}
|
||||
return this.#project(route);
|
||||
}
|
||||
|
||||
/** Forget runtime outcomes when the active provider client is replaced. */
|
||||
clear(): boolean {
|
||||
this.#operations.clear();
|
||||
this.#providers.clear();
|
||||
if (this.#projected.size === 0) return false;
|
||||
this.#projected = new Map();
|
||||
return true;
|
||||
}
|
||||
|
||||
#project(route: Iterable<RoutedOperation>): boolean {
|
||||
const projected = new Map<string, { readonly reason: string }>();
|
||||
for (const descriptor of route) {
|
||||
const reason = this.#unavailableReason(descriptor);
|
||||
if (reason !== undefined) projected.set(descriptor.operation, { reason });
|
||||
}
|
||||
const changed =
|
||||
projected.size !== this.#projected.size ||
|
||||
[...projected].some(
|
||||
([operation, { reason }]) =>
|
||||
this.#projected.get(operation)?.reason !== reason,
|
||||
);
|
||||
this.#projected = projected;
|
||||
return changed;
|
||||
}
|
||||
|
||||
#unavailableReason(descriptor: RoutedOperation): string | undefined {
|
||||
const operation = this.#operations.get(descriptor.operation);
|
||||
if (
|
||||
operation?.failure !== undefined &&
|
||||
operation.failure.dispatch > operation.succeeded
|
||||
)
|
||||
return operation.failure.reason;
|
||||
// A declared refusal already explains its operation and is never dispatched.
|
||||
if (!descriptor.available) return undefined;
|
||||
const provider = this.#providers.get(descriptor.provider.id);
|
||||
return provider?.failure !== undefined &&
|
||||
provider.failure.dispatch > provider.succeeded
|
||||
? inferredUnavailabilityReason(descriptor.provider.id, provider.failure)
|
||||
: undefined;
|
||||
}
|
||||
}
|
||||
|
||||
const outcomes = <F extends Failure>(
|
||||
owners: Map<string, Outcomes<F>>,
|
||||
key: string,
|
||||
): Outcomes<F> => {
|
||||
const existing = owners.get(key);
|
||||
if (existing !== undefined) return existing;
|
||||
const created: Outcomes<F> = { failure: undefined, succeeded: 0 };
|
||||
owners.set(key, created);
|
||||
return created;
|
||||
};
|
||||
|
||||
const newer = <F extends Failure>(current: F | undefined, next: F): F =>
|
||||
current !== undefined && current.dispatch > next.dispatch ? current : next;
|
||||
Reference in new issue
Block a user