mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-27 04:47:03 -06:00
fix(sessions): prevent stale upstream activity after session changes (#116864)
* fix(sessions): fence upstream monitor results * fix(sessions): recheck provenance ownership
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
36cc7bb105
commit
39d4ca8f4e
@@ -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: {
|
||||
|
||||
@@ -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<Array<{ kind: "missing"; sessionKey: string }>>();
|
||||
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<string[]>();
|
||||
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";
|
||||
|
||||
@@ -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<SessionUpstreamProbe, "sessionKey" | "agentId">,
|
||||
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<SessionUpstreamProbe, "sessionKey" | "agentId">,
|
||||
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<SessionUpstreamProbe, "ownRecentUserTexts">,
|
||||
entry: SessionEntry,
|
||||
@@ -113,19 +159,21 @@ async function loadOwnRecentUserTexts(
|
||||
|
||||
async function probeProvenanceUnchanged(
|
||||
probe: SessionUpstreamProbe,
|
||||
expectedSessionId: string | undefined,
|
||||
options: SessionUpstreamMonitorOptions,
|
||||
): Promise<boolean> {
|
||||
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<string, SessionUpstreamMissingCounter> = new Map(),
|
||||
): Promise<void> {
|
||||
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<string, string>();
|
||||
for (const link of links) {
|
||||
const probe = {
|
||||
sessionKey: link.sessionKey,
|
||||
@@ -166,24 +218,21 @@ async function runSessionUpstreamMonitorTick(
|
||||
} satisfies Omit<SessionUpstreamProbe, "ownRecentUserTexts">;
|
||||
// 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<string, SessionUpstreamMissingCounter>();
|
||||
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);
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user