refactor(sessions): replace ambient watch sentinels with provenance (#110428)

* refactor(sessions): store watch provenance in cursors

* style(docs): align database schema table

* refactor(state): use Kysely for watch migration

* style(state): remove redundant timestamp casts

docs/reference/database-schemas.md

@@ -52,6 +52,7 @@ Version 3 was an unshipped development step folded into version 4.
 | 1       | Initial shared state database                                                                            | `v2026.5.30-beta.1` |
 | 2       | Metadata-only message audit events ([#103903](https://github.com/openclaw/openclaw/pull/103903))         | `v2026.7.2-beta.1`  |
 | 3       | `STRICT` tables and schema-drift hardening ([#108663](https://github.com/openclaw/openclaw/pull/108663)) | `v2026.7.2-beta.2`  |
+| 4       | Session watch provenance replaces encoded sentinel rows                                                  | Unreleased          |
 
 ## Integrity checks
 

extensions/voice-call/doctor-contract-api.ts

@@ -147,6 +147,8 @@ function describeVoiceCallSchemaMigration(migration: OpenClawStateDatabaseSchema
       return "audit event ledger -> versioned message lifecycle schema";
     case "operator-approvals-system-agent":
       return "operator approvals -> OpenClaw system changes";
+    case "session-watch-cursor-provenance-v4":
+      return "session watch cursors -> provenance column";
     case "strict-tables-v3":
       return "tables -> SQLite STRICT typing";
   }

package.json

@@ -3,7 +3,7 @@
   "version": "2026.7.2",
   "openclaw": {
     "schemaVersions": {
-      "state": 3,
+      "state": 4,
       "agent": 11
     }
   },

src/infra/state-migrations.doctor.ts

@@ -154,6 +154,8 @@ function describeStateSchemaMigration(migration: OpenClawStateDatabaseSchemaMigr
       return "audit event ledger → versioned message lifecycle schema";
     case "operator-approvals-system-agent":
       return "operator approvals → OpenClaw system changes";
+    case "session-watch-cursor-provenance-v4":
+      return "session watch cursors → provenance column";
     case "strict-tables-v3":
       return "tables → SQLite STRICT typing";
   }

src/plugin-sdk/session-visibility.ts

@@ -324,9 +324,9 @@ export function createSessionVisibilityRowChecker(params: {
     const targetAgentId = row.agentId ?? resolveAgentIdFromSessionKey(targetSessionKey);
     const isRequesterSession =
       targetSessionKey === params.requesterSessionKey || targetSessionKey === "current";
-    // Only a durable ambient-group marker makes the paired target
-    // ownership-equivalent for same-agent reads. Explicit A2A watches, absent
-    // markers, send access, and cross-agent targets remain fail-closed.
+    // Only durable ambient-group provenance makes the target ownership-equivalent
+    // for same-agent reads. Explicit A2A watches, send access, and cross-agent
+    // targets remain fail-closed.
     const isWatchedRead =
       params.action !== "send" &&
       params.visibility === "tree" &&

src/sessions/session-state-events.test.ts

@@ -558,7 +558,7 @@ describe("session state events", () => {
 
     const rows = openOpenClawStateDatabase(database)
       .db.prepare(
-        `SELECT watcher_session_key, target_session_key, updated_at
+        `SELECT watcher_session_key, target_session_key, provenance, updated_at
          FROM session_watch_cursors
          WHERE watcher_session_key = ?`,
       )
@@ -567,6 +567,7 @@ describe("session state events", () => {
       {
         watcher_session_key: watcher,
         target_session_key: group,
+        provenance: "ambient-group",
         updated_at: 100,
       },
     ]);
@@ -639,7 +640,7 @@ describe("session state events", () => {
     expect(wakes).not.toHaveBeenCalled();
   });
 
-  it("prunes orphaned ambient markers while retaining markers for active cursors", () => {
+  it("prunes dormant ambient cursors while retaining active cursors", () => {
     const database = createDatabaseOptions();
     const dormantGroup = "agent:main:slack:channel:dormant";
     const registeredAt = 100;
@@ -669,7 +670,7 @@ describe("session state events", () => {
     const cursors = openOpenClawStateDatabase(database)
       .db.prepare("SELECT COUNT(*) AS count FROM session_watch_cursors")
       .get() as { count: number };
-    expect(cursors.count).toBe(2);
+    expect(cursors.count).toBe(1);
   });
 
   it("keeps explicit A2A group watches on the immediate wake path", async () => {
@@ -716,6 +717,14 @@ describe("session state events", () => {
 
     registerSessionStateWatch({ watcherSessionKey: watcher, targetSessionKey: group }, database);
     expect(listAmbientGroupWatchTargets(watcher, database)).toEqual(new Set());
+    expect(
+      openOpenClawStateDatabase(database)
+        .db.prepare(
+          `SELECT provenance FROM session_watch_cursors
+           WHERE watcher_session_key = ? AND target_session_key = ?`,
+        )
+        .get(watcher, group),
+    ).toEqual({ provenance: "explicit" });
     // Later inbound group registration must not downgrade the explicit watch.
     registerMainSessionGroupWatch(
       { sessionKey: group, agentId: "main", dmScope: "main" },

src/sessions/session-state-events.ts

@@ -18,6 +18,11 @@ import {
   runOpenClawStateWriteTransaction,
   type OpenClawStateDatabaseOptions,
 } from "../state/openclaw-state-db.js";
+import {
+  SESSION_WATCH_PROVENANCE_AMBIENT_GROUP,
+  SESSION_WATCH_PROVENANCE_EXPLICIT,
+  type SessionWatchCursorProvenance,
+} from "../state/session-watch-cursor-provenance.js";
 import { classifySessionKind } from "./classify-session-kind.js";
 import type { InputProvenance } from "./input-provenance.js";
 import {
@@ -70,7 +75,6 @@ type SessionWatchCursorRow = Selectable<OpenClawStateKyselyDatabase["session_wat
 const SESSION_STATE_RETENTION_MS = 30 * 24 * 60 * 60_000;
 const SESSION_STATE_MAX_ROWS = 50_000;
 const SESSION_STATE_PRUNE_INTERVAL_MS = 60 * 60_000;
-const AMBIENT_GROUP_WATCH_MARKER_PREFIX = "ambient-group-watch:";
 const log = createSubsystemLogger("sessions/state-events");
 let lastPruneAt = 0;
 
@@ -149,24 +153,8 @@ function readCursor(
   );
 }
 
-function ambientGroupWatchMarkerKey(watcherSessionKey: string): string {
-  return `${AMBIENT_GROUP_WATCH_MARKER_PREFIX}${Buffer.from(watcherSessionKey, "utf8").toString("hex")}`;
-}
-
-function decodeAmbientGroupWatchMarkerKey(markerKey: string): string | undefined {
-  const encoded = markerKey.slice(AMBIENT_GROUP_WATCH_MARKER_PREFIX.length);
-  if (!encoded || encoded.length % 2 !== 0 || !/^[0-9a-f]+$/.test(encoded)) {
-    return undefined;
-  }
-  return Buffer.from(encoded, "hex").toString("utf8");
-}
-
-function hasAmbientGroupWatchMarker(
-  db: DatabaseSync,
-  watcherSessionKey: string,
-  targetSessionKey: string,
-): boolean {
-  return Boolean(readCursor(db, ambientGroupWatchMarkerKey(watcherSessionKey), targetSessionKey));
+function isAmbientGroupWatchCursor(row: SessionWatchCursorRow | undefined): boolean {
+  return row?.provenance === SESSION_WATCH_PROVENANCE_AMBIENT_GROUP;
 }
 
 function upsertSeedCursor(params: {
@@ -175,6 +163,7 @@ function upsertSeedCursor(params: {
   targetSessionKey: string;
   sequence: number;
   now: number;
+  provenance?: SessionWatchCursorProvenance;
 }): void {
   executeSqliteQuerySync(
     params.db,
@@ -186,6 +175,7 @@ function upsertSeedCursor(params: {
         last_seen_sequence: params.sequence,
         notified_sequence: params.sequence,
         material_sequence: params.sequence,
+        provenance: params.provenance ?? SESSION_WATCH_PROVENANCE_EXPLICIT,
         updated_at: params.now,
       })
       .onConflict((conflict) =>
@@ -205,7 +195,7 @@ function updateMaterialCursor(params: {
   targetSessionKey: string;
   sequence: number;
   now: number;
-}): number {
+}): { lastSeenSequence: number; queueOnly: boolean } {
   const current = readCursor(params.db, params.watcherSessionKey, params.targetSessionKey);
   const lastSeen = normalizeOptionalSqliteNumber(current?.last_seen_sequence) ?? 0;
   const notified = normalizeOptionalSqliteNumber(current?.notified_sequence) ?? 0;
@@ -220,6 +210,7 @@ function updateMaterialCursor(params: {
         last_seen_sequence: lastSeen,
         notified_sequence: frozenNotified,
         material_sequence: params.sequence,
+        provenance: SESSION_WATCH_PROVENANCE_EXPLICIT,
         updated_at: params.now,
       })
       .onConflict((conflict) =>
@@ -230,7 +221,7 @@ function updateMaterialCursor(params: {
         }),
       ),
   );
-  return lastSeen;
+  return { lastSeenSequence: lastSeen, queueOnly: isAmbientGroupWatchCursor(current) };
 }
 
 /** Classify the actor once at producer boundaries; missing provenance is interactive human input. */
@@ -357,7 +348,7 @@ export function recordSessionStateEvent(
         if (!NOTIFY_BY_KIND[input.kind] || input.actorId === watcherSessionKey) {
           continue;
         }
-        const lastSeenSequence = updateMaterialCursor({
+        const materialCursor = updateMaterialCursor({
           db,
           watcherSessionKey,
           targetSessionKey: input.sessionKey,
@@ -367,8 +358,8 @@ export function recordSessionStateEvent(
         notices.push({
           watcherSessionKey,
           targetSessionKey: input.sessionKey,
-          lastSeenSequence,
-          queueOnly: hasAmbientGroupWatchMarker(db, watcherSessionKey, input.sessionKey),
+          lastSeenSequence: materialCursor.lastSeenSequence,
+          queueOnly: materialCursor.queueOnly,
         });
       }
 
@@ -557,7 +548,7 @@ export function acknowledgeSessionStateNotices(
             watcherSessionKey,
             targetSessionKey,
             lastSeenSequence: notified,
-            queueOnly: hasAmbientGroupWatchMarker(db, watcherSessionKey, targetSessionKey),
+            queueOnly: isAmbientGroupWatchCursor(row),
           });
         }
       }
@@ -577,18 +568,11 @@ export function handleSessionStateSessionReset(
 ): void {
   try {
     runOpenClawStateWriteTransaction(({ db }) => {
-      // Notifiable cursor rows use agent-qualified keys. The encoded ambient
-      // marker is deleted alongside its owner without interpreting bare keys.
       executeSqliteQuerySync(
         db,
         getSessionStateKysely(db)
           .deleteFrom("session_watch_cursors")
-          .where((eb) =>
-            eb.or([
-              eb("watcher_session_key", "=", sessionKey),
-              eb("watcher_session_key", "=", ambientGroupWatchMarkerKey(sessionKey)),
-            ]),
-          ),
+          .where("watcher_session_key", "=", sessionKey),
       );
     }, options);
   } catch (error) {
@@ -627,7 +611,6 @@ export function handleSessionStateSessionDeleted(
           .where((eb) =>
             eb.or([
               eb("watcher_session_key", "=", sessionKey),
-              eb("watcher_session_key", "=", ambientGroupWatchMarkerKey(sessionKey)),
               eb("target_session_key", "=", sessionKey),
             ]),
           ),
@@ -677,7 +660,7 @@ export function sweepSessionStateWatchNotices(
         watcherSessionKey: row.watcher_session_key,
         targetSessionKey: row.target_session_key,
         lastSeenSequence: normalizeSqliteNumber(row.last_seen_sequence) ?? 0,
-        queueOnly: hasAmbientGroupWatchMarker(db, row.watcher_session_key, row.target_session_key),
+        queueOnly: isAmbientGroupWatchCursor(row),
       });
     }
     pruneSessionStateEvents({ ...options, now });
@@ -751,35 +734,8 @@ function pruneSessionStateEvents(
       const cursorCutoff = now - SESSION_STATE_RETENTION_MS;
       executeSqliteQuerySync(
         db,
-        kysely
-          .deleteFrom("session_watch_cursors")
-          .where("updated_at", "<", cursorCutoff)
-          .where("watcher_session_key", "not like", `${AMBIENT_GROUP_WATCH_MARKER_PREFIX}%`),
+        kysely.deleteFrom("session_watch_cursors").where("updated_at", "<", cursorCutoff),
       );
-      const staleMarkers = executeSqliteQuerySync(
-        db,
-        kysely
-          .selectFrom("session_watch_cursors")
-          .select(["watcher_session_key", "target_session_key"])
-          .where("updated_at", "<", cursorCutoff)
-          .where("watcher_session_key", "like", `${AMBIENT_GROUP_WATCH_MARKER_PREFIX}%`),
-      ).rows;
-      for (const marker of staleMarkers) {
-        const watcherSessionKey = decodeAmbientGroupWatchMarkerKey(marker.watcher_session_key);
-        if (
-          watcherSessionKey &&
-          readCursor(db, watcherSessionKey, marker.target_session_key) !== undefined
-        ) {
-          continue;
-        }
-        executeSqliteQuerySync(
-          db,
-          kysely
-            .deleteFrom("session_watch_cursors")
-            .where("watcher_session_key", "=", marker.watcher_session_key)
-            .where("target_session_key", "=", marker.target_session_key),
-        );
-      }
     }, options);
     lastPruneAt = now;
   } catch (error) {
@@ -847,7 +803,6 @@ function hasSessionStateWatchers(
         .selectFrom("session_watch_cursors")
         .select("watcher_session_key")
         .where("target_session_key", "=", targetSessionKey)
-        .where("watcher_session_key", "not like", `${AMBIENT_GROUP_WATCH_MARKER_PREFIX}%`)
         .limit(1),
     );
     return row !== undefined;
@@ -870,13 +825,10 @@ export function listAmbientGroupWatchTargets(
       getSessionStateKysely(db)
         .selectFrom("session_watch_cursors")
         .select("target_session_key")
-        .where("watcher_session_key", "=", ambientGroupWatchMarkerKey(watcherSessionKey)),
+        .where("watcher_session_key", "=", watcherSessionKey)
+        .where("provenance", "=", SESSION_WATCH_PROVENANCE_AMBIENT_GROUP),
     ).rows;
-    return new Set(
-      rows
-        .map((row) => row.target_session_key)
-        .filter((targetSessionKey) => readCursor(db, watcherSessionKey, targetSessionKey)),
-    );
+    return new Set(rows.map((row) => row.target_session_key));
   } catch (error) {
     log.warn(`failed to list ambient group watch targets: ${String(error)}`);
     return new Set();
@@ -898,17 +850,19 @@ export function registerSessionStateWatch(
   try {
     let registered = false;
     runOpenClawStateWriteTransaction(({ db }) => {
-      // An explicit watch promotes an ambient group watch back to the normal
-      // immediate-wake path and removes its tree-read authorization marker.
-      executeSqliteQuerySync(
-        db,
-        getSessionStateKysely(db)
-          .deleteFrom("session_watch_cursors")
-          .where("watcher_session_key", "=", ambientGroupWatchMarkerKey(params.watcherSessionKey))
-          .where("target_session_key", "=", params.targetSessionKey),
-      );
       // Re-watching must not clobber pending-notice cursor state.
-      if (readCursor(db, params.watcherSessionKey, params.targetSessionKey)) {
+      const existing = readCursor(db, params.watcherSessionKey, params.targetSessionKey);
+      if (existing) {
+        if (existing.provenance !== SESSION_WATCH_PROVENANCE_EXPLICIT) {
+          executeSqliteQuerySync(
+            db,
+            getSessionStateKysely(db)
+              .updateTable("session_watch_cursors")
+              .set({ provenance: SESSION_WATCH_PROVENANCE_EXPLICIT })
+              .where("watcher_session_key", "=", params.watcherSessionKey)
+              .where("target_session_key", "=", params.targetSessionKey),
+          );
+        }
         registered = true;
         return;
       }
@@ -958,14 +912,13 @@ export function registerMainSessionGroupWatch(
   try {
     const { db: readDb } = openOpenClawStateDatabase(options);
     if (params.dmScope !== "main") {
-      const markerKey = ambientGroupWatchMarkerKey(watcherSessionKey);
-      if (!readCursor(readDb, markerKey, params.sessionKey)) {
+      if (!isAmbientGroupWatchCursor(readCursor(readDb, watcherSessionKey, params.sessionKey))) {
         return false;
       }
       runOpenClawStateWriteTransaction(({ db }) => {
         // Recheck provenance in the write transaction: an explicit registration
         // may have promoted this pair after the read-only preflight.
-        if (!readCursor(db, markerKey, params.sessionKey)) {
+        if (!isAmbientGroupWatchCursor(readCursor(db, watcherSessionKey, params.sessionKey))) {
           return;
         }
         executeSqliteQuerySync(
@@ -973,7 +926,8 @@ export function registerMainSessionGroupWatch(
           getSessionStateKysely(db)
             .deleteFrom("session_watch_cursors")
             .where("target_session_key", "=", params.sessionKey)
-            .where("watcher_session_key", "in", [watcherSessionKey, markerKey]),
+            .where("watcher_session_key", "=", watcherSessionKey)
+            .where("provenance", "=", SESSION_WATCH_PROVENANCE_AMBIENT_GROUP),
         );
       }, options);
       return false;
@@ -986,11 +940,9 @@ export function registerMainSessionGroupWatch(
     let registered = false;
     runOpenClawStateWriteTransaction(({ db }) => {
       const existing = readCursor(db, watcherSessionKey, params.sessionKey);
-      const markerKey = ambientGroupWatchMarkerKey(watcherSessionKey);
-      const marker = readCursor(db, markerKey, params.sessionKey);
       if (existing) {
-        // Missing marker means an explicit watch already owns this pair. Do not
-        // downgrade it when later human group turns revisit registration.
+        // An explicit watch already owns this pair. Do not downgrade it when
+        // later human group turns revisit registration.
         registered = true;
         return;
       }
@@ -1009,18 +961,8 @@ export function registerMainSessionGroupWatch(
         targetSessionKey: params.sessionKey,
         sequence,
         now,
+        provenance: SESSION_WATCH_PROVENANCE_AMBIENT_GROUP,
       });
-      if (!marker) {
-        // The paired marker is durable provenance, not a notifiable watcher. It
-        // authorizes tree reads and queue-only delivery only for this auto-watch.
-        upsertSeedCursor({
-          db,
-          watcherSessionKey: markerKey,
-          targetSessionKey: params.sessionKey,
-          sequence,
-          now,
-        });
-      }
       registered = true;
     }, options);
     return registered;

src/sessions/session-state-notices.ts

@@ -36,7 +36,7 @@ function shouldWakeWatcher(watcherSessionKey: string): boolean {
 // for one agent's child could be drained and acknowledged by another agent's global
 // turn — a cross-A2A metadata leak plus a lost notification. Until watcher identity
 // is agent-scoped end-to-end, such watchers get durable events and changesSince but
-// no notices. Non-notifiable ambient marker rows are also deliberately ignored.
+// no notices.
 export function isNotifiableWatcherKey(watcherSessionKey: string): boolean {
   return parseAgentSessionKey(watcherSessionKey) != null;
 }

src/state/openclaw-state-db-session-watch-migration.ts

@@ -0,0 +1,133 @@
+// Schema-v4 migration for legacy ambient session-watch sentinel rows.
+import type { DatabaseSync } from "node:sqlite";
+import {
+  executeSqliteQuerySync,
+  executeSqliteQueryTakeFirstSync,
+  getNodeSqliteKysely,
+} from "../infra/kysely-sync.js";
+import { ensureColumn, tableExists, tableHasColumn } from "./openclaw-state-db-schema-helpers.js";
+import type { DB as OpenClawStateKyselyDatabase } from "./openclaw-state-db.generated.js";
+import {
+  SESSION_WATCH_PROVENANCE_AMBIENT_GROUP,
+  SESSION_WATCH_PROVENANCE_EXPLICIT,
+} from "./session-watch-cursor-provenance.js";
+
+const SESSION_WATCH_PROVENANCE_SCHEMA_VERSION = 4;
+const LEGACY_AMBIENT_GROUP_WATCH_MARKER_PREFIX = "ambient-group-watch:";
+const SESSION_WATCH_PROVENANCE_COLUMN_SQL =
+  `provenance TEXT NOT NULL DEFAULT '${SESSION_WATCH_PROVENANCE_EXPLICIT}' ` +
+  `CHECK (provenance IN ('${SESSION_WATCH_PROVENANCE_EXPLICIT}', '${SESSION_WATCH_PROVENANCE_AMBIENT_GROUP}'))`;
+
+type SessionWatchCursorDatabase = Pick<OpenClawStateKyselyDatabase, "session_watch_cursors">;
+
+type SessionWatchCursorProvenanceMigrationResult = {
+  addedColumn: boolean;
+  migratedAmbientWatches: number;
+  removedLegacySentinels: number;
+};
+
+function getSessionWatchCursorKysely(db: DatabaseSync) {
+  return getNodeSqliteKysely<SessionWatchCursorDatabase>(db);
+}
+
+function hasLegacyAmbientWatchSentinels(db: DatabaseSync): boolean {
+  if (!tableExists(db, "session_watch_cursors")) {
+    return false;
+  }
+  return (
+    executeSqliteQueryTakeFirstSync(
+      db,
+      getSessionWatchCursorKysely(db)
+        .selectFrom("session_watch_cursors")
+        .select("watcher_session_key")
+        .where("watcher_session_key", "like", `${LEGACY_AMBIENT_GROUP_WATCH_MARKER_PREFIX}%`)
+        .limit(1),
+    ) !== undefined
+  );
+}
+
+export function needsSessionWatchCursorProvenanceMigration(
+  db: DatabaseSync,
+  userVersion: number,
+): boolean {
+  if (!tableExists(db, "session_watch_cursors")) {
+    return false;
+  }
+  return (
+    userVersion < SESSION_WATCH_PROVENANCE_SCHEMA_VERSION ||
+    !tableHasColumn(db, "session_watch_cursors", "provenance") ||
+    hasLegacyAmbientWatchSentinels(db)
+  );
+}
+
+function decodeLegacyAmbientWatchMarkerKey(markerKey: string): string | undefined {
+  const encoded = markerKey.slice(LEGACY_AMBIENT_GROUP_WATCH_MARKER_PREFIX.length);
+  if (!encoded || encoded.length % 2 !== 0 || !/^[0-9a-f]+$/.test(encoded)) {
+    return undefined;
+  }
+  return Buffer.from(encoded, "hex").toString("utf8");
+}
+
+export function migrateSessionWatchCursorProvenance(
+  db: DatabaseSync,
+): SessionWatchCursorProvenanceMigrationResult {
+  if (!tableExists(db, "session_watch_cursors")) {
+    return { addedColumn: false, migratedAmbientWatches: 0, removedLegacySentinels: 0 };
+  }
+
+  const addedColumn = ensureColumn(
+    db,
+    "session_watch_cursors",
+    SESSION_WATCH_PROVENANCE_COLUMN_SQL,
+  );
+  const kysely = getSessionWatchCursorKysely(db);
+  const legacyMarkers = executeSqliteQuerySync(
+    db,
+    kysely
+      .selectFrom("session_watch_cursors")
+      .select(["watcher_session_key", "target_session_key", "updated_at"])
+      .where("watcher_session_key", "like", `${LEGACY_AMBIENT_GROUP_WATCH_MARKER_PREFIX}%`),
+  ).rows;
+  let migratedAmbientWatches = 0;
+  for (const marker of legacyMarkers) {
+    const watcherSessionKey = decodeLegacyAmbientWatchMarkerKey(marker.watcher_session_key);
+    if (watcherSessionKey) {
+      // Startup and doctor callers hold BEGIN IMMEDIATE across this migration,
+      // so the paired timestamp read and update cannot lose a concurrent write.
+      const watch = executeSqliteQueryTakeFirstSync(
+        db,
+        kysely
+          .selectFrom("session_watch_cursors")
+          .select("updated_at")
+          .where("watcher_session_key", "=", watcherSessionKey)
+          .where("target_session_key", "=", marker.target_session_key),
+      );
+      if (watch) {
+        const promoted = executeSqliteQuerySync(
+          db,
+          kysely
+            .updateTable("session_watch_cursors")
+            .set({
+              provenance: SESSION_WATCH_PROVENANCE_AMBIENT_GROUP,
+              updated_at: Math.max(watch.updated_at, marker.updated_at),
+            })
+            .where("watcher_session_key", "=", watcherSessionKey)
+            .where("target_session_key", "=", marker.target_session_key),
+        );
+        migratedAmbientWatches += Number(promoted.numAffectedRows ?? 0n);
+      }
+    }
+    executeSqliteQuerySync(
+      db,
+      kysely
+        .deleteFrom("session_watch_cursors")
+        .where("watcher_session_key", "=", marker.watcher_session_key)
+        .where("target_session_key", "=", marker.target_session_key),
+    );
+  }
+  return {
+    addedColumn,
+    migratedAmbientWatches,
+    removedLegacySentinels: legacyMarkers.length,
+  };
+}

src/state/openclaw-state-db.generated.d.ts

@@ -907,6 +907,7 @@ export interface SessionWatchCursors {
   last_seen_sequence: Generated<number>;
   material_sequence: Generated<number>;
   notified_sequence: Generated<number>;
+  provenance: Generated<string>;
   target_session_key: string;
   updated_at: number;
   watcher_session_key: string;

src/state/openclaw-state-db.test.ts

@@ -48,6 +48,68 @@ function createTempStateDir(): string {
   return makeTempDir(stateDbTempDirs, "openclaw-state-db-");
 }
 
+const LEGACY_SESSION_WATCH_SCHEMA_VERSION = 3;
+const LEGACY_AMBIENT_WATCH_PREFIX = "ambient-group-watch:";
+
+function seedLegacySessionWatchCursorSchema(stateDir: string): {
+  ambientTarget: string;
+  databasePath: string;
+  explicitTarget: string;
+  watcherSessionKey: string;
+} {
+  const options = { env: { OPENCLAW_STATE_DIR: stateDir } };
+  const databasePath = openOpenClawStateDatabase(options).path;
+  closeOpenClawStateDatabaseForTest();
+
+  const watcherSessionKey = "agent:main:main";
+  const ambientTarget = "agent:main:telegram:group:ambient";
+  const explicitTarget = "agent:main:subagent:explicit";
+  const markerKey = `${LEGACY_AMBIENT_WATCH_PREFIX}${Buffer.from(watcherSessionKey, "utf8").toString("hex")}`;
+  const orphanMarkerKey = `${LEGACY_AMBIENT_WATCH_PREFIX}${Buffer.from("agent:main:orphan", "utf8").toString("hex")}`;
+  const { DatabaseSync } = requireNodeSqlite();
+  const legacy = new DatabaseSync(databasePath);
+  try {
+    legacy.exec(`
+      PRAGMA foreign_keys = OFF;
+      BEGIN IMMEDIATE;
+      DROP INDEX idx_session_watch_cursors_target;
+      ALTER TABLE session_watch_cursors RENAME TO session_watch_cursors_v4;
+      CREATE TABLE session_watch_cursors (
+        watcher_session_key TEXT NOT NULL,
+        target_session_key TEXT NOT NULL,
+        last_seen_sequence INTEGER NOT NULL DEFAULT 0,
+        notified_sequence INTEGER NOT NULL DEFAULT 0,
+        material_sequence INTEGER NOT NULL DEFAULT 0,
+        updated_at INTEGER NOT NULL,
+        PRIMARY KEY (watcher_session_key, target_session_key)
+      ) STRICT;
+      DROP TABLE session_watch_cursors_v4;
+      CREATE INDEX idx_session_watch_cursors_target
+        ON session_watch_cursors(target_session_key);
+      PRAGMA user_version = ${LEGACY_SESSION_WATCH_SCHEMA_VERSION};
+      UPDATE schema_meta
+      SET schema_version = ${LEGACY_SESSION_WATCH_SCHEMA_VERSION}
+      WHERE meta_key = 'primary';
+      COMMIT;
+      PRAGMA foreign_keys = ON;
+    `);
+    const insert = legacy.prepare(`
+      INSERT INTO session_watch_cursors (
+        watcher_session_key, target_session_key, last_seen_sequence,
+        notified_sequence, material_sequence, updated_at
+      ) VALUES (?, ?, ?, ?, ?, ?)
+    `);
+    insert.run(watcherSessionKey, ambientTarget, 7, 8, 9, 200);
+    insert.run(watcherSessionKey, explicitTarget, 3, 4, 5, 300);
+    insert.run(markerKey, ambientTarget, 7, 7, 7, 400);
+    insert.run(orphanMarkerKey, "agent:main:telegram:group:orphan", 1, 1, 1, 100);
+    insert.run(`${LEGACY_AMBIENT_WATCH_PREFIX}not-hex`, ambientTarget, 1, 1, 1, 100);
+  } finally {
+    legacy.close();
+  }
+  return { ambientTarget, databasePath, explicitTarget, watcherSessionKey };
+}
+
 type PlacementConstraintProbe = {
   sessionId: string;
   state: string;
@@ -834,9 +896,13 @@ describe("openclaw state database", () => {
 
     expect(detectOpenClawStateDatabaseSchemaMigrations(options)).toEqual([
       { kind: "strict-tables-v3", path: databasePath },
+      { kind: "session-watch-cursor-provenance-v4", path: databasePath },
     ]);
     expect(repairOpenClawStateDatabaseSchema(options)).toEqual({
-      changes: ["Migrated shared state tables to SQLite STRICT typing (1)"],
+      changes: [
+        "Migrated shared state session watch cursors → provenance column (0 ambient, 0 sentinels removed)",
+        "Migrated shared state tables to SQLite STRICT typing (1)",
+      ],
       warnings: [],
     });
     expect(detectOpenClawStateDatabaseSchemaMigrations(options)).toEqual([]);
@@ -856,6 +922,83 @@ describe("openclaw state database", () => {
     });
   });
 
+  it("doctor migrates version 3 ambient watch sentinels into cursor provenance", () => {
+    const stateDir = createTempStateDir();
+    const options = { env: { OPENCLAW_STATE_DIR: stateDir } };
+    const seeded = seedLegacySessionWatchCursorSchema(stateDir);
+
+    expect(detectOpenClawStateDatabaseSchemaMigrations(options)).toEqual([
+      { kind: "session-watch-cursor-provenance-v4", path: seeded.databasePath },
+    ]);
+    expect(repairOpenClawStateDatabaseSchema(options)).toEqual({
+      changes: [
+        "Migrated shared state session watch cursors → provenance column (1 ambient, 3 sentinels removed)",
+      ],
+      warnings: [],
+    });
+
+    const migrated = openOpenClawStateDatabase(options);
+    expect(
+      migrated.db
+        .prepare(
+          `SELECT watcher_session_key, target_session_key, last_seen_sequence,
+                  notified_sequence, material_sequence, provenance, updated_at
+           FROM session_watch_cursors
+           ORDER BY target_session_key`,
+        )
+        .all(),
+    ).toEqual([
+      {
+        watcher_session_key: seeded.watcherSessionKey,
+        target_session_key: seeded.explicitTarget,
+        last_seen_sequence: 3,
+        notified_sequence: 4,
+        material_sequence: 5,
+        provenance: "explicit",
+        updated_at: 300,
+      },
+      {
+        watcher_session_key: seeded.watcherSessionKey,
+        target_session_key: seeded.ambientTarget,
+        last_seen_sequence: 7,
+        notified_sequence: 8,
+        material_sequence: 9,
+        provenance: "ambient-group",
+        updated_at: 400,
+      },
+    ]);
+    expect(readSqliteNumberPragma(migrated.db, "user_version")).toBe(OPENCLAW_STATE_SCHEMA_VERSION);
+    expect(
+      migrated.db
+        .prepare("SELECT schema_version FROM schema_meta WHERE meta_key = 'primary'")
+        .get(),
+    ).toEqual({ schema_version: OPENCLAW_STATE_SCHEMA_VERSION });
+    closeOpenClawStateDatabaseForTest();
+    expect(repairOpenClawStateDatabaseSchema(options)).toEqual({ changes: [], warnings: [] });
+  });
+
+  it("automatically migrates version 3 ambient watch sentinels on database open", () => {
+    const stateDir = createTempStateDir();
+    const options = { env: { OPENCLAW_STATE_DIR: stateDir } };
+    const seeded = seedLegacySessionWatchCursorSchema(stateDir);
+
+    const migrated = openOpenClawStateDatabase(options);
+    expect(
+      migrated.db
+        .prepare(
+          `SELECT target_session_key, provenance
+           FROM session_watch_cursors
+           ORDER BY target_session_key`,
+        )
+        .all(),
+    ).toEqual([
+      { target_session_key: seeded.explicitTarget, provenance: "explicit" },
+      { target_session_key: seeded.ambientTarget, provenance: "ambient-group" },
+    ]);
+    expect(readSqliteNumberPragma(migrated.db, "user_version")).toBe(OPENCLAW_STATE_SCHEMA_VERSION);
+    expect(detectOpenClawStateDatabaseSchemaMigrations(options)).toEqual([]);
+  });
+
   it("rejects a placement turn claim tuple without an owner", () => {
     const database = openOpenClawStateDatabase({
       env: { OPENCLAW_STATE_DIR: createTempStateDir() },

src/state/openclaw-state-db.ts

@@ -49,6 +49,7 @@ import {
   tableHasColumn,
   tablePrimaryKeyColumns,
 } from "./openclaw-state-db-schema-helpers.js";
+import * as sessionWatchMigration from "./openclaw-state-db-session-watch-migration.js";
 import type { DB as OpenClawStateKyselyDatabase } from "./openclaw-state-db.generated.js";
 import {
   resolveOpenClawStateSqliteDir,
@@ -63,9 +64,10 @@ import { OPENCLAW_STATE_SCHEMA_SQL } from "./openclaw-state-schema.generated.js"
  * tables, private file permissions, cached handles, and audit rows for
  * migrations/backups that operate on local state.
  */
-// v3 rebuilds every OpenClaw-owned table with SQLite STRICT type enforcement.
-// database_verifications is additive derived cache; no bump preserves safe downgrades.
-export const OPENCLAW_STATE_SCHEMA_VERSION = 3;
+// v4 replaces ambient session-watch sentinel rows with cursor provenance.
+// database_verifications remains additive derived cache within this version.
+export const OPENCLAW_STATE_SCHEMA_VERSION = 4;
+const OPENCLAW_STATE_STRICT_SCHEMA_VERSION = 3;
 /** Maximum time one synchronous SQLite call may wait for a lock. */
 export const OPENCLAW_SQLITE_BUSY_TIMEOUT_MS = 5_000;
 /** User-facing guide for schema refusals; lives here so error sites avoid import cycles. */
@@ -130,6 +132,7 @@ export type OpenClawStateDatabaseSchemaMigration = {
     | "agent-databases-composite-primary-key"
     | "audit-events-v2"
     | "operator-approvals-system-agent"
+    | "session-watch-cursor-provenance-v4"
     | "strict-tables-v3";
   path: string;
 };
@@ -918,18 +921,19 @@ export function detectOpenClawStateDatabaseSchemaMigrations(
   const db = new sqlite.DatabaseSync(pathname, { readOnly: true });
   try {
     const migrations: OpenClawStateDatabaseSchemaMigration[] = [];
+    const userVersion = readSqliteUserVersion(db);
     if (!hasCanonicalAgentDatabasesPrimaryKey(db)) {
       migrations.push({ kind: "agent-databases-composite-primary-key", path: pathname });
     }
     if (!hasCanonicalAuditEventsSchema(db)) {
       migrations.push({ kind: "audit-events-v2", path: pathname });
     }
-    if (
-      tableExists(db, "audit_events") &&
-      readSqliteUserVersion(db) < OPENCLAW_STATE_SCHEMA_VERSION
-    ) {
+    if (tableExists(db, "audit_events") && userVersion < OPENCLAW_STATE_STRICT_SCHEMA_VERSION) {
       migrations.push({ kind: "strict-tables-v3", path: pathname });
     }
+    if (sessionWatchMigration.needsSessionWatchCursorProvenanceMigration(db, userVersion)) {
+      migrations.push({ kind: "session-watch-cursor-provenance-v4", path: pathname });
+    }
     migrations.push(
       ...operatorApprovalMigration.detectOperatorApprovalSchemaMigration(db, pathname),
     );
@@ -959,6 +963,7 @@ export function repairOpenClawStateDatabaseSchema(options: OpenClawStateDatabase
       db,
       () => {
         const applied: string[] = [];
+        const previousVersion = readSqliteUserVersion(db);
         if (repairAgentDatabasesCompositePrimaryKey(db)) {
           applied.push(`Migrated shared state agent database registry primary key → agent_id,path`);
         }
@@ -968,6 +973,14 @@ export function repairOpenClawStateDatabaseSchema(options: OpenClawStateDatabase
           );
         }
         applied.push(...operatorApprovalMigration.repairOperatorApprovalSchema(db));
+        const needsSessionWatchMigration =
+          sessionWatchMigration.needsSessionWatchCursorProvenanceMigration(db, previousVersion);
+        const sessionWatchResult = sessionWatchMigration.migrateSessionWatchCursorProvenance(db);
+        if (needsSessionWatchMigration) {
+          applied.push(
+            `Migrated shared state session watch cursors → provenance column (${sessionWatchResult.migratedAmbientWatches} ambient, ${sessionWatchResult.removedLegacySentinels} sentinels removed)`,
+          );
+        }
         assertCanonicalStateSchemaShape(db, pathname);
         if (tableExists(db, "audit_events")) {
           ensureAdditiveStateColumns(db);
@@ -1743,10 +1756,11 @@ function ensureSchema(db: DatabaseSync, pathname: string): void {
         assertSupportedSchemaVersion(db, pathname);
         const previousVersion = readSqliteUserVersion(db);
         ensureAdditiveStateColumns(db);
+        sessionWatchMigration.migrateSessionWatchCursorProvenance(db);
         assertCanonicalStateSchemaShape(db, pathname);
         db.exec(OPENCLAW_STATE_SCHEMA_SQL);
         migrateLegacyCronRunLogsToTaskRuns(db);
-        if (previousVersion < OPENCLAW_STATE_SCHEMA_VERSION) {
+        if (previousVersion < OPENCLAW_STATE_STRICT_SCHEMA_VERSION) {
           migrateSqliteSchemaToStrictInTransaction(db, OPENCLAW_STATE_SCHEMA_SQL, {
             databaseLabel: pathname,
           });

src/state/openclaw-state-schema.generated.ts

@@ -168,15 +168,17 @@ CREATE TABLE IF NOT EXISTS session_state_heads (
 ) STRICT;
 
 -- Notifiable watcher identity is the bare session key, matching the process-local
--- system-event queue it feeds. Ambient group watches also own non-notifiable marker
--- rows. Other bare keys (session.scope="global") are ambiguous across agents and
--- excluded until watcher identity is agent-scoped end-to-end.
+-- system-event queue it feeds. Provenance distinguishes explicit immediate-wake
+-- watches from ambient queue-only group watches. Other bare keys
+-- (session.scope="global") are ambiguous across agents and excluded until watcher
+-- identity is agent-scoped end-to-end.
 CREATE TABLE IF NOT EXISTS session_watch_cursors (
   watcher_session_key TEXT NOT NULL,
   target_session_key TEXT NOT NULL,
   last_seen_sequence INTEGER NOT NULL DEFAULT 0,
   notified_sequence INTEGER NOT NULL DEFAULT 0,
   material_sequence INTEGER NOT NULL DEFAULT 0,
+  provenance TEXT NOT NULL DEFAULT 'explicit' CHECK (provenance IN ('explicit', 'ambient-group')),
   updated_at INTEGER NOT NULL,
   PRIMARY KEY (watcher_session_key, target_session_key)
 ) STRICT;

src/state/openclaw-state-schema.sql

@@ -163,15 +163,17 @@ CREATE TABLE IF NOT EXISTS session_state_heads (
 ) STRICT;
 
 -- Notifiable watcher identity is the bare session key, matching the process-local
--- system-event queue it feeds. Ambient group watches also own non-notifiable marker
--- rows. Other bare keys (session.scope="global") are ambiguous across agents and
--- excluded until watcher identity is agent-scoped end-to-end.
+-- system-event queue it feeds. Provenance distinguishes explicit immediate-wake
+-- watches from ambient queue-only group watches. Other bare keys
+-- (session.scope="global") are ambiguous across agents and excluded until watcher
+-- identity is agent-scoped end-to-end.
 CREATE TABLE IF NOT EXISTS session_watch_cursors (
   watcher_session_key TEXT NOT NULL,
   target_session_key TEXT NOT NULL,
   last_seen_sequence INTEGER NOT NULL DEFAULT 0,
   notified_sequence INTEGER NOT NULL DEFAULT 0,
   material_sequence INTEGER NOT NULL DEFAULT 0,
+  provenance TEXT NOT NULL DEFAULT 'explicit' CHECK (provenance IN ('explicit', 'ambient-group')),
   updated_at INTEGER NOT NULL,
   PRIMARY KEY (watcher_session_key, target_session_key)
 ) STRICT;

src/state/session-watch-cursor-provenance.ts

@@ -0,0 +1,7 @@
+// Canonical provenance values for durable session watch cursors.
+export const SESSION_WATCH_PROVENANCE_EXPLICIT = "explicit";
+export const SESSION_WATCH_PROVENANCE_AMBIENT_GROUP = "ambient-group";
+
+export type SessionWatchCursorProvenance =
+  | typeof SESSION_WATCH_PROVENANCE_EXPLICIT
+  | typeof SESSION_WATCH_PROVENANCE_AMBIENT_GROUP;