feat: parents get durable state-change notices when humans interact with child sessions (#104636)

* feat(sessions): durable session state events with parent invalidation

Parents no longer act on stale child-session state: a durable, typed
signal log (session_state_events + per-agent heads) records direct human
messages to children, child spawn/terminal outcomes, goal changes, and
compaction at the seams where those facts are created. A frozen-watermark
cursor protocol (session_watch_cursors) delivers one coalesced system
notice to the parent through the existing system-event + heartbeat wake
idiom, acknowledged at the shared generic drain, with a deferred startup
sweep for restart recovery. session_status gains stateVersion and
changesSince (with exact pruned-history gap signaling); sessions_list
rows gain stateVersion.

Closes #104565

* chore: regenerate docs map and prompt snapshots for session_status changesSince

docs/concepts/session-tool.md

@@ -81,6 +81,16 @@ When route metadata is available, `session_status` also includes a visible `Rout
 - `active` is the current live-run route. It is only reported for the live or current session being handled now.
 - `deliveryContext` is the persisted delivery route stored on the session, which OpenClaw can reuse for later delivery even when the active surface differs.
 
+## Session state changes
+
+OpenClaw keeps a best-effort signal log for selected session state changes: direct human messages to child sessions, child-run completion or failure, child creation, goal changes, and compaction. The log contains metadata and one-line summaries, never message content. Its `stateVersion` is the session's signal-log head, not a transactional change-data-capture version; the session-store mutation and signal append use separate storage, so a failed append is logged without failing the originating turn.
+
+`sessions_list` includes `stateVersion` on rows with logged changes. `session_status` always returns `stateVersion` in structured details. Pass `changesSince: <previousStateVersion>` to retrieve up to 200 retained events after that version; this read does not acknowledge or advance parent notification cursors. A `historyGap: true` result means the requested version predates retained history, so refresh the whole session state instead of treating the response as an exact delta.
+
+When another actor sends a direct human turn to a watched child or changes its goal, the parent receives a system notice telling it to call `session_status` with its last-seen version. Main-session parents are proactively woken. Nested sub-agent parents receive the notice on their next turn because heartbeat routing cannot target their queue directly. Completion announcements remain the owner for ordinary child-run completion delivery.
+
+History is bounded to 30 days and 50,000 rows, while per-session heads remain monotonic after pruning. Notice delivery uses the gateway's in-memory system-event queue and assumes one gateway process owns delivery for the shared state database. Multiple gateways still share the durable log and `changesSince` reconciliation surface, but v1 does not push notices across processes. Parent notices require an agent-qualified parent session key; under `session.scope="global"` the shared `global` key is ambiguous across agents, so those parents get the durable log and `changesSince` but no proactive notices in v1.
+
 `sessions_yield` intentionally ends the current turn so the next message can be the follow-up event you are waiting for. Use it after spawning sub-agents when you want completion results to arrive as the next message instead of building poll loops.
 
 `subagents` is the visibility helper for already spawned OpenClaw sub-agents. It supports `action: "list"` to inspect active/recent runs.
@@ -119,6 +129,7 @@ Default is `tree`. Sandboxed sessions are clamped to `tree` regardless of config
 ## Further reading
 
 - [Session Management](/concepts/session): routing, lifecycle, maintenance
+- [Sub-agents](/tools/subagents): child-session lifecycle and delivery
 - [ACP Agents](/tools/acp-agents): external harness spawning
 - [Multi-agent](/concepts/multi-agent): multi-agent architecture
 - [Gateway Configuration](/gateway/configuration): session tool config knobs

docs/docs_map.md

@@ -2859,6 +2859,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`.
   - H2: Listing and reading sessions
   - H2: Sending cross-session messages
   - H2: Status and orchestration helpers
+  - H2: Session state changes
   - H2: Spawning sub-agents
   - H2: Visibility
   - H2: Further reading

docs/tools/subagents.md

@@ -652,6 +652,7 @@ still need normal device approval for scope upgrades.
 
 ## Related
 
+- [Session tools and state changes](/concepts/session-tool)
 - [ACP agents](/tools/acp-agents)
 - [Agent send](/tools/agent-send)
 - [Background tasks](/automation/tasks)

src/acp/control-plane/manager.cancel-session.test.ts

@@ -46,6 +46,7 @@ describe("AcpSessionManager cancelSession", () => {
 
       const manager = new AcpSessionManager();
       const runPromise = manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:child-1",
         text: "long task",

src/acp/control-plane/manager.core.ts

@@ -87,16 +87,18 @@ export class AcpSessionManager {
         sessionKey,
       };
     }
-    const acp = this.deps.readSessionEntry({
+    const stored = this.deps.readSessionEntry({
       cfg: params.cfg,
       sessionKey,
       clone: false,
-    })?.acp;
+    });
+    const acp = stored?.acp;
     if (acp) {
       return {
         kind: "ready",
         sessionKey,
         meta: acp,
+        entry: stored.entry,
       };
     }
     if (isAcpSessionKey(sessionKey)) {

src/acp/control-plane/manager.failover.test.ts

@@ -103,6 +103,7 @@ describe("AcpSessionManager backend failover", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: harness.cfg,
       sessionKey: harness.sessionKey,
       text: "use primary",
@@ -127,6 +128,7 @@ describe("AcpSessionManager backend failover", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: harness.cfg,
       sessionKey: harness.sessionKey,
       text: "use fallback",
@@ -138,6 +140,7 @@ describe("AcpSessionManager backend failover", () => {
 
     harness.fallbackRuntime.close.mockClear();
     await manager.runTurn({
+      provenance: "system",
       cfg: harness.cfg,
       sessionKey: harness.sessionKey,
       text: "return to primary",
@@ -166,6 +169,7 @@ describe("AcpSessionManager backend failover", () => {
     const manager = new AcpSessionManager();
     await expect(
       manager.runTurn({
+        provenance: "system",
         cfg: harness.cfg,
         sessionKey: harness.sessionKey,
         text: "fallback",
@@ -193,6 +197,7 @@ describe("AcpSessionManager backend failover", () => {
     const manager = new AcpSessionManager();
     await expect(
       manager.runTurn({
+        provenance: "system",
         cfg: harness.cfg,
         sessionKey: harness.sessionKey,
         text: "fallback",
@@ -218,6 +223,7 @@ describe("AcpSessionManager backend failover", () => {
     const manager = new AcpSessionManager();
     await expect(
       manager.runTurn({
+        provenance: "system",
         cfg: harness.cfg,
         sessionKey: harness.sessionKey,
         text: "fallback",
@@ -241,6 +247,7 @@ describe("AcpSessionManager backend failover", () => {
     const manager = new AcpSessionManager();
     await expect(
       manager.runTurn({
+        provenance: "system",
         cfg: harness.cfg,
         sessionKey: harness.sessionKey,
         text: "do not duplicate",

src/acp/control-plane/manager.runtime-config.test.ts

@@ -100,6 +100,7 @@ describe("AcpSessionManager runtime config", () => {
     expect(runtimeState.setMode).not.toHaveBeenCalled();
 
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "do work",
@@ -171,6 +172,7 @@ describe("AcpSessionManager runtime config", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "do work",
@@ -245,6 +247,7 @@ describe("AcpSessionManager runtime config", () => {
     });
 
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "do work",
@@ -334,6 +337,7 @@ describe("AcpSessionManager runtime config", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey,
       text: "learn prompt session",
@@ -409,6 +413,7 @@ describe("AcpSessionManager runtime config", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "do work",
@@ -464,6 +469,7 @@ describe("AcpSessionManager runtime config", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:opencode:acp:session-1",
       text: "do work",
@@ -503,6 +509,7 @@ describe("AcpSessionManager runtime config", () => {
     const manager = new AcpSessionManager();
     await expectRejectedRecord(
       manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:opencode:acp:session-1",
         text: "do work",
@@ -549,6 +556,7 @@ describe("AcpSessionManager runtime config", () => {
     const manager = new AcpSessionManager();
     await expectRejectedRecord(
       manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:opencode:acp:session-1",
         text: "do work",
@@ -586,6 +594,7 @@ describe("AcpSessionManager runtime config", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:claude:acp:session-1",
       text: "do work",
@@ -628,6 +637,7 @@ describe("AcpSessionManager runtime config", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:gemini:acp:session-1",
       text: "do work",
@@ -691,6 +701,7 @@ describe("AcpSessionManager runtime config", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey,
       text: "first",
@@ -714,6 +725,7 @@ describe("AcpSessionManager runtime config", () => {
     expect(currentEntry.acp.cwd).toBe("/workspace/next");
 
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey,
       text: "second",

src/acp/control-plane/manager.runtime-handles.test.ts

@@ -31,6 +31,7 @@ describe("AcpSessionManager runtime handles", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "first",
@@ -38,6 +39,7 @@ describe("AcpSessionManager runtime handles", () => {
       requestId: "r1",
     });
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "second",
@@ -81,6 +83,7 @@ describe("AcpSessionManager runtime handles", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: allowlistCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "first",
@@ -88,6 +91,7 @@ describe("AcpSessionManager runtime handles", () => {
       requestId: "r1",
     });
     await manager.runTurn({
+      provenance: "system",
       cfg: denyCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "second",
@@ -129,6 +133,7 @@ describe("AcpSessionManager runtime handles", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "first",
@@ -136,6 +141,7 @@ describe("AcpSessionManager runtime handles", () => {
       requestId: "r1",
     });
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "second",
@@ -190,6 +196,7 @@ describe("AcpSessionManager runtime handles", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "first",
@@ -210,6 +217,7 @@ describe("AcpSessionManager runtime handles", () => {
     });
 
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "second",
@@ -235,6 +243,7 @@ describe("AcpSessionManager runtime handles", () => {
 
     const managerA = new AcpSessionManager();
     await managerA.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "before restart",
@@ -243,6 +252,7 @@ describe("AcpSessionManager runtime handles", () => {
     });
     const managerB = new AcpSessionManager();
     await managerB.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "after restart",
@@ -280,6 +290,7 @@ describe("AcpSessionManager runtime handles", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey,
       text: "after restart",
@@ -323,6 +334,7 @@ describe("AcpSessionManager runtime handles", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey,
       text: "after restart",
@@ -361,6 +373,7 @@ describe("AcpSessionManager runtime handles", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey,
       text: "after restart",
@@ -397,6 +410,7 @@ describe("AcpSessionManager runtime handles", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey,
       text: "after restart",
@@ -433,6 +447,7 @@ describe("AcpSessionManager runtime handles", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey,
       text: "after restart",
@@ -474,6 +489,7 @@ describe("AcpSessionManager runtime handles", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey,
       text: "after restart",
@@ -562,6 +578,7 @@ describe("AcpSessionManager runtime handles", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey,
       text: "after restart",

src/acp/control-plane/manager.test.ts

@@ -7,6 +7,8 @@ import {
   requireTaskByRunId,
   withAcpManagerTaskStateDir,
 } from "../../../test/helpers/acp-manager-task-state.js";
+import { listSessionStateEventsSince } from "../../sessions/session-state-events.js";
+import { closeOpenClawStateDatabaseForTest } from "../../state/openclaw-state-db.js";
 import { isAcpTurnActive } from "./active-turns.js";
 import {
   AcpRuntimeError,
@@ -22,6 +24,7 @@ import {
   hoisted,
   installAcpSessionManagerTestLifecycle,
   mockCallArg,
+  mockParentedAcpSessionEntries,
   readySessionMeta,
   type OpenClawConfig,
   resetAcpSessionManagerForTests,
@@ -82,6 +85,7 @@ describe("AcpSessionManager", () => {
     } as OpenClawConfig;
 
     await manager.runTurn({
+      provenance: "system",
       cfg,
       sessionKey: "main",
       text: "after restart",
@@ -103,6 +107,46 @@ describe("AcpSessionManager", () => {
     ]);
   });
 
+  it("records parented ACP turns only for human provenance", async () => {
+    await withAcpManagerTaskStateDir(async () => {
+      const runtimeState = createRuntime();
+      hoisted.requireAcpRuntimeBackendMock.mockReturnValue({
+        id: "acpx",
+        runtime: runtimeState.runtime,
+      });
+      const childSessionKey = "agent:main:acp:child-state";
+      mockParentedAcpSessionEntries({
+        childSessionKey,
+        parentSessionKey: "agent:main:main",
+      });
+      const manager = new AcpSessionManager();
+
+      await manager.runTurn({
+        provenance: "human",
+        cfg: baseCfg,
+        sessionKey: childSessionKey,
+        text: "human turn",
+        mode: "prompt",
+        requestId: "human-state-turn",
+      });
+      await manager.runTurn({
+        provenance: "system",
+        cfg: baseCfg,
+        sessionKey: childSessionKey,
+        text: "system turn",
+        mode: "prompt",
+        requestId: "system-state-turn",
+      });
+
+      expect(listSessionStateEventsSince(childSessionKey, "main", 0, 200).events).toMatchObject([
+        { kind: "human_direct_message", runId: "human-state-turn" },
+        { kind: "run_completed", runId: "human-state-turn" },
+        { kind: "run_completed", runId: "system-state-turn" },
+      ]);
+      closeOpenClawStateDatabaseForTest();
+    });
+  });
+
   it("tracks parented direct ACP turns in the task registry", async () => {
     await withAcpManagerTaskStateDir(async () => {
       const runtimeState = createRuntime();
@@ -173,6 +217,7 @@ describe("AcpSessionManager", () => {
 
       const manager = new AcpSessionManager();
       await manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:child-1",
         text: "Implement the feature and report back",
@@ -259,6 +304,7 @@ describe("AcpSessionManager", () => {
 
       const manager = new AcpSessionManager();
       await manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:child-1",
         text: "Print the current directory in Korean",
@@ -310,6 +356,7 @@ describe("AcpSessionManager", () => {
 
     const manager = new AcpSessionManager();
     const first = manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "first",
@@ -323,6 +370,7 @@ describe("AcpSessionManager", () => {
       { interval: 1 },
     );
     const second = manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "second",
@@ -377,6 +425,7 @@ describe("AcpSessionManager", () => {
 
       const manager = new AcpSessionManager();
       const turn = manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:child-1",
         text: "long running",
@@ -455,6 +504,7 @@ describe("AcpSessionManager", () => {
 
       const manager = new AcpSessionManager();
       const turn = manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:child-1",
         text: "slow init",
@@ -527,6 +577,7 @@ describe("AcpSessionManager", () => {
       const manager = new AcpSessionManager();
       await expect(
         manager.runTurn({
+          provenance: "system",
           cfg: baseCfg,
           sessionKey: "agent:codex:acp:child-1",
           text: "stale resume",
@@ -569,6 +620,7 @@ describe("AcpSessionManager", () => {
 
     const manager = new AcpSessionManager();
     const first = manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "first",
@@ -584,6 +636,7 @@ describe("AcpSessionManager", () => {
 
     const abortController = new AbortController();
     const second = manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "second",
@@ -659,6 +712,7 @@ describe("AcpSessionManager", () => {
       } as OpenClawConfig;
 
       const first = manager.runTurn({
+        provenance: "system",
         cfg,
         sessionKey: "agent:codex:acp:session-1",
         text: "first",
@@ -674,6 +728,7 @@ describe("AcpSessionManager", () => {
       );
 
       const second = manager.runTurn({
+        provenance: "system",
         cfg,
         sessionKey: "agent:codex:acp:session-1",
         text: "second",
@@ -731,6 +786,7 @@ describe("AcpSessionManager", () => {
     try {
       const manager = new AcpSessionManager();
       await manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:session-1",
         text: "first",
@@ -789,6 +845,7 @@ describe("AcpSessionManager", () => {
       } as OpenClawConfig;
 
       const first = manager.runTurn({
+        provenance: "system",
         cfg,
         sessionKey: "agent:codex:acp:session-a",
         text: "first",
@@ -813,6 +870,7 @@ describe("AcpSessionManager", () => {
 
       await expectRejectedRecord(
         manager.runTurn({
+          provenance: "system",
           cfg,
           sessionKey: "agent:codex:acp:session-b",
           text: "second",
@@ -869,6 +927,7 @@ describe("AcpSessionManager", () => {
     await Promise.race([
       Promise.all([
         manager.runTurn({
+          provenance: "system",
           cfg: baseCfg,
           sessionKey: "agent:codex:acp:session-a",
           text: "first",
@@ -876,6 +935,7 @@ describe("AcpSessionManager", () => {
           requestId: "r1",
         }),
         manager.runTurn({
+          provenance: "system",
           cfg: baseCfg,
           sessionKey: "agent:codex:acp:session-b",
           text: "second",
@@ -917,6 +977,7 @@ describe("AcpSessionManager", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: limitedCfg,
       sessionKey: "agent:codex:acp:session-a",
       text: "first",
@@ -926,6 +987,7 @@ describe("AcpSessionManager", () => {
 
     await expectRejectedRecord(
       manager.runTurn({
+        provenance: "system",
         cfg: limitedCfg,
         sessionKey: "agent:codex:acp:session-b",
         text: "second",
@@ -975,6 +1037,7 @@ describe("AcpSessionManager", () => {
     const manager = new AcpSessionManager();
     await expect(
       manager.runTurn({
+        provenance: "system",
         cfg,
         sessionKey,
         text: "hello",
@@ -1029,6 +1092,7 @@ describe("AcpSessionManager", () => {
 
       const manager = new AcpSessionManager();
       await manager.runTurn({
+        provenance: "system",
         cfg: limitedCfg,
         sessionKey: "agent:codex:acp:session-a",
         text: "first",
@@ -1047,6 +1111,7 @@ describe("AcpSessionManager", () => {
 
       await expect(
         manager.runTurn({
+          provenance: "system",
           cfg: limitedCfg,
           sessionKey: "agent:codex:acp:session-b",
           text: "second",
@@ -1247,6 +1312,7 @@ describe("AcpSessionManager", () => {
     const manager = new AcpSessionManager();
     await expect(
       manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey,
         text: "who are you?",
@@ -1362,6 +1428,7 @@ describe("AcpSessionManager", () => {
 
       const manager = new AcpSessionManager();
       await manager.runTurn({
+        provenance: "system",
         cfg,
         sessionKey: "agent:codex:acp:session-a",
         text: "first",
@@ -1371,6 +1438,7 @@ describe("AcpSessionManager", () => {
 
       vi.advanceTimersByTime(2_000);
       await manager.runTurn({
+        provenance: "system",
         cfg,
         sessionKey: "agent:codex:acp:session-b",
         text: "second",
@@ -1417,6 +1485,7 @@ describe("AcpSessionManager", () => {
 
     const manager = new AcpSessionManager();
     await manager.runTurn({
+      provenance: "system",
       cfg: baseCfg,
       sessionKey: "agent:codex:acp:session-1",
       text: "ok",
@@ -1425,6 +1494,7 @@ describe("AcpSessionManager", () => {
     });
     await expectRejectedRecord(
       manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:session-1",
         text: "boom",
@@ -1461,6 +1531,7 @@ describe("AcpSessionManager", () => {
     const manager = new AcpSessionManager();
     await expectRejectedRecord(
       manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:session-1",
         text: "do work",

src/acp/control-plane/manager.turn-results.test.ts

@@ -84,6 +84,7 @@ describe("AcpSessionManager turn results", () => {
       const manager = new AcpSessionManager();
       await expect(
         manager.runTurn({
+          provenance: "system",
           cfg: baseCfg,
           sessionKey: "agent:codex:acp:child-1",
           text: "Investigate and report back",
@@ -169,6 +170,7 @@ describe("AcpSessionManager turn results", () => {
       const events: string[] = [];
       const manager = new AcpSessionManager();
       await manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:child-1",
         text: "Print the current directory",
@@ -252,6 +254,7 @@ describe("AcpSessionManager turn results", () => {
 
       const manager = new AcpSessionManager();
       await manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:child-1",
         text: "Inspect and report back",
@@ -327,6 +330,7 @@ describe("AcpSessionManager turn results", () => {
 
       const manager = new AcpSessionManager();
       await manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:child-1",
         text: "Inspect and report back",
@@ -402,6 +406,7 @@ describe("AcpSessionManager turn results", () => {
 
       const manager = new AcpSessionManager();
       await manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:child-1",
         text: "Inspect and report back",
@@ -477,6 +482,7 @@ describe("AcpSessionManager turn results", () => {
       const events: string[] = [];
       const manager = new AcpSessionManager();
       await manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:child-1",
         text: "Inspect and report back",
@@ -550,6 +556,7 @@ describe("AcpSessionManager turn results", () => {
       const events: string[] = [];
       const manager = new AcpSessionManager();
       await manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:child-1",
         text: "Produce a final result",
@@ -643,6 +650,7 @@ describe("AcpSessionManager turn results", () => {
       const events: string[] = [];
       const manager = new AcpSessionManager();
       await manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:child-1",
         text: "Investigate and report back",
@@ -689,6 +697,7 @@ describe("AcpSessionManager turn results", () => {
       const events: AcpRuntimeEvent[] = [];
       const manager = new AcpSessionManager();
       await manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:child-1",
         text: "long task",
@@ -741,6 +750,7 @@ describe("AcpSessionManager turn results", () => {
     const manager = new AcpSessionManager();
     await expect(
       manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:session-1",
         text: "do work",
@@ -827,6 +837,7 @@ describe("AcpSessionManager turn results", () => {
       const manager = new AcpSessionManager();
       await expect(
         manager.runTurn({
+          provenance: "system",
           cfg: baseCfg,
           sessionKey: "agent:codex:acp:child-1",
           text: "Investigate and report back",
@@ -864,6 +875,7 @@ describe("AcpSessionManager turn results", () => {
     const manager = new AcpSessionManager();
     await expect(
       manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:session-1",
         text: "do work",
@@ -900,6 +912,7 @@ describe("AcpSessionManager turn results", () => {
     const manager = new AcpSessionManager();
     await expectRejectedRecord(
       manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey: "agent:codex:acp:session-1",
         text: "do work",
@@ -944,6 +957,7 @@ describe("AcpSessionManager turn results", () => {
       const manager = new AcpSessionManager();
       await expect(
         manager.runTurn({
+          provenance: "system",
           cfg: baseCfg,
           sessionKey: "agent:codex:acp:session-1",
           text: "do work",
@@ -1027,6 +1041,7 @@ describe("AcpSessionManager turn results", () => {
     const manager = new AcpSessionManager();
     const runTurn = () =>
       manager.runTurn({
+        provenance: "system",
         cfg: baseCfg,
         sessionKey,
         text: "do work",

src/acp/control-plane/manager.turn-runner.ts

@@ -1,6 +1,10 @@
 /** Runs ACP turns, failover, timeout cleanup, and detached-task progress mirroring. */
 import type { AcpRuntime, AcpRuntimeHandle } from "@openclaw/acp-core/runtime/types";
 import { logVerbose } from "../../globals.js";
+import {
+  recordSessionHumanDirectMessage,
+  recordSubagentTerminalState,
+} from "../../sessions/session-state-events.js";
 import { AcpRuntimeError, formatAcpErrorChain, toAcpRuntimeError } from "../runtime/errors.js";
 import { clearAcpTurnActive, markAcpTurnActive } from "./active-turns.js";
 import {
@@ -88,6 +92,19 @@ export async function runManagerTurn(params: {
     sessionKey,
   });
   const initialMeta = requireReadySessionMeta(initialResolution);
+  recordSessionHumanDirectMessage({
+    sessionKey,
+    entry: initialResolution.kind === "ready" ? initialResolution.entry : undefined,
+    actor: { actorType: input.provenance },
+    channel: "acp",
+    runId: input.requestId,
+  });
+  // ACP children bypass the subagent registry; terminal outcomes are projected into
+  // the signal log here so changesSince histories are not spawn-only for ACP runs.
+  const spawnedByWatcher =
+    initialResolution.kind === "ready"
+      ? (initialResolution.entry?.spawnedBy ?? initialResolution.entry?.parentSessionKey)
+      : undefined;
   const { candidateBackends, describeBackendCandidate } = resolveBackendCandidatePlan({
     configuredPrimaryBackend: input.cfg.acp?.backend,
     resolvedPrimaryBackend: initialMeta.backend,
@@ -110,15 +127,24 @@ export async function runManagerTurn(params: {
       errorCode: errorToRecord.code,
     });
     if (taskContext) {
+      const failureStatus = resolveBackgroundTaskFailureStatus(errorToRecord);
       markBackgroundTaskTerminal(taskContext.runId, {
         sessionKey,
-        status: resolveBackgroundTaskFailureStatus(errorToRecord),
+        status: failureStatus,
         endedAt: Date.now(),
         lastEventAt: Date.now(),
         error: formatAcpErrorChain(errorToRecord),
         progressSummary: taskProgressSummary || null,
         terminalSummary: null,
       });
+      if (spawnedByWatcher) {
+        recordSubagentTerminalState({
+          childSessionKey: sessionKey,
+          runId: taskContext.runId,
+          requesterSessionKey: spawnedByWatcher,
+          outcomeStatus: failureStatus === "timed_out" ? "timeout" : "error",
+        });
+      }
     }
     await params.setSessionState({
       cfg: input.cfg,
@@ -304,6 +330,14 @@ export async function runManagerTurn(params: {
               terminalSummary: terminalResult.terminalSummary ?? null,
               terminalOutcome: terminalResult.terminalOutcome,
             });
+            if (spawnedByWatcher) {
+              recordSubagentTerminalState({
+                childSessionKey: sessionKey,
+                runId: taskContext.runId,
+                requesterSessionKey: spawnedByWatcher,
+                outcomeStatus: turnOutcome.terminalStatus === "cancelled" ? "error" : "ok",
+              });
+            }
           }
           await params.setSessionState({
             cfg: input.cfg,

src/acp/control-plane/manager.types.ts

@@ -38,6 +38,7 @@ export type AcpSessionResolution =
       kind: "ready";
       sessionKey: string;
       meta: SessionAcpMeta;
+      entry?: SessionEntry;
     };
 
 /** Input required to create or resume an ACP runtime session. */
@@ -61,6 +62,7 @@ export type AcpTurnAttachment = {
 export type AcpRunTurnInput = {
   cfg: OpenClawConfig;
   sessionKey: string;
+  provenance: "human" | "agent" | "system";
   text: string;
   attachments?: AcpTurnAttachment[];
   mode: AcpRuntimePromptMode;

src/agents/acp-spawn.ts

@@ -71,6 +71,7 @@ import {
   parseAgentSessionKey,
   resolveAgentIdFromSessionKey,
 } from "../routing/session-key.js";
+import { recordSubagentSpawned } from "../sessions/session-state-events.js";
 import { createRunningTaskRun } from "../tasks/detached-task-runtime.js";
 import { listTasksForOwnerKey } from "../tasks/runtime-internal.js";
 import { deliveryContextFromSession, normalizeDeliveryContext } from "../utils/delivery-context.js";
@@ -1533,6 +1534,14 @@ export async function spawnAcpDirect(
   });
   const childIdem = crypto.randomUUID();
   let childRunId: string = childIdem;
+  // ACP children take this branch instead of spawnSubagentDirect; without this the
+  // signal log has no child_spawned event and the parent cursor is never seeded.
+  recordSubagentSpawned({
+    childSessionKey: sessionKey,
+    childRunId: childIdem,
+    requesterSessionKey: requesterInternalKey,
+    agentId: targetAgentId,
+  });
   const streamLogPath =
     effectiveStreamToParent && parentSessionKey
       ? resolveAcpSpawnStreamLogPath({

src/agents/agent-command.ts

@@ -63,6 +63,10 @@ import {
 } from "../sessions/model-overrides.js";
 import { resolveSendPolicy } from "../sessions/send-policy.js";
 import { beginSessionWorkAdmission } from "../sessions/session-lifecycle-admission.js";
+import {
+  classifySessionStateActor,
+  recordSessionHumanDirectMessage,
+} from "../sessions/session-state-events.js";
 import { createUserTurnTranscriptRecorder } from "../sessions/user-turn-transcript.js";
 import { createLazyImportLoader } from "../shared/lazy-promise.js";
 import { resolveEffectiveAgentSkillFilter } from "../skills/discovery/agent-filter.js";
@@ -991,6 +995,15 @@ async function agentCommandInternal(
   assertAgentRunLifecycleGenerationCurrent(lifecycleGeneration);
   const effectiveCwd = cwd ? resolveUserPath(cwd) : workspaceDir;
   let sessionEntry = prepared.sessionEntry;
+  const sessionStateActor = classifySessionStateActor({
+    inputProvenance: opts.inputProvenance,
+    internalEvents: opts.internalEvents,
+    sessionEffects: opts.sessionEffects,
+  });
+  // Subagent-lane turns are the parent's own task dispatch into the child (they
+  // carry no inter_session provenance today); classifying them as human would tell
+  // the parent a human interjected on every spawn, for embedded and ACP children alike.
+  const isSubagentLaneTurn = normalizeOptionalString(opts.lane) === AGENT_LANE_SUBAGENT;
   let sessionReboundDuringRun = false;
   let trackedRestartRecoveryDeliveryContext = false;
   let currentRunDeliveryContext: DeliveryContext | undefined;
@@ -1178,6 +1191,7 @@ async function agentCommandInternal(
           await acpManager.runTurn({
             cfg,
             sessionKey,
+            provenance: isSubagentLaneTurn ? "agent" : sessionStateActor.actorType,
             text: body,
             attachments: acpImageAttachments.length > 0 ? acpImageAttachments : undefined,
             mode: "prompt",
@@ -1464,6 +1478,16 @@ async function agentCommandInternal(
         });
         sessionEntry = persisted ?? sessionEntry;
       }
+      if (sessionKey && !isSubagentLaneTurn) {
+        recordSessionHumanDirectMessage({
+          sessionKey,
+          entry: sessionEntry,
+          agentId: sessionAgentId,
+          actor: sessionStateActor,
+          channel: opts.channel,
+          runId,
+        });
+      }
 
       const configuredDefaultRef = resolveDefaultModelForAgent({
         cfg,

src/agents/embedded-agent-subscribe.handlers.compaction.ts

@@ -5,6 +5,7 @@
  */
 import { emitAgentEvent } from "../infra/agent-events.js";
 import { getGlobalHookRunner } from "../plugins/hook-runner-global.js";
+import { recordSessionCompacted } from "../sessions/session-state-events.js";
 import { stripStaleAssistantUsageBeforeLatestCompaction } from "./compaction-usage.js";
 import { runBestEffortCallback } from "./embedded-agent-subscribe.callback.js";
 import type { EmbeddedAgentSubscribeContext } from "./embedded-agent-subscribe.handlers.types.js";
@@ -114,6 +115,12 @@ export function handleCompactionEnd(ctx: EmbeddedAgentSubscribeContext, evt: Com
         : undefined;
     ctx.noteCompactionTokensAfter(tokensAfter);
     const observedCompactionCount = ctx.getCompactionCount();
+    recordSessionCompacted({
+      sessionKey: ctx.params.sessionKey,
+      operationId: `${ctx.params.runId}:${observedCompactionCount}`,
+      agentId: ctx.params.agentId,
+      runId: ctx.params.runId,
+    });
     ctx.log.info(`embedded run ${kind} complete`, {
       event: "embedded_run_compaction_end",
       runId: ctx.params.runId,

src/agents/openclaw-tools.session-status.test.ts

@@ -33,6 +33,17 @@ const resolveEnvApiKeyMock = vi.hoisted(() =>
 const resolveUsableCustomProviderApiKeyMock = vi.hoisted(() =>
   vi.fn((_params?: { provider?: string }) => null as { apiKey: string; source: string } | null),
 );
+const getSessionStateVersionMock = vi.hoisted(() =>
+  vi.fn((_sessionKey: string, _agentId: string) => 0),
+);
+const listSessionStateEventsSinceMock = vi.hoisted(() =>
+  vi.fn((_sessionKey: string, _agentId: string, _after: number, _limit: number) => ({
+    events: [] as Array<Record<string, unknown>>,
+    truncated: false,
+    earliestAvailableSequence: 0,
+    historyGap: false,
+  })),
+);
 const emptyPluginMetadataSnapshot = vi.hoisted(() => ({
   configFingerprint: "session-status-test-empty-plugin-metadata",
   plugins: [],
@@ -335,6 +346,16 @@ vi.mock("../tasks/task-owner-access.js", () => ({
       now: TASK_STATUS_SNAPSHOT_NOW,
     }),
 }));
+vi.mock("../sessions/session-state-events.js", () => ({
+  getSessionStateVersion: (sessionKey: string, agentId: string) =>
+    getSessionStateVersionMock(sessionKey, agentId),
+  listSessionStateEventsSince: (
+    sessionKey: string,
+    agentId: string,
+    after: number,
+    limit: number,
+  ) => listSessionStateEventsSinceMock(sessionKey, agentId, after, limit),
+}));
 
 let createSessionStatusTool: typeof import("./tools/session-status-tool.js").createSessionStatusTool;
 
@@ -365,6 +386,15 @@ function resetSessionStore(store: Record<string, SessionEntry>) {
   callGatewayMock.mockClear();
   listTasksForRelatedSessionKeyForOwnerMock.mockClear();
   listTasksForRelatedSessionKeyForOwnerMock.mockReturnValue([]);
+  getSessionStateVersionMock.mockReset();
+  getSessionStateVersionMock.mockReturnValue(0);
+  listSessionStateEventsSinceMock.mockReset();
+  listSessionStateEventsSinceMock.mockReturnValue({
+    events: [],
+    truncated: false,
+    earliestAvailableSequence: 0,
+    historyGap: false,
+  });
   loadSessionStoreMock.mockReturnValue(store);
   callGatewayMock.mockImplementation(async (opts: unknown) => {
     const request = opts as { method?: string; params?: Record<string, unknown> };
@@ -503,6 +533,32 @@ describe("session_status tool", () => {
     expect(details.statusText).not.toContain("OAuth/token status");
   });
 
+  it("returns read-only state changes and the signal-log head", async () => {
+    resetSessionStore({
+      main: {
+        sessionId: "s1",
+        updatedAt: 10,
+      },
+    });
+    getSessionStateVersionMock.mockReturnValue(12);
+    listSessionStateEventsSinceMock.mockReturnValue({
+      events: [{ sequence: 12, kind: "goal_changed", summary: "goal created" }],
+      truncated: false,
+      earliestAvailableSequence: 12,
+      historyGap: true,
+    });
+
+    const result = await getSessionStatusTool().execute("call-state", { changesSince: 3 });
+    const details = result.details as Record<string, unknown>;
+    const text = (result.content?.[0] as { text?: string } | undefined)?.text ?? "";
+
+    expect(getSessionStateVersionMock).toHaveBeenCalledWith("main", "main");
+    expect(listSessionStateEventsSinceMock).toHaveBeenCalledWith("main", "main", 3, 200);
+    expect(details.stateVersion).toBe(12);
+    expect(details.stateChanges).toMatchObject({ historyGap: true });
+    expect(text).toContain("Session state changes:");
+  });
+
   it("enables transcript usage fallback for session_status", async () => {
     resetSessionStore({
       main: {

src/agents/subagent-registry-lifecycle.ts

@@ -14,6 +14,7 @@ import {
 } from "../process/gateway-work-admission.js";
 import { defaultRuntime } from "../runtime.js";
 import { emitSessionLifecycleEvent } from "../sessions/session-lifecycle-events.js";
+import { recordSubagentTerminalState } from "../sessions/session-state-events.js";
 import { extractTextFromChatContent } from "../shared/chat-content.js";
 import { createLazyImportLoader } from "../shared/lazy-promise.js";
 import type { DetachedTaskFindResult } from "../tasks/detached-task-runtime-contract.js";
@@ -1897,6 +1898,17 @@ export function createSubagentRegistryLifecycleController(params: {
       return;
     }
     const isProvisionalKill = entry.killReconciliation !== undefined;
+    // Record only the current, non-superseded callback with a committed outcome; the
+    // run-terminal dedupe key is first-write-wins, so a provisional/stale status here
+    // would permanently mislabel the signal-log terminal kind.
+    if (!isProvisionalKill && entry.outcome?.status && entry.outcome.status !== "unknown") {
+      recordSubagentTerminalState({
+        childSessionKey: entry.childSessionKey,
+        runId: entry.runId,
+        requesterSessionKey: entry.requesterSessionKey,
+        outcomeStatus: entry.outcome.status,
+      });
+    }
 
     if (!completeParams.suppressSessionEffects) {
       try {

src/agents/subagent-spawn.ts

@@ -34,6 +34,7 @@ import { stringifyRouteThreadId } from "../plugin-sdk/channel-route.js";
 import { listRegisteredPluginAgentPromptGuidance } from "../plugins/command-registry-state.js";
 import type { SubagentLifecycleHookRunner } from "../plugins/hooks.js";
 import { isValidAgentId, normalizeAgentId, parseAgentSessionKey } from "../routing/session-key.js";
+import { recordSubagentSpawned } from "../sessions/session-state-events.js";
 import { resolveUserPath } from "../utils.js";
 import type { DeliveryContext } from "../utils/delivery-context.types.js";
 import { listAgentIds, resolveAgentDir } from "./agent-scope-config.js";
@@ -1491,6 +1492,8 @@ export async function spawnSubagentDirect(
     task,
   });
 
+  const childIdem = crypto.randomUUID();
+  let childRunId: string = childIdem;
   const spawnedMetadata = normalizeSpawnedRunMetadata({
     spawnedBy: spawnedByKey,
     ...toolSpawnMetadata,
@@ -1514,6 +1517,12 @@ export async function spawnSubagentDirect(
       childSessionKey,
     };
   }
+  recordSubagentSpawned({
+    childSessionKey,
+    childRunId,
+    requesterSessionKey: requesterInternalKey,
+    agentId: targetAgentId,
+  });
   const contextEnginePrepareResult =
     params.lightContext && preparedSpawnContext.mode === "isolated"
       ? ({ status: "ok", preparation: undefined } as const)
@@ -1539,8 +1548,6 @@ export async function spawnSubagentDirect(
   }
   const contextEnginePreparation = contextEnginePrepareResult.preparation;
 
-  const childIdem = crypto.randomUUID();
-  let childRunId: string = childIdem;
   const deliverInitialChildRunDirectly =
     requestThreadBinding && spawnMode === "session" && hasBoundThreadDeliveryOrigin;
   const shouldAnnounceCompletion = deliverInitialChildRunDirectly

src/agents/tools/goal-tools.ts

@@ -31,6 +31,7 @@ type GoalToolOptions = {
 
 type GoalSessionScope = {
   sessionKey: string;
+  agentId: string;
   storePath: string;
 };
 
@@ -65,6 +66,7 @@ function resolveGoalSessionScope(options: GoalToolOptions): GoalSessionScope {
   );
   return {
     sessionKey,
+    agentId,
     storePath: resolveStorePath(options.config?.session?.store, {
       agentId,
     }),
@@ -106,8 +108,10 @@ export function createCreateGoalTool(options: GoalToolOptions): AnyAgentTool {
         // Budgets are positive limits; zero would immediately make accounting ambiguous.
         throw new ToolInputError("token_budget must be positive");
       }
+      const scope = resolveGoalSessionScope(options);
       const goal = await createSessionGoal({
-        ...resolveGoalSessionScope(options),
+        ...scope,
+        actor: { type: "agent", id: scope.sessionKey },
         objective,
         ...(tokenBudget !== undefined ? { tokenBudget } : {}),
       });
@@ -138,8 +142,10 @@ export function createUpdateGoalTool(options: GoalToolOptions): AnyAgentTool {
         );
       }
       const note = readStringParam(params, "note");
+      const scope = resolveGoalSessionScope(options);
       const goal = await updateSessionGoalStatus({
-        ...resolveGoalSessionScope(options),
+        ...scope,
+        actor: { type: "agent", id: scope.sessionKey },
         status: status as (typeof MODEL_UPDATABLE_SESSION_GOAL_STATUSES)[number],
         ...(note ? { note } : {}),
       });

src/agents/tools/session-status-tool.ts

@@ -27,6 +27,10 @@ import {
   resolveAgentIdFromSessionKey,
 } from "../../routing/session-key.js";
 import { applyModelOverrideToSessionEntry } from "../../sessions/model-overrides.js";
+import {
+  getSessionStateVersion,
+  listSessionStateEventsSince,
+} from "../../sessions/session-state-events.js";
 import { createLazyImportLoader } from "../../shared/lazy-promise.js";
 import type { BuildStatusTextParams } from "../../status/status-text.types.js";
 import { buildTaskStatusSnapshotForRelatedSessionKeyForOwner } from "../../tasks/task-owner-access.js";
@@ -55,7 +59,11 @@ import {
   SESSION_STATUS_TOOL_DISPLAY_SUMMARY,
 } from "../tool-description-presets.js";
 import type { AnyAgentTool } from "./common.js";
-import { normalizeToolModelOverride, readStringParam } from "./common.js";
+import {
+  normalizeToolModelOverride,
+  readNonNegativeIntegerParam,
+  readStringParam,
+} from "./common.js";
 import {
   listImplicitDefaultDirectFallbackKeys,
   resolveImplicitCurrentSessionFallback,
@@ -76,6 +84,7 @@ import {
 const SessionStatusToolSchema = Type.Object({
   sessionKey: Type.Optional(Type.String()),
   model: Type.Optional(Type.String()),
+  changesSince: Type.Optional(Type.Integer({ minimum: 0 })),
 });
 
 type CommandsStatusRuntimeModule = {
@@ -226,6 +235,16 @@ ${JSON.stringify(details, null, 2)}
 \`\`\``;
 }
 
+function formatSessionStateChanges(details: {
+  stateVersion: number;
+  stateChanges: ReturnType<typeof listSessionStateEventsSince>;
+}): string {
+  return `Session state changes:
+\`\`\`json
+${JSON.stringify(details, null, 2)}
+\`\`\``;
+}
+
 function resolveActiveStatusModelIdentity(params: {
   activeModelId?: string;
   activeModelProvider?: string;
@@ -395,6 +414,7 @@ export function createSessionStatusTool(opts?: {
     parameters: SessionStatusToolSchema,
     execute: async (_toolCallId, args) => {
       const params = args as Record<string, unknown>;
+      const changesSince = readNonNegativeIntegerParam(params, "changesSince");
       const cfg = opts?.config ?? getRuntimeConfig();
       const { mainKey, alias, effectiveRequesterKey } = resolveSandboxedSessionToolContext({
         cfg,
@@ -866,11 +886,19 @@ export function createSessionStatusTool(opts?: {
         isLiveRunSession: isLiveRouteSession,
       });
       const routeContextText = formatSessionStatusRouteContext(routeDetails);
-      const visibleStatusText = routeContextText
-        ? `${fullStatusText}
-
-${routeContextText}`
-        : fullStatusText;
+      const stateVersion = getSessionStateVersion(resolved.key, agentId);
+      const stateChanges =
+        changesSince !== undefined
+          ? listSessionStateEventsSince(resolved.key, agentId, changesSince, 200)
+          : undefined;
+      const extraBlocks = [
+        routeContextText,
+        stateChanges ? formatSessionStateChanges({ stateVersion, stateChanges }) : undefined,
+      ].filter((block): block is string => Boolean(block));
+      const visibleStatusText =
+        extraBlocks.length > 0
+          ? `${fullStatusText}\n\n${extraBlocks.join("\n\n")}`
+          : fullStatusText;
       const modelOverrideForResult =
         modelRaw === undefined
           ? undefined
@@ -886,6 +914,8 @@ ${routeContextText}`
           ok: true,
           sessionKey: resolved.key,
           changedModel,
+          stateVersion,
+          ...(stateChanges ? { stateChanges } : {}),
           ...(modelRaw !== undefined
             ? {
                 model: resultOverrideModel ?? defaultModelForCard,

src/agents/tools/sessions-helpers.ts

@@ -62,6 +62,7 @@ export type SessionListRow = {
   pinned?: boolean;
   pinnedAt?: number;
   sessionId?: string;
+  stateVersion?: number;
   model?: string;
   contextTokens?: number | null;
   totalTokens?: number | null;

src/agents/tools/sessions-list-tool.test.ts

@@ -16,12 +16,21 @@ const mocks = vi.hoisted(() => ({
     requesterInternalKey: undefined,
     restrictToSpawned: false,
   })),
+  getSessionStateVersions: vi.fn(
+    (_refs: Array<{ sessionKey: string; agentId: string }>) =>
+      ({}) as Record<string, Record<string, number>>,
+  ),
 }));
 
 vi.mock("../../gateway/call.js", () => ({
   callGateway: (opts: unknown) => mocks.gatewayCall(opts),
 }));
 
+vi.mock("../../sessions/session-state-events.js", () => ({
+  getSessionStateVersions: (refs: Array<{ sessionKey: string; agentId: string }>) =>
+    mocks.getSessionStateVersions(refs),
+}));
+
 vi.mock("./sessions-helpers.js", async (importActual) => {
   const actual = await importActual<typeof import("./sessions-helpers.js")>();
   return {
@@ -50,6 +59,7 @@ type SessionsListDetails = {
     archivedAt?: number;
     pinned?: boolean;
     pinnedAt?: number;
+    stateVersion?: number;
     reasoningLevel?: string;
     responseUsage?: string;
     thinkingLevel?: string;
@@ -75,6 +85,29 @@ describe("sessions-list-tool", () => {
       requesterInternalKey: undefined,
       restrictToSpawned: false,
     });
+    mocks.getSessionStateVersions.mockReturnValue({});
+  });
+
+  it("adds nonzero state versions with one batch lookup", async () => {
+    mocks.gatewayCall.mockResolvedValue({
+      path: "/tmp/sessions.json",
+      sessions: [
+        { key: "agent:main:main", kind: "main", sessionId: "main-1" },
+        { key: "agent:main:subagent:child", kind: "other", sessionId: "child-1" },
+      ],
+    });
+    mocks.getSessionStateVersions.mockReturnValue({
+      main: { "agent:main:main": 7, "agent:main:subagent:child": 0 },
+    });
+
+    const result = await createSessionsListTool({ config: {} as never }).execute("call-state", {});
+
+    expect(mocks.getSessionStateVersions).toHaveBeenCalledWith([
+      { sessionKey: "agent:main:main", agentId: "main" },
+      { sessionKey: "agent:main:subagent:child", agentId: "main" },
+    ]);
+    expect(getSessionsListDetails(result).sessions?.[0]?.stateVersion).toBe(7);
+    expect(getSessionsListDetails(result).sessions?.[1]?.stateVersion).toBeUndefined();
   });
 
   it("keeps deliveryContext.threadId in sessions_list results", async () => {

src/agents/tools/sessions-list-tool.ts

@@ -22,6 +22,7 @@ import { callGateway } from "../../gateway/call.js";
 import { readSessionTitleFieldsFromTranscriptAsync } from "../../gateway/session-transcript-readers.js";
 import { deriveSessionTitle } from "../../gateway/session-utils.js";
 import { resolveAgentIdFromSessionKey } from "../../routing/session-key.js";
+import { getSessionStateVersions } from "../../sessions/session-state-events.js";
 import { normalizeFastModeAutoOnSeconds, normalizeFastModeSource } from "../../shared/fast-mode.js";
 import { deliveryContextFromSession } from "../../utils/delivery-context.shared.js";
 import {
@@ -149,6 +150,21 @@ export function createSessionsListTool(opts?: {
       });
 
       const sessions = Array.isArray(list?.sessions) ? list.sessions : [];
+      const stateVersions = getSessionStateVersions(
+        sessions.flatMap((entry) =>
+          entry && typeof entry === "object" && typeof entry.key === "string"
+            ? [
+                {
+                  sessionKey: entry.key,
+                  agentId:
+                    typeof entry.agentId === "string" && entry.agentId
+                      ? entry.agentId
+                      : resolveAgentIdFromSessionKey(entry.key),
+                },
+              ]
+            : [],
+        ),
+      );
       const storePath = typeof list?.path === "string" ? list.path : undefined;
       const visibilityGuard = createSessionVisibilityRowChecker({
         action: "list",
@@ -272,6 +288,11 @@ export function createSessionsListTool(opts?: {
         const effectiveFastMode = normalizeFastMode(entry.effectiveFastMode);
         const effectiveFastModeSource = normalizeFastModeSource(entry.effectiveFastModeSource);
         const fastAutoOnSeconds = normalizeFastModeAutoOnSeconds(entry.fastAutoOnSeconds);
+        // Version lookup keys on the store-owning agent (gateway row agentId), not the
+        // key-derived agent: bare "global" keys parse to the default agent id.
+        const stateVersionAgentId =
+          typeof entry.agentId === "string" && entry.agentId ? entry.agentId : resolvedAgentId;
+        const stateVersion = stateVersions[stateVersionAgentId]?.[key];
         const row: SessionListRow = {
           key: displayKey,
           agentId: resolvedAgentId,
@@ -320,6 +341,7 @@ export function createSessionsListTool(opts?: {
           pinned: entry.pinned === true,
           pinnedAt: typeof entry.pinnedAt === "number" ? entry.pinnedAt : undefined,
           sessionId,
+          ...(stateVersion ? { stateVersion } : {}),
           model: readStringValue(entry.model),
           contextTokens: typeof entry.contextTokens === "number" ? entry.contextTokens : undefined,
           totalTokens: typeof entry.totalTokens === "number" ? entry.totalTokens : undefined,

src/auto-reply/reply/commands-acp/lifecycle.ts

@@ -775,6 +775,7 @@ async function runAcpSteer(params: {
   await acpManager.runTurn({
     cfg: params.cfg,
     sessionKey: params.sessionKey,
+    provenance: "agent",
     text: params.instruction,
     mode: "steer",
     requestId: params.requestId,

src/auto-reply/reply/commands-goal.ts

@@ -166,6 +166,8 @@ export const handleGoalCommand: CommandHandler = async (params, allowTextCommand
   if (unauthorized) {
     return unauthorized;
   }
+  const actor = { type: "human" as const };
+  const goalAgentId = params.agentId;
 
   try {
     switch (parsed.action) {
@@ -191,6 +193,8 @@ export const handleGoalCommand: CommandHandler = async (params, allowTextCommand
           storePath: params.storePath,
           objective,
           fallbackEntry: params.sessionEntry,
+          actor,
+          agentId: goalAgentId,
         });
         syncGoalSessionEntry(params);
         markCommandSessionMetadataChanged(params);
@@ -206,6 +210,8 @@ export const handleGoalCommand: CommandHandler = async (params, allowTextCommand
           sessionKey: params.sessionKey,
           storePath: params.storePath,
           objective,
+          actor,
+          agentId: goalAgentId,
         });
         syncGoalSessionEntry(params);
         markCommandSessionMetadataChanged(params);
@@ -216,6 +222,8 @@ export const handleGoalCommand: CommandHandler = async (params, allowTextCommand
           sessionKey: params.sessionKey,
           storePath: params.storePath,
           status: "paused",
+          actor,
+          agentId: goalAgentId,
           ...(parsed.text ? { note: parsed.text } : {}),
         });
         syncGoalSessionEntry(params);
@@ -227,6 +235,8 @@ export const handleGoalCommand: CommandHandler = async (params, allowTextCommand
           sessionKey: params.sessionKey,
           storePath: params.storePath,
           status: "active",
+          actor,
+          agentId: goalAgentId,
           ...(parsed.text ? { note: parsed.text } : {}),
         });
         syncGoalSessionEntry(params);
@@ -241,6 +251,8 @@ export const handleGoalCommand: CommandHandler = async (params, allowTextCommand
           sessionKey: params.sessionKey,
           storePath: params.storePath,
           status: "complete",
+          actor,
+          agentId: goalAgentId,
           ...(parsed.text ? { note: parsed.text } : {}),
         });
         syncGoalSessionEntry(params);
@@ -253,6 +265,8 @@ export const handleGoalCommand: CommandHandler = async (params, allowTextCommand
           sessionKey: params.sessionKey,
           storePath: params.storePath,
           status: "blocked",
+          actor,
+          agentId: goalAgentId,
           ...(parsed.text ? { note: parsed.text } : {}),
         });
         syncGoalSessionEntry(params);
@@ -263,6 +277,8 @@ export const handleGoalCommand: CommandHandler = async (params, allowTextCommand
         const removed = await clearSessionGoal({
           sessionKey: params.sessionKey,
           storePath: params.storePath,
+          actor,
+          agentId: goalAgentId,
         });
         syncGoalSessionEntry(params);
         if (removed) {

src/auto-reply/reply/dispatch-acp.ts

@@ -28,6 +28,7 @@ import {
   type ExtractedFileImage,
 } from "../../media-understanding/extracted-file-images.js";
 import { resolveAgentIdFromSessionKey } from "../../routing/session-key.js";
+import { classifySessionStateActor } from "../../sessions/session-state-events.js";
 import { createLazyImportLoader } from "../../shared/lazy-promise.js";
 import { resolveStatusTtsSnapshot } from "../../tts/status-config.js";
 import { resolveConfiguredTtsMode } from "../../tts/tts-config.js";
@@ -711,6 +712,10 @@ export async function tryDispatchAcpReply(params: {
     await acpManager.runTurn({
       cfg: params.cfg,
       sessionKey: canonicalSessionKey,
+      provenance: classifySessionStateActor({
+        inputProvenance: params.ctx.InputProvenance,
+        sessionEffects: params.ctx.InboundEventKind === "room_event" ? "internal" : "visible",
+      }).actorType,
       text: resolveAcpTurnText({
         promptText: turnPromptText,
         sourceReplyDeliveryMode: params.sourceReplyDeliveryMode,

src/auto-reply/reply/session-system-events.ts

@@ -17,6 +17,10 @@ import {
   peekSystemEventEntries,
   type SystemEvent,
 } from "../../infra/system-events.js";
+import {
+  acknowledgeSessionStateNotices,
+  decodeSessionStateNoticeContextKey,
+} from "../../sessions/session-state-events.js";
 
 function isCronContextSystemEvent(event: SystemEvent): boolean {
   return event.contextKey?.startsWith("cron:") ?? false;
@@ -117,6 +121,14 @@ export async function drainFormattedSystemEvents(params: {
       suppressHeartbeatOwnedEvents: params.suppressHeartbeatOwnedEvents,
     }),
   );
+  const sessionStateTargets = queued
+    .map((event) =>
+      event.contextKey ? decodeSessionStateNoticeContextKey(event.contextKey) : undefined,
+    )
+    .filter((target): target is string => target !== undefined);
+  if (sessionStateTargets.length > 0) {
+    acknowledgeSessionStateNotices(params.sessionKey, sessionStateTargets);
+  }
   for (const event of queued) {
     const compacted = compactSystemEvent(event.text);
     if (!compacted) {

src/config/sessions/goals.ts

@@ -1,5 +1,9 @@
 // Session goal state tracks objective progress and token budgets in the session store.
 import crypto from "node:crypto";
+import {
+  recordSessionGoalChanged,
+  type SessionStateActorType,
+} from "../../sessions/session-state-events.js";
 import { formatTokenCount } from "../../utils/token-format.js";
 import { loadSessionEntry, patchSessionEntry } from "./session-accessor.js";
 import { resolveFreshSessionTotalTokens } from "./types.js";
@@ -16,6 +20,8 @@ type SessionGoalStoreOptions = {
   now?: number;
   fallbackEntry?: SessionEntry;
   persist?: boolean;
+  actor?: { type: SessionStateActorType; id?: string };
+  agentId?: string;
 };
 
 type CreateSessionGoalOptions = SessionGoalStoreOptions & {
@@ -63,6 +69,20 @@ function cloneGoal(goal: SessionGoal): SessionGoal {
   return { ...goal };
 }
 
+function recordGoalChange(
+  options: SessionGoalStoreOptions,
+  entry: SessionEntry,
+  summary: string,
+): void {
+  recordSessionGoalChanged({
+    sessionKey: options.sessionKey,
+    entry,
+    actor: options.actor,
+    agentId: options.agentId,
+    summary,
+  });
+}
+
 export function resolveSessionGoalDisplayState(
   entry: Pick<SessionEntry, "goal" | "totalTokens" | "totalTokensFresh">,
   now?: number,
@@ -223,6 +243,7 @@ export async function createSessionGoal(options: CreateSessionGoalOptions): Prom
   if (!result || !created) {
     throw new Error("session not found");
   }
+  recordGoalChange(options, result, "goal created");
   return cloneGoal(created);
 }
 
@@ -281,6 +302,7 @@ export async function updateSessionGoalStatus(
   if (!result || !updated) {
     throw new Error(foundSession ? "goal not found" : "session not found");
   }
+  recordGoalChange(options, result, `goal status changed to ${updated.status}`);
   return cloneGoal(updated);
 }
 
@@ -313,6 +335,7 @@ export async function updateSessionGoalObjective(
   if (!result || !updated) {
     throw new Error(foundSession ? "goal not found" : "session not found");
   }
+  recordGoalChange(options, result, "goal objective changed");
   return cloneGoal(updated);
 }
 
@@ -328,5 +351,8 @@ export async function clearSessionGoal(options: SessionGoalStoreOptions): Promis
       return { goal: undefined };
     },
   );
+  if (result && removed) {
+    recordGoalChange(options, result, "goal cleared");
+  }
   return Boolean(result && removed);
 }

src/gateway/server-methods/sessions.ts

@@ -85,6 +85,10 @@ import {
   runExclusiveSessionLifecycleMutation,
   SESSION_WORK_ADMISSION_DRAIN_TIMEOUT_MS,
 } from "../../sessions/session-lifecycle-admission.js";
+import {
+  handleSessionStateSessionDeleted,
+  recordSessionCompacted,
+} from "../../sessions/session-state-events.js";
 import { createLazyRuntimeModule } from "../../shared/lazy-runtime.js";
 import { ADMIN_SCOPE } from "../operator-scopes.js";
 import { resolveSessionKeyForRun } from "../server-session-key.js";
@@ -2657,6 +2661,12 @@ export const sessionsHandlers: GatewayRequestHandlers = {
     // operator UIs can point at the preserved checkout instead of orphaning it.
     let worktreePreserved: { id: string; branch: string; path: string } | undefined;
     if (deleted) {
+      // requestedAgentId wins: "global" canonical keys resolve to the default store
+      // agent, which would purge the wrong agent's rows for explicit-agent deletes.
+      handleSessionStateSessionDeleted(
+        target.canonicalKey ?? key,
+        requestedAgentId ?? resolveSessionStoreAgentId(cfg, target.canonicalKey ?? key),
+      );
       try {
         const worktree = managedWorktrees.findLiveByOwner("session", target.canonicalKey);
         if (worktree && !(await managedWorktrees.removeIfLossless(worktree.id))) {
@@ -3014,6 +3024,7 @@ export const sessionsHandlers: GatewayRequestHandlers = {
             return;
           }
 
+          const operationId = randomUUID();
           if (maxLines !== undefined) {
             const trimResult = await trimSessionTranscriptForManualCompact(
               {
@@ -3039,6 +3050,12 @@ export const sessionsHandlers: GatewayRequestHandlers = {
               undefined,
             );
             if (trimResult.compacted) {
+              recordSessionCompacted({
+                sessionKey: target.canonicalKey,
+                operationId,
+                sessionId,
+                agentId: target.agentId ?? requestedAgentId,
+              });
               emitSessionsChanged(context, {
                 sessionKey: target.canonicalKey,
                 ...(target.canonicalKey === "global" && target.agentId
@@ -3078,7 +3095,6 @@ export const sessionsHandlers: GatewayRequestHandlers = {
               workspaceDir: latestEntry.spawnedWorkspaceDir,
               cwd: latestEntry.spawnedCwd,
             }) ?? resolveAgentWorkspaceDir(cfg, target.agentId);
-          const operationId = randomUUID();
           emitSessionOperation(context, {
             operationId,
             operation: "compact",
@@ -3195,6 +3211,12 @@ export const sessionsHandlers: GatewayRequestHandlers = {
               );
               return;
             }
+            recordSessionCompacted({
+              sessionKey: target.canonicalKey,
+              operationId,
+              sessionId: result.result?.sessionId ?? sessionId,
+              agentId: target.agentId ?? requestedAgentId,
+            });
           }
 
           emitCompactionEnd(result.ok && result.compacted, result.reason);

src/gateway/server-startup-post-attach.ts

@@ -17,6 +17,7 @@ import { getPluginModuleLoaderStats } from "../plugins/plugin-module-loader-cach
 import type { PluginRegistry } from "../plugins/registry.js";
 import type { PluginServicesHandle } from "../plugins/services.js";
 import { runWithGatewayIndependentRootWorkAdmission } from "../process/gateway-work-admission.js";
+import { sweepSessionStateWatchNotices } from "../sessions/session-state-events.js";
 import { createLazyRuntimeModule } from "../shared/lazy-runtime.js";
 import {
   GATEWAY_EVENT_UPDATE_AVAILABLE,
@@ -1390,6 +1391,10 @@ export async function startGatewayPostAttachRuntime(
         log: params.log,
         refreshLatestUpdateRestartSentinel: runtimeDeps.refreshLatestUpdateRestartSentinel,
       });
+      const sessionStateSweepHandle = setImmediate(() => {
+        sweepSessionStateWatchNotices();
+      });
+      sessionStateSweepHandle.unref?.();
       if (!hasGatewayStartHooks(sidecarsResult.pluginRegistry)) {
         return;
       }

src/gateway/session-reset-service.ts

@@ -68,6 +68,7 @@ import {
   runExclusiveSessionLifecycleMutation,
   SESSION_WORK_ADMISSION_DRAIN_TIMEOUT_MS,
 } from "../sessions/session-lifecycle-admission.js";
+import { handleSessionStateSessionReset } from "../sessions/session-state-events.js";
 import {
   forgetActiveSessionForShutdown,
   listActiveSessionsForShutdown,
@@ -1252,6 +1253,7 @@ export async function performGatewaySessionReset(params: {
           });
         },
       });
+      handleSessionStateSessionReset(target.canonicalKey ?? params.key);
       const next = lifecycle.nextEntry;
       const selectedModel = resolveSessionModelRef(cfg, next, target.agentId);
       const resolved = {

src/infra/heartbeat-runner.session-state.test.ts

@@ -0,0 +1,20 @@
+import { describe, expect, it } from "vitest";
+import { testing } from "./heartbeat-runner.js";
+
+describe("session-state heartbeat wakes", () => {
+  it("infers the source and marks the wake as payload-bearing", () => {
+    expect(testing.inferHeartbeatWakeSourceFromReason("session-state:agent:main:child")).toBe(
+      "session-state",
+    );
+    expect(
+      testing.resolveHeartbeatWakePayloadFlags({
+        reason: "session-state:agent:main:child",
+      }),
+    ).toMatchObject({ isWakePayload: true });
+    expect(
+      testing.resolveHeartbeatWakePayloadFlags({
+        source: "session-state",
+      }),
+    ).toMatchObject({ isWakePayload: true });
+  });
+});

src/infra/heartbeat-runner.ts

@@ -1020,6 +1020,9 @@ function inferHeartbeatWakeSourceFromReason(reason?: string): HeartbeatWakeSourc
   if (trimmed.startsWith("acp:spawn:")) {
     return "acp-spawn";
   }
+  if (trimmed.startsWith("session-state:")) {
+    return "session-state";
+  }
   return undefined;
 }
 
@@ -1032,7 +1035,11 @@ function resolveHeartbeatWakePayloadFlags(params: {
   return {
     isExecEventWake: source === "exec-event",
     isCronWake: source === "cron",
-    isWakePayload: source === "hook" || source === "acp-spawn" || reason === "wake",
+    isWakePayload:
+      source === "hook" ||
+      source === "acp-spawn" ||
+      source === "session-state" ||
+      reason === "wake",
   };
 }
 
@@ -2388,7 +2395,11 @@ export async function runHeartbeatOnce(opts: {
   }
 }
 
-export const testing = { truncateHeartbeatPreview };
+export const testing = {
+  inferHeartbeatWakeSourceFromReason,
+  resolveHeartbeatWakePayloadFlags,
+  truncateHeartbeatPreview,
+};
 
 export function startHeartbeatRunner(opts: {
   cfg?: OpenClawConfig;

src/infra/heartbeat-wake.ts

@@ -39,6 +39,7 @@ export type HeartbeatWakeSource =
   | "background-task"
   | "background-task-blocked"
   | "acp-spawn"
+  | "session-state"
   | "cli-watchdog"
   | "restart-sentinel"
   | "retry"

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

@@ -0,0 +1,513 @@
+import { afterAll, afterEach, describe, expect, it, vi } from "vitest";
+import { cleanupTempDirs, makeTempDir } from "../../test/helpers/temp-dir.js";
+import { drainFormattedSystemEvents } from "../auto-reply/reply/session-system-events.js";
+import { upsertSessionEntry } from "../config/sessions/session-accessor.js";
+import type { OpenClawConfig } from "../config/types.openclaw.js";
+import {
+  resetHeartbeatWakeStateForTests,
+  setHeartbeatWakeHandler,
+} from "../infra/heartbeat-wake.js";
+import {
+  enqueueSystemEvent,
+  peekSystemEventEntries,
+  resetSystemEventsForTest,
+} from "../infra/system-events.js";
+import {
+  closeOpenClawStateDatabaseForTest,
+  openOpenClawStateDatabase,
+} from "../state/openclaw-state-db.js";
+import {
+  acknowledgeSessionStateNotices,
+  classifySessionStateActor,
+  getSessionStateVersion,
+  handleSessionStateSessionDeleted,
+  handleSessionStateSessionReset,
+  listSessionStateEventsSince,
+  pruneSessionStateEvents,
+  recordSessionCompacted,
+  recordSessionGoalChanged,
+  recordSessionStateEvent,
+  recordSubagentSpawned,
+  recordSubagentTerminalState,
+  sessionStateEventStoreLimits,
+  sweepSessionStateWatchNotices,
+} from "./session-state-events.js";
+
+const tempDirs: string[] = [];
+const watcher = "agent:main:main";
+const nestedWatcher = "agent:main:subagent:parent";
+const child = "agent:main:subagent:child";
+const cfg = {} as OpenClawConfig;
+
+function createDatabaseOptions() {
+  const stateDir = makeTempDir(tempDirs, "openclaw-session-state-");
+  vi.stubEnv("OPENCLAW_STATE_DIR", stateDir);
+  return { env: { ...process.env, OPENCLAW_STATE_DIR: stateDir } };
+}
+
+function eventInput(
+  overrides: Partial<Parameters<typeof recordSessionStateEvent>[0]> = {},
+): Parameters<typeof recordSessionStateEvent>[0] {
+  return {
+    sessionKey: child,
+    sessionId: "session-child",
+    agentId: "main",
+    kind: "human_direct_message",
+    actorType: "human",
+    summary: "human message via test",
+    watcherSessionKeys: [watcher],
+    ...overrides,
+  };
+}
+
+function readCursor(
+  database: ReturnType<typeof createDatabaseOptions>,
+  watcherSessionKey = watcher,
+) {
+  return openOpenClawStateDatabase(database)
+    .db.prepare(
+      `SELECT last_seen_sequence, notified_sequence, material_sequence
+       FROM session_watch_cursors
+       WHERE watcher_session_key = ? AND target_session_key = ?`,
+    )
+    .get(watcherSessionKey, child) as
+    | {
+        last_seen_sequence: number;
+        notified_sequence: number;
+        material_sequence: number;
+      }
+    | undefined;
+}
+
+function seedChild(
+  database: ReturnType<typeof createDatabaseOptions>,
+  watcherSessionKey = watcher,
+) {
+  return recordSessionStateEvent(
+    eventInput({
+      kind: "child_spawned",
+      actorType: "agent",
+      actorId: watcherSessionKey,
+      dedupeKey: `child-spawned:${watcherSessionKey}`,
+      watcherSessionKeys: [watcherSessionKey],
+    }),
+    database,
+  );
+}
+
+async function createWatcherSession(
+  database: ReturnType<typeof createDatabaseOptions>,
+  watcherSessionKey = watcher,
+) {
+  await upsertSessionEntry(
+    { sessionKey: watcherSessionKey, env: database.env },
+    { sessionId: `session-${watcherSessionKey}`, updatedAt: Date.now() },
+  );
+}
+
+afterEach(() => {
+  closeOpenClawStateDatabaseForTest();
+  resetSystemEventsForTest();
+  resetHeartbeatWakeStateForTests();
+  vi.unstubAllEnvs();
+  vi.useRealTimers();
+});
+
+afterAll(() => {
+  cleanupTempDirs(tempDirs);
+});
+
+describe("session state events", () => {
+  it("bumps a durable head that survives pruning all retained rows", () => {
+    const database = createDatabaseOptions();
+    const now = Date.now();
+    const event = recordSessionStateEvent(eventInput(), { ...database, now });
+    expect(getSessionStateVersion(child, "main", database)).toBe(event?.sequence);
+
+    pruneSessionStateEvents({
+      ...database,
+      now: now + sessionStateEventStoreLimits.retentionMs + 1,
+    });
+
+    expect(listSessionStateEventsSince(child, "main", 0, 200, database).events).toEqual([]);
+    expect(getSessionStateVersion(child, "main", database)).toBe(event?.sequence);
+  });
+
+  it("freezes one notice watermark while material events continue", () => {
+    const database = createDatabaseOptions();
+    seedChild(database);
+    const first = recordSessionStateEvent(eventInput(), database)!;
+    recordSessionStateEvent(eventInput(), database);
+    const third = recordSessionStateEvent(eventInput(), database)!;
+
+    expect(peekSystemEventEntries(watcher)).toHaveLength(1);
+    expect(readCursor(database)).toEqual({
+      last_seen_sequence: first.sequence - 1,
+      notified_sequence: first.sequence,
+      material_sequence: third.sequence,
+    });
+  });
+
+  it("opens a fresh notice for material work interleaved before ack", () => {
+    const database = createDatabaseOptions();
+    seedChild(database);
+    const frozen = recordSessionStateEvent(eventInput(), database)!;
+    const interleaved = recordSessionStateEvent(eventInput(), database)!;
+    resetSystemEventsForTest();
+
+    acknowledgeSessionStateNotices(watcher, [child], database);
+
+    expect(readCursor(database)).toEqual({
+      last_seen_sequence: frozen.sequence,
+      notified_sequence: interleaved.sequence,
+      material_sequence: interleaved.sequence,
+    });
+    expect(peekSystemEventEntries(watcher)).toHaveLength(1);
+    expect(peekSystemEventEntries(watcher)[0]?.text).toContain(`changesSince ${frozen.sequence}`);
+  });
+
+  it("does not reopen an acked notice for log-only events or during sweep", async () => {
+    const database = createDatabaseOptions();
+    await createWatcherSession(database);
+    seedChild(database);
+    const material = recordSessionStateEvent(eventInput(), database)!;
+    recordSessionStateEvent(
+      eventInput({ kind: "run_completed", actorType: "system", runId: "run-log-only" }),
+      database,
+    );
+    resetSystemEventsForTest();
+
+    acknowledgeSessionStateNotices(watcher, [child], database);
+    expect(readCursor(database)).toEqual({
+      last_seen_sequence: material.sequence,
+      notified_sequence: material.sequence,
+      material_sequence: material.sequence,
+    });
+    expect(peekSystemEventEntries(watcher)).toEqual([]);
+
+    sweepSessionStateWatchNotices(database);
+    expect(peekSystemEventEntries(watcher)).toEqual([]);
+  });
+
+  it("wakes main watchers but only queues notices for nested watchers", async () => {
+    vi.useFakeTimers();
+    const wakes = vi.fn(async () => ({ status: "ran" as const, durationMs: 1 }));
+    setHeartbeatWakeHandler(wakes);
+    const database = createDatabaseOptions();
+    seedChild(database, nestedWatcher);
+
+    recordSessionStateEvent(eventInput({ watcherSessionKeys: [nestedWatcher] }), database);
+    await vi.advanceTimersByTimeAsync(300);
+    expect(peekSystemEventEntries(nestedWatcher)).toHaveLength(1);
+    expect(wakes).not.toHaveBeenCalled();
+
+    seedChild(database, watcher);
+    recordSessionStateEvent(eventInput(), database);
+    await vi.advanceTimersByTimeAsync(300);
+    expect(wakes).toHaveBeenCalledWith(
+      expect.objectContaining({ source: "session-state", sessionKey: watcher }),
+    );
+  });
+
+  it("suppresses watcher-originated material events", () => {
+    const database = createDatabaseOptions();
+    const seeded = seedChild(database)!;
+    recordSessionStateEvent(eventInput({ actorType: "agent", actorId: watcher }), database);
+
+    expect(readCursor(database)).toEqual({
+      last_seen_sequence: seeded.sequence,
+      notified_sequence: seeded.sequence,
+      material_sequence: seeded.sequence,
+    });
+    expect(peekSystemEventEntries(watcher)).toEqual([]);
+  });
+
+  it("records log-only kinds without queueing notices", () => {
+    const database = createDatabaseOptions();
+    const event = recordSessionStateEvent(
+      eventInput({ kind: "compacted", actorType: "system" }),
+      database,
+    );
+
+    expect(getSessionStateVersion(child, "main", database)).toBe(event?.sequence);
+    expect(peekSystemEventEntries(watcher)).toEqual([]);
+  });
+
+  it("returns the existing row for a duplicate dedupe key", () => {
+    const database = createDatabaseOptions();
+    const input = eventInput({
+      kind: "run_failed",
+      actorType: "system",
+      runId: "run-1",
+      dedupeKey: "run-terminal:run-1",
+    });
+    const first = recordSessionStateEvent(input, database);
+    const duplicate = recordSessionStateEvent(input, database);
+
+    expect(duplicate?.sequence).toBe(first?.sequence);
+    expect(listSessionStateEventsSince(child, "main", 0, 200, database).events).toHaveLength(1);
+  });
+
+  it("re-enqueues and re-freezes pending notices after restart", async () => {
+    const database = createDatabaseOptions();
+    await createWatcherSession(database);
+    seedChild(database);
+    const material = recordSessionStateEvent(eventInput(), database)!;
+    resetSystemEventsForTest();
+
+    sweepSessionStateWatchNotices(database);
+
+    expect(peekSystemEventEntries(watcher)).toHaveLength(1);
+    expect(readCursor(database)?.notified_sequence).toBe(material.sequence);
+  });
+
+  it("self-heals a lost queued notice on the next material event", () => {
+    const database = createDatabaseOptions();
+    seedChild(database);
+    recordSessionStateEvent(eventInput(), database);
+    resetSystemEventsForTest();
+
+    recordSessionStateEvent(eventInput(), database);
+
+    expect(peekSystemEventEntries(watcher)).toHaveLength(1);
+  });
+
+  it("prunes retention and cap rows while keeping monotonic autoincrement heads", () => {
+    const database = createDatabaseOptions();
+    const now = Date.now();
+    const { db } = openOpenClawStateDatabase(database);
+    db.exec(`
+      WITH RECURSIVE rows(value) AS (
+        SELECT 1 UNION ALL SELECT value + 1 FROM rows WHERE value <= ${sessionStateEventStoreLimits.maxRows}
+      )
+      INSERT INTO session_state_events (
+        session_key, agent_id, kind, actor_type, occurred_at, summary
+      )
+      SELECT 'bulk', 'main', 'compacted', 'system', ${now}, 'bulk' FROM rows;
+    `);
+    const before = db
+      .prepare("SELECT max(sequence) AS sequence FROM session_state_events")
+      .get() as { sequence: number };
+
+    pruneSessionStateEvents({ ...database, now });
+    const count = db.prepare("SELECT count(*) AS count FROM session_state_events").get() as {
+      count: number;
+    };
+    expect(count.count).toBe(sessionStateEventStoreLimits.maxRows);
+
+    const next = recordSessionStateEvent(eventInput(), { ...database, now: now + 1 })!;
+    expect(next.sequence).toBeGreaterThan(before.sequence);
+    expect(getSessionStateVersion(child, "main", database)).toBe(next.sequence);
+  });
+
+  it("lists typed ascending deltas with truncation and history-gap signaling", () => {
+    const database = createDatabaseOptions();
+    const now = Date.now();
+    const first = recordSessionStateEvent(eventInput({ summary: "first" }), {
+      ...database,
+      now,
+    })!;
+    recordSessionStateEvent(eventInput({ summary: "second", payload: { status: "active" } }), {
+      ...database,
+      now: now + 1,
+    });
+    recordSessionStateEvent(eventInput({ summary: "third" }), { ...database, now: now + 2 });
+
+    const page = listSessionStateEventsSince(child, "main", 0, 2, database);
+    expect(page.events.map((event) => event.summary)).toEqual(["first", "second"]);
+    expect(page.events[1]?.payload).toEqual({ status: "active" });
+    expect(page.truncated).toBe(true);
+
+    // A manually removed row is not a retention gap: only pruning stamps the
+    // per-session watermark that historyGap may consult.
+    openOpenClawStateDatabase(database)
+      .db.prepare("DELETE FROM session_state_events WHERE sequence = ?")
+      .run(first.sequence);
+    expect(listSessionStateEventsSince(child, "main", 0, 200, database).historyGap).toBe(false);
+  });
+
+  it("reports history gaps only for actually pruned events, not sparse global sequences", () => {
+    const database = createDatabaseOptions();
+    const now = Date.now();
+    // Other sessions consume early global sequences; the child starts high.
+    for (let index = 0; index < 3; index += 1) {
+      recordSessionStateEvent(
+        eventInput({ sessionKey: "agent:main:subagent:noise", watcherSessionKeys: [] }),
+        { ...database, now },
+      );
+    }
+    const old = recordSessionStateEvent(eventInput({ summary: "old" }), { ...database, now })!;
+    expect(old.sequence).toBeGreaterThan(1);
+    expect(listSessionStateEventsSince(child, "main", 0, 200, database).historyGap).toBe(false);
+
+    const later = now + sessionStateEventStoreLimits.retentionMs + 1;
+    const fresh = recordSessionStateEvent(eventInput({ summary: "fresh" }), {
+      ...database,
+      now: later,
+    })!;
+    pruneSessionStateEvents({ ...database, now: later });
+
+    const sincePruned = listSessionStateEventsSince(child, "main", 0, 200, database);
+    expect(sincePruned.historyGap).toBe(true);
+    expect(sincePruned.events.map((event) => event.summary)).toEqual(["fresh"]);
+    expect(listSessionStateEventsSince(child, "main", old.sequence, 200, database).historyGap).toBe(
+      false,
+    );
+    expect(getSessionStateVersion(child, "main", database)).toBe(fresh.sequence);
+  });
+
+  it("suppresses cursors and notices for agent-ambiguous bare watcher keys", () => {
+    const database = createDatabaseOptions();
+    const event = recordSessionStateEvent(
+      eventInput({ watcherSessionKeys: ["global"] }),
+      database,
+    )!;
+    expect(event.sequence).toBeGreaterThan(0);
+    expect(peekSystemEventEntries("global")).toEqual([]);
+    const cursorRow = openOpenClawStateDatabase(database)
+      .db.prepare("SELECT COUNT(*) AS n FROM session_watch_cursors")
+      .get() as { n: number };
+    expect(cursorRow.n).toBe(0);
+  });
+
+  it("keeps same-keyed global sessions independent across agents", () => {
+    const database = createDatabaseOptions();
+    const mainEvent = recordSessionStateEvent(
+      eventInput({
+        sessionKey: "global",
+        agentId: "main",
+        kind: "goal_changed",
+        actorType: "human",
+        watcherSessionKeys: [],
+      }),
+      database,
+    )!;
+    const opsEvent = recordSessionStateEvent(
+      eventInput({
+        sessionKey: "global",
+        agentId: "ops",
+        kind: "goal_changed",
+        actorType: "human",
+        watcherSessionKeys: [],
+      }),
+      database,
+    )!;
+
+    expect(getSessionStateVersion("global", "main", database)).toBe(mainEvent.sequence);
+    expect(getSessionStateVersion("global", "ops", database)).toBe(opsEvent.sequence);
+    expect(
+      listSessionStateEventsSince("global", "main", 0, 200, database).events.map(
+        (event) => event.sequence,
+      ),
+    ).toEqual([mainEvent.sequence]);
+
+    handleSessionStateSessionDeleted("global", "ops", database);
+    expect(getSessionStateVersion("global", "ops", database)).toBe(0);
+    expect(getSessionStateVersion("global", "main", database)).toBe(mainEvent.sequence);
+  });
+
+  it("acks only drained session-state entries and ignores ordinary events", async () => {
+    const database = createDatabaseOptions();
+    seedChild(database);
+    const material = recordSessionStateEvent(eventInput(), database)!;
+    enqueueSystemEvent("Cron completed", { sessionKey: watcher, contextKey: "cron:job-1" });
+
+    await drainFormattedSystemEvents({
+      cfg,
+      sessionKey: watcher,
+      isMainSession: false,
+      isNewSession: false,
+    });
+    expect(readCursor(database)?.last_seen_sequence).toBe(material.sequence);
+
+    recordSessionStateEvent(eventInput(), database);
+    resetSystemEventsForTest();
+    enqueueSystemEvent("Exec completed", { sessionKey: watcher, contextKey: "exec:job-1" });
+    await drainFormattedSystemEvents({
+      cfg,
+      sessionKey: watcher,
+      isMainSession: false,
+      isNewSession: false,
+    });
+    expect(readCursor(database)?.last_seen_sequence).toBe(material.sequence);
+  });
+
+  it("keeps target history on reset and removes all ownership on delete", () => {
+    const database = createDatabaseOptions();
+    seedChild(database);
+    recordSessionStateEvent(eventInput(), database);
+
+    handleSessionStateSessionReset(watcher, database);
+    expect(readCursor(database)).toBeUndefined();
+    expect(
+      listSessionStateEventsSince(child, "main", 0, 200, database).events.length,
+    ).toBeGreaterThan(0);
+
+    handleSessionStateSessionDeleted(child, "main", database);
+    expect(getSessionStateVersion(child, "main", database)).toBe(0);
+    expect(listSessionStateEventsSince(child, "main", 0, 200, database).events).toEqual([]);
+  });
+
+  it("classifies missing provenance as human and inter-session provenance as agent", () => {
+    expect(classifySessionStateActor({})).toEqual({ actorType: "human" });
+    expect(
+      classifySessionStateActor({
+        inputProvenance: {
+          kind: "inter_session",
+          sourceSessionKey: "agent:main:source",
+        },
+      }),
+    ).toEqual({ actorType: "agent", actorId: "agent:main:source" });
+    expect(classifySessionStateActor({ internalEvents: [{}] })).toEqual({
+      actorType: "system",
+    });
+  });
+
+  it("projects spawn, terminal, goal, and compaction producer helpers", () => {
+    const database = createDatabaseOptions();
+    recordSubagentSpawned({
+      childSessionKey: child,
+      childRunId: "run-child",
+      requesterSessionKey: watcher,
+      agentId: "main",
+    });
+    recordSubagentTerminalState({
+      childSessionKey: child,
+      runId: "run-child",
+      requesterSessionKey: watcher,
+      outcomeStatus: "ok",
+    });
+    recordSubagentTerminalState({
+      childSessionKey: child,
+      runId: "run-child",
+      requesterSessionKey: watcher,
+      outcomeStatus: "ok",
+    });
+    recordSessionGoalChanged({
+      sessionKey: child,
+      entry: {
+        sessionId: "session-child",
+        updatedAt: Date.now(),
+        spawnedBy: watcher,
+      },
+      actor: { type: "human" },
+      summary: "goal created",
+    });
+    recordSessionCompacted({
+      sessionKey: child,
+      operationId: "compact-1",
+      sessionId: "session-child",
+    });
+    recordSessionCompacted({
+      sessionKey: child,
+      operationId: "compact-1",
+      sessionId: "session-child",
+    });
+
+    expect(
+      listSessionStateEventsSince(child, "main", 0, 200, database).events.map(
+        (event) => event.kind,
+      ),
+    ).toEqual(["child_spawned", "run_completed", "goal_changed", "compacted"]);
+  });
+});

src/sessions/session-state-events.ts

@@ -0,0 +1,884 @@
+/** Best-effort durable signal log for session state changes. */
+import type { DatabaseSync } from "node:sqlite";
+import type { Insertable, Selectable } from "kysely";
+import { loadSessionEntry } from "../config/sessions/session-accessor.js";
+import type { SessionEntry } from "../config/sessions/types.js";
+import { requestHeartbeat } from "../infra/heartbeat-wake.js";
+import {
+  executeSqliteQuerySync,
+  executeSqliteQueryTakeFirstSync,
+  getNodeSqliteKysely,
+} from "../infra/kysely-sync.js";
+import { normalizeSqliteNumber } from "../infra/sqlite-number.js";
+import { enqueueSystemEvent } from "../infra/system-events.js";
+import { createSubsystemLogger } from "../logging/subsystem.js";
+import {
+  isSubagentSessionKey,
+  parseAgentSessionKey,
+  resolveAgentIdFromSessionKey,
+} from "../routing/session-key.js";
+import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js";
+import {
+  openOpenClawStateDatabase,
+  runOpenClawStateWriteTransaction,
+  type OpenClawStateDatabaseOptions,
+} from "../state/openclaw-state-db.js";
+import type { InputProvenance } from "./input-provenance.js";
+
+export type SessionStateActorType = "human" | "agent" | "system";
+export type SessionStateEventKind =
+  | "human_direct_message"
+  | "run_completed"
+  | "run_failed"
+  | "child_spawned"
+  | "goal_changed"
+  | "compacted";
+
+export type SessionStateEventInput = {
+  sessionKey: string;
+  sessionId?: string;
+  agentId: string;
+  kind: SessionStateEventKind;
+  actorType: SessionStateActorType;
+  actorId?: string;
+  runId?: string;
+  dedupeKey?: string;
+  summary: string;
+  payload?: Record<string, unknown>;
+  watcherSessionKeys?: readonly string[];
+};
+
+export type SessionStateEventRecord = {
+  sequence: number;
+  sessionKey: string;
+  sessionId?: string;
+  agentId: string;
+  kind: SessionStateEventKind;
+  actorType: SessionStateActorType;
+  actorId?: string;
+  runId?: string;
+  occurredAt: number;
+  summary: string;
+  payload?: Record<string, unknown>;
+};
+
+type SessionStateDatabase = Pick<
+  OpenClawStateKyselyDatabase,
+  "session_state_events" | "session_state_heads" | "session_watch_cursors"
+>;
+type SessionStateEventsTable = OpenClawStateKyselyDatabase["session_state_events"];
+type SessionStateEventRow = Selectable<SessionStateEventsTable>;
+type SessionWatchCursorRow = Selectable<OpenClawStateKyselyDatabase["session_watch_cursors"]>;
+
+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 SESSION_STATE_CONTEXT_PREFIX = "session-state:";
+const log = createSubsystemLogger("sessions/state-events");
+let lastPruneAt = 0;
+
+// Future utility-model materiality belongs at this single deterministic seam; no config until then.
+const NOTIFY_BY_KIND: Record<SessionStateEventKind, boolean> = {
+  human_direct_message: true,
+  goal_changed: true,
+  run_completed: false,
+  run_failed: false,
+  child_spawned: false,
+  compacted: false,
+};
+
+function getSessionStateKysely(db: DatabaseSync) {
+  return getNodeSqliteKysely<SessionStateDatabase>(db);
+}
+
+function normalizeOptionalSqliteNumber(
+  value: number | bigint | null | undefined,
+): number | undefined {
+  return value === undefined ? undefined : normalizeSqliteNumber(value);
+}
+
+function parsePayload(value: string | null): Record<string, unknown> | undefined {
+  if (!value) {
+    return undefined;
+  }
+  try {
+    const parsed: unknown = JSON.parse(value);
+    return parsed && typeof parsed === "object" && !Array.isArray(parsed)
+      ? (parsed as Record<string, unknown>)
+      : undefined;
+  } catch {
+    return undefined;
+  }
+}
+
+function rowToSessionStateEvent(row: SessionStateEventRow): SessionStateEventRecord {
+  const payload = parsePayload(row.payload_json);
+  return {
+    sequence: normalizeSqliteNumber(row.sequence) ?? 0,
+    sessionKey: row.session_key,
+    ...(row.session_id ? { sessionId: row.session_id } : {}),
+    agentId: row.agent_id,
+    kind: row.kind as SessionStateEventKind,
+    actorType: row.actor_type as SessionStateActorType,
+    ...(row.actor_id ? { actorId: row.actor_id } : {}),
+    ...(row.run_id ? { runId: row.run_id } : {}),
+    occurredAt: normalizeSqliteNumber(row.occurred_at) ?? 0,
+    summary: row.summary,
+    ...(payload ? { payload } : {}),
+  };
+}
+
+function bindSessionStateEvent(
+  input: SessionStateEventInput,
+  occurredAt: number,
+): Insertable<SessionStateEventsTable> {
+  return {
+    dedupe_key: input.dedupeKey ?? null,
+    session_key: input.sessionKey,
+    session_id: input.sessionId ?? null,
+    agent_id: input.agentId,
+    kind: input.kind,
+    actor_type: input.actorType,
+    actor_id: input.actorId ?? null,
+    run_id: input.runId ?? null,
+    occurred_at: occurredAt,
+    summary: input.summary,
+    payload_json: input.payload ? JSON.stringify(input.payload) : null,
+  };
+}
+
+function encodeNoticeTarget(sessionKey: string): string {
+  return Buffer.from(sessionKey, "utf8").toString("hex");
+}
+
+export function decodeSessionStateNoticeContextKey(contextKey: string): string | undefined {
+  if (!contextKey.startsWith(SESSION_STATE_CONTEXT_PREFIX)) {
+    return undefined;
+  }
+  const encoded = contextKey.slice(SESSION_STATE_CONTEXT_PREFIX.length);
+  if (!encoded || encoded.length % 2 !== 0 || !/^[0-9a-f]+$/.test(encoded)) {
+    return undefined;
+  }
+  return Buffer.from(encoded, "hex").toString("utf8");
+}
+
+function sessionStateNoticeText(targetSessionKey: string, lastSeenSequence: number): string {
+  return `Another actor has interacted with child session "${targetSessionKey}" since you last synced. Your assumptions about that session may be stale. Call session_status with sessionKey "${targetSessionKey}" and changesSince ${lastSeenSequence} before acting on it.`;
+}
+
+function shouldWakeWatcher(watcherSessionKey: string): boolean {
+  return !isSubagentSessionKey(watcherSessionKey);
+}
+
+// Bare keys (session.scope="global") are store-local per agent, but cursors, the
+// system-event queue, and heartbeat wakes are keyed by session key alone. A notice
+// 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 or cursors.
+function isNotifiableWatcherKey(watcherSessionKey: string): boolean {
+  return parseAgentSessionKey(watcherSessionKey) != null;
+}
+
+function enqueueSessionStateNotice(params: {
+  watcherSessionKey: string;
+  targetSessionKey: string;
+  lastSeenSequence: number;
+}): void {
+  enqueueSystemEvent(sessionStateNoticeText(params.targetSessionKey, params.lastSeenSequence), {
+    sessionKey: params.watcherSessionKey,
+    contextKey: `${SESSION_STATE_CONTEXT_PREFIX}${encodeNoticeTarget(params.targetSessionKey)}`,
+  });
+  if (!shouldWakeWatcher(params.watcherSessionKey)) {
+    return;
+  }
+  requestHeartbeat({
+    source: "session-state",
+    intent: "event",
+    reason: `session-state:${params.targetSessionKey}`,
+    sessionKey: params.watcherSessionKey,
+  });
+}
+
+function readCursor(
+  db: DatabaseSync,
+  watcherSessionKey: string,
+  targetSessionKey: string,
+): SessionWatchCursorRow | undefined {
+  return executeSqliteQueryTakeFirstSync(
+    db,
+    getSessionStateKysely(db)
+      .selectFrom("session_watch_cursors")
+      .selectAll()
+      .where("watcher_session_key", "=", watcherSessionKey)
+      .where("target_session_key", "=", targetSessionKey),
+  );
+}
+
+function upsertSeedCursor(params: {
+  db: DatabaseSync;
+  watcherSessionKey: string;
+  targetSessionKey: string;
+  sequence: number;
+  now: number;
+}): void {
+  executeSqliteQuerySync(
+    params.db,
+    getSessionStateKysely(params.db)
+      .insertInto("session_watch_cursors")
+      .values({
+        watcher_session_key: params.watcherSessionKey,
+        target_session_key: params.targetSessionKey,
+        last_seen_sequence: params.sequence,
+        notified_sequence: params.sequence,
+        material_sequence: params.sequence,
+        updated_at: params.now,
+      })
+      .onConflict((conflict) =>
+        conflict.columns(["watcher_session_key", "target_session_key"]).doUpdateSet({
+          last_seen_sequence: params.sequence,
+          notified_sequence: params.sequence,
+          material_sequence: params.sequence,
+          updated_at: params.now,
+        }),
+      ),
+  );
+}
+
+function updateMaterialCursor(params: {
+  db: DatabaseSync;
+  watcherSessionKey: string;
+  targetSessionKey: string;
+  sequence: number;
+  now: number;
+}): number {
+  const current = readCursor(params.db, params.watcherSessionKey, params.targetSessionKey);
+  const lastSeen = normalizeOptionalSqliteNumber(current?.last_seen_sequence) ?? 0;
+  const notified = normalizeOptionalSqliteNumber(current?.notified_sequence) ?? 0;
+  const frozenNotified = notified === lastSeen ? params.sequence : notified;
+  executeSqliteQuerySync(
+    params.db,
+    getSessionStateKysely(params.db)
+      .insertInto("session_watch_cursors")
+      .values({
+        watcher_session_key: params.watcherSessionKey,
+        target_session_key: params.targetSessionKey,
+        last_seen_sequence: lastSeen,
+        notified_sequence: frozenNotified,
+        material_sequence: params.sequence,
+        updated_at: params.now,
+      })
+      .onConflict((conflict) =>
+        conflict.columns(["watcher_session_key", "target_session_key"]).doUpdateSet({
+          notified_sequence: frozenNotified,
+          material_sequence: params.sequence,
+          updated_at: params.now,
+        }),
+      ),
+  );
+  return lastSeen;
+}
+
+/** Classify the actor once at producer boundaries; missing provenance is interactive human input. */
+export function classifySessionStateActor(opts: {
+  inputProvenance?: InputProvenance;
+  internalEvents?: readonly unknown[];
+  sessionEffects?: "visible" | "internal";
+  humanActorId?: string;
+}): { actorType: SessionStateActorType; actorId?: string } {
+  if (opts.inputProvenance?.kind === "inter_session") {
+    return {
+      actorType: "agent",
+      ...(opts.inputProvenance.sourceSessionKey
+        ? { actorId: opts.inputProvenance.sourceSessionKey }
+        : {}),
+    };
+  }
+  if (
+    opts.inputProvenance?.kind === "internal_system" ||
+    (opts.internalEvents?.length ?? 0) > 0 ||
+    opts.sessionEffects === "internal"
+  ) {
+    return { actorType: "system" };
+  }
+  return { actorType: "human", ...(opts.humanActorId ? { actorId: opts.humanActorId } : {}) };
+}
+
+/** Append a signal-log event without allowing signaling failure to fail the originating action. */
+export function recordSessionStateEvent(
+  input: SessionStateEventInput,
+  options: OpenClawStateDatabaseOptions & { now?: number } = {},
+): SessionStateEventRecord | undefined {
+  const occurredAt = options.now ?? Date.now();
+  const notices: Array<{
+    watcherSessionKey: string;
+    targetSessionKey: string;
+    lastSeenSequence: number;
+  }> = [];
+  try {
+    const event = runOpenClawStateWriteTransaction(({ db }) => {
+      const insert = executeSqliteQuerySync(
+        db,
+        getSessionStateKysely(db)
+          .insertInto("session_state_events")
+          .values(bindSessionStateEvent(input, occurredAt))
+          .onConflict((conflict) => conflict.column("dedupe_key").doNothing()),
+      );
+      const insertedSequence = insert.insertId ? Number(insert.insertId) : undefined;
+      if (insertedSequence === undefined) {
+        if (!input.dedupeKey) {
+          return undefined;
+        }
+        const existing = executeSqliteQueryTakeFirstSync(
+          db,
+          getSessionStateKysely(db)
+            .selectFrom("session_state_events")
+            .selectAll()
+            .where("dedupe_key", "=", input.dedupeKey),
+        );
+        return existing ? rowToSessionStateEvent(existing) : undefined;
+      }
+
+      executeSqliteQuerySync(
+        db,
+        getSessionStateKysely(db)
+          .insertInto("session_state_heads")
+          .values({
+            session_key: input.sessionKey,
+            agent_id: input.agentId,
+            last_sequence: insertedSequence,
+            updated_at: occurredAt,
+          })
+          .onConflict((conflict) =>
+            // (session_key, agent_id) composite identity: under session.scope="global"
+            // every agent owns a session-store row keyed "global"; a key-only head
+            // would let agents overwrite each other's version heads.
+            conflict.columns(["session_key", "agent_id"]).doUpdateSet({
+              last_sequence: insertedSequence,
+              updated_at: occurredAt,
+            }),
+          ),
+      );
+
+      const watcherSessionKeys = [...new Set(input.watcherSessionKeys ?? [])].filter(
+        (key) => Boolean(key) && isNotifiableWatcherKey(key),
+      );
+      for (const watcherSessionKey of watcherSessionKeys) {
+        if (input.kind === "child_spawned") {
+          upsertSeedCursor({
+            db,
+            watcherSessionKey,
+            targetSessionKey: input.sessionKey,
+            sequence: insertedSequence,
+            now: occurredAt,
+          });
+          continue;
+        }
+        if (!NOTIFY_BY_KIND[input.kind] || input.actorId === watcherSessionKey) {
+          continue;
+        }
+        const lastSeenSequence = updateMaterialCursor({
+          db,
+          watcherSessionKey,
+          targetSessionKey: input.sessionKey,
+          sequence: insertedSequence,
+          now: occurredAt,
+        });
+        notices.push({ watcherSessionKey, targetSessionKey: input.sessionKey, lastSeenSequence });
+      }
+
+      const row = executeSqliteQueryTakeFirstSync(
+        db,
+        getSessionStateKysely(db)
+          .selectFrom("session_state_events")
+          .selectAll()
+          .where("sequence", "=", insertedSequence),
+      );
+      return row ? rowToSessionStateEvent(row) : undefined;
+    }, options);
+
+    for (const notice of notices) {
+      enqueueSessionStateNotice(notice);
+    }
+    if (occurredAt - lastPruneAt > SESSION_STATE_PRUNE_INTERVAL_MS) {
+      pruneSessionStateEvents({ ...options, now: occurredAt });
+    }
+    return event;
+  } catch (error) {
+    log.warn(`failed to record session state event: ${String(error)}`);
+    return undefined;
+  }
+}
+
+/** Return the durable signal-log head for one session; degrades to 0 on read failure. */
+export function getSessionStateVersion(
+  sessionKey: string,
+  agentId: string,
+  options: OpenClawStateDatabaseOptions = {},
+): number {
+  try {
+    const { db } = openOpenClawStateDatabase(options);
+    const row = executeSqliteQueryTakeFirstSync(
+      db,
+      getSessionStateKysely(db)
+        .selectFrom("session_state_heads")
+        .select("last_sequence")
+        .where("session_key", "=", sessionKey)
+        .where("agent_id", "=", agentId),
+    );
+    return normalizeOptionalSqliteNumber(row?.last_sequence) ?? 0;
+  } catch (error) {
+    // Best-effort log: enrichment reads must never fail core session tools.
+    log.warn(`failed to read session state version: ${String(error)}`);
+    return 0;
+  }
+}
+
+/** Batch durable signal-log heads for session-list enrichment, keyed agent → session key. */
+export function getSessionStateVersions(
+  refs: ReadonlyArray<{ sessionKey: string; agentId: string }>,
+  options: OpenClawStateDatabaseOptions = {},
+): Record<string, Record<string, number>> {
+  const keys = [...new Set(refs.map((ref) => ref.sessionKey).filter(Boolean))];
+  if (keys.length === 0) {
+    return {};
+  }
+  const byAgent: Record<string, Record<string, number>> = {};
+  try {
+    const { db } = openOpenClawStateDatabase(options);
+    // Chunk IN() binds: sessions_list accepts arbitrary limits and SQLite caps
+    // host parameters per statement.
+    for (let offset = 0; offset < keys.length; offset += 500) {
+      const rows = executeSqliteQuerySync(
+        db,
+        getSessionStateKysely(db)
+          .selectFrom("session_state_heads")
+          .select(["session_key", "agent_id", "last_sequence"])
+          .where("session_key", "in", keys.slice(offset, offset + 500)),
+      ).rows;
+      for (const row of rows) {
+        (byAgent[row.agent_id] ??= {})[row.session_key] =
+          normalizeSqliteNumber(row.last_sequence) ?? 0;
+      }
+    }
+  } catch (error) {
+    // Best-effort log: enrichment reads must never fail core session tools.
+    log.warn(`failed to read session state versions: ${String(error)}`);
+  }
+  return byAgent;
+}
+
+/** List retained signal-log events after a version without advancing watcher cursors. */
+export function listSessionStateEventsSince(
+  sessionKey: string,
+  agentId: string,
+  afterSequence: number,
+  limit = 200,
+  options: OpenClawStateDatabaseOptions = {},
+): {
+  events: SessionStateEventRecord[];
+  truncated: boolean;
+  earliestAvailableSequence: number;
+  historyGap: boolean;
+} {
+  try {
+    const boundedLimit = Math.max(1, Math.min(200, Math.floor(limit)));
+    const { db } = openOpenClawStateDatabase(options);
+    const kysely = getSessionStateKysely(db);
+    const rows = executeSqliteQuerySync(
+      db,
+      kysely
+        .selectFrom("session_state_events")
+        .selectAll()
+        .where("session_key", "=", sessionKey)
+        .where("agent_id", "=", agentId)
+        .where("sequence", ">", afterSequence)
+        .orderBy("sequence", "asc")
+        .limit(boundedLimit + 1),
+    ).rows;
+    const earliest = executeSqliteQueryTakeFirstSync(
+      db,
+      kysely
+        .selectFrom("session_state_events")
+        .select((eb) => eb.fn.min<number>("sequence").as("sequence"))
+        .where("session_key", "=", sessionKey)
+        .where("agent_id", "=", agentId),
+    );
+    const headRow = executeSqliteQueryTakeFirstSync(
+      db,
+      kysely
+        .selectFrom("session_state_heads")
+        .select(["last_sequence", "pruned_max_sequence"])
+        .where("session_key", "=", sessionKey)
+        .where("agent_id", "=", agentId),
+    );
+    const head = normalizeOptionalSqliteNumber(headRow?.last_sequence) ?? 0;
+    const prunedMax = normalizeOptionalSqliteNumber(headRow?.pruned_max_sequence) ?? 0;
+    const earliestAvailableSequence =
+      normalizeOptionalSqliteNumber(earliest?.sequence) ?? (head > 0 ? head + 1 : 0);
+    return {
+      events: rows.slice(0, boundedLimit).map(rowToSessionStateEvent),
+      truncated: rows.length > boundedLimit,
+      earliestAvailableSequence,
+      // Sequences are globally sparse, so distance from earliest retained proves nothing.
+      // Only the per-session pruned watermark stamped by pruneSessionStateEvents can say
+      // whether events this cursor never saw were actually removed.
+      historyGap: afterSequence < prunedMax,
+    };
+  } catch (error) {
+    // Best-effort log: enrichment reads must never fail core session tools.
+    log.warn(`failed to list session state events: ${String(error)}`);
+    return { events: [], truncated: false, earliestAvailableSequence: 0, historyGap: false };
+  }
+}
+
+/** Ack only the frozen notice watermark; advancing to head would lose an interleaved event. */
+export function acknowledgeSessionStateNotices(
+  watcherSessionKey: string,
+  targetSessionKeys: readonly string[],
+  options: OpenClawStateDatabaseOptions & { now?: number } = {},
+): void {
+  const now = options.now ?? Date.now();
+  const followups: Array<{
+    watcherSessionKey: string;
+    targetSessionKey: string;
+    lastSeenSequence: number;
+  }> = [];
+  try {
+    runOpenClawStateWriteTransaction(({ db }) => {
+      for (const targetSessionKey of new Set(targetSessionKeys)) {
+        const row = readCursor(db, watcherSessionKey, targetSessionKey);
+        if (!row) {
+          continue;
+        }
+        const notified = normalizeSqliteNumber(row.notified_sequence) ?? 0;
+        const material = normalizeSqliteNumber(row.material_sequence) ?? 0;
+        const nextNotified = material > notified ? material : notified;
+        executeSqliteQuerySync(
+          db,
+          getSessionStateKysely(db)
+            .updateTable("session_watch_cursors")
+            .set({
+              last_seen_sequence: notified,
+              notified_sequence: nextNotified,
+              updated_at: now,
+            })
+            .where("watcher_session_key", "=", watcherSessionKey)
+            .where("target_session_key", "=", targetSessionKey),
+        );
+        if (material > notified) {
+          followups.push({
+            watcherSessionKey,
+            targetSessionKey,
+            lastSeenSequence: notified,
+          });
+        }
+      }
+    }, options);
+    for (const followup of followups) {
+      enqueueSessionStateNotice(followup);
+    }
+  } catch (error) {
+    log.warn(`failed to acknowledge session state notices: ${String(error)}`);
+  }
+}
+
+/** Reset parent-side assumptions while retaining target history across session incarnations. */
+export function handleSessionStateSessionReset(
+  sessionKey: string,
+  options: OpenClawStateDatabaseOptions = {},
+): void {
+  try {
+    runOpenClawStateWriteTransaction(({ db }) => {
+      // Cursor rows only exist for agent-qualified watcher keys (see
+      // isNotifiableWatcherKey), so a bare-key reset cannot cross agents here.
+      executeSqliteQuerySync(
+        db,
+        getSessionStateKysely(db)
+          .deleteFrom("session_watch_cursors")
+          .where("watcher_session_key", "=", sessionKey),
+      );
+    }, options);
+  } catch (error) {
+    log.warn(`failed to reset session state cursors: ${String(error)}`);
+  }
+}
+
+/** Delete all signal-log and cursor state owned by a deleted session key. */
+export function handleSessionStateSessionDeleted(
+  sessionKey: string,
+  agentId: string,
+  options: OpenClawStateDatabaseOptions = {},
+): void {
+  try {
+    runOpenClawStateWriteTransaction(({ db }) => {
+      const kysely = getSessionStateKysely(db);
+      executeSqliteQuerySync(
+        db,
+        kysely
+          .deleteFrom("session_state_events")
+          .where("session_key", "=", sessionKey)
+          .where("agent_id", "=", agentId),
+      );
+      executeSqliteQuerySync(
+        db,
+        kysely
+          .deleteFrom("session_state_heads")
+          .where("session_key", "=", sessionKey)
+          .where("agent_id", "=", agentId),
+      );
+      executeSqliteQuerySync(
+        db,
+        kysely
+          .deleteFrom("session_watch_cursors")
+          .where((eb) =>
+            eb.or([
+              eb("watcher_session_key", "=", sessionKey),
+              eb("target_session_key", "=", sessionKey),
+            ]),
+          ),
+      );
+    }, options);
+  } catch (error) {
+    log.warn(`failed to delete session state history: ${String(error)}`);
+  }
+}
+
+function sessionExists(sessionKey: string, env?: NodeJS.ProcessEnv): boolean {
+  try {
+    return Boolean(loadSessionEntry({ sessionKey, clone: false, env }));
+  } catch {
+    return false;
+  }
+}
+
+/** Re-materialize pending notices after the in-memory queue is lost on restart. */
+export function sweepSessionStateWatchNotices(
+  options: OpenClawStateDatabaseOptions & { now?: number } = {},
+): void {
+  const now = options.now ?? Date.now();
+  try {
+    const { db } = openOpenClawStateDatabase(options);
+    const pendingRows = executeSqliteQuerySync(
+      db,
+      getSessionStateKysely(db)
+        .selectFrom("session_watch_cursors")
+        .selectAll()
+        .whereRef("material_sequence", ">", "last_seen_sequence"),
+    ).rows.filter((row) => sessionExists(row.watcher_session_key, options.env));
+    runOpenClawStateWriteTransaction(({ db: writeDb }) => {
+      for (const row of pendingRows) {
+        executeSqliteQuerySync(
+          writeDb,
+          getSessionStateKysely(writeDb)
+            .updateTable("session_watch_cursors")
+            .set({ notified_sequence: row.material_sequence, updated_at: now })
+            .where("watcher_session_key", "=", row.watcher_session_key)
+            .where("target_session_key", "=", row.target_session_key),
+        );
+      }
+    }, options);
+    for (const row of pendingRows) {
+      enqueueSessionStateNotice({
+        watcherSessionKey: row.watcher_session_key,
+        targetSessionKey: row.target_session_key,
+        lastSeenSequence: normalizeSqliteNumber(row.last_seen_sequence) ?? 0,
+      });
+    }
+    pruneSessionStateEvents({ ...options, now });
+  } catch (error) {
+    log.warn(`failed to sweep session state notices: ${String(error)}`);
+  }
+}
+
+/** Enforce bounded retained history without regressing durable per-session heads. */
+export function pruneSessionStateEvents(
+  options: OpenClawStateDatabaseOptions & { now?: number } = {},
+): void {
+  const now = options.now ?? Date.now();
+  try {
+    runOpenClawStateWriteTransaction(({ db }) => {
+      const kysely = getSessionStateKysely(db);
+      // Stamp per-session pruned watermarks BEFORE deleting: historyGap can only be
+      // answered from what pruning actually removed for that session, never inferred
+      // from globally sparse sequence arithmetic.
+      const stampPrunedWatermarks = (predicate: {
+        occurredBefore?: number;
+        sequenceAtOrBelow?: number;
+      }) => {
+        let query = kysely
+          .selectFrom("session_state_events")
+          .select(["session_key", "agent_id"])
+          .select((eb) => eb.fn.max<number>("sequence").as("max_sequence"))
+          .groupBy(["session_key", "agent_id"]);
+        if (predicate.occurredBefore !== undefined) {
+          query = query.where("occurred_at", "<", predicate.occurredBefore);
+        }
+        if (predicate.sequenceAtOrBelow !== undefined) {
+          query = query.where("sequence", "<=", predicate.sequenceAtOrBelow);
+        }
+        for (const row of executeSqliteQuerySync(db, query).rows) {
+          const maxSequence = normalizeSqliteNumber(row.max_sequence) ?? 0;
+          executeSqliteQuerySync(
+            db,
+            kysely
+              .updateTable("session_state_heads")
+              .set({ pruned_max_sequence: maxSequence, updated_at: now })
+              .where("session_key", "=", row.session_key)
+              .where("agent_id", "=", row.agent_id)
+              .where("pruned_max_sequence", "<", maxSequence),
+          );
+        }
+      };
+      const retentionCutoff = now - SESSION_STATE_RETENTION_MS;
+      stampPrunedWatermarks({ occurredBefore: retentionCutoff });
+      executeSqliteQuerySync(
+        db,
+        kysely.deleteFrom("session_state_events").where("occurred_at", "<", retentionCutoff),
+      );
+      const overflowRow = executeSqliteQueryTakeFirstSync(
+        db,
+        kysely
+          .selectFrom("session_state_events")
+          .select("sequence")
+          .orderBy("sequence", "desc")
+          .offset(SESSION_STATE_MAX_ROWS)
+          .limit(1),
+      );
+      const sequenceCutoff = normalizeOptionalSqliteNumber(overflowRow?.sequence);
+      if (sequenceCutoff !== undefined) {
+        stampPrunedWatermarks({ sequenceAtOrBelow: sequenceCutoff });
+        executeSqliteQuerySync(
+          db,
+          kysely.deleteFrom("session_state_events").where("sequence", "<=", sequenceCutoff),
+        );
+      }
+      executeSqliteQuerySync(
+        db,
+        kysely
+          .deleteFrom("session_watch_cursors")
+          .where("updated_at", "<", now - SESSION_STATE_RETENTION_MS),
+      );
+    }, options);
+    lastPruneAt = now;
+  } catch (error) {
+    log.warn(`failed to prune session state history: ${String(error)}`);
+  }
+}
+
+/** Record one successful compaction from the two concrete v1 owners. */
+export function recordSessionCompacted(params: {
+  sessionKey?: string;
+  operationId: string;
+  sessionId?: string;
+  agentId?: string;
+  runId?: string;
+}): void {
+  if (!params.sessionKey) {
+    return;
+  }
+  // Native-harness-only compaction remains log-incomplete in v1; this signal is reconciliation aid.
+  recordSessionStateEvent({
+    sessionKey: params.sessionKey,
+    sessionId: params.sessionId,
+    agentId: params.agentId ?? resolveAgentIdFromSessionKey(params.sessionKey),
+    kind: "compacted",
+    actorType: "system",
+    runId: params.runId,
+    dedupeKey: `compacted:${params.operationId}`,
+    summary: "session compacted",
+  });
+}
+
+/** Record a persisted goal mutation using lineage already available at the session-store seam. */
+export function recordSessionGoalChanged(params: {
+  sessionKey: string;
+  entry: SessionEntry;
+  actor?: { type: SessionStateActorType; id?: string };
+  agentId?: string;
+  summary: string;
+}): void {
+  const watcherSessionKey = params.entry.spawnedBy ?? params.entry.parentSessionKey;
+  // Callers that own an explicit store agent must pass it: bare "global" keys
+  // parse to the default agent and would misattribute the event.
+  recordSessionStateEvent({
+    sessionKey: params.sessionKey,
+    sessionId: params.entry.sessionId,
+    agentId: params.agentId ?? resolveAgentIdFromSessionKey(params.sessionKey),
+    kind: "goal_changed",
+    actorType: params.actor?.type ?? "system",
+    ...(params.actor?.id ? { actorId: params.actor.id } : {}),
+    summary: params.summary,
+    ...(watcherSessionKey ? { watcherSessionKeys: [watcherSessionKey] } : {}),
+  });
+}
+
+/** Record a direct human turn only when the target has an implicit parent watcher. */
+export function recordSessionHumanDirectMessage(params: {
+  sessionKey: string;
+  entry?: SessionEntry;
+  agentId?: string;
+  actor: { actorType: SessionStateActorType; actorId?: string };
+  channel?: string;
+  runId?: string;
+}): void {
+  const watcherSessionKey = params.entry?.spawnedBy ?? params.entry?.parentSessionKey;
+  if (params.actor.actorType !== "human" || !watcherSessionKey) {
+    return;
+  }
+  recordSessionStateEvent({
+    sessionKey: params.sessionKey,
+    sessionId: params.entry?.sessionId,
+    agentId: params.agentId ?? resolveAgentIdFromSessionKey(params.sessionKey),
+    kind: "human_direct_message",
+    actorType: "human",
+    ...(params.actor.actorId ? { actorId: params.actor.actorId } : {}),
+    runId: params.runId,
+    summary: `human message via ${params.channel?.trim() || "unknown"}`,
+    watcherSessionKeys: [watcherSessionKey],
+  });
+}
+
+/** Seed the parent cursor at the child-spawn version. */
+export function recordSubagentSpawned(params: {
+  childSessionKey: string;
+  childRunId: string;
+  requesterSessionKey: string;
+  agentId: string;
+}): void {
+  recordSessionStateEvent({
+    sessionKey: params.childSessionKey,
+    agentId: params.agentId,
+    kind: "child_spawned",
+    actorType: "agent",
+    actorId: params.requesterSessionKey,
+    runId: params.childRunId,
+    dedupeKey: `child-spawned:${params.childRunId}`,
+    summary: "child session spawned",
+    watcherSessionKeys: [params.requesterSessionKey],
+  });
+}
+
+/** Project an already-normalized subagent terminal outcome into the signal log. */
+export function recordSubagentTerminalState(params: {
+  childSessionKey: string;
+  runId: string;
+  requesterSessionKey: string;
+  outcomeStatus: "ok" | "error" | "timeout";
+}): void {
+  recordSessionStateEvent({
+    sessionKey: params.childSessionKey,
+    agentId: resolveAgentIdFromSessionKey(params.childSessionKey),
+    kind: params.outcomeStatus === "ok" ? "run_completed" : "run_failed",
+    actorType: "system",
+    runId: params.runId,
+    dedupeKey: `run-terminal:${params.runId}`,
+    summary: params.outcomeStatus === "ok" ? "child run completed" : "child run failed",
+    watcherSessionKeys: [params.requesterSessionKey],
+  });
+}
+
+export const sessionStateEventStoreLimits = {
+  maxRows: SESSION_STATE_MAX_ROWS,
+  retentionMs: SESSION_STATE_RETENTION_MS,
+} as const;

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

@@ -801,6 +801,38 @@ export interface SessionGroups {
   position: number;
 }
 
+export interface SessionStateEvents {
+  actor_id: string | null;
+  actor_type: string;
+  agent_id: string;
+  dedupe_key: string | null;
+  kind: string;
+  occurred_at: number;
+  payload_json: string | null;
+  run_id: string | null;
+  sequence: Generated<number>;
+  session_id: string | null;
+  session_key: string;
+  summary: string;
+}
+
+export interface SessionStateHeads {
+  agent_id: string;
+  last_sequence: number;
+  pruned_max_sequence: Generated<number>;
+  session_key: string;
+  updated_at: number;
+}
+
+export interface SessionWatchCursors {
+  last_seen_sequence: Generated<number>;
+  material_sequence: Generated<number>;
+  notified_sequence: Generated<number>;
+  target_session_key: string;
+  updated_at: number;
+  watcher_session_key: string;
+}
+
 export interface SkillCuratorState {
   id: Generated<number>;
   last_attempt_at_ms: number;
@@ -1118,6 +1150,9 @@ export interface DB {
   sandbox_registry_entries: SandboxRegistryEntries;
   schema_meta: SchemaMeta;
   session_groups: SessionGroups;
+  session_state_events: SessionStateEvents;
+  session_state_heads: SessionStateHeads;
+  session_watch_cursors: SessionWatchCursors;
   skill_curator_state: SkillCuratorState;
   skill_lifecycle: SkillLifecycle;
   skill_uploads: SkillUploads;

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

@@ -103,6 +103,53 @@ CREATE INDEX IF NOT EXISTS idx_audit_events_kind_sequence
 CREATE INDEX IF NOT EXISTS idx_audit_events_status_sequence
   ON audit_events(status, sequence DESC);
 
+CREATE TABLE IF NOT EXISTS session_state_events (
+  sequence INTEGER PRIMARY KEY AUTOINCREMENT,
+  dedupe_key TEXT UNIQUE,
+  session_key TEXT NOT NULL,
+  session_id TEXT,
+  agent_id TEXT NOT NULL,
+  kind TEXT NOT NULL,
+  actor_type TEXT NOT NULL,
+  actor_id TEXT,
+  run_id TEXT,
+  occurred_at INTEGER NOT NULL,
+  summary TEXT NOT NULL,
+  payload_json TEXT
+);
+
+CREATE INDEX IF NOT EXISTS idx_session_state_events_session_sequence
+  ON session_state_events(session_key, sequence DESC);
+
+CREATE INDEX IF NOT EXISTS idx_session_state_events_time
+  ON session_state_events(occurred_at DESC, sequence DESC);
+
+CREATE TABLE IF NOT EXISTS session_state_heads (
+  session_key TEXT NOT NULL,
+  agent_id TEXT NOT NULL,
+  last_sequence INTEGER NOT NULL,
+  pruned_max_sequence INTEGER NOT NULL DEFAULT 0,
+  updated_at INTEGER NOT NULL,
+  PRIMARY KEY (session_key, agent_id)
+);
+
+-- Watcher identity is the bare session key, matching the process-local system-event
+-- queue it feeds. Producers only create rows for agent-qualified watcher keys;
+-- bare keys (session.scope="global") are ambiguous across agents and are excluded
+-- from the notice protocol 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,
+  updated_at INTEGER NOT NULL,
+  PRIMARY KEY (watcher_session_key, target_session_key)
+);
+
+CREATE INDEX IF NOT EXISTS idx_session_watch_cursors_target
+  ON session_watch_cursors(target_session_key);
+
 CREATE TABLE IF NOT EXISTS diagnostic_stability_bundles (
   bundle_key TEXT NOT NULL PRIMARY KEY,
   reason TEXT NOT NULL,

src/state/openclaw-state-schema.sql

@@ -98,6 +98,53 @@ CREATE INDEX IF NOT EXISTS idx_audit_events_kind_sequence
 CREATE INDEX IF NOT EXISTS idx_audit_events_status_sequence
   ON audit_events(status, sequence DESC);
 
+CREATE TABLE IF NOT EXISTS session_state_events (
+  sequence INTEGER PRIMARY KEY AUTOINCREMENT,
+  dedupe_key TEXT UNIQUE,
+  session_key TEXT NOT NULL,
+  session_id TEXT,
+  agent_id TEXT NOT NULL,
+  kind TEXT NOT NULL,
+  actor_type TEXT NOT NULL,
+  actor_id TEXT,
+  run_id TEXT,
+  occurred_at INTEGER NOT NULL,
+  summary TEXT NOT NULL,
+  payload_json TEXT
+);
+
+CREATE INDEX IF NOT EXISTS idx_session_state_events_session_sequence
+  ON session_state_events(session_key, sequence DESC);
+
+CREATE INDEX IF NOT EXISTS idx_session_state_events_time
+  ON session_state_events(occurred_at DESC, sequence DESC);
+
+CREATE TABLE IF NOT EXISTS session_state_heads (
+  session_key TEXT NOT NULL,
+  agent_id TEXT NOT NULL,
+  last_sequence INTEGER NOT NULL,
+  pruned_max_sequence INTEGER NOT NULL DEFAULT 0,
+  updated_at INTEGER NOT NULL,
+  PRIMARY KEY (session_key, agent_id)
+);
+
+-- Watcher identity is the bare session key, matching the process-local system-event
+-- queue it feeds. Producers only create rows for agent-qualified watcher keys;
+-- bare keys (session.scope="global") are ambiguous across agents and are excluded
+-- from the notice protocol 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,
+  updated_at INTEGER NOT NULL,
+  PRIMARY KEY (watcher_session_key, target_session_key)
+);
+
+CREATE INDEX IF NOT EXISTS idx_session_watch_cursors_target
+  ON session_watch_cursors(target_session_key);
+
 CREATE TABLE IF NOT EXISTS diagnostic_stability_bundles (
   bundle_key TEXT NOT NULL PRIMARY KEY,
   reason TEXT NOT NULL,

src/tui/embedded-backend.test.ts

@@ -837,6 +837,7 @@ describe("EmbeddedTuiBackend", () => {
       sessionKey: "agent:main:main",
       storePath: "/tmp/openclaw-sessions.json",
       objective: "Ship Goal",
+      actor: { type: "human" },
       fallbackEntry: {
         sessionId: expect.any(String),
         updatedAt: expect.any(Number),

src/tui/embedded-backend.ts

@@ -815,6 +815,8 @@ export class EmbeddedTuiBackend implements TuiBackend {
           storePath,
           objective,
           fallbackEntry,
+          actor: { type: "human" },
+          agentId: opts.agentId,
         });
         return { text: `Goal started: ${goal.objective}` };
       }
@@ -823,7 +825,13 @@ export class EmbeddedTuiBackend implements TuiBackend {
         if (!objective) {
           return { text: "Usage: /goal edit <objective>" };
         }
-        const goal = await updateSessionGoalObjective({ sessionKey, storePath, objective });
+        const goal = await updateSessionGoalObjective({
+          sessionKey,
+          storePath,
+          objective,
+          actor: { type: "human" },
+          agentId: opts.agentId,
+        });
         return { text: `Goal updated: ${goal.objective}` };
       }
       case "pause": {
@@ -831,6 +839,8 @@ export class EmbeddedTuiBackend implements TuiBackend {
           sessionKey,
           storePath,
           status: "paused",
+          actor: { type: "human" },
+          agentId: opts.agentId,
           ...(parsed.text ? { note: parsed.text } : {}),
         });
         return { text: `Goal paused: ${goal.objective}` };
@@ -840,6 +850,8 @@ export class EmbeddedTuiBackend implements TuiBackend {
           sessionKey,
           storePath,
           status: "active",
+          actor: { type: "human" },
+          agentId: opts.agentId,
           ...(parsed.text ? { note: parsed.text } : {}),
         });
         return { text: `Goal resumed: ${goal.objective}` };
@@ -850,6 +862,8 @@ export class EmbeddedTuiBackend implements TuiBackend {
           sessionKey,
           storePath,
           status: "complete",
+          actor: { type: "human" },
+          agentId: opts.agentId,
           ...(parsed.text ? { note: parsed.text } : {}),
         });
         return { text: `Goal complete: ${goal.objective}\nTokens used: ${goal.tokensUsed}` };
@@ -860,12 +874,19 @@ export class EmbeddedTuiBackend implements TuiBackend {
           sessionKey,
           storePath,
           status: "blocked",
+          actor: { type: "human" },
+          agentId: opts.agentId,
           ...(parsed.text ? { note: parsed.text } : {}),
         });
         return { text: `Goal blocked: ${goal.objective}` };
       }
       case "clear": {
-        const removed = await clearSessionGoal({ sessionKey, storePath });
+        const removed = await clearSessionGoal({
+          sessionKey,
+          storePath,
+          actor: { type: "human" },
+          agentId: opts.agentId,
+        });
         return { text: removed ? "Goal cleared." : "No goal to clear." };
       }
       default:

test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.discord-group.json

@@ -1317,6 +1317,10 @@
         "description": "Show /status-like card for current/visible session: model, usage, time, cost, tasks. Use `sessionKey=\"current\"` for current session; UI labels like `openclaw-tui` are not keys. `model` sets session override; `model=default` resets. Use for active model/session config questions.",
         "inputSchema": {
           "properties": {
+            "changesSince": {
+              "minimum": 0,
+              "type": "integer"
+            },
             "model": {
               "type": "string"
             },

test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.heartbeat-turn.json

@@ -1349,6 +1349,10 @@
         "description": "Show /status-like card for current/visible session: model, usage, time, cost, tasks. Use `sessionKey=\"current\"` for current session; UI labels like `openclaw-tui` are not keys. `model` sets session override; `model=default` resets. Use for active model/session config questions.",
         "inputSchema": {
           "properties": {
+            "changesSince": {
+              "minimum": 0,
+              "type": "integer"
+            },
             "model": {
               "type": "string"
             },

test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.telegram-direct.json

@@ -1313,6 +1313,10 @@
         "description": "Show /status-like card for current/visible session: model, usage, time, cost, tasks. Use `sessionKey=\"current\"` for current session; UI labels like `openclaw-tui` are not keys. `model` sets session override; `model=default` resets. Use for active model/session config questions.",
         "inputSchema": {
           "properties": {
+            "changesSince": {
+              "minimum": 0,
+              "type": "integer"
+            },
             "model": {
               "type": "string"
             },

test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/discord-group-codex-message-tool.md

@@ -207,8 +207,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the
     "roughTokens": 0
   },
   "dynamicToolsJson": {
-    "chars": 53645,
-    "roughTokens": 13412
+    "chars": 53750,
+    "roughTokens": 13438
   },
   "openClawDeveloperInstructions": {
     "chars": 3245,
@@ -219,8 +219,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the
     "roughTokens": 6943
   },
   "totalWithDynamicToolsJson": {
-    "chars": 81417,
-    "roughTokens": 20355
+    "chars": 81522,
+    "roughTokens": 20381
   },
   "userInputText": {
     "chars": 1442,

test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-direct-codex-message-tool.md

@@ -207,8 +207,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the
     "roughTokens": 0
   },
   "dynamicToolsJson": {
-    "chars": 53334,
-    "roughTokens": 13334
+    "chars": 53439,
+    "roughTokens": 13360
   },
   "openClawDeveloperInstructions": {
     "chars": 2136,
@@ -219,8 +219,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the
     "roughTokens": 6563
   },
   "totalWithDynamicToolsJson": {
-    "chars": 79588,
-    "roughTokens": 19897
+    "chars": 79693,
+    "roughTokens": 19924
   },
   "userInputText": {
     "chars": 1033,

test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-heartbeat-codex-tool.md

@@ -208,8 +208,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the
     "roughTokens": 0
   },
   "dynamicToolsJson": {
-    "chars": 54624,
-    "roughTokens": 13656
+    "chars": 54729,
+    "roughTokens": 13683
   },
   "openClawDeveloperInstructions": {
     "chars": 2155,
@@ -220,8 +220,8 @@ This is the deterministic model-bound layer stack OpenClaw can snapshot for the
     "roughTokens": 6799
   },
   "totalWithDynamicToolsJson": {
-    "chars": 81821,
-    "roughTokens": 20456
+    "chars": 81926,
+    "roughTokens": 20482
   },
   "userInputText": {
     "chars": 1271,