From 2bb79d10fd6510d1fe7e4574c4695ca90f01c7ac Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 11 Jul 2026 12:19:02 -0700 Subject: [PATCH] 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 | 11 + docs/docs_map.md | 1 + docs/tools/subagents.md | 1 + .../manager.cancel-session.test.ts | 1 + src/acp/control-plane/manager.core.ts | 6 +- .../control-plane/manager.failover.test.ts | 7 + .../manager.runtime-config.test.ts | 12 + .../manager.runtime-handles.test.ts | 17 + src/acp/control-plane/manager.test.ts | 71 ++ .../manager.turn-results.test.ts | 15 + src/acp/control-plane/manager.turn-runner.ts | 36 +- src/acp/control-plane/manager.types.ts | 2 + src/agents/acp-spawn.ts | 9 + src/agents/agent-command.ts | 24 + ...ded-agent-subscribe.handlers.compaction.ts | 7 + .../openclaw-tools.session-status.test.ts | 56 ++ src/agents/subagent-registry-lifecycle.ts | 12 + src/agents/subagent-spawn.ts | 11 +- src/agents/tools/goal-tools.ts | 10 +- src/agents/tools/session-status-tool.ts | 42 +- src/agents/tools/sessions-helpers.ts | 1 + src/agents/tools/sessions-list-tool.test.ts | 33 + src/agents/tools/sessions-list-tool.ts | 22 + .../reply/commands-acp/lifecycle.ts | 1 + src/auto-reply/reply/commands-goal.ts | 16 + src/auto-reply/reply/dispatch-acp.ts | 5 + src/auto-reply/reply/session-system-events.ts | 12 + src/config/sessions/goals.ts | 26 + src/gateway/server-methods/sessions.ts | 24 +- src/gateway/server-startup-post-attach.ts | 5 + src/gateway/session-reset-service.ts | 2 + .../heartbeat-runner.session-state.test.ts | 20 + src/infra/heartbeat-runner.ts | 15 +- src/infra/heartbeat-wake.ts | 1 + src/sessions/session-state-events.test.ts | 513 ++++++++++ src/sessions/session-state-events.ts | 884 ++++++++++++++++++ src/state/openclaw-state-db.generated.d.ts | 35 + src/state/openclaw-state-schema.generated.ts | 47 + src/state/openclaw-state-schema.sql | 47 + src/tui/embedded-backend.test.ts | 1 + src/tui/embedded-backend.ts | 25 +- .../codex-dynamic-tools.discord-group.json | 4 + .../codex-dynamic-tools.heartbeat-turn.json | 4 + .../codex-dynamic-tools.telegram-direct.json | 4 + .../discord-group-codex-message-tool.md | 8 +- .../telegram-direct-codex-message-tool.md | 8 +- .../telegram-heartbeat-codex-tool.md | 8 +- 47 files changed, 2092 insertions(+), 30 deletions(-) create mode 100644 src/infra/heartbeat-runner.session-state.test.ts create mode 100644 src/sessions/session-state-events.test.ts create mode 100644 src/sessions/session-state-events.ts diff --git a/docs/concepts/session-tool.md b/docs/concepts/session-tool.md index 6414deca9bfa..84899bdf6c82 100644 --- a/docs/concepts/session-tool.md +++ b/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: ` 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 diff --git a/docs/docs_map.md b/docs/docs_map.md index 560e5e94d575..02ec87f04b63 100644 --- a/docs/docs_map.md +++ b/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 diff --git a/docs/tools/subagents.md b/docs/tools/subagents.md index 0d2c5cadfced..0369196ad2f5 100644 --- a/docs/tools/subagents.md +++ b/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) diff --git a/src/acp/control-plane/manager.cancel-session.test.ts b/src/acp/control-plane/manager.cancel-session.test.ts index 416e511c1664..447d76357af5 100644 --- a/src/acp/control-plane/manager.cancel-session.test.ts +++ b/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", diff --git a/src/acp/control-plane/manager.core.ts b/src/acp/control-plane/manager.core.ts index eb1184a6c44d..3672461b59ae 100644 --- a/src/acp/control-plane/manager.core.ts +++ b/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)) { diff --git a/src/acp/control-plane/manager.failover.test.ts b/src/acp/control-plane/manager.failover.test.ts index 4b366fc5b444..5595d4d38fee 100644 --- a/src/acp/control-plane/manager.failover.test.ts +++ b/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", diff --git a/src/acp/control-plane/manager.runtime-config.test.ts b/src/acp/control-plane/manager.runtime-config.test.ts index 023630b466a8..b6100fed605f 100644 --- a/src/acp/control-plane/manager.runtime-config.test.ts +++ b/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", diff --git a/src/acp/control-plane/manager.runtime-handles.test.ts b/src/acp/control-plane/manager.runtime-handles.test.ts index 07daba509c52..7bfbe49d38f6 100644 --- a/src/acp/control-plane/manager.runtime-handles.test.ts +++ b/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", diff --git a/src/acp/control-plane/manager.test.ts b/src/acp/control-plane/manager.test.ts index e30d2807ce4e..12f399d39a99 100644 --- a/src/acp/control-plane/manager.test.ts +++ b/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", diff --git a/src/acp/control-plane/manager.turn-results.test.ts b/src/acp/control-plane/manager.turn-results.test.ts index 6088ccc11f32..f64f2004480b 100644 --- a/src/acp/control-plane/manager.turn-results.test.ts +++ b/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", diff --git a/src/acp/control-plane/manager.turn-runner.ts b/src/acp/control-plane/manager.turn-runner.ts index 6c7f37ccafa5..d8bc9f216d1b 100644 --- a/src/acp/control-plane/manager.turn-runner.ts +++ b/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, diff --git a/src/acp/control-plane/manager.types.ts b/src/acp/control-plane/manager.types.ts index 13d3fca3cd07..da9c35f43d13 100644 --- a/src/acp/control-plane/manager.types.ts +++ b/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; diff --git a/src/agents/acp-spawn.ts b/src/agents/acp-spawn.ts index d463046cd1b4..3be1e9a18733 100644 --- a/src/agents/acp-spawn.ts +++ b/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({ diff --git a/src/agents/agent-command.ts b/src/agents/agent-command.ts index 08da1c15a207..4b294db2f595 100644 --- a/src/agents/agent-command.ts +++ b/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, diff --git a/src/agents/embedded-agent-subscribe.handlers.compaction.ts b/src/agents/embedded-agent-subscribe.handlers.compaction.ts index ac3f735ca8c0..d082a36f729c 100644 --- a/src/agents/embedded-agent-subscribe.handlers.compaction.ts +++ b/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, diff --git a/src/agents/openclaw-tools.session-status.test.ts b/src/agents/openclaw-tools.session-status.test.ts index 30b8b889c499..84726a295dfe 100644 --- a/src/agents/openclaw-tools.session-status.test.ts +++ b/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>, + 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) { 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 }; @@ -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; + 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: { diff --git a/src/agents/subagent-registry-lifecycle.ts b/src/agents/subagent-registry-lifecycle.ts index 957164db5c16..b0a0bd76fab5 100644 --- a/src/agents/subagent-registry-lifecycle.ts +++ b/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 { diff --git a/src/agents/subagent-spawn.ts b/src/agents/subagent-spawn.ts index 2ce1c4cda0fa..4751ba837384 100644 --- a/src/agents/subagent-spawn.ts +++ b/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 diff --git a/src/agents/tools/goal-tools.ts b/src/agents/tools/goal-tools.ts index b7b31d3a2fde..09815e532449 100644 --- a/src/agents/tools/goal-tools.ts +++ b/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 } : {}), }); diff --git a/src/agents/tools/session-status-tool.ts b/src/agents/tools/session-status-tool.ts index 78bf02d15471..18056aeddc4a 100644 --- a/src/agents/tools/session-status-tool.ts +++ b/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; +}): 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; + 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, diff --git a/src/agents/tools/sessions-helpers.ts b/src/agents/tools/sessions-helpers.ts index 63085edb8fe6..4784e52977a5 100644 --- a/src/agents/tools/sessions-helpers.ts +++ b/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; diff --git a/src/agents/tools/sessions-list-tool.test.ts b/src/agents/tools/sessions-list-tool.test.ts index 5dfd675ceffc..6cc8f46fdbea 100644 --- a/src/agents/tools/sessions-list-tool.test.ts +++ b/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>, + ), })); 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(); 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 () => { diff --git a/src/agents/tools/sessions-list-tool.ts b/src/agents/tools/sessions-list-tool.ts index 7568bb678150..90db0f0f3d11 100644 --- a/src/agents/tools/sessions-list-tool.ts +++ b/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, diff --git a/src/auto-reply/reply/commands-acp/lifecycle.ts b/src/auto-reply/reply/commands-acp/lifecycle.ts index 5e3ac82ef31f..4588c8310dde 100644 --- a/src/auto-reply/reply/commands-acp/lifecycle.ts +++ b/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, diff --git a/src/auto-reply/reply/commands-goal.ts b/src/auto-reply/reply/commands-goal.ts index e1b45c31c203..d42dcea6960b 100644 --- a/src/auto-reply/reply/commands-goal.ts +++ b/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) { diff --git a/src/auto-reply/reply/dispatch-acp.ts b/src/auto-reply/reply/dispatch-acp.ts index 92d821b3506c..89dd732c38cf 100644 --- a/src/auto-reply/reply/dispatch-acp.ts +++ b/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, diff --git a/src/auto-reply/reply/session-system-events.ts b/src/auto-reply/reply/session-system-events.ts index ac5ea9a700a4..bbd8c1320562 100644 --- a/src/auto-reply/reply/session-system-events.ts +++ b/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) { diff --git a/src/config/sessions/goals.ts b/src/config/sessions/goals.ts index b71ac0c8ecb0..f3ab6a09dc1f 100644 --- a/src/config/sessions/goals.ts +++ b/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, 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); } diff --git a/src/gateway/server-methods/sessions.ts b/src/gateway/server-methods/sessions.ts index 8e99b7242ce7..d26996fddf38 100644 --- a/src/gateway/server-methods/sessions.ts +++ b/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); diff --git a/src/gateway/server-startup-post-attach.ts b/src/gateway/server-startup-post-attach.ts index 5bbfa3a6c4a7..bb37deefd4f3 100644 --- a/src/gateway/server-startup-post-attach.ts +++ b/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; } diff --git a/src/gateway/session-reset-service.ts b/src/gateway/session-reset-service.ts index e24da74dd66a..abf3f575de7e 100644 --- a/src/gateway/session-reset-service.ts +++ b/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 = { diff --git a/src/infra/heartbeat-runner.session-state.test.ts b/src/infra/heartbeat-runner.session-state.test.ts new file mode 100644 index 000000000000..ceb6c41f8519 --- /dev/null +++ b/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 }); + }); +}); diff --git a/src/infra/heartbeat-runner.ts b/src/infra/heartbeat-runner.ts index 89e370ab1a05..e74173020552 100644 --- a/src/infra/heartbeat-runner.ts +++ b/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; diff --git a/src/infra/heartbeat-wake.ts b/src/infra/heartbeat-wake.ts index a6c8e29b14e2..449240903a06 100644 --- a/src/infra/heartbeat-wake.ts +++ b/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" diff --git a/src/sessions/session-state-events.test.ts b/src/sessions/session-state-events.test.ts new file mode 100644 index 000000000000..71ffa27a8f47 --- /dev/null +++ b/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[0]> = {}, +): Parameters[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, + 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, + watcherSessionKey = watcher, +) { + return recordSessionStateEvent( + eventInput({ + kind: "child_spawned", + actorType: "agent", + actorId: watcherSessionKey, + dedupeKey: `child-spawned:${watcherSessionKey}`, + watcherSessionKeys: [watcherSessionKey], + }), + database, + ); +} + +async function createWatcherSession( + database: ReturnType, + 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"]); + }); +}); diff --git a/src/sessions/session-state-events.ts b/src/sessions/session-state-events.ts new file mode 100644 index 000000000000..b9a50930ad75 --- /dev/null +++ b/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; + 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; +}; + +type SessionStateDatabase = Pick< + OpenClawStateKyselyDatabase, + "session_state_events" | "session_state_heads" | "session_watch_cursors" +>; +type SessionStateEventsTable = OpenClawStateKyselyDatabase["session_state_events"]; +type SessionStateEventRow = Selectable; +type SessionWatchCursorRow = Selectable; + +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 = { + human_direct_message: true, + goal_changed: true, + run_completed: false, + run_failed: false, + child_spawned: false, + compacted: false, +}; + +function getSessionStateKysely(db: DatabaseSync) { + return getNodeSqliteKysely(db); +} + +function normalizeOptionalSqliteNumber( + value: number | bigint | null | undefined, +): number | undefined { + return value === undefined ? undefined : normalizeSqliteNumber(value); +} + +function parsePayload(value: string | null): Record | undefined { + if (!value) { + return undefined; + } + try { + const parsed: unknown = JSON.parse(value); + return parsed && typeof parsed === "object" && !Array.isArray(parsed) + ? (parsed as Record) + : 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 { + 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> { + const keys = [...new Set(refs.map((ref) => ref.sessionKey).filter(Boolean))]; + if (keys.length === 0) { + return {}; + } + const byAgent: Record> = {}; + 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("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("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; diff --git a/src/state/openclaw-state-db.generated.d.ts b/src/state/openclaw-state-db.generated.d.ts index 365349cfb30a..7a5b97a2c7be 100644 --- a/src/state/openclaw-state-db.generated.d.ts +++ b/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; + session_id: string | null; + session_key: string; + summary: string; +} + +export interface SessionStateHeads { + agent_id: string; + last_sequence: number; + pruned_max_sequence: Generated; + session_key: string; + updated_at: number; +} + +export interface SessionWatchCursors { + last_seen_sequence: Generated; + material_sequence: Generated; + notified_sequence: Generated; + target_session_key: string; + updated_at: number; + watcher_session_key: string; +} + export interface SkillCuratorState { id: Generated; 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; diff --git a/src/state/openclaw-state-schema.generated.ts b/src/state/openclaw-state-schema.generated.ts index fc212c3a0f06..886a9c007153 100644 --- a/src/state/openclaw-state-schema.generated.ts +++ b/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, diff --git a/src/state/openclaw-state-schema.sql b/src/state/openclaw-state-schema.sql index 6a2ab256197a..61c85eb4a58e 100644 --- a/src/state/openclaw-state-schema.sql +++ b/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, diff --git a/src/tui/embedded-backend.test.ts b/src/tui/embedded-backend.test.ts index 5bf614df5e6f..d2b4c99583e8 100644 --- a/src/tui/embedded-backend.test.ts +++ b/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), diff --git a/src/tui/embedded-backend.ts b/src/tui/embedded-backend.ts index d8365f0e1590..b0a16dbddbb0 100644 --- a/src/tui/embedded-backend.ts +++ b/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 " }; } - 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: diff --git a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.discord-group.json b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.discord-group.json index 00525f5ff7f6..8a36df4607ed 100644 --- a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.discord-group.json +++ b/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" }, diff --git a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.heartbeat-turn.json b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.heartbeat-turn.json index 401fbbad9b3d..d224edeb7672 100644 --- a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.heartbeat-turn.json +++ b/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" }, diff --git a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.telegram-direct.json b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.telegram-direct.json index c0e0be26714d..6c058c540cb3 100644 --- a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/codex-dynamic-tools.telegram-direct.json +++ b/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" }, diff --git a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/discord-group-codex-message-tool.md b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/discord-group-codex-message-tool.md index 37b06fcb3f45..bf108054e598 100644 --- a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/discord-group-codex-message-tool.md +++ b/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, diff --git a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-direct-codex-message-tool.md b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-direct-codex-message-tool.md index 83456e2201c7..6b80c3f168c6 100644 --- a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-direct-codex-message-tool.md +++ b/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, diff --git a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-heartbeat-codex-tool.md b/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-heartbeat-codex-tool.md index 87620d3e964f..6600222afb81 100644 --- a/test/fixtures/agents/prompt-snapshots/codex-runtime-happy-path/telegram-heartbeat-codex-tool.md +++ b/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,