feat(sessions): signal watchers when an adopted upstream disappears (#107356)

2026-07-21 10:16:44 +00:00 · 2026-07-14 03:51:20 -07:00
parent c5da3462b8
commit e488dc0012
11 changed files with 581 additions and 25 deletions
@@ -23,6 +23,7 @@ OpenClaw appends a typed event to the shared state database (`session_state_even
 | Kind                   | Recorded when                                            | Notifies watchers |
 | ---------------------- | -------------------------------------------------------- | ----------------- |
 | `human_direct_message` | A human sends a turn directly to a watched session       | Yes               |
+| `upstream_missing`     | An adopted session's upstream source disappears          | Yes               |
 | `goal_changed`         | The session's goal state is created, updated, or cleared | Yes               |
 | `child_spawned`        | A sub-agent or ACP child session is created              | No (seeds cursor) |
 | `run_completed`        | A child run ends successfully                            | No (log only)     |
@@ -49,6 +50,8 @@ Watches clean themselves up: cursor rows expire with signal-log retention, are r

 Watched sessions adopted from a session catalog are checked for direct upstream human activity on a fixed cadence. Detected activity enters the same signal log and watcher flow as other direct human turns.

+If an adopted session's upstream source is deleted externally, three consecutive missing checks (about three monitor ticks) produce one `upstream_missing` signal for its watchers and remove the upstream link. Continuing the catalog session again creates a fresh link.
+
 ## Notices: one, not many

 When a notify-eligible event lands and a watcher's cursor is behind, the watcher receives one system notice on its next turn:
@@ -103,6 +106,7 @@ Current limits:
 - Upstream self-echo detection compares normalized user text. An external prompt matching one of the session's 10 most recent OpenClaw-side user messages is treated as self-echo.
 - A single local Claude JSONL row larger than the 1 MiB per-cadence scan cap blocks that session's cursor in v1; unclassified bytes are never skipped.
 - Paired-node Claude checks classify the latest 50 transcript items per cadence. Larger bursts can fall outside the v1 scan window.
+- Paired-node Claude history reads do not expose a definitive thread-not-found result, so remote Claude deletions are not classified as `upstream_missing` in v1.
 - Catalog sessions that have not been adopted remain outside the awareness layer in v1.
 - Sessions adopted before this feature carry no upstream link; continue them from the catalog once to start upstream monitoring.
 - Upstream links assume each adopted session key maps to one owning agent (adoption uses the default store agent). Multi-agent adoption of the same external thread is not monitored in v1.
@@ -2,7 +2,7 @@ import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
 import type { SessionUpstreamProbe } from "openclaw/plugin-sdk/session-catalog";
-import { afterAll, describe, expect, it } from "vitest";
+import { afterAll, afterEach, describe, expect, it, vi } from "vitest";
 import {
  MAX_CLAUDE_UPSTREAM_SCAN_BYTES,
  checkClaudeSessionUpstreamActivity,
@@ -37,6 +37,10 @@ afterAll(async () => {
  await Promise.all(tempDirs.map((dir) => fs.rm(dir, { recursive: true, force: true })));
 });

+afterEach(() => {
+  vi.restoreAllMocks();
+});
+
 describe("Claude upstream activity", () => {
  it("counts only external user rows after the byte marker", async () => {
    const dir = await makeTempDir("openclaw-claude-upstream-");
@@ -93,6 +97,7 @@ describe("Claude upstream activity", () => {
    const activity = await checkClaudeSessionUpstreamActivity(probe);

    expect(activity).toEqual({
+      kind: "activity",
      sessionKey: probe.sessionKey,
      occurredAt: Date.parse("2026-07-13T10:05:00.000Z"),
      humanTurns: 1,
@@ -143,12 +148,54 @@ describe("Claude upstream activity", () => {
        ownRecentUserTexts: ["same prompt"],
      }),
    ).resolves.toEqual({
+      kind: "activity",
      sessionKey: "agent:main:adopted:claude-provenance",
      humanTurns: 0,
      nextMarker: { offset: (await fs.stat(filePath)).size },
    });
  });

+  it("returns missing for an absent local transcript", async () => {
+    const filePath = path.join(
+      await makeTempDir("openclaw-claude-upstream-missing-"),
+      "gone.jsonl",
+    );
+
+    await expect(
+      checkClaudeSessionUpstreamActivity({
+        sessionKey: "agent:main:adopted:claude-missing",
+        agentId: "main",
+        threadId: "thread-missing",
+        hostId: "gateway:local",
+        upstreamKind: "claude-cli",
+        upstreamRef: { filePath },
+        marker: { offset: 3 },
+        ownRecentUserTexts: [],
+      }),
+    ).resolves.toEqual({
+      kind: "missing",
+      sessionKey: "agent:main:adopted:claude-missing",
+    });
+  });
+
+  it("swallows non-missing local transcript errors", async () => {
+    const error = Object.assign(new Error("permission denied"), { code: "EACCES" });
+    vi.spyOn(fs, "open").mockRejectedValueOnce(error);
+
+    await expect(
+      checkClaudeSessionUpstreamActivity({
+        sessionKey: "agent:main:adopted:claude-permission",
+        agentId: "main",
+        threadId: "thread-permission",
+        hostId: "gateway:local",
+        upstreamKind: "claude-cli",
+        upstreamRef: { filePath: "/unreadable/thread.jsonl" },
+        marker: { offset: 3 },
+        ownRecentUserTexts: [],
+      }),
+    ).resolves.toBeUndefined();
+  });
+
  it("isolates a missing transcript from healthy probes", async () => {
    const dir = await makeTempDir("openclaw-claude-upstream-batch-");
    const filePath = path.join(dir, "thread-good.jsonl");
@@ -176,7 +223,10 @@ describe("Claude upstream activity", () => {
        { ...baseProbe, sessionKey: "stale", upstreamRef: { filePath: `${filePath}.missing` } },
        baseProbe,
      ]),
-    ).resolves.toEqual([expect.objectContaining({ sessionKey: "healthy", humanTurns: 1 })]);
+    ).resolves.toEqual([
+      { kind: "missing", sessionKey: "stale" },
+      expect.objectContaining({ kind: "activity", sessionKey: "healthy", humanTurns: 1 }),
+    ]);
  });

  it("keeps continuation successful when baseline enumeration fails", async () => {
@@ -212,6 +262,7 @@ describe("Claude upstream activity", () => {
      ]),
    ).resolves.toEqual([
      {
+        kind: "activity",
        sessionKey: "remote",
        occurredAt: Date.parse("2026-07-13T10:07:00.000Z"),
        humanTurns: 1,
@@ -253,14 +304,19 @@ describe("Claude upstream activity", () => {

    const first = await checkClaudeSessionUpstreamActivity(baseProbe);
    expect(first).toEqual({
+      kind: "activity",
      sessionKey: baseProbe.sessionKey,
      humanTurns: 0,
      nextMarker: { offset: Buffer.byteLength(firstRow) },
    });
+    if (first?.kind !== "activity") {
+      throw new Error("expected activity marker");
+    }
    await expect(
-      checkClaudeSessionUpstreamActivity({ ...baseProbe, marker: first?.nextMarker ?? null }),
+      checkClaudeSessionUpstreamActivity({ ...baseProbe, marker: first.nextMarker }),
    ).resolves.toEqual(
      expect.objectContaining({
+        kind: "activity",
        humanTurns: 1,
        nextMarker: { offset: Buffer.byteLength(firstRow + userRow + finalRow) },
      }),
@@ -296,7 +352,10 @@ describe("Claude upstream activity", () => {
      marker: { size: Buffer.byteLength(baseline) },
    });
    expect(sizeResult).toEqual(offsetResult);
-    expect(offsetResult?.nextMarker).toEqual({ offset: (await fs.stat(filePath)).size });
+    expect(offsetResult?.kind).toBe("activity");
+    if (offsetResult?.kind === "activity") {
+      expect(offsetResult.nextMarker).toEqual({ offset: (await fs.stat(filePath)).size });
+    }
  });
  it("declines a remote link when the newest history item lacks a UUID", async () => {
    const readRemote = async () => [{ type: "userMessage", text: "hi" }] as never;
@@ -124,10 +124,20 @@ export async function checkClaudeSessionUpstreamActivity(
  if (!filePath || markerOffset === undefined) {
    return undefined;
  }
-  const handle = await fs.open(filePath, "r");
+  let handle: Awaited<ReturnType<typeof fs.open>>;
+  try {
+    handle = await fs.open(filePath, "r");
+  } catch (error) {
+    return isRecord(error) && error.code === "ENOENT"
+      ? { kind: "missing", sessionKey: probe.sessionKey }
+      : undefined;
+  }
  try {
    const stat = await handle.stat();
-    if (!stat.isFile() || stat.size <= markerOffset) {
+    if (!stat.isFile()) {
+      return { kind: "missing", sessionKey: probe.sessionKey };
+    }
+    if (stat.size <= markerOffset) {
      return undefined;
    }
    const readLength = Math.min(stat.size - markerOffset, MAX_CLAUDE_UPSTREAM_SCAN_BYTES);
@@ -160,6 +170,7 @@ export async function checkClaudeSessionUpstreamActivity(
    }
    const nextOffset = markerOffset + lastNewline + 1;
    return {
+      kind: "activity",
      sessionKey: probe.sessionKey,
      humanTurns,
      nextMarker: { offset: nextOffset },
@@ -224,6 +235,7 @@ async function checkRemoteClaudeSessionUpstreamActivity(
  }
  const activityId = newest.uuid;
  return {
+    kind: "activity",
    sessionKey: probe.sessionKey,
    humanTurns,
    nextMarker: { uuid: activityId },
@@ -1,5 +1,6 @@
 import type { SessionUpstreamProbe } from "openclaw/plugin-sdk/session-catalog";
 import { describe, expect, it, vi } from "vitest";
+import { CodexAppServerRpcError } from "./app-server/client.js";
 import type { CodexTurn } from "./app-server/protocol.js";
 import {
  checkCodexUpstreamActivity,
@@ -45,6 +46,7 @@ describe("Codex upstream activity", () => {
        ],
      }),
    ).toEqual({
+      kind: "activity",
      sessionKey: "agent:main:adopted:codex",
      occurredAt: 300_000,
      humanTurns: 1,
@@ -53,13 +55,28 @@ describe("Codex upstream activity", () => {
    });
  });

+  it("leaves empty-page existence classification to the provider", () => {
+    expect(classifyCodexUpstreamTurns({ probe: probe(), turns: [] })).toBeUndefined();
+  });
+
+  it("accepts an empty page for a thread with no materialized turn", () => {
+    expect(
+      classifyCodexUpstreamTurns({
+        probe: probe({ marker: { turnId: null, userMessageCount: 0 } }),
+        turns: [],
+      }),
+    ).toBeUndefined();
+  });
+
  it("uses the pinned connection and a bounded descending summary page", async () => {
    const listTurnPage = vi.fn(async () => ({
      data: [turn("turn-2", ["userMessage"], 200), turn("turn-1", [], 100)],
    }));
+    const readThread = vi.fn(async () => ({}) as never);
    const control: Parameters<typeof checkCodexUpstreamActivity>[1] = {
      connectionFingerprint: "connection-1",
      listTurnPage,
+      readThread,
      withPinnedConnection: async (run) => await run(control),
    };

@@ -72,6 +89,95 @@ describe("Codex upstream activity", () => {
      sortDirection: "desc",
      itemsView: "full",
    });
+    expect(readThread).not.toHaveBeenCalled();
+  });
+
+  it("keeps a rolled-back thread linked when thread/read succeeds", async () => {
+    const readThread = vi.fn(async () => ({}) as never);
+    const control: Parameters<typeof checkCodexUpstreamActivity>[1] = {
+      connectionFingerprint: "connection-1",
+      listTurnPage: async () => ({ data: [] }),
+      readThread,
+      withPinnedConnection: async (run) => await run(control),
+    };
+
+    await expect(
+      checkCodexUpstreamActivity([probe()], control, async () => "thread-canonical"),
+    ).resolves.toEqual([]);
+    expect(readThread).toHaveBeenCalledWith("thread-canonical", false);
+  });
+
+  it("reports missing when thread/read rejects with the definitive not-found code", async () => {
+    const readThread = vi.fn(async () => {
+      throw new CodexAppServerRpcError(
+        { code: -32600, message: "thread not loaded: thread-1" },
+        "thread/read",
+      );
+    });
+    const control: Parameters<typeof checkCodexUpstreamActivity>[1] = {
+      connectionFingerprint: "connection-1",
+      listTurnPage: async () => ({ data: [] }),
+      readThread,
+      withPinnedConnection: async (run) => await run(control),
+    };
+
+    await expect(checkCodexUpstreamActivity([probe()], control)).resolves.toEqual([
+      { kind: "missing", sessionKey: "agent:main:adopted:codex" },
+    ]);
+    expect(readThread).toHaveBeenCalledWith("thread-1", false);
+  });
+
+  it("treats non-definitive thread/read failures as inconclusive", async () => {
+    const readThread = vi.fn(async () => {
+      throw new CodexAppServerRpcError({ code: -32603, message: "store hiccup" }, "thread/read");
+    });
+    const control: Parameters<typeof checkCodexUpstreamActivity>[1] = {
+      connectionFingerprint: "connection-1",
+      listTurnPage: async () => ({ data: [] }),
+      readThread,
+      withPinnedConnection: async (run) => await run(control),
+    };
+
+    await expect(checkCodexUpstreamActivity([probe()], control)).resolves.toEqual([]);
+  });
+
+  it("treats other invalid-request thread/read failures as inconclusive", async () => {
+    const readThread = vi.fn(async () => {
+      throw new CodexAppServerRpcError(
+        { code: -32600, message: "some other validation failure" },
+        "thread/read",
+      );
+    });
+    const control: Parameters<typeof checkCodexUpstreamActivity>[1] = {
+      connectionFingerprint: "connection-1",
+      listTurnPage: async () => ({ data: [] }),
+      readThread,
+      withPinnedConnection: async (run) => await run(control),
+    };
+
+    await expect(checkCodexUpstreamActivity([probe()], control)).resolves.toEqual([]);
+  });
+
+  it("detects deletion of a thread baselined with no materialized turn", async () => {
+    const readThread = vi.fn(async () => {
+      throw new CodexAppServerRpcError(
+        { code: -32600, message: "thread not loaded: thread-1" },
+        "thread/read",
+      );
+    });
+    const control: Parameters<typeof checkCodexUpstreamActivity>[1] = {
+      connectionFingerprint: "connection-1",
+      listTurnPage: async () => ({ data: [] }),
+      readThread,
+      withPinnedConnection: async (run) => await run(control),
+    };
+
+    await expect(
+      checkCodexUpstreamActivity(
+        [probe({ marker: { turnId: null, userMessageCount: 0 } })],
+        control,
+      ),
+    ).resolves.toEqual([{ kind: "missing", sessionKey: "agent:main:adopted:codex" }]);
  });

  it("isolates a stale thread from healthy probes", async () => {
@@ -84,6 +190,7 @@ describe("Codex upstream activity", () => {
    const control: Parameters<typeof checkCodexUpstreamActivity>[1] = {
      connectionFingerprint: "connection-1",
      listTurnPage,
+      readThread: async () => ({}) as never,
      withPinnedConnection: async (run) => await run(control),
    };

@@ -102,6 +209,7 @@ describe("Codex upstream activity", () => {
        turns: [turn("turn-1", ["userMessage", "agentMessage", "userMessage"], 100)],
      }),
    ).toEqual({
+      kind: "activity",
      sessionKey: "agent:main:adopted:codex",
      occurredAt: 100_000,
      humanTurns: 1,
@@ -117,6 +225,7 @@ describe("Codex upstream activity", () => {
        turns: [turn("turn-1", ["userMessage", "agentMessage", "userMessage"], 100)],
      }),
    ).toEqual({
+      kind: "activity",
      sessionKey: "agent:main:adopted:codex",
      humanTurns: 0,
      nextMarker: { turnId: "turn-1", userMessageCount: 2 },
@@ -145,6 +254,7 @@ describe("Codex upstream activity", () => {
        ],
      }),
    ).toEqual({
+      kind: "activity",
      sessionKey: "agent:main:adopted:codex",
      humanTurns: 0,
      nextMarker: { turnId: "turn-1", userMessageCount: 1 },
@@ -176,6 +286,7 @@ describe("Codex upstream activity", () => {
        ],
      }),
    ).toEqual({
+      kind: "activity",
      sessionKey: "agent:main:adopted:codex",
      humanTurns: 0,
      nextMarker: { turnId: "turn-1", userMessageCount: 1 },
@@ -6,7 +6,9 @@ import type {
  SessionUpstreamProbe,
 } from "openclaw/plugin-sdk/session-catalog";
 import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime";
+import { CodexAppServerRpcError } from "./app-server/client.js";
 import type {
+  CodexThread,
  CodexThreadTurnsListParams,
  CodexThreadTurnsListResponse,
  CodexTurn,
@@ -19,11 +21,26 @@ import {
 import type { CodexSessionCatalogControl } from "./session-catalog-types.js";

 const CODEX_UPSTREAM_TURN_LIMIT = 100;
+// codex-rs app-server thread/read maps a gone rollout to JSON-RPC invalid_request
+// with exactly this message prefix (read_thread_view "thread not loaded"). The code
+// alone is generic (other store validation reuses it), so both must match; a harness
+// message rename degrades to the old silent gap instead of unlinking live threads.
+const CODEX_APP_SERVER_INVALID_REQUEST_CODE = -32600;
+const CODEX_THREAD_NOT_LOADED_MESSAGE_PREFIX = "thread not loaded:";
+
+function isCodexThreadGoneError(error: unknown): boolean {
+  return (
+    error instanceof CodexAppServerRpcError &&
+    error.code === CODEX_APP_SERVER_INVALID_REQUEST_CODE &&
+    error.message.startsWith(CODEX_THREAD_NOT_LOADED_MESSAGE_PREFIX)
+  );
+}

 type CodexUpstreamControl = {
  connectionFingerprint?: string;
  withPinnedConnection<T>(run: (control: CodexUpstreamControl) => Promise<T>): Promise<T>;
  listTurnPage(params: CodexThreadTurnsListParams): Promise<CodexThreadTurnsListResponse>;
+  readThread(threadId: string, includeTurns?: boolean): Promise<CodexThread>;
 };

 type CodexUpstreamMarker = {
@@ -61,8 +78,11 @@ export function classifyCodexUpstreamTurns(params: {
  now?: number;
 }): SessionUpstreamActivity | undefined {
  const marker = readMarker(params.probe);
+  if (!marker) {
+    return undefined;
+  }
  const newest = params.turns[0];
-  if (!marker || !newest?.id) {
+  if (!newest?.id) {
    return undefined;
  }
  const markerIndex =
@@ -103,6 +123,7 @@ export function classifyCodexUpstreamTurns(params: {
  }
  const activityId = `${newest.id}:${newestUserMessageCount}`;
  return {
+    kind: "activity",
    sessionKey: params.probe.sessionKey,
    humanTurns,
    nextMarker: { turnId: newest.id, userMessageCount: newestUserMessageCount },
@@ -146,18 +167,34 @@ export async function checkCodexUpstreamActivity(
        continue;
      }
      try {
+        const threadId = await resolveThreadId(probe);
        const page = await pinned.listTurnPage({
-          threadId: await resolveThreadId(probe),
+          threadId,
          limit: CODEX_UPSTREAM_TURN_LIMIT,
          sortDirection: "desc",
          itemsView: "full",
        });
+        const marker = readMarker(probe);
+        if (page.data.length === 0 && marker) {
+          // Deleted threads do NOT reject turns/list: codex-rs load_thread_turns_list_history
+          // swallows ThreadNotFound/no-rollout and returns an empty page, and rollback can
+          // empty a live thread too. thread/read is the existence oracle: it still succeeds
+          // after rollback and rejects "thread not loaded" only once the rollout is gone.
+          try {
+            await pinned.readThread(threadId, false);
+          } catch (error) {
+            if (isCodexThreadGoneError(error)) {
+              activities.push({ kind: "missing", sessionKey: probe.sessionKey });
+            }
+          }
+          continue;
+        }
        const activity = classifyCodexUpstreamTurns({ probe, turns: page.data });
        if (activity) {
          activities.push(activity);
        }
      } catch {
-        // Stale links must not suppress healthy sessions in the same provider batch.
+        // One transient probe failure must not suppress healthy sessions in the same batch.
      }
    }
    return activities;
@@ -550,7 +550,7 @@ describe("session_status tool", () => {
    });
    getSessionStateVersionMock.mockReturnValue(12);
    listSessionStateEventsSinceMock.mockReturnValue({
-      events: [{ sequence: 12, kind: "goal_changed", summary: "goal created" }],
+      events: [{ sequence: 12, kind: "upstream_missing", summary: "upstream missing via codex" }],
      truncated: false,
      earliestAvailableSequence: 12,
      historyGap: true,
@@ -565,6 +565,7 @@ describe("session_status tool", () => {
    expect(details.stateVersion).toBe(12);
    expect(details.stateChanges).toMatchObject({ historyGap: true });
    expect(text).toContain("Session state changes:");
+    expect(text).toContain('"kind": "upstream_missing"');
  });

  it("enables transcript usage fallback for session_status", async () => {
@@ -60,13 +60,16 @@ export type SessionUpstreamProbe = {
  ownRecentUserTexts: string[];
 };

-export type SessionUpstreamActivity = {
-  sessionKey: string;
-  humanTurns: number;
-  nextMarker: SessionUpstreamJsonValue;
-  occurredAt?: number;
-  dedupeId?: string;
-};
+export type SessionUpstreamActivity =
+  | {
+      kind: "activity";
+      sessionKey: string;
+      humanTurns: number;
+      nextMarker: SessionUpstreamJsonValue;
+      occurredAt?: number;
+      dedupeId?: string;
+    }
+  | { kind: "missing"; sessionKey: string };

 export type SessionCatalogContinueProviderResult = {
  sessionKey: string;
@@ -7,11 +7,13 @@ export type SessionStateEventKind =
  | "run_failed"
  | "child_spawned"
  | "goal_changed"
+  | "upstream_missing"
  | "compacted";

 // Future utility-model materiality belongs at this deterministic seam; no config until then.
 export const NOTIFY_BY_SESSION_STATE_EVENT_KIND: Record<SessionStateEventKind, boolean> = {
  human_direct_message: true,
+  upstream_missing: true,
  adopted: false,
  goal_changed: true,
  run_completed: false,
@@ -243,6 +243,20 @@ describe("session state events", () => {
    expect(peekSystemEventEntries(watcher)).toEqual([]);
  });

+  it("notifies watchers when an upstream session disappears", () => {
+    const database = createDatabaseOptions();
+    seedChild(database);
+    resetSystemEventsForTest();
+
+    const event = recordSessionStateEvent(
+      eventInput({ kind: "upstream_missing", actorType: "system" }),
+      database,
+    );
+
+    expect(event).toMatchObject({ kind: "upstream_missing", actorType: "system" });
+    expect(peekSystemEventEntries(watcher)).toHaveLength(1);
+  });
+
  it("returns the existing row for a duplicate dedupe key", () => {
    const database = createDatabaseOptions();
    const input = eventInput({
@@ -10,12 +10,20 @@ import {
  openOpenClawStateDatabase,
 } from "../state/openclaw-state-db.js";
 import { listSessionStateEventsSince, registerSessionStateWatch } from "./session-state-events.js";
-import { upsertSessionUpstreamLink } from "./session-upstream-links.js";
+import {
+  deleteSessionUpstreamLink,
+  readSessionUpstreamLink,
+  upsertSessionUpstreamLink,
+} from "./session-upstream-links.js";
 import { runSessionUpstreamMonitorTick } from "./session-upstream-monitor.js";

 const tempDirs: string[] = [];
 const watcherSessionKey = "agent:main:main";

+function createMissingCounts() {
+  return new Map<string, { count: number; linkUpdatedAt: number }>();
+}
+
 function createDatabaseOptions() {
  const stateDir = makeTempDir(tempDirs, "openclaw-session-upstream-monitor-");
  vi.stubEnv("OPENCLAW_STATE_DIR", stateDir);
@@ -77,6 +85,7 @@ describe("session upstream monitor", () => {
    createLink(unwatched, "claude", database, false);
    const checkUpstreamActivity = vi.fn(async (probes: SessionUpstreamProbe[]) =>
      probes.map((probe) => ({
+        kind: "activity" as const,
        sessionKey: probe.sessionKey,
        occurredAt: 2_000,
        humanTurns: 1,
@@ -127,6 +136,202 @@ describe("session upstream monitor", () => {
    expect(dedupeRow.dedupe_key).toMatch(new RegExp(`^upstream:${watched}:[0-9a-f]{16}:8$`));
  });

+  it("records one upstream-missing event after three misses and removes the link", async () => {
+    const database = createDatabaseOptions();
+    const sessionKey = "agent:main:adopted:missing";
+    createLink(sessionKey, "claude", database);
+    const check = vi.fn(async (probes: SessionUpstreamProbe[]) =>
+      probes.map((probe) => ({ kind: "missing" as const, sessionKey: probe.sessionKey })),
+    );
+    const options = {
+      ...database,
+      providers: [provider("claude", check)],
+      now: () => 3_000,
+      loadEntry: () => ({ sessionId: "session-missing" }) as never,
+      loadOwnRecentUserTexts: async () => [],
+    };
+    const missingCounts = createMissingCounts();
+
+    await runSessionUpstreamMonitorTick(options, missingCounts);
+    await runSessionUpstreamMonitorTick(options, missingCounts);
+    await runSessionUpstreamMonitorTick(options, missingCounts);
+    await runSessionUpstreamMonitorTick(options, missingCounts);
+
+    expect(check).toHaveBeenCalledTimes(3);
+    expect(readSessionUpstreamLink(sessionKey, "main", database)).toBeUndefined();
+    expect(missingCounts.size).toBe(0);
+    expect(listSessionStateEventsSince(sessionKey, "main", 0, 20, database).events).toEqual([
+      expect.objectContaining({
+        kind: "upstream_missing",
+        actorType: "system",
+        summary: "upstream missing via claude",
+        payload: { channel: "claude" },
+      }),
+    ]);
+  });
+
+  it("resets consecutive misses on activity", async () => {
+    const database = createDatabaseOptions();
+    const sessionKey = "agent:main:adopted:missing-reset";
+    createLink(sessionKey, "claude", database);
+    let scan = 0;
+    const check = vi.fn(async () => {
+      scan += 1;
+      return scan === 3
+        ? [
+            {
+              kind: "activity" as const,
+              sessionKey,
+              humanTurns: 0,
+              nextMarker: { offset: 3 },
+            },
+          ]
+        : [{ kind: "missing" as const, sessionKey }];
+    });
+    const options = {
+      ...database,
+      providers: [provider("claude", check)],
+      loadEntry: () => ({ sessionId: "session-missing-reset" }) as never,
+      loadOwnRecentUserTexts: async () => [],
+    };
+    const missingCounts = createMissingCounts();
+
+    for (let index = 0; index < 5; index += 1) {
+      await runSessionUpstreamMonitorTick(options, missingCounts);
+    }
+
+    expect(check).toHaveBeenCalledTimes(5);
+    expect(listSessionStateEventsSince(sessionKey, "main", 0, 20, database).events).toEqual([]);
+    expect(readSessionUpstreamLink(sessionKey, "main", database)).toBeDefined();
+    expect([...missingCounts.values()].map((counter) => counter.count)).toEqual([2]);
+  });
+
+  it("breaks a missing streak when a successful probe has no missing outcome", async () => {
+    const database = createDatabaseOptions();
+    const sessionKey = "agent:main:adopted:missing-quiet";
+    createLink(sessionKey, "claude", database);
+    let scan = 0;
+    const check = vi.fn(async () => {
+      scan += 1;
+      return scan === 2 ? [] : [{ kind: "missing" as const, sessionKey }];
+    });
+    const options = {
+      ...database,
+      providers: [provider("claude", check)],
+      loadEntry: () => ({ sessionId: "session-missing-quiet" }) as never,
+      loadOwnRecentUserTexts: async () => [],
+    };
+    const missingCounts = createMissingCounts();
+
+    for (let index = 0; index < 4; index += 1) {
+      await runSessionUpstreamMonitorTick(options, missingCounts);
+    }
+
+    expect(listSessionStateEventsSince(sessionKey, "main", 0, 20, database).events).toEqual([]);
+    expect(readSessionUpstreamLink(sessionKey, "main", database)).toBeDefined();
+    expect([...missingCounts.values()].map((counter) => counter.count)).toEqual([2]);
+  });
+
+  it("starts a fresh streak when Continue refreshes the same source", async () => {
+    const database = createDatabaseOptions();
+    const sessionKey = "agent:main:adopted:missing-same-source";
+    createLink(sessionKey, "claude", database);
+    const check = vi.fn(async () => [{ kind: "missing" as const, sessionKey }]);
+    const options = {
+      ...database,
+      providers: [provider("claude", check)],
+      loadEntry: () => ({ sessionId: "session-missing-same-source" }) as never,
+      loadOwnRecentUserTexts: async () => [],
+    };
+    const missingCounts = createMissingCounts();
+
+    await runSessionUpstreamMonitorTick(options, missingCounts);
+    await runSessionUpstreamMonitorTick(options, missingCounts);
+    upsertSessionUpstreamLink(
+      {
+        sessionKey,
+        agentId: "main",
+        catalogId: "claude",
+        hostId: "gateway:local",
+        threadId: "thread-claude",
+        upstreamKind: "claude-cli",
+        upstreamRef: { source: "claude" },
+        marker: { offset: 0 },
+      },
+      { ...database, now: 7_777 },
+    );
+    await runSessionUpstreamMonitorTick(options, missingCounts);
+
+    expect(listSessionStateEventsSince(sessionKey, "main", 0, 20, database).events).toEqual([]);
+    expect(readSessionUpstreamLink(sessionKey, "main", database)).toBeDefined();
+    expect([...missingCounts.values()].map((counter) => counter.count)).toEqual([1]);
+  });
+
+  it("aborts missing record and deletion when Continue changes the source", async () => {
+    const database = createDatabaseOptions();
+    const sessionKey = "agent:main:adopted:missing-refreshed";
+    createLink(sessionKey, "claude", database);
+    let scan = 0;
+    const check = vi.fn(async () => {
+      scan += 1;
+      if (scan === 3) {
+        upsertSessionUpstreamLink(
+          {
+            sessionKey,
+            agentId: "main",
+            catalogId: "claude",
+            hostId: "gateway:local",
+            threadId: "thread-refreshed",
+            upstreamKind: "claude-cli",
+            upstreamRef: { source: "refreshed" },
+            marker: { offset: 999 },
+          },
+          { ...database, now: 7_777 },
+        );
+      }
+      return [{ kind: "missing" as const, sessionKey }];
+    });
+    const options = {
+      ...database,
+      providers: [provider("claude", check)],
+      loadEntry: () => ({ sessionId: "session-missing-refreshed" }) as never,
+      loadOwnRecentUserTexts: async () => [],
+    };
+    const missingCounts = createMissingCounts();
+
+    await runSessionUpstreamMonitorTick(options, missingCounts);
+    await runSessionUpstreamMonitorTick(options, missingCounts);
+    await runSessionUpstreamMonitorTick(options, missingCounts);
+
+    expect(listSessionStateEventsSince(sessionKey, "main", 0, 20, database).events).toEqual([]);
+    expect(readSessionUpstreamLink(sessionKey, "main", database)).toEqual(
+      expect.objectContaining({ threadId: "thread-refreshed", marker: { offset: 999 } }),
+    );
+    expect(missingCounts.size).toBe(0);
+  });
+
+  it("prunes missing counters when a link leaves the watched set", async () => {
+    const database = createDatabaseOptions();
+    const sessionKey = "agent:main:adopted:missing-pruned";
+    createLink(sessionKey, "claude", database);
+    const check = vi.fn(async () => [{ kind: "missing" as const, sessionKey }]);
+    const options = {
+      ...database,
+      providers: [provider("claude", check)],
+      loadEntry: () => ({ sessionId: "session-missing-pruned" }) as never,
+      loadOwnRecentUserTexts: async () => [],
+    };
+    const missingCounts = createMissingCounts();
+
+    await runSessionUpstreamMonitorTick(options, missingCounts);
+    expect(missingCounts.size).toBe(1);
+    deleteSessionUpstreamLink(sessionKey, "main", database);
+    await runSessionUpstreamMonitorTick(options, missingCounts);
+
+    expect(check).toHaveBeenCalledOnce();
+    expect(missingCounts.size).toBe(0);
+  });
+
  it("clamps skewed upstream event times without touching bookkeeping clocks", async () => {
    const database = createDatabaseOptions();
    const watched = "agent:main:adopted:clamped";
@@ -135,6 +340,7 @@ describe("session upstream monitor", () => {
    const ancient = 1_000; // far beyond the 24h clamp window
    const claude = provider("claude", async (probes: SessionUpstreamProbe[]) =>
      probes.map((probe) => ({
+        kind: "activity" as const,
        sessionKey: probe.sessionKey,
        occurredAt: ancient,
        humanTurns: 1,
@@ -182,6 +388,7 @@ describe("session upstream monitor", () => {
        { ...database, now: 7_777 },
      );
      return probes.map((probe) => ({
+        kind: "activity" as const,
        sessionKey: probe.sessionKey,
        occurredAt: 2_000,
        humanTurns: 1,
@@ -213,6 +420,7 @@ describe("session upstream monitor", () => {
    createLink(healthy, "claude", database);
    const claude = provider("claude", async (probes: SessionUpstreamProbe[]) =>
      probes.map((probe) => ({
+        kind: "activity" as const,
        sessionKey: probe.sessionKey,
        occurredAt: 2_500,
        humanTurns: 1,
@@ -248,6 +456,7 @@ describe("session upstream monitor", () => {
      providers: [
        provider("codex", async () => [
          {
+            kind: "activity" as const,
            sessionKey,
            occurredAt: 2_000,
            humanTurns: 3,
@@ -275,6 +484,7 @@ describe("session upstream monitor", () => {
    createLink(codexSession, "codex", database);
    const codexCheck = vi.fn(async () => [
      {
+        kind: "activity" as const,
        sessionKey: codexSession,
        occurredAt: 5_000,
        humanTurns: 1,
@@ -337,6 +547,7 @@ describe("session upstream monitor", () => {
      active = true;
      return [
        {
+          kind: "activity" as const,
          sessionKey,
          occurredAt: 2_000,
          humanTurns: 1,
@@ -373,7 +584,9 @@ describe("session upstream monitor", () => {
    createLink(sessionKey, "claude", database);
    const check = vi
      .fn<NonNullable<SessionCatalogProvider["checkUpstreamActivity"]>>()
-      .mockResolvedValueOnce([{ sessionKey, humanTurns: 0, nextMarker: { offset: 12 } }])
+      .mockResolvedValueOnce([
+        { kind: "activity" as const, sessionKey, humanTurns: 0, nextMarker: { offset: 12 } },
+      ])
      .mockResolvedValueOnce([]);

    const options = {
@@ -415,6 +628,7 @@ describe("session upstream monitor", () => {
    createLink(sessionKey, "claude", database);
    const check = vi.fn(async (probes: SessionUpstreamProbe[]) => [
      {
+        kind: "activity" as const,
        sessionKey,
        humanTurns: probes[0]?.ownRecentUserTexts.includes("exact decorated prompt") ? 0 : 1,
        nextMarker: { offset: 20 },
@@ -443,6 +657,7 @@ describe("session upstream monitor", () => {
      providers: [
        provider("claude", async () => [
          {
+            kind: "activity" as const,
            sessionKey,
            occurredAt: 10_000,
            humanTurns: 1,
@@ -9,8 +9,12 @@ import { createSubsystemLogger } from "../logging/subsystem.js";
 import { getPluginRegistryState } from "../plugins/runtime-state.js";
 import type { SessionCatalogProvider, SessionUpstreamProbe } from "../plugins/session-catalog.js";
 import type { OpenClawStateDatabaseOptions } from "../state/openclaw-state-db.js";
-import { recordSessionHumanDirectMessage } from "./session-state-events.js";
 import {
+  recordSessionHumanDirectMessage,
+  recordSessionStateEvent,
+} from "./session-state-events.js";
+import {
+  deleteSessionUpstreamLink,
  listWatchedSessionUpstreamLinks,
  readSessionUpstreamLink,
  updateSessionUpstreamLinkMarker,
@@ -19,6 +23,7 @@ import {
 const SESSION_UPSTREAM_MONITOR_INTERVAL_MS = 60_000;
 const SESSION_UPSTREAM_MONITOR_INITIAL_DELAY_MS = 15_000;
 const SESSION_UPSTREAM_OWN_USER_TEXT_LIMIT = 10;
+const SESSION_UPSTREAM_MISSING_THRESHOLD = 3;

 const log = createSubsystemLogger("sessions/upstream-monitor");

@@ -35,6 +40,11 @@ type SessionUpstreamMonitorOptions = OpenClawStateDatabaseOptions & {

 export type SessionUpstreamMonitor = { stop: () => void };

+type SessionUpstreamMissingCounter = {
+  count: number;
+  linkUpdatedAt: number;
+};
+
 function currentProviders(): SessionCatalogProvider[] {
  return (getPluginRegistryState()?.activeRegistry?.sessionCatalogs ?? []).map(
    (registration) => registration.provider,
@@ -67,6 +77,16 @@ function upstreamSourceKey(probe: {
    .slice(0, 16);
 }

+function upstreamMonitorLinkKey(probe: {
+  sessionKey: string;
+  agentId: string;
+  hostId: string;
+  threadId: string;
+  upstreamRef: unknown;
+}): string {
+  return `${probe.sessionKey}\n${probe.agentId}\n${upstreamSourceKey(probe)}`;
+}
+
 async function loadOwnRecentUserTexts(
  probe: Omit<SessionUpstreamProbe, "ownRecentUserTexts">,
  entry: SessionEntry,
@@ -113,9 +133,19 @@ async function probeProvenanceUnchanged(

 export async function runSessionUpstreamMonitorTick(
  options: SessionUpstreamMonitorOptions = {},
+  missingCounts: Map<string, SessionUpstreamMissingCounter> = new Map(),
 ): Promise<void> {
  const dbOptions = databaseOptions(options);
  const linksByCatalog = listWatchedSessionUpstreamLinks(dbOptions);
+  const watchedLinkKeys = new Set(
+    [...linksByCatalog.values()].flatMap((links) => links.map(upstreamMonitorLinkKey)),
+  );
+  // Monitor-owned counters must follow the watched-link lifecycle or churn would leak keys.
+  for (const key of missingCounts.keys()) {
+    if (!watchedLinkKeys.has(key)) {
+      missingCounts.delete(key);
+    }
+  }
  const providers = options.providers ?? currentProviders();
  const providerById = new Map(providers.map((provider) => [provider.id, provider]));
  for (const [catalogId, links] of linksByCatalog) {
@@ -166,10 +196,77 @@ export async function runSessionUpstreamMonitorTick(
      links.map((link) => [link.sessionKey, link.updatedAt]),
    );
    try {
-      const activities = await provider.checkUpstreamActivity(probes);
-      for (const activity of activities) {
-        const probe = probeBySessionKey.get(activity.sessionKey);
-        if (!probe || !Number.isSafeInteger(activity.humanTurns) || activity.humanTurns < 0) {
+      const outcomes = await provider.checkUpstreamActivity(probes);
+      const missingSessionKeys = new Set(
+        outcomes
+          .filter((outcome) => outcome.kind === "missing")
+          .map((outcome) => outcome.sessionKey),
+      );
+      for (const probe of probes) {
+        if (!missingSessionKeys.has(probe.sessionKey)) {
+          missingCounts.delete(upstreamMonitorLinkKey(probe));
+        }
+      }
+      for (const outcome of outcomes) {
+        const probe = probeBySessionKey.get(outcome.sessionKey);
+        if (!probe) {
+          continue;
+        }
+        const missingCountKey = upstreamMonitorLinkKey(probe);
+        if (outcome.kind === "missing") {
+          const expectedUpdatedAt = linkUpdatedAtBySessionKey.get(outcome.sessionKey);
+          if (expectedUpdatedAt === undefined) {
+            missingCounts.delete(missingCountKey);
+            continue;
+          }
+          const previous = missingCounts.get(missingCountKey);
+          const missingCount = Math.min(
+            SESSION_UPSTREAM_MISSING_THRESHOLD,
+            (previous?.linkUpdatedAt === expectedUpdatedAt ? previous.count : 0) + 1,
+          );
+          missingCounts.set(missingCountKey, {
+            count: missingCount,
+            linkUpdatedAt: expectedUpdatedAt,
+          });
+          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(
+            {
+              sessionKey: probe.sessionKey,
+              agentId: probe.agentId,
+              kind: "upstream_missing",
+              actorType: "system",
+              dedupeKey: `upstream-missing:${probe.sessionKey}:${sourceKey}:${currentLink.updatedAt}`,
+              summary: `upstream missing via ${catalogId}`,
+              payload: { channel: catalogId },
+            },
+            { ...dbOptions, now: (options.now ?? Date.now)() },
+          );
+          if (!recorded) {
+            missingCounts.set(missingCountKey, {
+              count: SESSION_UPSTREAM_MISSING_THRESHOLD - 1,
+              linkUpdatedAt: expectedUpdatedAt,
+            });
+            continue;
+          }
+          deleteSessionUpstreamLink(probe.sessionKey, probe.agentId, dbOptions);
+          missingCounts.delete(missingCountKey);
+          continue;
+        }
+        missingCounts.delete(missingCountKey);
+        const activity = outcome;
+        if (!Number.isSafeInteger(activity.humanTurns) || activity.humanTurns < 0) {
          continue;
        }
        try {
@@ -250,12 +347,13 @@ export function startSessionUpstreamMonitor(
 ): SessionUpstreamMonitor {
  let stopped = false;
  let running = false;
+  const missingCounts = new Map<string, SessionUpstreamMissingCounter>();
  const run = () => {
    if (stopped || running) {
      return;
    }
    running = true;
-    void runSessionUpstreamMonitorTick(options)
+    void runSessionUpstreamMonitorTick(options, missingCounts)
      .catch((error: unknown) => {
        log.warn(`upstream monitor tick failed: ${String(error)}`);
      })