From 39d4ca8f4ee286d8e98d2bf1d23ba8181c9dbfe7 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 31 Jul 2026 08:08:22 -0700 Subject: [PATCH] fix(sessions): prevent stale upstream activity after session changes (#116864) * fix(sessions): fence upstream monitor results * fix(sessions): recheck provenance ownership --- .../session-upstream-monitor.test-support.ts | 1 + src/sessions/session-upstream-monitor.test.ts | 178 ++++++++++++++++++ src/sessions/session-upstream-monitor.ts | 145 ++++++++++---- 3 files changed, 284 insertions(+), 40 deletions(-) diff --git a/src/sessions/session-upstream-monitor.test-support.ts b/src/sessions/session-upstream-monitor.test-support.ts index 35c189ad5716..da8d1cb8493f 100644 --- a/src/sessions/session-upstream-monitor.test-support.ts +++ b/src/sessions/session-upstream-monitor.test-support.ts @@ -8,6 +8,7 @@ import "./session-upstream-monitor.js"; type SessionUpstreamMonitorOptions = OpenClawStateDatabaseOptions & { providers?: readonly SessionCatalogProvider[]; now?: () => number; + signal?: AbortSignal; loadEntry?: typeof loadSessionEntry; isRunActive?: typeof isEmbeddedAgentRunActive; loadOwnRecentUserTexts?: (params: { diff --git a/src/sessions/session-upstream-monitor.test.ts b/src/sessions/session-upstream-monitor.test.ts index 3005a859956d..767ba6720115 100644 --- a/src/sessions/session-upstream-monitor.test.ts +++ b/src/sessions/session-upstream-monitor.test.ts @@ -6,6 +6,7 @@ import { } from "../config/sessions/session-accessor.js"; import { importSessionCatalogHistory } from "../plugins/session-catalog-history-import.js"; import type { SessionCatalogProvider, SessionUpstreamProbe } from "../plugins/session-catalog.js"; +import { createDeferred } from "../shared/deferred.js"; import { closeOpenClawStateDatabaseForTest, openOpenClawStateDatabase, @@ -16,6 +17,7 @@ import { readSessionUpstreamLink, upsertSessionUpstreamLink, } from "./session-upstream-links.js"; +import { startSessionUpstreamMonitor } from "./session-upstream-monitor.js"; import { runSessionUpstreamMonitorTick } from "./session-upstream-monitor.test-support.js"; const tempDirs: string[] = []; @@ -69,6 +71,7 @@ function provider( } afterEach(() => { + vi.useRealTimers(); closeOpenClawStateDatabaseForTest(); vi.unstubAllEnvs(); }); @@ -171,6 +174,122 @@ describe("session upstream monitor", () => { ]); }); + it("defers a third missing result when a run starts during the provider scan", async () => { + const database = createDatabaseOptions(); + const sessionKey = "agent:main:adopted:missing-active-race"; + createLink(sessionKey, "claude", database); + let active = false; + let scan = 0; + const check = vi.fn(async () => { + scan += 1; + if (scan === 3) { + active = true; + } + return [{ kind: "missing" as const, sessionKey }]; + }); + const options = { + ...database, + providers: [provider("claude", check)], + loadEntry: () => ({ sessionId: "session-missing-active-race" }) as never, + isRunActive: () => active, + loadOwnRecentUserTexts: async () => [], + }; + const missingCounts = createMissingCounts(); + + await runSessionUpstreamMonitorTick(options, missingCounts); + await runSessionUpstreamMonitorTick(options, missingCounts); + await runSessionUpstreamMonitorTick(options, missingCounts); + + expect(readSessionUpstreamLink(sessionKey, "main", database)).toBeDefined(); + expect(listSessionStateEventsSince(sessionKey, "main", 0, 20, database).events).toEqual([]); + expect([...missingCounts.values()].map((counter) => counter.count)).toEqual([2]); + + active = false; + await runSessionUpstreamMonitorTick(options, missingCounts); + + expect(check).toHaveBeenCalledTimes(4); + expect(readSessionUpstreamLink(sessionKey, "main", database)).toBeUndefined(); + expect(listSessionStateEventsSince(sessionKey, "main", 0, 20, database).events).toEqual([ + expect.objectContaining({ kind: "upstream_missing" }), + ]); + }); + + it("resets a missing streak when the session is replaced during the provider scan", async () => { + const database = createDatabaseOptions(); + const sessionKey = "agent:main:adopted:missing-session-replaced"; + createLink(sessionKey, "claude", database); + let sessionId = "session-before"; + let scan = 0; + const check = vi.fn(async () => { + scan += 1; + if (scan === 3) { + sessionId = "session-after"; + } + return [{ kind: "missing" as const, sessionKey }]; + }); + const options = { + ...database, + providers: [provider("claude", check)], + loadEntry: () => ({ sessionId }) as never, + isRunActive: () => false, + loadOwnRecentUserTexts: async () => [], + }; + const missingCounts = createMissingCounts(); + + await runSessionUpstreamMonitorTick(options, missingCounts); + await runSessionUpstreamMonitorTick(options, missingCounts); + await runSessionUpstreamMonitorTick(options, missingCounts); + + expect(missingCounts.size).toBe(0); + expect(readSessionUpstreamLink(sessionKey, "main", database)).toBeDefined(); + expect(listSessionStateEventsSince(sessionKey, "main", 0, 20, database).events).toEqual([]); + + await runSessionUpstreamMonitorTick(options, missingCounts); + await runSessionUpstreamMonitorTick(options, missingCounts); + + expect([...missingCounts.values()].map((counter) => counter.count)).toEqual([2]); + expect(readSessionUpstreamLink(sessionKey, "main", database)).toBeDefined(); + }); + + it("does not publish a deferred third missing result after monitor stop", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + const database = createDatabaseOptions(); + const sessionKey = "agent:main:adopted:missing-stopped"; + createLink(sessionKey, "claude", database); + const thirdResult = createDeferred>(); + let scan = 0; + const check = vi.fn(async () => { + scan += 1; + return scan === 3 ? await thirdResult.promise : [{ kind: "missing" as const, sessionKey }]; + }); + const monitor = startSessionUpstreamMonitor({ + ...database, + providers: [provider("claude", check)], + loadEntry: () => ({ sessionId: "session-missing-stopped" }) as never, + isRunActive: () => false, + loadOwnRecentUserTexts: async () => [], + }); + + try { + await vi.advanceTimersByTimeAsync(15_000); + await vi.advanceTimersByTimeAsync(45_000); + await vi.advanceTimersByTimeAsync(60_000); + expect(check).toHaveBeenCalledTimes(3); + + monitor.stop(); + thirdResult.resolve([{ kind: "missing", sessionKey }]); + for (let flush = 0; flush < 10; flush += 1) { + await Promise.resolve(); + } + + expect(readSessionUpstreamLink(sessionKey, "main", database)).toBeDefined(); + expect(listSessionStateEventsSince(sessionKey, "main", 0, 20, database).events).toEqual([]); + } finally { + monitor.stop(); + } + }); + it("resets consecutive misses on activity", async () => { const database = createDatabaseOptions(); const sessionKey = "agent:main:adopted:missing-reset"; @@ -579,6 +698,65 @@ describe("session upstream monitor", () => { expect(listSessionStateEventsSince(sessionKey, "main", 0, 20, database).events).toEqual([]); }); + it.each([ + { + change: "a run starts", + mutate: (state: { active: boolean; sessionId: string }) => { + state.active = true; + }, + }, + { + change: "the session is replaced", + mutate: (state: { active: boolean; sessionId: string }) => { + state.sessionId = "session-after"; + }, + }, + ])("defers activity when $change during final provenance I/O", async ({ mutate }) => { + const database = createDatabaseOptions(); + const sessionKey = "agent:main:adopted:provenance-race"; + createLink(sessionKey, "claude", database); + const state = { active: false, sessionId: "session-before" }; + const provenanceReadStarted = createDeferred(); + const provenanceResult = createDeferred(); + let provenanceReads = 0; + const loadOwnRecentUserTexts = vi.fn(async () => { + provenanceReads += 1; + if (provenanceReads === 2) { + provenanceReadStarted.resolve(); + return await provenanceResult.promise; + } + return []; + }); + const check = vi.fn(async () => [ + { + kind: "activity" as const, + sessionKey, + occurredAt: 2_000, + humanTurns: 1, + nextMarker: { offset: 12 }, + dedupeId: "12", + }, + ]); + + const tick = runSessionUpstreamMonitorTick({ + ...database, + providers: [provider("claude", check)], + loadEntry: () => ({ sessionId: state.sessionId }) as never, + isRunActive: () => state.active, + loadOwnRecentUserTexts, + }); + await provenanceReadStarted.promise; + mutate(state); + provenanceResult.resolve([]); + await tick; + + expect(check).toHaveBeenCalledOnce(); + expect(readSessionUpstreamLink(sessionKey, "main", database)).toEqual( + expect.objectContaining({ marker: { offset: 0 } }), + ); + expect(listSessionStateEventsSince(sessionKey, "main", 0, 20, database).events).toEqual([]); + }); + it("advances scan-only markers without recording an event", async () => { const database = createDatabaseOptions(); const sessionKey = "agent:main:adopted:scan-only"; diff --git a/src/sessions/session-upstream-monitor.ts b/src/sessions/session-upstream-monitor.ts index 8ce6cc65d875..3cb44ce984f6 100644 --- a/src/sessions/session-upstream-monitor.ts +++ b/src/sessions/session-upstream-monitor.ts @@ -30,6 +30,7 @@ const log = createSubsystemLogger("sessions/upstream-monitor"); type SessionUpstreamMonitorOptions = OpenClawStateDatabaseOptions & { providers?: readonly SessionCatalogProvider[]; now?: () => number; + signal?: AbortSignal; loadEntry?: typeof loadSessionEntryReadOnly; isRunActive?: typeof isEmbeddedAgentRunActive; loadOwnRecentUserTexts?: (params: { @@ -87,6 +88,51 @@ function upstreamMonitorLinkKey(probe: { return `${probe.sessionKey}\n${probe.agentId}\n${upstreamSourceKey(probe)}`; } +function loadProbeSession( + probe: Pick, + options: SessionUpstreamMonitorOptions, +): SessionEntry | undefined { + if (options.signal?.aborted) { + return undefined; + } + const entry = (options.loadEntry ?? loadSessionEntryReadOnly)({ + sessionKey: probe.sessionKey, + agentId: probe.agentId, + clone: false, + ...(options.env ? { env: options.env } : {}), + }); + return entry?.sessionId ? entry : undefined; +} + +function loadIdleProbeSession( + probe: Pick, + options: SessionUpstreamMonitorOptions, + expectedSessionId?: string, +): SessionEntry | undefined { + const entry = loadProbeSession(probe, options); + if ( + !entry || + (expectedSessionId !== undefined && entry.sessionId !== expectedSessionId) || + (options.isRunActive ?? isEmbeddedAgentRunActive)(entry.sessionId) + ) { + return undefined; + } + return entry; +} + +function readMatchingProbeLink( + probe: SessionUpstreamProbe, + expectedUpdatedAt: number | undefined, + options: OpenClawStateDatabaseOptions, +) { + const currentLink = readSessionUpstreamLink(probe.sessionKey, probe.agentId, options); + return currentLink && + currentLink.updatedAt === expectedUpdatedAt && + upstreamSourceKey(currentLink) === upstreamSourceKey(probe) + ? currentLink + : undefined; +} + async function loadOwnRecentUserTexts( probe: Omit, entry: SessionEntry, @@ -113,19 +159,21 @@ async function loadOwnRecentUserTexts( async function probeProvenanceUnchanged( probe: SessionUpstreamProbe, + expectedSessionId: string | undefined, options: SessionUpstreamMonitorOptions, ): Promise { - const entry = (options.loadEntry ?? loadSessionEntryReadOnly)({ - sessionKey: probe.sessionKey, - agentId: probe.agentId, - clone: false, - ...(options.env ? { env: options.env } : {}), - }); - if (!entry?.sessionId || (options.isRunActive ?? isEmbeddedAgentRunActive)(entry.sessionId)) { + const entry = loadIdleProbeSession(probe, options, expectedSessionId); + if (!entry) { return false; } const current = await loadOwnRecentUserTexts(probe, entry, options); + // Transcript I/O can outlive run admission or session replacement. Recheck + // the current idle owner after the last await before committing its result. + if (!loadIdleProbeSession(probe, options, expectedSessionId)) { + return false; + } return ( + options.signal?.aborted !== true && current.length === probe.ownRecentUserTexts.length && current.every((text, index) => text === probe.ownRecentUserTexts[index]) ); @@ -135,6 +183,9 @@ async function runSessionUpstreamMonitorTick( options: SessionUpstreamMonitorOptions = {}, missingCounts: Map = new Map(), ): Promise { + if (options.signal?.aborted) { + return; + } const dbOptions = databaseOptions(options); const linksByCatalog = listWatchedSessionUpstreamLinks(dbOptions); const watchedLinkKeys = new Set( @@ -154,6 +205,7 @@ async function runSessionUpstreamMonitorTick( continue; } const probes: SessionUpstreamProbe[] = []; + const sessionIdBySessionKey = new Map(); for (const link of links) { const probe = { sessionKey: link.sessionKey, @@ -166,24 +218,21 @@ async function runSessionUpstreamMonitorTick( } satisfies Omit; // One corrupt session store must not reject the whole tick; skip that link only. try { - const entry = (options.loadEntry ?? loadSessionEntryReadOnly)({ - sessionKey: probe.sessionKey, - agentId: probe.agentId, - clone: false, - ...(options.env ? { env: options.env } : {}), - }); + const entry = loadIdleProbeSession(probe, options); // Active runs may still append upstream user items. Defer the scan so their // marker remains available for positive transcript-provenance matching. - if ( - !entry?.sessionId || - (options.isRunActive ?? isEmbeddedAgentRunActive)(entry.sessionId) - ) { + if (!entry) { continue; } + const ownRecentUserTexts = await loadOwnRecentUserTexts(probe, entry, options); + if (options.signal?.aborted) { + return; + } probes.push({ ...probe, - ownRecentUserTexts: await loadOwnRecentUserTexts(probe, entry, options), + ownRecentUserTexts, }); + sessionIdBySessionKey.set(probe.sessionKey, entry.sessionId); } catch (error) { log.warn(`upstream transcript provenance failed for ${probe.sessionKey}: ${String(error)}`); } @@ -197,6 +246,9 @@ async function runSessionUpstreamMonitorTick( ); try { const outcomes = await provider.checkUpstreamActivity(probes); + if (options.signal?.aborted) { + return; + } const missingSessionKeys = new Set( outcomes .filter((outcome) => outcome.kind === "missing") @@ -215,10 +267,29 @@ async function runSessionUpstreamMonitorTick( const missingCountKey = upstreamMonitorLinkKey(probe); if (outcome.kind === "missing") { const expectedUpdatedAt = linkUpdatedAtBySessionKey.get(outcome.sessionKey); - if (expectedUpdatedAt === undefined) { + const expectedSessionId = sessionIdBySessionKey.get(outcome.sessionKey); + // Provider I/O may outlive a new run or Continue. Only an idle current + // session and the exact scanned link can advance the missing streak. + if (expectedUpdatedAt === undefined || expectedSessionId === undefined) { missingCounts.delete(missingCountKey); continue; } + if (options.signal?.aborted) { + return; + } + const currentLink = readMatchingProbeLink(probe, expectedUpdatedAt, dbOptions); + if (!currentLink) { + missingCounts.delete(missingCountKey); + continue; + } + const currentSession = loadProbeSession(probe, options); + if (!currentSession || currentSession.sessionId !== expectedSessionId) { + missingCounts.delete(missingCountKey); + continue; + } + if ((options.isRunActive ?? isEmbeddedAgentRunActive)(currentSession.sessionId)) { + continue; + } const previous = missingCounts.get(missingCountKey); const missingCount = Math.min( SESSION_UPSTREAM_MISSING_THRESHOLD, @@ -231,15 +302,6 @@ async function runSessionUpstreamMonitorTick( if (missingCount < SESSION_UPSTREAM_MISSING_THRESHOLD) { continue; } - const currentLink = readSessionUpstreamLink(probe.sessionKey, probe.agentId, dbOptions); - if ( - !currentLink || - currentLink.updatedAt !== expectedUpdatedAt || - upstreamSourceKey(currentLink) !== upstreamSourceKey(probe) - ) { - missingCounts.delete(missingCountKey); - continue; - } const sourceKey = upstreamSourceKey(probe); const recorded = recordSessionStateEvent( { @@ -272,7 +334,13 @@ async function runSessionUpstreamMonitorTick( try { // A run can start while the provider is scanning. Recheck ownership and // provenance before any marker advance so its prompt remains deferred. - if (!(await probeProvenanceUnchanged(probe, options))) { + if ( + !(await probeProvenanceUnchanged( + probe, + sessionIdBySessionKey.get(probe.sessionKey), + options, + )) + ) { continue; } } catch (error) { @@ -286,19 +354,10 @@ async function runSessionUpstreamMonitorTick( // From here to the record the path is synchronous, so a stale scan can // neither record from the old source nor clobber the refreshed marker. const expectedUpdatedAt = linkUpdatedAtBySessionKey.get(activity.sessionKey); - const currentLink = readSessionUpstreamLink(probe.sessionKey, probe.agentId, dbOptions); // Compare source identity too: a same-millisecond Continue can refresh the // row without changing updated_at, so the timestamp alone is not a reliable // optimistic lock. - if ( - !currentLink || - currentLink.updatedAt !== expectedUpdatedAt || - upstreamSourceKey({ - hostId: currentLink.hostId, - threadId: currentLink.threadId, - upstreamRef: currentLink.upstreamRef, - }) !== upstreamSourceKey(probe) - ) { + if (!readMatchingProbeLink(probe, expectedUpdatedAt, dbOptions)) { continue; } if (activity.humanTurns === 0) { @@ -347,13 +406,15 @@ export function startSessionUpstreamMonitor( ): SessionUpstreamMonitor { let stopped = false; let running = false; + const lifecycle = new AbortController(); + const tickOptions = { ...options, signal: lifecycle.signal }; const missingCounts = new Map(); const run = () => { if (stopped || running) { return; } running = true; - void runSessionUpstreamMonitorTick(options, missingCounts) + void runSessionUpstreamMonitorTick(tickOptions, missingCounts) .catch((error: unknown) => { log.warn(`upstream monitor tick failed: ${String(error)}`); }) @@ -368,7 +429,11 @@ export function startSessionUpstreamMonitor( interval.unref?.(); return { stop: () => { + if (stopped) { + return; + } stopped = true; + lifecycle.abort(); clearTimeout(initialTimer); clearInterval(interval); },