mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-19 03:21:38 +00:00
refactor(gateway): centralize agent wait lifecycle (#104147)
* refactor(gateway): centralize agent wait lifecycle * fix(gateway): preserve agent wait freshness --------- Co-authored-by: Ayaan Zaidi <hi@obviy.us>
This commit is contained in:
Ayaan Zaidi
@@ -1,105 +1,154 @@
|
||||
// Agent job tracking caches terminal run snapshots so `agent.wait` can observe
|
||||
// recent run outcomes even after the live event stream has moved on.
|
||||
// Agent job tracking owns terminal run state and `agent.wait` resolution.
|
||||
// Gateway dedupe retains response payloads only for idempotent RPC replay.
|
||||
import { asFiniteNumber } from "@openclaw/normalization-core/number-coercion";
|
||||
import { asOptionalRecord } from "@openclaw/normalization-core/record-coerce";
|
||||
import {
|
||||
AGENT_RUN_TERMINAL_RETRY_GRACE_MS,
|
||||
buildAgentRunTerminalOutcome,
|
||||
isStickyAgentRunTerminalOutcome,
|
||||
mergeAgentRunTerminalOutcome,
|
||||
type AgentRunTerminalOutcome,
|
||||
} from "../../agents/agent-run-terminal-outcome.js";
|
||||
import { onAgentEvent } from "../../infra/agent-events.js";
|
||||
import { isNonTerminalAgentRunStatus } from "../../shared/agent-run-status.js";
|
||||
import { setSafeTimeout } from "../../utils/timer-delay.js";
|
||||
import type { AgentWaitTerminalSnapshot } from "./agent-wait-dedupe.js";
|
||||
import type { DedupeEntry } from "../server-shared.js";
|
||||
|
||||
const AGENT_RUN_CACHE_TTL_MS = 10 * 60_000;
|
||||
const AGENT_RUN_CACHE_MAX_ENTRIES = 5_000;
|
||||
const agentRunCache = new Map<string, AgentRunSnapshot>();
|
||||
|
||||
export type AgentJobTerminalSnapshot = {
|
||||
status: "ok" | "error" | "timeout";
|
||||
startedAt?: number;
|
||||
endedAt?: number;
|
||||
error?: string;
|
||||
stopReason?: string;
|
||||
livenessState?: string;
|
||||
yielded?: boolean;
|
||||
pendingError?: boolean;
|
||||
timeoutPhase?: AgentRunTerminalOutcome["timeoutPhase"];
|
||||
providerStarted?: boolean;
|
||||
};
|
||||
|
||||
type AgentJobSource = "agent" | "chat" | "lifecycle";
|
||||
type AgentRunObservation = AgentJobTerminalSnapshot & {
|
||||
runId: string;
|
||||
source: AgentJobSource;
|
||||
recordedAt: number;
|
||||
version: number;
|
||||
};
|
||||
type AgentRunSnapshot = AgentRunObservation & { cachedAt: number };
|
||||
type PendingAgentRunTerminal = {
|
||||
snapshot: AgentRunObservation;
|
||||
timer: NodeJS.Timeout;
|
||||
};
|
||||
type AgentJobRecord = {
|
||||
cachedAt: number;
|
||||
snapshotsBySource: Map<AgentJobSource, AgentRunSnapshot>;
|
||||
};
|
||||
type AgentJobWaiter = () => void;
|
||||
type DedupeObservation =
|
||||
| { state: "active" }
|
||||
| { state: "terminal"; snapshot: AgentJobTerminalSnapshot }
|
||||
| { state: "untracked" };
|
||||
|
||||
const agentJobs = new Map<string, AgentJobRecord>();
|
||||
const agentRunStarts = new Map<string, number>();
|
||||
const pendingAgentRunErrors = new Map<string, PendingAgentRunTerminal>();
|
||||
const pendingAgentRunTimeouts = new Map<string, PendingAgentRunTerminal>();
|
||||
const agentRunWaiterCounts = new Map<string, number>();
|
||||
const agentRunWaiters = new Map<string, Set<AgentJobWaiter>>();
|
||||
let agentRunListenerStarted = false;
|
||||
let agentRunVersion = 0;
|
||||
|
||||
type AgentRunSnapshot = AgentWaitTerminalSnapshot & {
|
||||
runId: string;
|
||||
ts: number;
|
||||
};
|
||||
|
||||
type PendingAgentRunTerminal = {
|
||||
snapshot: AgentRunSnapshot;
|
||||
dueAt: number;
|
||||
timer: NodeJS.Timeout;
|
||||
};
|
||||
|
||||
function pruneAgentRunCache(now = Date.now()) {
|
||||
for (const [runId, entry] of agentRunCache) {
|
||||
if (now - entry.ts > AGENT_RUN_CACHE_TTL_MS) {
|
||||
agentRunCache.delete(runId);
|
||||
}
|
||||
}
|
||||
function nextAgentRunVersion(): number {
|
||||
agentRunVersion += 1;
|
||||
return agentRunVersion;
|
||||
}
|
||||
|
||||
function recordAgentRunSnapshot(entry: AgentRunSnapshot) {
|
||||
pruneAgentRunCache(entry.ts);
|
||||
const existing = agentRunCache.get(entry.runId);
|
||||
if (existing && shouldPreserveTerminalSnapshot(existing, entry)) {
|
||||
agentRunCache.set(entry.runId, {
|
||||
...existing,
|
||||
ts: entry.ts,
|
||||
});
|
||||
return;
|
||||
function pruneAgentRunCache(now = Date.now()) {
|
||||
for (const [runId, job] of agentJobs) {
|
||||
if (now - job.cachedAt <= AGENT_RUN_CACHE_TTL_MS) {
|
||||
continue;
|
||||
}
|
||||
agentJobs.delete(runId);
|
||||
}
|
||||
agentRunCache.set(entry.runId, entry);
|
||||
// Time-based prune only fires on the TTL window; under high run fan-out a
|
||||
// burst can add far more entries than the window reclaims. Cap with a FIFO
|
||||
// drop so the cache cannot grow without bound between prunes.
|
||||
enforceAgentRunCacheMaxEntries();
|
||||
}
|
||||
|
||||
function enforceAgentRunCacheMaxEntries() {
|
||||
if (agentRunCache.size <= AGENT_RUN_CACHE_MAX_ENTRIES) {
|
||||
if (agentJobs.size <= AGENT_RUN_CACHE_MAX_ENTRIES) {
|
||||
return;
|
||||
}
|
||||
const toRemove = agentRunCache.size - AGENT_RUN_CACHE_MAX_ENTRIES;
|
||||
const toRemove = agentJobs.size - AGENT_RUN_CACHE_MAX_ENTRIES;
|
||||
let removed = 0;
|
||||
for (const runId of agentRunCache.keys()) {
|
||||
for (const runId of agentJobs.keys()) {
|
||||
if (removed >= toRemove) {
|
||||
break;
|
||||
}
|
||||
if ((agentRunWaiterCounts.get(runId) ?? 0) > 0) {
|
||||
if ((agentRunWaiters.get(runId)?.size ?? 0) > 0) {
|
||||
continue;
|
||||
}
|
||||
agentRunCache.delete(runId);
|
||||
agentJobs.delete(runId);
|
||||
removed += 1;
|
||||
}
|
||||
}
|
||||
|
||||
function shouldPreserveTerminalSnapshot(
|
||||
existing: AgentRunSnapshot,
|
||||
incoming: AgentRunSnapshot,
|
||||
): boolean {
|
||||
const existingOutcome = terminalOutcomeFromSnapshot(existing);
|
||||
const incomingOutcome = terminalOutcomeFromSnapshot(incoming);
|
||||
if (!existingOutcome) {
|
||||
return false;
|
||||
}
|
||||
if (!incomingOutcome) {
|
||||
return false;
|
||||
}
|
||||
const terminalOutcome = mergeAgentRunTerminalOutcome(existingOutcome, incomingOutcome);
|
||||
return terminalOutcome === existingOutcome;
|
||||
}
|
||||
|
||||
function terminalOutcomeFromSnapshot(
|
||||
snapshot: AgentRunSnapshot,
|
||||
snapshot: AgentJobTerminalSnapshot,
|
||||
): AgentRunTerminalOutcome | undefined {
|
||||
if (snapshot.pendingError) {
|
||||
// Pending errors are still inside retry grace; a lifecycle start can cancel
|
||||
// them, so they must not participate in sticky terminal precedence.
|
||||
return undefined;
|
||||
}
|
||||
return buildAgentRunTerminalOutcome(snapshot);
|
||||
}
|
||||
|
||||
function shouldPreserveTerminalSnapshot(
|
||||
existing: AgentJobTerminalSnapshot,
|
||||
incoming: AgentJobTerminalSnapshot,
|
||||
): boolean {
|
||||
const existingOutcome = terminalOutcomeFromSnapshot(existing);
|
||||
const incomingOutcome = terminalOutcomeFromSnapshot(incoming);
|
||||
if (!existingOutcome || !incomingOutcome) {
|
||||
return false;
|
||||
}
|
||||
return mergeAgentRunTerminalOutcome(existingOutcome, incomingOutcome) === existingOutcome;
|
||||
}
|
||||
|
||||
function mergeSnapshot(
|
||||
existing: AgentRunSnapshot | undefined,
|
||||
incoming: AgentRunSnapshot,
|
||||
): AgentRunSnapshot {
|
||||
if (!existing || !shouldPreserveTerminalSnapshot(existing, incoming)) {
|
||||
return incoming;
|
||||
}
|
||||
return { ...existing, cachedAt: incoming.cachedAt };
|
||||
}
|
||||
|
||||
function notifyAgentRunWaiters(runId: string) {
|
||||
for (const waiter of agentRunWaiters.get(runId) ?? []) {
|
||||
waiter();
|
||||
}
|
||||
}
|
||||
|
||||
function recordAgentRunSnapshot(
|
||||
snapshot: Omit<AgentRunObservation, "version">,
|
||||
version = nextAgentRunVersion(),
|
||||
) {
|
||||
const entry = { ...snapshot, cachedAt: Date.now(), version };
|
||||
pruneAgentRunCache(entry.cachedAt);
|
||||
|
||||
const existing = agentJobs.get(entry.runId);
|
||||
const snapshotsBySource =
|
||||
existing?.snapshotsBySource ?? new Map<AgentJobSource, AgentRunSnapshot>();
|
||||
const sourceSnapshot = mergeSnapshot(snapshotsBySource.get(entry.source), entry);
|
||||
snapshotsBySource.set(entry.source, sourceSnapshot);
|
||||
agentJobs.set(entry.runId, {
|
||||
cachedAt: entry.cachedAt,
|
||||
snapshotsBySource,
|
||||
});
|
||||
enforceAgentRunCacheMaxEntries();
|
||||
notifyAgentRunWaiters(entry.runId);
|
||||
}
|
||||
|
||||
function clearPendingAgentRunError(runId: string) {
|
||||
const pending = pendingAgentRunErrors.get(runId);
|
||||
if (!pending) {
|
||||
@@ -118,77 +167,63 @@ function clearPendingAgentRunTimeout(runId: string) {
|
||||
pendingAgentRunTimeouts.delete(runId);
|
||||
}
|
||||
|
||||
function schedulePendingAgentRunError(snapshot: AgentRunSnapshot) {
|
||||
const pendingTimeout = pendingAgentRunTimeouts.get(snapshot.runId);
|
||||
if (pendingTimeout && shouldPreserveTerminalSnapshot(pendingTimeout.snapshot, snapshot)) {
|
||||
// A late rejection can race in before the timeout grace publishes. Keep the
|
||||
// pending hard timeout so waiters observe the original terminal cause.
|
||||
function beginAgentJob(runId: string, startedAt?: number) {
|
||||
nextAgentRunVersion();
|
||||
clearPendingAgentRunError(runId);
|
||||
clearPendingAgentRunTimeout(runId);
|
||||
agentJobs.delete(runId);
|
||||
if (startedAt !== undefined) {
|
||||
agentRunStarts.set(runId, startedAt);
|
||||
}
|
||||
}
|
||||
|
||||
function schedulePendingAgentRunTerminal(
|
||||
pendingRuns: Map<string, PendingAgentRunTerminal>,
|
||||
snapshot: AgentRunObservation,
|
||||
) {
|
||||
const existing = pendingRuns.get(snapshot.runId);
|
||||
if (existing && shouldPreserveTerminalSnapshot(existing.snapshot, snapshot)) {
|
||||
return;
|
||||
}
|
||||
clearPendingAgentRunTimeout(snapshot.runId);
|
||||
clearPendingAgentRunError(snapshot.runId);
|
||||
const dueAt = Date.now() + AGENT_RUN_TERMINAL_RETRY_GRACE_MS;
|
||||
const timer = setTimeout(() => {
|
||||
const pending = pendingAgentRunErrors.get(snapshot.runId);
|
||||
if (pendingRuns === pendingAgentRunErrors) {
|
||||
clearPendingAgentRunError(snapshot.runId);
|
||||
} else {
|
||||
clearPendingAgentRunTimeout(snapshot.runId);
|
||||
}
|
||||
const timer = setSafeTimeout(() => {
|
||||
const pending = pendingRuns.get(snapshot.runId);
|
||||
if (!pending) {
|
||||
return;
|
||||
}
|
||||
pendingAgentRunErrors.delete(snapshot.runId);
|
||||
recordAgentRunSnapshot(pending.snapshot);
|
||||
pendingRuns.delete(snapshot.runId);
|
||||
recordAgentRunSnapshot(pending.snapshot, pending.snapshot.version);
|
||||
}, AGENT_RUN_TERMINAL_RETRY_GRACE_MS);
|
||||
timer.unref?.();
|
||||
pendingAgentRunErrors.set(snapshot.runId, { snapshot, dueAt, timer });
|
||||
pendingRuns.set(snapshot.runId, { snapshot, timer });
|
||||
}
|
||||
|
||||
function schedulePendingAgentRunTimeout(snapshot: AgentRunSnapshot) {
|
||||
function schedulePendingAgentRunError(snapshot: AgentRunObservation) {
|
||||
const pendingTimeout = pendingAgentRunTimeouts.get(snapshot.runId);
|
||||
if (pendingTimeout && shouldPreserveTerminalSnapshot(pendingTimeout.snapshot, snapshot)) {
|
||||
return;
|
||||
}
|
||||
clearPendingAgentRunTimeout(snapshot.runId);
|
||||
schedulePendingAgentRunTerminal(pendingAgentRunErrors, snapshot);
|
||||
}
|
||||
|
||||
function schedulePendingAgentRunTimeout(snapshot: AgentRunObservation) {
|
||||
const pendingTimeout = pendingAgentRunTimeouts.get(snapshot.runId);
|
||||
if (pendingTimeout && shouldPreserveTerminalSnapshot(pendingTimeout.snapshot, snapshot)) {
|
||||
// Keep the first hard timeout through retry grace; later timeout-shaped
|
||||
// cleanup events may lose provider attribution before the cache publishes.
|
||||
return;
|
||||
}
|
||||
clearPendingAgentRunError(snapshot.runId);
|
||||
clearPendingAgentRunTimeout(snapshot.runId);
|
||||
const dueAt = Date.now() + AGENT_RUN_TERMINAL_RETRY_GRACE_MS;
|
||||
const timer = setTimeout(() => {
|
||||
const pending = pendingAgentRunTimeouts.get(snapshot.runId);
|
||||
if (!pending) {
|
||||
return;
|
||||
}
|
||||
pendingAgentRunTimeouts.delete(snapshot.runId);
|
||||
recordAgentRunSnapshot(pending.snapshot);
|
||||
}, AGENT_RUN_TERMINAL_RETRY_GRACE_MS);
|
||||
timer.unref?.();
|
||||
pendingAgentRunTimeouts.set(snapshot.runId, { snapshot, dueAt, timer });
|
||||
schedulePendingAgentRunTerminal(pendingAgentRunTimeouts, snapshot);
|
||||
}
|
||||
|
||||
function getPendingAgentRunError(runId: string) {
|
||||
const pending = pendingAgentRunErrors.get(runId);
|
||||
if (!pending) {
|
||||
return undefined;
|
||||
}
|
||||
function createPendingErrorTimeoutSnapshot(
|
||||
snapshot: AgentJobTerminalSnapshot,
|
||||
): AgentJobTerminalSnapshot {
|
||||
return {
|
||||
snapshot: pending.snapshot,
|
||||
dueAt: pending.dueAt,
|
||||
};
|
||||
}
|
||||
|
||||
function getPendingAgentRunTimeout(runId: string) {
|
||||
const pending = pendingAgentRunTimeouts.get(runId);
|
||||
if (!pending) {
|
||||
return undefined;
|
||||
}
|
||||
return {
|
||||
snapshot: pending.snapshot,
|
||||
dueAt: pending.dueAt,
|
||||
};
|
||||
}
|
||||
|
||||
function createPendingErrorTimeoutSnapshot(snapshot: AgentRunSnapshot): AgentRunSnapshot {
|
||||
// Keep this non-terminal: the retry grace can still be canceled by a later
|
||||
// lifecycle start, so omit terminal fields such as endedAt and stopReason.
|
||||
return {
|
||||
runId: snapshot.runId,
|
||||
status: "timeout",
|
||||
startedAt: snapshot.startedAt,
|
||||
error: snapshot.error,
|
||||
@@ -196,7 +231,6 @@ function createPendingErrorTimeoutSnapshot(snapshot: AgentRunSnapshot): AgentRun
|
||||
...(snapshot.providerStarted !== undefined
|
||||
? { providerStarted: snapshot.providerStarted }
|
||||
: {}),
|
||||
ts: Date.now(),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -204,7 +238,7 @@ function createSnapshotFromLifecycleEvent(params: {
|
||||
runId: string;
|
||||
phase: "end" | "error";
|
||||
data?: Record<string, unknown>;
|
||||
}): AgentRunSnapshot {
|
||||
}): AgentRunObservation {
|
||||
const { runId, phase, data } = params;
|
||||
const startedAt =
|
||||
typeof data?.startedAt === "number" ? data.startedAt : agentRunStarts.get(runId);
|
||||
@@ -212,9 +246,8 @@ function createSnapshotFromLifecycleEvent(params: {
|
||||
const error = typeof data?.error === "string" ? data.error : undefined;
|
||||
const stopReason = typeof data?.stopReason === "string" ? data.stopReason : undefined;
|
||||
const livenessState = typeof data?.livenessState === "string" ? data.livenessState : undefined;
|
||||
const status = phase === "error" ? "error" : data?.aborted ? "timeout" : "ok";
|
||||
const terminalOutcome = buildAgentRunTerminalOutcome({
|
||||
status,
|
||||
status: phase === "error" ? "error" : data?.aborted ? "timeout" : "ok",
|
||||
error,
|
||||
stopReason,
|
||||
livenessState,
|
||||
@@ -225,6 +258,8 @@ function createSnapshotFromLifecycleEvent(params: {
|
||||
});
|
||||
return {
|
||||
runId,
|
||||
source: "lifecycle",
|
||||
recordedAt: Date.now(),
|
||||
status: terminalOutcome.status,
|
||||
startedAt,
|
||||
endedAt,
|
||||
@@ -236,7 +271,7 @@ function createSnapshotFromLifecycleEvent(params: {
|
||||
...(terminalOutcome.providerStarted !== undefined
|
||||
? { providerStarted: terminalOutcome.providerStarted }
|
||||
: {}),
|
||||
ts: Date.now(),
|
||||
version: nextAgentRunVersion(),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -246,21 +281,13 @@ function ensureAgentRunListener() {
|
||||
}
|
||||
agentRunListenerStarted = true;
|
||||
onAgentEvent((evt) => {
|
||||
if (!evt) {
|
||||
return;
|
||||
}
|
||||
if (evt.stream !== "lifecycle") {
|
||||
if (!evt || evt.stream !== "lifecycle") {
|
||||
return;
|
||||
}
|
||||
const phase = evt.data?.phase;
|
||||
if (phase === "start") {
|
||||
const startedAt = typeof evt.data?.startedAt === "number" ? evt.data.startedAt : undefined;
|
||||
agentRunStarts.set(evt.runId, startedAt ?? Date.now());
|
||||
clearPendingAgentRunError(evt.runId);
|
||||
clearPendingAgentRunTimeout(evt.runId);
|
||||
// A new start means this run is active again (or retried). Drop stale
|
||||
// terminal snapshots so waiters don't resolve from old state.
|
||||
agentRunCache.delete(evt.runId);
|
||||
const startedAt = typeof evt.data?.startedAt === "number" ? evt.data.startedAt : Date.now();
|
||||
beginAgentJob(evt.runId, startedAt);
|
||||
return;
|
||||
}
|
||||
if (phase !== "end" && phase !== "error") {
|
||||
@@ -286,228 +313,270 @@ function ensureAgentRunListener() {
|
||||
}
|
||||
clearPendingAgentRunError(evt.runId);
|
||||
clearPendingAgentRunTimeout(evt.runId);
|
||||
recordAgentRunSnapshot(snapshot);
|
||||
recordAgentRunSnapshot(snapshot, snapshot.version);
|
||||
});
|
||||
}
|
||||
|
||||
function getCachedAgentRun(runId: string) {
|
||||
pruneAgentRunCache();
|
||||
return agentRunCache.get(runId);
|
||||
function asString(value: unknown): string | undefined {
|
||||
return typeof value === "string" && value.trim() ? value : undefined;
|
||||
}
|
||||
|
||||
function addAgentRunWaiter(runId: string): () => void {
|
||||
agentRunWaiterCounts.set(runId, (agentRunWaiterCounts.get(runId) ?? 0) + 1);
|
||||
let removed = false;
|
||||
function parseDedupeObservation(entry: DedupeEntry): DedupeObservation {
|
||||
const payload = entry.payload as
|
||||
| {
|
||||
status?: unknown;
|
||||
startedAt?: unknown;
|
||||
endedAt?: unknown;
|
||||
error?: unknown;
|
||||
summary?: unknown;
|
||||
stopReason?: unknown;
|
||||
livenessState?: unknown;
|
||||
yielded?: unknown;
|
||||
timeoutPhase?: unknown;
|
||||
providerStarted?: unknown;
|
||||
result?: unknown;
|
||||
}
|
||||
| undefined;
|
||||
const status = typeof payload?.status === "string" ? payload.status : undefined;
|
||||
if (isNonTerminalAgentRunStatus(status)) {
|
||||
return { state: "active" };
|
||||
}
|
||||
|
||||
const terminalStatus =
|
||||
status === "ok" || status === "timeout" || status === "error"
|
||||
? status
|
||||
: entry.ok
|
||||
? undefined
|
||||
: "error";
|
||||
if (!terminalStatus) {
|
||||
return { state: "untracked" };
|
||||
}
|
||||
|
||||
const resultMeta = asOptionalRecord(asOptionalRecord(payload?.result)?.meta);
|
||||
const startedAt = asFiniteNumber(payload?.startedAt);
|
||||
const endedAt = asFiniteNumber(payload?.endedAt) ?? entry.ts;
|
||||
const stopReason = asString(payload?.stopReason) ?? asString(resultMeta?.stopReason);
|
||||
const livenessState = asString(payload?.livenessState) ?? asString(resultMeta?.livenessState);
|
||||
const terminalOutcome = buildAgentRunTerminalOutcome({
|
||||
status: terminalStatus,
|
||||
startedAt,
|
||||
endedAt,
|
||||
error:
|
||||
typeof payload?.error === "string"
|
||||
? payload.error
|
||||
: typeof payload?.summary === "string"
|
||||
? payload.summary
|
||||
: entry.error?.message,
|
||||
stopReason,
|
||||
livenessState,
|
||||
timeoutPhase: payload?.timeoutPhase ?? resultMeta?.timeoutPhase,
|
||||
providerStarted: payload?.providerStarted ?? resultMeta?.providerStarted,
|
||||
});
|
||||
return {
|
||||
state: "terminal",
|
||||
snapshot: {
|
||||
status: terminalOutcome.status,
|
||||
startedAt,
|
||||
endedAt,
|
||||
error: terminalOutcome.status === "ok" ? undefined : terminalOutcome.error,
|
||||
stopReason,
|
||||
livenessState,
|
||||
...(payload?.yielded === true || resultMeta?.yielded === true ? { yielded: true } : {}),
|
||||
...(terminalOutcome.timeoutPhase ? { timeoutPhase: terminalOutcome.timeoutPhase } : {}),
|
||||
...(terminalOutcome.providerStarted !== undefined
|
||||
? { providerStarted: terminalOutcome.providerStarted }
|
||||
: {}),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function parseDedupeKey(key: string): { runId: string; source: "agent" | "chat" } | undefined {
|
||||
const separator = key.indexOf(":");
|
||||
if (separator === -1) {
|
||||
return undefined;
|
||||
}
|
||||
const source = key.slice(0, separator);
|
||||
const runId = key.slice(separator + 1);
|
||||
if ((source !== "agent" && source !== "chat") || !runId) {
|
||||
return undefined;
|
||||
}
|
||||
return { runId, source };
|
||||
}
|
||||
|
||||
export function setGatewayDedupeEntry(params: {
|
||||
dedupe: Map<string, DedupeEntry>;
|
||||
key: string;
|
||||
entry: DedupeEntry;
|
||||
}) {
|
||||
const existing = params.dedupe.get(params.key);
|
||||
const existingObservation = existing ? parseDedupeObservation(existing) : undefined;
|
||||
const incomingObservation = parseDedupeObservation(params.entry);
|
||||
const existingOutcome =
|
||||
existingObservation?.state === "terminal"
|
||||
? terminalOutcomeFromSnapshot(existingObservation.snapshot)
|
||||
: undefined;
|
||||
const incomingOutcome =
|
||||
incomingObservation.state === "terminal"
|
||||
? terminalOutcomeFromSnapshot(incomingObservation.snapshot)
|
||||
: undefined;
|
||||
if (existingOutcome && isStickyAgentRunTerminalOutcome(existingOutcome) && !incomingOutcome) {
|
||||
return;
|
||||
}
|
||||
if (existingOutcome && incomingOutcome && isStickyAgentRunTerminalOutcome(existingOutcome)) {
|
||||
if (mergeAgentRunTerminalOutcome(existingOutcome, incomingOutcome) === existingOutcome) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
params.dedupe.set(params.key, params.entry);
|
||||
const key = parseDedupeKey(params.key);
|
||||
if (!key) {
|
||||
return;
|
||||
}
|
||||
if (incomingObservation.state === "active") {
|
||||
beginAgentJob(key.runId);
|
||||
return;
|
||||
}
|
||||
if (incomingObservation.state === "terminal") {
|
||||
recordAgentRunSnapshot({
|
||||
...incomingObservation.snapshot,
|
||||
runId: key.runId,
|
||||
source: key.source,
|
||||
recordedAt: params.entry.ts,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
function getFreshestDedupeSnapshot(
|
||||
snapshotsBySource: Map<AgentJobSource, AgentRunSnapshot>,
|
||||
): AgentRunSnapshot | undefined {
|
||||
const agent = snapshotsBySource.get("agent");
|
||||
const chat = snapshotsBySource.get("chat");
|
||||
if (agent && chat) {
|
||||
return chat.recordedAt > agent.recordedAt ? chat : agent;
|
||||
}
|
||||
return agent ?? chat;
|
||||
}
|
||||
|
||||
function getCanonicalAgentRunSnapshot(
|
||||
snapshotsBySource: Map<AgentJobSource, AgentRunSnapshot>,
|
||||
): AgentRunSnapshot | undefined {
|
||||
const dedupe = getFreshestDedupeSnapshot(snapshotsBySource);
|
||||
const lifecycle = snapshotsBySource.get("lifecycle");
|
||||
if (!dedupe || !lifecycle) {
|
||||
return dedupe ?? lifecycle;
|
||||
}
|
||||
return dedupe.version > lifecycle.version
|
||||
? mergeSnapshot(lifecycle, dedupe)
|
||||
: mergeSnapshot(dedupe, lifecycle);
|
||||
}
|
||||
|
||||
function getAgentRunSnapshot(params: {
|
||||
runId: string;
|
||||
source?: "chat";
|
||||
afterVersion: number;
|
||||
}): AgentRunSnapshot | undefined {
|
||||
pruneAgentRunCache();
|
||||
const job = agentJobs.get(params.runId);
|
||||
const snapshot = params.source
|
||||
? job?.snapshotsBySource.get(params.source)
|
||||
: job
|
||||
? getCanonicalAgentRunSnapshot(job.snapshotsBySource)
|
||||
: undefined;
|
||||
return snapshot && snapshot.version > params.afterVersion ? snapshot : undefined;
|
||||
}
|
||||
|
||||
function addAgentRunWaiter(runId: string, waiter: AgentJobWaiter): () => void {
|
||||
const waiters = agentRunWaiters.get(runId) ?? new Set<AgentJobWaiter>();
|
||||
waiters.add(waiter);
|
||||
agentRunWaiters.set(runId, waiters);
|
||||
return () => {
|
||||
if (removed) {
|
||||
return;
|
||||
waiters.delete(waiter);
|
||||
if (waiters.size === 0) {
|
||||
agentRunWaiters.delete(runId);
|
||||
}
|
||||
removed = true;
|
||||
const nextCount = (agentRunWaiterCounts.get(runId) ?? 1) - 1;
|
||||
if (nextCount <= 0) {
|
||||
agentRunWaiterCounts.delete(runId);
|
||||
return;
|
||||
}
|
||||
agentRunWaiterCounts.set(runId, nextCount);
|
||||
};
|
||||
}
|
||||
|
||||
function publicSnapshot(snapshot: AgentRunObservation): AgentJobTerminalSnapshot {
|
||||
return {
|
||||
status: snapshot.status,
|
||||
startedAt: snapshot.startedAt,
|
||||
endedAt: snapshot.endedAt,
|
||||
error: snapshot.error,
|
||||
stopReason: snapshot.stopReason,
|
||||
livenessState: snapshot.livenessState,
|
||||
yielded: snapshot.yielded,
|
||||
pendingError: snapshot.pendingError,
|
||||
timeoutPhase: snapshot.timeoutPhase,
|
||||
providerStarted: snapshot.providerStarted,
|
||||
};
|
||||
}
|
||||
|
||||
export async function waitForAgentJob(params: {
|
||||
runId: string;
|
||||
timeoutMs: number;
|
||||
signal?: AbortSignal;
|
||||
ignoreCachedSnapshot?: boolean;
|
||||
}): Promise<AgentRunSnapshot | null> {
|
||||
const { runId, timeoutMs, signal, ignoreCachedSnapshot = false } = params;
|
||||
source?: "chat";
|
||||
}): Promise<AgentJobTerminalSnapshot | null> {
|
||||
ensureAgentRunListener();
|
||||
const cached = ignoreCachedSnapshot ? undefined : getCachedAgentRun(runId);
|
||||
const afterVersion = params.ignoreCachedSnapshot ? agentRunVersion : -1;
|
||||
const cached = getAgentRunSnapshot({
|
||||
runId: params.runId,
|
||||
source: params.source,
|
||||
afterVersion,
|
||||
});
|
||||
if (cached) {
|
||||
return cached;
|
||||
return publicSnapshot(cached);
|
||||
}
|
||||
if (timeoutMs <= 0 || signal?.aborted) {
|
||||
if (params.timeoutMs <= 0) {
|
||||
return null;
|
||||
}
|
||||
|
||||
return await new Promise((resolve) => {
|
||||
let settled = false;
|
||||
let pendingErrorTimer: NodeJS.Timeout | undefined;
|
||||
let pendingErrorSnapshot: AgentRunSnapshot | undefined;
|
||||
let pendingTimeoutTimer: NodeJS.Timeout | undefined;
|
||||
let pendingTimeoutSnapshot: AgentRunSnapshot | undefined;
|
||||
let removeWaiter = () => {};
|
||||
|
||||
const clearPendingErrorTimer = () => {
|
||||
if (!pendingErrorTimer) {
|
||||
return;
|
||||
}
|
||||
clearTimeout(pendingErrorTimer);
|
||||
pendingErrorTimer = undefined;
|
||||
pendingErrorSnapshot = undefined;
|
||||
};
|
||||
|
||||
const clearPendingTimeoutTimer = () => {
|
||||
if (!pendingTimeoutTimer) {
|
||||
return;
|
||||
}
|
||||
clearTimeout(pendingTimeoutTimer);
|
||||
pendingTimeoutTimer = undefined;
|
||||
pendingTimeoutSnapshot = undefined;
|
||||
};
|
||||
|
||||
const finish = (entry: AgentRunSnapshot | null) => {
|
||||
const finish = (snapshot: AgentJobTerminalSnapshot | null) => {
|
||||
if (settled) {
|
||||
return;
|
||||
}
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
clearPendingErrorTimer();
|
||||
clearPendingTimeoutTimer();
|
||||
unsubscribe();
|
||||
clearTimeout(timeoutHandle);
|
||||
removeWaiter();
|
||||
if (onAbort) {
|
||||
signal?.removeEventListener("abort", onAbort);
|
||||
}
|
||||
resolve(entry);
|
||||
resolve(snapshot);
|
||||
};
|
||||
|
||||
const scheduleTerminalFinish = (
|
||||
kind: "error" | "timeout",
|
||||
snapshot: AgentRunSnapshot,
|
||||
delayMs: number,
|
||||
) => {
|
||||
if (
|
||||
pendingTimeoutSnapshot &&
|
||||
shouldPreserveTerminalSnapshot(pendingTimeoutSnapshot, snapshot)
|
||||
) {
|
||||
// Mirror the shared pending map: while this waiter holds a hard timeout
|
||||
// in grace, late terminal events must not replace the original cause.
|
||||
return;
|
||||
}
|
||||
clearPendingErrorTimer();
|
||||
clearPendingTimeoutTimer();
|
||||
const timerRef = setSafeTimeout(() => {
|
||||
const latest = ignoreCachedSnapshot ? undefined : getCachedAgentRun(runId);
|
||||
if (latest) {
|
||||
finish(latest);
|
||||
return;
|
||||
}
|
||||
recordAgentRunSnapshot(snapshot);
|
||||
finish(snapshot);
|
||||
}, delayMs);
|
||||
timerRef.unref?.();
|
||||
if (kind === "error") {
|
||||
pendingErrorTimer = timerRef;
|
||||
pendingErrorSnapshot = snapshot;
|
||||
} else {
|
||||
pendingTimeoutTimer = timerRef;
|
||||
pendingTimeoutSnapshot = snapshot;
|
||||
}
|
||||
};
|
||||
|
||||
const scheduleErrorFinish = (
|
||||
snapshot: AgentRunSnapshot,
|
||||
delayMs = AGENT_RUN_TERMINAL_RETRY_GRACE_MS,
|
||||
) => {
|
||||
scheduleTerminalFinish("error", snapshot, delayMs);
|
||||
};
|
||||
|
||||
const scheduleTimeoutFinish = (
|
||||
snapshot: AgentRunSnapshot,
|
||||
delayMs = AGENT_RUN_TERMINAL_RETRY_GRACE_MS,
|
||||
) => {
|
||||
scheduleTerminalFinish("timeout", snapshot, delayMs);
|
||||
};
|
||||
|
||||
if (!ignoreCachedSnapshot) {
|
||||
const pendingError = getPendingAgentRunError(runId);
|
||||
if (pendingError) {
|
||||
scheduleErrorFinish(pendingError.snapshot, pendingError.dueAt - Date.now());
|
||||
}
|
||||
const pendingTimeout = getPendingAgentRunTimeout(runId);
|
||||
if (pendingTimeout) {
|
||||
scheduleTimeoutFinish(pendingTimeout.snapshot, pendingTimeout.dueAt - Date.now());
|
||||
}
|
||||
}
|
||||
|
||||
const unsubscribe = onAgentEvent((evt) => {
|
||||
if (!evt || evt.stream !== "lifecycle") {
|
||||
return;
|
||||
}
|
||||
if (evt.runId !== runId) {
|
||||
return;
|
||||
}
|
||||
const phase = evt.data?.phase;
|
||||
if (phase === "start") {
|
||||
clearPendingErrorTimer();
|
||||
clearPendingTimeoutTimer();
|
||||
return;
|
||||
}
|
||||
if (phase !== "end" && phase !== "error") {
|
||||
return;
|
||||
}
|
||||
const latest = ignoreCachedSnapshot ? undefined : getCachedAgentRun(runId);
|
||||
if (latest) {
|
||||
if (
|
||||
pendingTimeoutSnapshot &&
|
||||
shouldPreserveTerminalSnapshot(pendingTimeoutSnapshot, latest)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
finish(latest);
|
||||
return;
|
||||
}
|
||||
const snapshot = createSnapshotFromLifecycleEvent({
|
||||
runId: evt.runId,
|
||||
phase,
|
||||
data: evt.data,
|
||||
const onWake = () => {
|
||||
const snapshot = getAgentRunSnapshot({
|
||||
runId: params.runId,
|
||||
source: params.source,
|
||||
afterVersion,
|
||||
});
|
||||
if (phase === "error") {
|
||||
scheduleErrorFinish(snapshot);
|
||||
return;
|
||||
if (snapshot) {
|
||||
finish(publicSnapshot(snapshot));
|
||||
}
|
||||
if (snapshot.status === "timeout") {
|
||||
scheduleTimeoutFinish(snapshot);
|
||||
return;
|
||||
}
|
||||
if (
|
||||
pendingTimeoutSnapshot &&
|
||||
shouldPreserveTerminalSnapshot(pendingTimeoutSnapshot, snapshot)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
recordAgentRunSnapshot(snapshot);
|
||||
finish(snapshot);
|
||||
});
|
||||
removeWaiter = addAgentRunWaiter(runId);
|
||||
|
||||
const timer = setSafeTimeout(() => {
|
||||
// Fresh waits can only consume terminal state observed by this waiter;
|
||||
// the shared maps may still hold a previous attempt inside retry grace.
|
||||
const pendingErrorCandidate = ignoreCachedSnapshot
|
||||
? pendingErrorSnapshot
|
||||
: getPendingAgentRunError(runId)?.snapshot;
|
||||
if (pendingErrorCandidate) {
|
||||
finish(createPendingErrorTimeoutSnapshot(pendingErrorCandidate));
|
||||
return;
|
||||
}
|
||||
const pendingTimeoutCandidate = ignoreCachedSnapshot
|
||||
? pendingTimeoutSnapshot
|
||||
: getPendingAgentRunTimeout(runId)?.snapshot;
|
||||
// Forward only canonical hard timeouts. Reuse the shared terminal-outcome
|
||||
// classifier (which excludes restart-cancelled and soft queue/draining
|
||||
// snapshots) instead of rederiving the hard/soft gate here, so those stay
|
||||
// correctable via retry grace.
|
||||
if (
|
||||
pendingTimeoutCandidate &&
|
||||
terminalOutcomeFromSnapshot(pendingTimeoutCandidate)?.reason === "hard_timeout"
|
||||
) {
|
||||
finish(pendingTimeoutCandidate);
|
||||
return;
|
||||
};
|
||||
removeWaiter = addAgentRunWaiter(params.runId, onWake);
|
||||
const timeoutHandle = setSafeTimeout(() => {
|
||||
if (!params.source) {
|
||||
const pendingError = pendingAgentRunErrors.get(params.runId)?.snapshot;
|
||||
if (pendingError && pendingError.version > afterVersion) {
|
||||
finish(createPendingErrorTimeoutSnapshot(pendingError));
|
||||
return;
|
||||
}
|
||||
const pendingTimeout = pendingAgentRunTimeouts.get(params.runId)?.snapshot;
|
||||
if (
|
||||
pendingTimeout &&
|
||||
pendingTimeout.version > afterVersion &&
|
||||
terminalOutcomeFromSnapshot(pendingTimeout)?.reason === "hard_timeout"
|
||||
) {
|
||||
finish(publicSnapshot(pendingTimeout));
|
||||
return;
|
||||
}
|
||||
}
|
||||
finish(null);
|
||||
}, timeoutMs);
|
||||
const onAbort: (() => void) | undefined = () => finish(null);
|
||||
signal?.addEventListener("abort", onAbort, { once: true });
|
||||
}, params.timeoutMs);
|
||||
timeoutHandle.unref?.();
|
||||
onWake();
|
||||
});
|
||||
}
|
||||
|
||||
@@ -516,22 +585,37 @@ ensureAgentRunListener();
|
||||
export const testing = {
|
||||
getWaiterCount(runId?: string): number {
|
||||
if (runId) {
|
||||
return agentRunWaiterCounts.get(runId) ?? 0;
|
||||
return agentRunWaiters.get(runId)?.size ?? 0;
|
||||
}
|
||||
let total = 0;
|
||||
for (const count of agentRunWaiterCounts.values()) {
|
||||
total += count;
|
||||
for (const waiters of agentRunWaiters.values()) {
|
||||
total += waiters.size;
|
||||
}
|
||||
return total;
|
||||
},
|
||||
resetWaiters(): void {
|
||||
agentRunWaiterCounts.clear();
|
||||
},
|
||||
getAgentRunCacheSize(): number {
|
||||
return agentRunCache.size;
|
||||
return agentJobs.size;
|
||||
},
|
||||
resetAgentRunCache(): void {
|
||||
agentRunCache.clear();
|
||||
agentJobs.clear();
|
||||
},
|
||||
resetAgentJobs(): void {
|
||||
agentJobs.clear();
|
||||
agentRunStarts.clear();
|
||||
agentRunVersion = 0;
|
||||
for (const pending of pendingAgentRunErrors.values()) {
|
||||
clearTimeout(pending.timer);
|
||||
}
|
||||
for (const pending of pendingAgentRunTimeouts.values()) {
|
||||
clearTimeout(pending.timer);
|
||||
}
|
||||
pendingAgentRunErrors.clear();
|
||||
pendingAgentRunTimeouts.clear();
|
||||
agentRunWaiters.clear();
|
||||
},
|
||||
getTerminalSnapshot(runId: string, source?: "chat"): AgentJobTerminalSnapshot | null {
|
||||
const snapshot = getAgentRunSnapshot({ runId, source, afterVersion: -1 });
|
||||
return snapshot ? publicSnapshot(snapshot) : null;
|
||||
},
|
||||
agentRunCacheMaxEntries: AGENT_RUN_CACHE_MAX_ENTRIES,
|
||||
};
|
||||
|
||||
@@ -1,25 +1,15 @@
|
||||
/**
|
||||
* Tests agent wait dedupe behavior for repeated gateway wait requests.
|
||||
* Tests gateway dedupe observations feeding the canonical agent job registry.
|
||||
*/
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { AGENT_RUN_ABORTED_ERROR } from "../../agents/run-termination.js";
|
||||
import { emitAgentEvent } from "../../infra/agent-events.js";
|
||||
import type { DedupeEntry } from "../server-shared.js";
|
||||
import {
|
||||
testing,
|
||||
readTerminalSnapshotFromGatewayDedupe,
|
||||
setGatewayDedupeEntry,
|
||||
waitForTerminalGatewayDedupe,
|
||||
} from "./agent-wait-dedupe.js";
|
||||
import { testing, setGatewayDedupeEntry, waitForAgentJob } from "./agent-job.js";
|
||||
|
||||
type DedupeKind = "agent" | "chat";
|
||||
type SnapshotReadOptions = Omit<
|
||||
Parameters<typeof readTerminalSnapshotFromGatewayDedupe>[0],
|
||||
"dedupe" | "runId"
|
||||
>;
|
||||
type SnapshotWaitOptions = Omit<
|
||||
Parameters<typeof waitForTerminalGatewayDedupe>[0],
|
||||
"dedupe" | "runId"
|
||||
>;
|
||||
type SnapshotReadOptions = { source?: "chat" };
|
||||
type SnapshotWaitOptions = Omit<Parameters<typeof waitForAgentJob>[0], "runId">;
|
||||
type RunEntryParams = {
|
||||
dedupe: Map<string, DedupeEntry>;
|
||||
kind: DedupeKind;
|
||||
@@ -29,7 +19,7 @@ type RunEntryParams = {
|
||||
payload: Record<string, unknown>;
|
||||
};
|
||||
|
||||
describe("agent wait dedupe helper", () => {
|
||||
describe("agent job gateway dedupe observations", () => {
|
||||
function setRunEntry(params: RunEntryParams) {
|
||||
setGatewayDedupeEntry({
|
||||
dedupe: params.dedupe,
|
||||
@@ -102,40 +92,19 @@ describe("agent wait dedupe helper", () => {
|
||||
}
|
||||
|
||||
function expectTerminalSnapshot(
|
||||
dedupe: Map<string, DedupeEntry>,
|
||||
runId: string,
|
||||
snapshot: Record<string, unknown>,
|
||||
options: SnapshotReadOptions = {},
|
||||
) {
|
||||
expect(
|
||||
readTerminalSnapshotFromGatewayDedupe({
|
||||
dedupe,
|
||||
runId,
|
||||
...options,
|
||||
}),
|
||||
).toEqual(snapshot);
|
||||
expect(testing.getTerminalSnapshot(runId, options.source)).toEqual(snapshot);
|
||||
}
|
||||
|
||||
function expectNoTerminalSnapshot(
|
||||
dedupe: Map<string, DedupeEntry>,
|
||||
runId: string,
|
||||
options: SnapshotReadOptions = {},
|
||||
) {
|
||||
expect(
|
||||
readTerminalSnapshotFromGatewayDedupe({
|
||||
dedupe,
|
||||
runId,
|
||||
...options,
|
||||
}),
|
||||
).toBeNull();
|
||||
function expectNoTerminalSnapshot(runId: string, options: SnapshotReadOptions = {}) {
|
||||
expect(testing.getTerminalSnapshot(runId, options.source)).toBeNull();
|
||||
}
|
||||
|
||||
function waitForTerminalSnapshot(
|
||||
dedupe: Map<string, DedupeEntry>,
|
||||
runId: string,
|
||||
options: Partial<SnapshotWaitOptions> = {},
|
||||
) {
|
||||
return waitForTerminalGatewayDedupe({ dedupe, runId, timeoutMs: 1_000, ...options });
|
||||
function waitForTerminalSnapshot(runId: string, options: Partial<SnapshotWaitOptions> = {}) {
|
||||
return waitForAgentJob({ runId, timeoutMs: 1_000, ...options });
|
||||
}
|
||||
|
||||
const RPC_QUEUE_CANCEL_SNAPSHOT = {
|
||||
@@ -150,19 +119,19 @@ describe("agent wait dedupe helper", () => {
|
||||
} as const;
|
||||
|
||||
beforeEach(() => {
|
||||
testing.resetWaiters();
|
||||
testing.resetAgentJobs();
|
||||
vi.useFakeTimers();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
testing.resetWaiters();
|
||||
testing.resetAgentJobs();
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("unblocks waiters when a terminal chat dedupe entry is written", async () => {
|
||||
const dedupe = new Map();
|
||||
const runId = "run-chat-terminal";
|
||||
const waiter = waitForTerminalSnapshot(dedupe, runId);
|
||||
const waiter = waitForTerminalSnapshot(runId);
|
||||
|
||||
await Promise.resolve();
|
||||
expect(testing.getWaiterCount(runId)).toBe(1);
|
||||
@@ -192,7 +161,6 @@ describe("agent wait dedupe helper", () => {
|
||||
});
|
||||
|
||||
expectTerminalSnapshot(
|
||||
dedupe,
|
||||
runId,
|
||||
okSnapshot({
|
||||
startedAt: 100,
|
||||
@@ -223,7 +191,7 @@ describe("agent wait dedupe helper", () => {
|
||||
),
|
||||
});
|
||||
|
||||
expectTerminalSnapshot(dedupe, runId, {
|
||||
expectTerminalSnapshot(runId, {
|
||||
status: "timeout",
|
||||
startedAt: 100,
|
||||
endedAt: 200,
|
||||
@@ -254,7 +222,7 @@ describe("agent wait dedupe helper", () => {
|
||||
),
|
||||
});
|
||||
|
||||
expectTerminalSnapshot(dedupe, runId, {
|
||||
expectTerminalSnapshot(runId, {
|
||||
status: "timeout",
|
||||
startedAt: 100,
|
||||
endedAt: 200,
|
||||
@@ -283,7 +251,7 @@ describe("agent wait dedupe helper", () => {
|
||||
),
|
||||
});
|
||||
|
||||
expectTerminalSnapshot(dedupe, runId, {
|
||||
expectTerminalSnapshot(runId, {
|
||||
status: "error",
|
||||
startedAt: 100,
|
||||
endedAt: 200,
|
||||
@@ -304,7 +272,7 @@ describe("agent wait dedupe helper", () => {
|
||||
}),
|
||||
});
|
||||
|
||||
expectTerminalSnapshot(dedupe, runId, {
|
||||
expectTerminalSnapshot(runId, {
|
||||
status: "error",
|
||||
startedAt: 100,
|
||||
endedAt: 200,
|
||||
@@ -316,7 +284,7 @@ describe("agent wait dedupe helper", () => {
|
||||
it("unblocks waiters with normalized aborted snapshots", async () => {
|
||||
const dedupe = new Map();
|
||||
const runId = "run-wait-aborted";
|
||||
const waiter = waitForTerminalSnapshot(dedupe, runId);
|
||||
const waiter = waitForTerminalSnapshot(runId);
|
||||
|
||||
await Promise.resolve();
|
||||
expect(testing.getWaiterCount(runId)).toBe(1);
|
||||
@@ -358,9 +326,9 @@ describe("agent wait dedupe helper", () => {
|
||||
},
|
||||
});
|
||||
|
||||
expectNoTerminalSnapshot(dedupe, runId);
|
||||
expectNoTerminalSnapshot(runId);
|
||||
|
||||
const blockedWait = waitForTerminalSnapshot(dedupe, runId, { timeoutMs: 25 });
|
||||
const blockedWait = waitForTerminalSnapshot(runId, { timeoutMs: 25 });
|
||||
await vi.advanceTimersByTimeAsync(30);
|
||||
await expect(blockedWait).resolves.toBeNull();
|
||||
expect(testing.getWaiterCount(runId)).toBe(0);
|
||||
@@ -385,7 +353,7 @@ describe("agent wait dedupe helper", () => {
|
||||
payload: okPayload(runId, { startedAt: 1, endedAt: 2 }),
|
||||
});
|
||||
|
||||
expectTerminalSnapshot(dedupe, runId, okSnapshot({ startedAt: 1, endedAt: 2 }));
|
||||
expectTerminalSnapshot(runId, okSnapshot({ startedAt: 1, endedAt: 2 }));
|
||||
});
|
||||
|
||||
it("ignores stale agent snapshots when waiting for an active chat run", async () => {
|
||||
@@ -397,10 +365,10 @@ describe("agent wait dedupe helper", () => {
|
||||
payload: okPayload(runId),
|
||||
});
|
||||
|
||||
expectNoTerminalSnapshot(dedupe, runId, { ignoreAgentTerminalSnapshot: true });
|
||||
expectNoTerminalSnapshot(runId, { source: "chat" });
|
||||
|
||||
const wait = waitForTerminalSnapshot(dedupe, runId, {
|
||||
ignoreAgentTerminalSnapshot: true,
|
||||
const wait = waitForTerminalSnapshot(runId, {
|
||||
source: "chat",
|
||||
});
|
||||
await Promise.resolve();
|
||||
expect(testing.getWaiterCount(runId)).toBe(1);
|
||||
@@ -432,7 +400,7 @@ describe("agent wait dedupe helper", () => {
|
||||
payload: { runId, status: "error", startedAt: 30, endedAt: 40, error: "chat failed" },
|
||||
});
|
||||
|
||||
expectTerminalSnapshot(dedupe, runId, {
|
||||
expectTerminalSnapshot(runId, {
|
||||
status: "error",
|
||||
startedAt: 30,
|
||||
endedAt: 40,
|
||||
@@ -453,7 +421,7 @@ describe("agent wait dedupe helper", () => {
|
||||
payload: { runId, status: "timeout", startedAt: 3, endedAt: 4, error: "still running" },
|
||||
});
|
||||
|
||||
expectTerminalSnapshot(dedupeReverse, runId, {
|
||||
expectTerminalSnapshot(runId, {
|
||||
status: "timeout",
|
||||
startedAt: 3,
|
||||
endedAt: 4,
|
||||
@@ -461,6 +429,75 @@ describe("agent wait dedupe helper", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("uses the newer source when an older cross-source terminal is sticky", () => {
|
||||
const runId = "run-sticky-collision";
|
||||
const newerChat = new Map();
|
||||
|
||||
setRpcQueueTimeoutEntry({
|
||||
dedupe: newerChat,
|
||||
kind: "agent",
|
||||
runId,
|
||||
ts: 100,
|
||||
});
|
||||
setChatEntry({
|
||||
dedupe: newerChat,
|
||||
runId,
|
||||
ts: 200,
|
||||
payload: okPayload(runId, { startedAt: 150, endedAt: 200 }),
|
||||
});
|
||||
|
||||
expectTerminalSnapshot(runId, okSnapshot({ startedAt: 150, endedAt: 200 }));
|
||||
|
||||
const newerAgentRunId = `${runId}-reverse`;
|
||||
const newerAgent = new Map();
|
||||
setAgentEntry({
|
||||
dedupe: newerAgent,
|
||||
runId: newerAgentRunId,
|
||||
ts: 200,
|
||||
payload: okPayload(newerAgentRunId, { startedAt: 150, endedAt: 200 }),
|
||||
});
|
||||
setChatEntry({
|
||||
dedupe: newerAgent,
|
||||
runId: newerAgentRunId,
|
||||
ts: 100,
|
||||
payload: {
|
||||
runId: newerAgentRunId,
|
||||
status: "timeout",
|
||||
startedAt: 50,
|
||||
endedAt: 100,
|
||||
timeoutPhase: "provider",
|
||||
providerStarted: true,
|
||||
},
|
||||
});
|
||||
|
||||
expectTerminalSnapshot(newerAgentRunId, okSnapshot({ startedAt: 150, endedAt: 200 }));
|
||||
});
|
||||
|
||||
it("lets a fresh wait observe lifecycle completion after a dedupe snapshot", async () => {
|
||||
const dedupe = new Map();
|
||||
const runId = "run-lifecycle-after-dedupe";
|
||||
emitAgentEvent({
|
||||
runId,
|
||||
stream: "lifecycle",
|
||||
data: { phase: "start", startedAt: 100 },
|
||||
});
|
||||
setAgentEntry({
|
||||
dedupe,
|
||||
runId,
|
||||
ts: 150,
|
||||
payload: okPayload(runId, { startedAt: 100, endedAt: 150 }),
|
||||
});
|
||||
|
||||
const wait = waitForTerminalSnapshot(runId, { ignoreCachedSnapshot: true });
|
||||
emitAgentEvent({
|
||||
runId,
|
||||
stream: "lifecycle",
|
||||
data: { phase: "end", startedAt: 100, endedAt: 200 },
|
||||
});
|
||||
|
||||
await expect(wait).resolves.toEqual(okSnapshot({ startedAt: 100, endedAt: 200 }));
|
||||
});
|
||||
|
||||
it("preserves an RPC cancel snapshot when late completion writes the same key", () => {
|
||||
const dedupe = new Map();
|
||||
const runId = "run-cancel-wins";
|
||||
@@ -477,7 +514,7 @@ describe("agent wait dedupe helper", () => {
|
||||
payload: okPayload(runId, { endedAt: 200 }),
|
||||
});
|
||||
|
||||
expectTerminalSnapshot(dedupe, runId, RPC_QUEUE_CANCEL_SNAPSHOT);
|
||||
expectTerminalSnapshot(runId, RPC_QUEUE_CANCEL_SNAPSHOT);
|
||||
});
|
||||
|
||||
it("preserves an RPC cancel snapshot when a later accepted write reuses the key", () => {
|
||||
@@ -496,7 +533,7 @@ describe("agent wait dedupe helper", () => {
|
||||
payload: { runId, status: "accepted" },
|
||||
});
|
||||
|
||||
expectTerminalSnapshot(dedupe, runId, RPC_QUEUE_CANCEL_SNAPSHOT);
|
||||
expectTerminalSnapshot(runId, RPC_QUEUE_CANCEL_SNAPSHOT);
|
||||
});
|
||||
|
||||
it("lets an earlier terminal completion correct a provisional timeout snapshot", () => {
|
||||
@@ -522,7 +559,7 @@ describe("agent wait dedupe helper", () => {
|
||||
payload: okPayload(runId, { startedAt: 100, endedAt: 190 }),
|
||||
});
|
||||
|
||||
expectTerminalSnapshot(dedupe, runId, okSnapshot({ startedAt: 100, endedAt: 190 }));
|
||||
expectTerminalSnapshot(runId, okSnapshot({ startedAt: 100, endedAt: 190 }));
|
||||
});
|
||||
|
||||
it("does not make bare queue timeouts sticky", () => {
|
||||
@@ -542,7 +579,7 @@ describe("agent wait dedupe helper", () => {
|
||||
payload: okPayload(runId, { endedAt: 200 }),
|
||||
});
|
||||
|
||||
expectTerminalSnapshot(dedupe, runId, okSnapshot({ endedAt: 200 }));
|
||||
expectTerminalSnapshot(runId, okSnapshot({ endedAt: 200 }));
|
||||
});
|
||||
|
||||
it("preserves an RPC cancel snapshot when late rejection writes the same chat key", () => {
|
||||
@@ -562,14 +599,14 @@ describe("agent wait dedupe helper", () => {
|
||||
payload: { runId, status: "error", summary: "late failure", endedAt: 200 },
|
||||
});
|
||||
|
||||
expectTerminalSnapshot(dedupe, runId, RPC_QUEUE_CANCEL_SNAPSHOT);
|
||||
expectTerminalSnapshot(runId, RPC_QUEUE_CANCEL_SNAPSHOT);
|
||||
});
|
||||
|
||||
it("resolves multiple waiters for the same run id", async () => {
|
||||
const dedupe = new Map();
|
||||
const runId = "run-multi";
|
||||
const first = waitForTerminalSnapshot(dedupe, runId);
|
||||
const second = waitForTerminalSnapshot(dedupe, runId);
|
||||
const first = waitForTerminalSnapshot(runId);
|
||||
const second = waitForTerminalSnapshot(runId);
|
||||
|
||||
await Promise.resolve();
|
||||
expect(testing.getWaiterCount(runId)).toBe(2);
|
||||
@@ -593,9 +630,8 @@ describe("agent wait dedupe helper", () => {
|
||||
});
|
||||
|
||||
it("cleans up waiter registration on timeout", async () => {
|
||||
const dedupe = new Map();
|
||||
const runId = "run-timeout";
|
||||
const wait = waitForTerminalSnapshot(dedupe, runId, { timeoutMs: 20 });
|
||||
const wait = waitForTerminalSnapshot(runId, { timeoutMs: 20 });
|
||||
|
||||
await Promise.resolve();
|
||||
expect(testing.getWaiterCount(runId)).toBe(1);
|
||||
|
||||
@@ -1,357 +0,0 @@
|
||||
// Agent wait dedupe helpers normalize terminal run snapshots and wake waiters
|
||||
// that are blocked on active agent/chat dedupe entries.
|
||||
import { asFiniteNumber } from "@openclaw/normalization-core/number-coercion";
|
||||
import { asOptionalRecord } from "@openclaw/normalization-core/record-coerce";
|
||||
import {
|
||||
buildAgentRunTerminalOutcome,
|
||||
isStickyAgentRunTerminalOutcome,
|
||||
mergeAgentRunTerminalOutcome,
|
||||
type AgentRunTerminalOutcome,
|
||||
} from "../../agents/agent-run-terminal-outcome.js";
|
||||
import { normalizeBlockedLivenessWaitStatus } from "../../shared/agent-liveness.js";
|
||||
import { isNonTerminalAgentRunStatus } from "../../shared/agent-run-status.js";
|
||||
import { setSafeTimeout } from "../../utils/timer-delay.js";
|
||||
import type { DedupeEntry } from "../server-shared.js";
|
||||
|
||||
export type AgentWaitTerminalSnapshot = {
|
||||
status: "ok" | "error" | "timeout";
|
||||
startedAt?: number;
|
||||
endedAt?: number;
|
||||
error?: string;
|
||||
stopReason?: string;
|
||||
livenessState?: string;
|
||||
yielded?: boolean;
|
||||
pendingError?: boolean;
|
||||
timeoutPhase?: AgentRunTerminalOutcome["timeoutPhase"];
|
||||
providerStarted?: boolean;
|
||||
};
|
||||
|
||||
const AGENT_WAITERS_BY_RUN_ID = new Map<string, Set<() => void>>();
|
||||
|
||||
function normalizeTerminalOutcomeForWaitSnapshot(outcome: AgentRunTerminalOutcome): {
|
||||
status: AgentWaitTerminalSnapshot["status"];
|
||||
error?: string;
|
||||
} {
|
||||
if (outcome.reason === "hard_timeout") {
|
||||
return { status: outcome.status, error: outcome.error };
|
||||
}
|
||||
return normalizeBlockedLivenessWaitStatus(outcome);
|
||||
}
|
||||
|
||||
function parseRunIdFromDedupeKey(key: string): string | null {
|
||||
if (key.startsWith("agent:")) {
|
||||
return key.slice("agent:".length) || null;
|
||||
}
|
||||
if (key.startsWith("chat:")) {
|
||||
return key.slice("chat:".length) || null;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function asString(value: unknown): string | undefined {
|
||||
return typeof value === "string" && value.trim() ? value : undefined;
|
||||
}
|
||||
|
||||
function buildDedupeTerminalSnapshot(params: {
|
||||
status: AgentRunTerminalOutcome["status"];
|
||||
startedAt?: number;
|
||||
endedAt: number;
|
||||
error?: string;
|
||||
stopReason?: string;
|
||||
livenessState?: string;
|
||||
yielded: boolean;
|
||||
timeoutPhase: unknown;
|
||||
providerStarted: unknown;
|
||||
}): AgentWaitTerminalSnapshot {
|
||||
const terminalOutcome = buildAgentRunTerminalOutcome({
|
||||
status: params.status,
|
||||
livenessState: params.livenessState,
|
||||
error: params.error,
|
||||
stopReason: params.stopReason,
|
||||
timeoutPhase: params.timeoutPhase,
|
||||
providerStarted: params.providerStarted,
|
||||
startedAt: params.startedAt,
|
||||
endedAt: params.endedAt,
|
||||
});
|
||||
const normalized = normalizeTerminalOutcomeForWaitSnapshot(terminalOutcome);
|
||||
return {
|
||||
status: normalized.status,
|
||||
startedAt: params.startedAt,
|
||||
endedAt: params.endedAt,
|
||||
error:
|
||||
normalized.status === "error"
|
||||
? normalized.error
|
||||
: normalized.status === "timeout"
|
||||
? terminalOutcome.error
|
||||
: undefined,
|
||||
stopReason: params.stopReason,
|
||||
livenessState: params.livenessState,
|
||||
...(params.yielded ? { yielded: params.yielded } : {}),
|
||||
...(terminalOutcome.timeoutPhase ? { timeoutPhase: terminalOutcome.timeoutPhase } : {}),
|
||||
...(terminalOutcome.providerStarted !== undefined
|
||||
? { providerStarted: terminalOutcome.providerStarted }
|
||||
: {}),
|
||||
};
|
||||
}
|
||||
|
||||
function removeWaiter(runId: string, waiter: () => void): void {
|
||||
const waiters = AGENT_WAITERS_BY_RUN_ID.get(runId);
|
||||
if (!waiters) {
|
||||
return;
|
||||
}
|
||||
waiters.delete(waiter);
|
||||
if (waiters.size === 0) {
|
||||
AGENT_WAITERS_BY_RUN_ID.delete(runId);
|
||||
}
|
||||
}
|
||||
|
||||
function addWaiter(runId: string, waiter: () => void): () => void {
|
||||
const normalizedRunId = runId.trim();
|
||||
if (!normalizedRunId) {
|
||||
return () => {};
|
||||
}
|
||||
const existing = AGENT_WAITERS_BY_RUN_ID.get(normalizedRunId);
|
||||
if (existing) {
|
||||
existing.add(waiter);
|
||||
return () => removeWaiter(normalizedRunId, waiter);
|
||||
}
|
||||
AGENT_WAITERS_BY_RUN_ID.set(normalizedRunId, new Set([waiter]));
|
||||
return () => removeWaiter(normalizedRunId, waiter);
|
||||
}
|
||||
|
||||
// Waiters are keyed only by run id so chat and agent dedupe entries can wake
|
||||
// the same `agent.wait` request regardless of which path finishes first.
|
||||
function notifyWaiters(runId: string): void {
|
||||
const normalizedRunId = runId.trim();
|
||||
if (!normalizedRunId) {
|
||||
return;
|
||||
}
|
||||
const waiters = AGENT_WAITERS_BY_RUN_ID.get(normalizedRunId);
|
||||
if (!waiters || waiters.size === 0) {
|
||||
return;
|
||||
}
|
||||
for (const waiter of waiters) {
|
||||
waiter();
|
||||
}
|
||||
}
|
||||
|
||||
function readTerminalSnapshotFromDedupeEntry(entry: DedupeEntry): AgentWaitTerminalSnapshot | null {
|
||||
const payload = entry.payload as
|
||||
| {
|
||||
status?: unknown;
|
||||
startedAt?: unknown;
|
||||
endedAt?: unknown;
|
||||
error?: unknown;
|
||||
summary?: unknown;
|
||||
stopReason?: unknown;
|
||||
livenessState?: unknown;
|
||||
yielded?: unknown;
|
||||
timeoutPhase?: unknown;
|
||||
providerStarted?: unknown;
|
||||
result?: unknown;
|
||||
}
|
||||
| undefined;
|
||||
const status = typeof payload?.status === "string" ? payload.status : undefined;
|
||||
if (isNonTerminalAgentRunStatus(status)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const startedAt = asFiniteNumber(payload?.startedAt);
|
||||
const endedAt = asFiniteNumber(payload?.endedAt) ?? entry.ts;
|
||||
const resultMeta = asOptionalRecord(asOptionalRecord(payload?.result)?.meta);
|
||||
const stopReason = asString(payload?.stopReason) ?? asString(resultMeta?.stopReason);
|
||||
const livenessState = asString(payload?.livenessState) ?? asString(resultMeta?.livenessState);
|
||||
const yielded = payload?.yielded === true || resultMeta?.yielded === true;
|
||||
const timeoutPhase = payload?.timeoutPhase ?? resultMeta?.timeoutPhase;
|
||||
const providerStarted = payload?.providerStarted ?? resultMeta?.providerStarted;
|
||||
const errorMessage =
|
||||
typeof payload?.error === "string"
|
||||
? payload.error
|
||||
: typeof payload?.summary === "string"
|
||||
? payload.summary
|
||||
: entry.error?.message;
|
||||
|
||||
const terminalStatus =
|
||||
status === "ok" || status === "timeout" || status === "error"
|
||||
? status
|
||||
: entry.ok
|
||||
? null
|
||||
: "error";
|
||||
if (!terminalStatus) {
|
||||
return null;
|
||||
}
|
||||
return buildDedupeTerminalSnapshot({
|
||||
status: terminalStatus,
|
||||
startedAt,
|
||||
endedAt,
|
||||
error: errorMessage,
|
||||
stopReason,
|
||||
livenessState,
|
||||
yielded,
|
||||
timeoutPhase,
|
||||
providerStarted,
|
||||
});
|
||||
}
|
||||
|
||||
function terminalOutcomeFromWaitSnapshot(
|
||||
snapshot: AgentWaitTerminalSnapshot,
|
||||
): AgentRunTerminalOutcome | undefined {
|
||||
if (snapshot.pendingError) {
|
||||
return undefined;
|
||||
}
|
||||
return buildAgentRunTerminalOutcome(snapshot);
|
||||
}
|
||||
|
||||
export function readTerminalSnapshotFromGatewayDedupe(params: {
|
||||
dedupe: Map<string, DedupeEntry>;
|
||||
runId: string;
|
||||
ignoreAgentTerminalSnapshot?: boolean;
|
||||
}): AgentWaitTerminalSnapshot | null {
|
||||
// Agent and chat handlers both cache terminal state. Project them into one
|
||||
// wait result while preserving stronger terminal outcomes such as hard timeout.
|
||||
if (params.ignoreAgentTerminalSnapshot) {
|
||||
const chatEntry = params.dedupe.get(`chat:${params.runId}`);
|
||||
if (!chatEntry) {
|
||||
return null;
|
||||
}
|
||||
return readTerminalSnapshotFromDedupeEntry(chatEntry);
|
||||
}
|
||||
|
||||
const chatEntry = params.dedupe.get(`chat:${params.runId}`);
|
||||
const chatSnapshot = chatEntry ? readTerminalSnapshotFromDedupeEntry(chatEntry) : null;
|
||||
|
||||
const agentEntry = params.dedupe.get(`agent:${params.runId}`);
|
||||
const agentSnapshot = agentEntry ? readTerminalSnapshotFromDedupeEntry(agentEntry) : null;
|
||||
if (agentEntry) {
|
||||
if (!agentSnapshot) {
|
||||
// If agent is still in-flight, only trust chat if it was written after
|
||||
// this agent entry (indicating a newer completed chat run reused runId).
|
||||
if (chatSnapshot && chatEntry && chatEntry.ts > agentEntry.ts) {
|
||||
return chatSnapshot;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
if (agentSnapshot && chatSnapshot && agentEntry && chatEntry) {
|
||||
// Reused idempotency keys can leave both records present. Prefer the
|
||||
// freshest terminal snapshot so callers observe the latest run outcome.
|
||||
return chatEntry.ts > agentEntry.ts ? chatSnapshot : agentSnapshot;
|
||||
}
|
||||
|
||||
return agentSnapshot ?? chatSnapshot;
|
||||
}
|
||||
|
||||
export async function waitForTerminalGatewayDedupe(params: {
|
||||
dedupe: Map<string, DedupeEntry>;
|
||||
runId: string;
|
||||
timeoutMs: number;
|
||||
signal?: AbortSignal;
|
||||
ignoreAgentTerminalSnapshot?: boolean;
|
||||
}): Promise<AgentWaitTerminalSnapshot | null> {
|
||||
const initial = readTerminalSnapshotFromGatewayDedupe(params);
|
||||
if (initial) {
|
||||
return initial;
|
||||
}
|
||||
if (params.timeoutMs <= 0 || params.signal?.aborted) {
|
||||
return null;
|
||||
}
|
||||
|
||||
return await new Promise((resolve) => {
|
||||
let settled = false;
|
||||
|
||||
// Always re-read from the dedupe map on wake; waiters are notifications,
|
||||
// not carriers of terminal data, so stale callbacks cannot resolve a run.
|
||||
const finish = (snapshot: AgentWaitTerminalSnapshot | null) => {
|
||||
if (settled) {
|
||||
return;
|
||||
}
|
||||
settled = true;
|
||||
if (timeoutHandle) {
|
||||
clearTimeout(timeoutHandle);
|
||||
}
|
||||
if (onAbort) {
|
||||
params.signal?.removeEventListener("abort", onAbort);
|
||||
}
|
||||
removeWaiterLocal?.();
|
||||
resolve(snapshot);
|
||||
};
|
||||
|
||||
const onWake = () => {
|
||||
const snapshot = readTerminalSnapshotFromGatewayDedupe(params);
|
||||
if (snapshot) {
|
||||
finish(snapshot);
|
||||
}
|
||||
};
|
||||
|
||||
const removeWaiterLocal: (() => void) | undefined = addWaiter(params.runId, onWake);
|
||||
onWake();
|
||||
if (settled) {
|
||||
return;
|
||||
}
|
||||
|
||||
const timeoutHandle: NodeJS.Timeout | undefined = setSafeTimeout(
|
||||
() => finish(null),
|
||||
params.timeoutMs,
|
||||
);
|
||||
timeoutHandle.unref?.();
|
||||
|
||||
const onAbort: (() => void) | undefined = () => finish(null);
|
||||
params.signal?.addEventListener("abort", onAbort, { once: true });
|
||||
});
|
||||
}
|
||||
|
||||
export function setGatewayDedupeEntry(params: {
|
||||
dedupe: Map<string, DedupeEntry>;
|
||||
key: string;
|
||||
entry: DedupeEntry;
|
||||
}) {
|
||||
// Preserve sticky terminal outcomes before publishing the new entry. This
|
||||
// protects waiters from late accepted/in-flight rewrites for the same run id.
|
||||
const existing = params.dedupe.get(params.key);
|
||||
const existingSnapshot = existing ? readTerminalSnapshotFromDedupeEntry(existing) : null;
|
||||
const incomingSnapshot = readTerminalSnapshotFromDedupeEntry(params.entry);
|
||||
const existingOutcome = existingSnapshot
|
||||
? terminalOutcomeFromWaitSnapshot(existingSnapshot)
|
||||
: undefined;
|
||||
const incomingOutcome = incomingSnapshot
|
||||
? terminalOutcomeFromWaitSnapshot(incomingSnapshot)
|
||||
: undefined;
|
||||
if (existingOutcome && isStickyAgentRunTerminalOutcome(existingOutcome) && !incomingOutcome) {
|
||||
// Accepted/in-flight rewrites are not evidence against a terminal hard
|
||||
// timeout or explicit cancellation already stored for this run id.
|
||||
return;
|
||||
}
|
||||
if (existingOutcome && incomingOutcome && isStickyAgentRunTerminalOutcome(existingOutcome)) {
|
||||
const merged = mergeAgentRunTerminalOutcome(existingOutcome, incomingOutcome);
|
||||
if (merged === existingOutcome) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
params.dedupe.set(params.key, params.entry);
|
||||
const runId = parseRunIdFromDedupeKey(params.key);
|
||||
if (!runId) {
|
||||
return;
|
||||
}
|
||||
if (!incomingSnapshot) {
|
||||
return;
|
||||
}
|
||||
notifyWaiters(runId);
|
||||
}
|
||||
|
||||
export const testing = {
|
||||
getWaiterCount(runId?: string): number {
|
||||
if (runId) {
|
||||
return AGENT_WAITERS_BY_RUN_ID.get(runId)?.size ?? 0;
|
||||
}
|
||||
let total = 0;
|
||||
for (const waiters of AGENT_WAITERS_BY_RUN_ID.values()) {
|
||||
total += waiters.size;
|
||||
}
|
||||
return total;
|
||||
},
|
||||
resetWaiters() {
|
||||
AGENT_WAITERS_BY_RUN_ID.clear();
|
||||
},
|
||||
};
|
||||
export { testing as __testing };
|
||||
@@ -45,7 +45,7 @@ import {
|
||||
} from "../../tasks/task-registry.js";
|
||||
import { withTempDir } from "../../test-helpers/temp-dir.js";
|
||||
import { captureEnv, setTestEnvValue } from "../../test-utils/env.js";
|
||||
import { setGatewayDedupeEntry } from "./agent-wait-dedupe.js";
|
||||
import { setGatewayDedupeEntry } from "./agent-job.js";
|
||||
import { agentHandlers } from "./agent.js";
|
||||
import { chatHandlers } from "./chat.js";
|
||||
import { expectSubagentFollowupReactivation } from "./subagent-followup.test-helpers.js";
|
||||
|
||||
@@ -192,13 +192,7 @@ import {
|
||||
resolveSessionModelRef,
|
||||
} from "../session-utils.js";
|
||||
import { formatForLog } from "../ws-log.js";
|
||||
import { waitForAgentJob } from "./agent-job.js";
|
||||
import {
|
||||
readTerminalSnapshotFromGatewayDedupe,
|
||||
setGatewayDedupeEntry,
|
||||
type AgentWaitTerminalSnapshot,
|
||||
waitForTerminalGatewayDedupe,
|
||||
} from "./agent-wait-dedupe.js";
|
||||
import { setGatewayDedupeEntry, waitForAgentJob } from "./agent-job.js";
|
||||
import { normalizeRpcAttachmentsToChatAttachments } from "./attachment-normalize.js";
|
||||
import { emitSessionsChanged } from "./session-change-event.js";
|
||||
import type {
|
||||
@@ -4056,90 +4050,12 @@ export const agentHandlers: GatewayRequestHandlers = {
|
||||
const activeChatEntry = context.chatAbortControllers.get(runId);
|
||||
const hasActiveChatRun = activeChatEntry !== undefined && activeChatEntry.kind !== "agent";
|
||||
|
||||
const cachedGatewaySnapshot = readTerminalSnapshotFromGatewayDedupe({
|
||||
dedupe: context.dedupe,
|
||||
runId,
|
||||
ignoreAgentTerminalSnapshot: hasActiveChatRun,
|
||||
});
|
||||
if (cachedGatewaySnapshot) {
|
||||
respond(true, {
|
||||
runId,
|
||||
status: cachedGatewaySnapshot.status,
|
||||
startedAt: cachedGatewaySnapshot.startedAt,
|
||||
endedAt: cachedGatewaySnapshot.endedAt,
|
||||
error: cachedGatewaySnapshot.error,
|
||||
stopReason: cachedGatewaySnapshot.stopReason,
|
||||
livenessState: cachedGatewaySnapshot.livenessState,
|
||||
yielded: cachedGatewaySnapshot.yielded,
|
||||
pendingError: cachedGatewaySnapshot.pendingError,
|
||||
timeoutPhase: cachedGatewaySnapshot.timeoutPhase,
|
||||
providerStarted: cachedGatewaySnapshot.providerStarted,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const lifecycleAbortController = new AbortController();
|
||||
const dedupeAbortController = new AbortController();
|
||||
const dedupePromise = waitForTerminalGatewayDedupe({
|
||||
dedupe: context.dedupe,
|
||||
const snapshot = await waitForAgentJob({
|
||||
runId,
|
||||
timeoutMs,
|
||||
signal: dedupeAbortController.signal,
|
||||
ignoreAgentTerminalSnapshot: hasActiveChatRun,
|
||||
...(hasActiveChatRun ? { source: "chat" } : {}),
|
||||
});
|
||||
|
||||
if (hasActiveChatRun) {
|
||||
const snapshot = await dedupePromise;
|
||||
dedupeAbortController.abort();
|
||||
if (!snapshot) {
|
||||
respond(true, {
|
||||
runId,
|
||||
status: "timeout",
|
||||
timeoutPhase: "gateway_draining",
|
||||
});
|
||||
return;
|
||||
}
|
||||
respond(true, {
|
||||
runId,
|
||||
status: snapshot.status,
|
||||
startedAt: snapshot.startedAt,
|
||||
endedAt: snapshot.endedAt,
|
||||
error: snapshot.error,
|
||||
stopReason: snapshot.stopReason,
|
||||
livenessState: snapshot.livenessState,
|
||||
yielded: snapshot.yielded,
|
||||
pendingError: snapshot.pendingError,
|
||||
timeoutPhase: snapshot.timeoutPhase,
|
||||
providerStarted: snapshot.providerStarted,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const lifecyclePromise = waitForAgentJob({
|
||||
runId,
|
||||
timeoutMs,
|
||||
signal: lifecycleAbortController.signal,
|
||||
});
|
||||
|
||||
const first = await Promise.race([
|
||||
lifecyclePromise.then((snapshot) => ({ source: "lifecycle" as const, snapshot })),
|
||||
dedupePromise.then((snapshot) => ({ source: "dedupe" as const, snapshot })),
|
||||
]);
|
||||
|
||||
let snapshot: AgentWaitTerminalSnapshot | Awaited<ReturnType<typeof waitForAgentJob>> =
|
||||
first.snapshot;
|
||||
if (snapshot) {
|
||||
if (first.source === "lifecycle") {
|
||||
dedupeAbortController.abort();
|
||||
} else {
|
||||
lifecycleAbortController.abort();
|
||||
}
|
||||
} else {
|
||||
snapshot = first.source === "lifecycle" ? await dedupePromise : await lifecyclePromise;
|
||||
lifecycleAbortController.abort();
|
||||
dedupeAbortController.abort();
|
||||
}
|
||||
|
||||
if (!snapshot) {
|
||||
const activeRunRegistered = activeChatEntry !== undefined;
|
||||
respond(true, {
|
||||
|
||||
@@ -217,7 +217,7 @@ import {
|
||||
resolveSessionStoreKey,
|
||||
} from "../session-utils.js";
|
||||
import { formatForLog } from "../ws-log.js";
|
||||
import { setGatewayDedupeEntry } from "./agent-wait-dedupe.js";
|
||||
import { setGatewayDedupeEntry } from "./agent-job.js";
|
||||
import { normalizeRpcAttachmentsToChatAttachments } from "./attachment-normalize.js";
|
||||
import { normalizeWebchatReplyMediaPathsForDisplay } from "./chat-reply-media.js";
|
||||
import {
|
||||
|
||||
@@ -27,8 +27,7 @@ export function clearNodeWakeState(nodeId: string): void {
|
||||
}
|
||||
|
||||
// Narrow read-only seam for tests that assert nodeWakeById is cleaned up on
|
||||
// early-return paths. Mirrors the pattern used in agent-wait-dedupe.ts:223
|
||||
// and agents.ts:78 — keep production surface untouched and do not expose the
|
||||
// early-return paths. Keep production surface untouched and do not expose the
|
||||
// underlying Map reference.
|
||||
export const testing = {
|
||||
getNodeWakeByIdSize(): number {
|
||||
|
||||
@@ -14,8 +14,7 @@
|
||||
//
|
||||
// CAL-003 compliance: the null-registration branch is already exercised by
|
||||
// existing nodes.invoke-wake.test.ts cases. The test just observes that the
|
||||
// Map size returns to 0, using a minimal read-only testing seam mirrored on
|
||||
// agent-wait-dedupe.ts:223 and agents.ts:78.
|
||||
// Map size returns to 0 through a minimal read-only testing seam.
|
||||
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
|
||||
@@ -135,7 +135,7 @@ import {
|
||||
} from "../session-utils.js";
|
||||
import { projectSessionsPatchEntry } from "../sessions-patch.js";
|
||||
import { resolveSessionKeyFromResolveParams } from "../sessions-resolve.js";
|
||||
import { setGatewayDedupeEntry } from "./agent-wait-dedupe.js";
|
||||
import { setGatewayDedupeEntry } from "./agent-job.js";
|
||||
import { chatHandlers } from "./chat.js";
|
||||
import { loadOptionalServerMethodModelCatalog } from "./optional-model-catalog.js";
|
||||
import {
|
||||
|
||||
Reference in New Issue
Block a user