diff --git a/src/agents/run-session-target.test.ts b/src/agents/run-session-target.test.ts index d308e313d137..e7665a8cfb56 100644 --- a/src/agents/run-session-target.test.ts +++ b/src/agents/run-session-target.test.ts @@ -301,7 +301,7 @@ describe("agent run session target", () => { }); }); - it("recovers the stored key from a legacy SQLite marker", async () => { + it("recovers the persisted owner from a legacy SQLite marker", async () => { const storePath = path.join(tempDir, "legacy", "sessions.json"); const sessionKey = "agent:main:dashboard:legacy-session"; await upsertSessionEntryCore( @@ -334,7 +334,7 @@ describe("agent run session target", () => { storePath, }), }), - ).resolves.toMatchObject({ sessionKey, storePath }); + ).resolves.toMatchObject({ sessionKey: "agent:main:legacy-session", storePath }); }); it("uses the marker store for a compatible partial typed target", async () => { diff --git a/src/channels/feedback-reflection.test.ts b/src/channels/feedback-reflection.test.ts index 5b1ccc255a0f..d2f89f2c405e 100644 --- a/src/channels/feedback-reflection.test.ts +++ b/src/channels/feedback-reflection.test.ts @@ -6,6 +6,7 @@ const appendTranscriptEvent = vi.hoisted(() => vi.fn(async () => undefined)); const dispatchRoutedChannelTurn = vi.hoisted(() => vi.fn()); const loadSessionEntry = vi.hoisted(() => vi.fn()); const readSessionUpdatedAtCore = vi.hoisted(() => vi.fn()); +const resolveSessionTranscriptRuntimeTarget = vi.hoisted(() => vi.fn()); const resolveStorePath = vi.hoisted(() => vi.fn(() => "/state/main/sessions.json")); vi.mock("../config/sessions/paths.js", () => ({ @@ -16,6 +17,7 @@ vi.mock("../config/sessions/session-accessor.js", () => ({ loadSessionEntry, loadSessionEntryReadOnly: loadSessionEntry, readSessionUpdatedAtCore, + resolveSessionTranscriptRuntimeTarget, })); vi.mock("./turn/lifecycle.js", () => ({ dispatchRoutedChannelTurn })); @@ -114,8 +116,14 @@ describe("channel feedback reflection", () => { ).resolves.toMatchObject({ status: "complete", followUp: false }); }); - it("records feedback through the canonical transcript accessor", async () => { + it("records feedback through the persisted transcript owner", async () => { loadSessionEntry.mockReturnValue({ sessionId: "session-1" }); + resolveSessionTranscriptRuntimeTarget.mockResolvedValue({ + agentId: "main", + sessionId: "session-1", + sessionKey: "agent:main:main", + storePath: "/state/main/sessions.json", + }); const event = { type: "custom", event: "feedback", ts: 1 }; await expect( @@ -130,7 +138,7 @@ describe("channel feedback reflection", () => { { agentId: "main", sessionId: "session-1", - sessionKey: "agent:main:msteams:feedback-2", + sessionKey: "agent:main:main", storePath: "/state/main/sessions.json", }, event, diff --git a/src/channels/feedback-reflection.ts b/src/channels/feedback-reflection.ts index 85a2d7460c77..443622ab0d5b 100644 --- a/src/channels/feedback-reflection.ts +++ b/src/channels/feedback-reflection.ts @@ -3,6 +3,7 @@ import { resolveSessionStorePathCore } from "../config/sessions/paths.js"; import { appendTranscriptEvent, loadSessionEntryReadOnly, + resolveSessionTranscriptRuntimeTarget, } from "../config/sessions/session-accessor.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; import { buildChannelInboundEventContext } from "./inbound-event/context.js"; @@ -31,15 +32,13 @@ export async function recordChannelFeedbackEvent(params: { if (!entry?.sessionId) { return false; } - await appendTranscriptEvent( - { - agentId: params.agentId, - sessionId: entry.sessionId, - sessionKey: params.sessionKey, - storePath, - }, - params.event, - ); + const target = await resolveSessionTranscriptRuntimeTarget({ + agentId: params.agentId, + sessionId: entry.sessionId, + sessionKey: params.sessionKey, + storePath, + }); + await appendTranscriptEvent(target, params.event); return true; } diff --git a/src/commands/doctor-session-canonical-keys.owner-repair.test.ts b/src/commands/doctor-session-canonical-keys.owner-repair.test.ts new file mode 100644 index 000000000000..f51aec1b55ec --- /dev/null +++ b/src/commands/doctor-session-canonical-keys.owner-repair.test.ts @@ -0,0 +1,182 @@ +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { resolveSessionStorePathCore } from "../config/sessions/paths.js"; +import { + loadExactSessionEntryReadOnly, + loadTranscriptEvents, +} from "../config/sessions/session-accessor.js"; +import { resolveSqliteTargetFromSessionStorePath } from "../config/sessions/session-sqlite-target.js"; +import type { OpenClawConfig } from "../config/types.openclaw.js"; +import { + closeOpenClawAgentDatabasesForTest, + openOpenClawAgentDatabase, +} from "../state/openclaw-agent-db.js"; +import { withStateDirEnv } from "../test-helpers/state-dir-env.js"; +import { repairCanonicalSessionKeys } from "./doctor-session-canonical-keys.js"; +import { insertLegacySession } from "./doctor-session-canonical-keys.test-support.js"; + +afterEach(() => closeOpenClawAgentDatabasesForTest()); + +function insertEmptyAlias(params: { + agentId: string; + env: NodeJS.ProcessEnv; + sessionId: string; + sessionKey: string; + storePath: string; + updatedAt: number; +}) { + const database = openOpenClawAgentDatabase({ + agentId: params.agentId, + env: params.env, + path: resolveSqliteTargetFromSessionStorePath(params.storePath, { + agentId: params.agentId, + env: params.env, + }).path, + }); + database.db + .prepare( + "INSERT INTO session_nodes (session_key, current_session_id, entry_json, updated_at) VALUES (?, ?, '{}', ?)", + ) + .run(params.sessionKey, params.sessionId, params.updatedAt); + return database; +} + +describe("doctor transcript owner repair", () => { + it("restores a valid node after an empty alias steals its transcript window", async () => { + await withStateDirEnv("openclaw-doctor-transcript-owner-", async ({ stateDir }) => { + const env = { ...process.env, OPENCLAW_STATE_DIR: stateDir }; + const storeTemplate = path.join(stateDir, "agents", "{agentId}", "sessions.json"); + const storePath = resolveSessionStorePathCore(storeTemplate, { agentId: "main", env }); + const cfg = { + agents: { list: [{ id: "main", default: true }] }, + session: { store: storeTemplate }, + } as OpenClawConfig; + const canonicalKey = "agent:main:main"; + const staleKey = "agent:main:telegram:default:direct:fixture-peer"; + const sessionId = "stolen-owner-session"; + insertLegacySession({ + agentId: "main", + entry: { label: "canonical metadata", sessionId, updatedAt: 20 }, + env, + eventText: "preserved history", + sessionKey: canonicalKey, + storePath, + }); + const database = insertEmptyAlias({ + agentId: "main", + env, + sessionId, + sessionKey: staleKey, + storePath, + updatedAt: 30, + }); + database.db + .prepare("UPDATE session_nodes SET entry_valid = 1 WHERE session_key = ?") + .run(canonicalKey); + database.db + .prepare("UPDATE session_windows SET session_key = ? WHERE session_id = ?") + .run(staleKey, sessionId); + + expect(await repairCanonicalSessionKeys({ apply: false, cfg, env })).toMatchObject({ + foundGroups: 1, + repairedGroups: 0, + }); + expect(await repairCanonicalSessionKeys({ apply: true, cfg, env })).toMatchObject({ + foundGroups: 1, + removedRows: 1, + repairedGroups: 1, + }); + expect( + loadExactSessionEntryReadOnly({ agentId: "main", env, sessionKey: staleKey, storePath }), + ).toBeUndefined(); + expect( + loadExactSessionEntryReadOnly({ agentId: "main", env, sessionKey: canonicalKey, storePath }) + ?.entry, + ).toMatchObject({ label: "canonical metadata", sessionId }); + expect( + database.db + .prepare("SELECT session_key FROM session_windows WHERE session_id = ?") + .get(sessionId), + ).toEqual({ session_key: canonicalKey }); + await expect( + loadTranscriptEvents({ + agentId: "main", + env, + sessionId, + sessionKey: canonicalKey, + storePath, + }), + ).resolves.toEqual([ + expect.objectContaining({ + message: expect.objectContaining({ content: "preserved history" }), + }), + ]); + expect(await repairCanonicalSessionKeys({ apply: true, cfg, env })).toMatchObject({ + foundGroups: 0, + repairedGroups: 0, + }); + }); + }); + + it("follows alias ownership transitively to the configured canonical key", async () => { + await withStateDirEnv("openclaw-doctor-transcript-owner-chain-", async ({ stateDir }) => { + const env = { ...process.env, OPENCLAW_STATE_DIR: stateDir }; + const storeTemplate = path.join(stateDir, "agents", "{agentId}", "sessions.json"); + const storePath = resolveSessionStorePathCore(storeTemplate, { agentId: "main", env }); + const cfg = { + agents: { list: [{ id: "main", default: true }] }, + session: { mainKey: "work", store: storeTemplate }, + } as OpenClawConfig; + const staleKey = "agent:main:telegram:default:direct:fixture-peer"; + const intermediateKey = "agent:main:main"; + const canonicalKey = "agent:main:work"; + const sessionId = "owner-chain-session"; + insertLegacySession({ + agentId: "main", + entry: { label: "intermediate metadata", sessionId, updatedAt: 20 }, + env, + eventText: "chain history", + sessionKey: intermediateKey, + storePath, + }); + insertEmptyAlias({ + agentId: "main", + env, + sessionId, + sessionKey: staleKey, + storePath, + updatedAt: 30, + }); + + expect(await repairCanonicalSessionKeys({ apply: true, cfg, env })).toMatchObject({ + foundGroups: 1, + removedRows: 2, + repairedGroups: 1, + }); + for (const sessionKey of [staleKey, intermediateKey]) { + expect( + loadExactSessionEntryReadOnly({ agentId: "main", env, sessionKey, storePath }), + ).toBeUndefined(); + } + expect( + loadExactSessionEntryReadOnly({ agentId: "main", env, sessionKey: canonicalKey, storePath }) + ?.entry, + ).toMatchObject({ label: "intermediate metadata", sessionId }); + await expect( + loadTranscriptEvents({ + agentId: "main", + env, + sessionId, + sessionKey: canonicalKey, + storePath, + }), + ).resolves.toEqual([ + expect.objectContaining({ message: expect.objectContaining({ content: "chain history" }) }), + ]); + expect(await repairCanonicalSessionKeys({ apply: true, cfg, env })).toMatchObject({ + foundGroups: 0, + repairedGroups: 0, + }); + }); + }); +}); diff --git a/src/commands/doctor-session-canonical-keys.ts b/src/commands/doctor-session-canonical-keys.ts index 83cf1650cf3c..4ea00273c1d5 100644 --- a/src/commands/doctor-session-canonical-keys.ts +++ b/src/commands/doctor-session-canonical-keys.ts @@ -4,7 +4,6 @@ import { resolveSessionStorePathCore } from "../config/sessions/paths.js"; import { applySessionEntryLifecycleMutation, copySessionOwnedStateForCanonicalRepair, - listSessionEntriesForCanonicalRepair, listSessionGenerationIdsForCanonicalRepair, loadTranscriptEvents, rehomeSessionDeliveryReferencesForCanonicalRepair, @@ -12,6 +11,7 @@ import { type SessionEntryLifecycleRemoval, } from "../config/sessions/session-accessor.js"; import { writeTranscriptArchive } from "../config/sessions/session-accessor.sqlite-archive.js"; +import { listSqliteSessionEntriesWithCanonicalOwnerEvidence } from "../config/sessions/session-accessor.sqlite-canonical-inventory.js"; import { copySessionNodeArtifactsForRepair, deleteSessionMembersForRepair, @@ -33,6 +33,7 @@ import { openOpenClawAgentDatabase, type OpenClawAgentDatabase, } from "../state/openclaw-agent-db.js"; +import { applyCanonicalOwnerEvidence } from "./doctor-session-canonical-owner-evidence.js"; import { resolveTargetSqlitePath } from "./doctor-session-sqlite-readers.js"; type CanonicalSessionCandidate = { @@ -41,6 +42,7 @@ type CanonicalSessionCandidate = { entry: SessionEntry; expectedEntry: SessionEntry; lineageRepairRequired: boolean; + ownerEvidenceOnly: boolean; rawEntryJson?: string; sessionKey: string; sqlitePath: string; @@ -112,11 +114,11 @@ function collectCanonicalSessionCandidates( stores: readonly CanonicalSessionStore[], ): CanonicalSessionCandidate[] { const inventory = stores.flatMap((target) => - listSessionEntriesForCanonicalRepair({ + listSqliteSessionEntriesWithCanonicalOwnerEvidence({ agentId: target.agentId, clone: false, storePath: target.storePath, - }).map(({ entry, rawEntryJson, sessionKey }) => { + }).map(({ canonicalOwnerSessionKey, entry, rawEntryJson, sessionKey }) => { const storedKey = resolveStoredSessionKeyForAgentStore({ cfg: params.cfg, agentId: target.agentId, @@ -127,6 +129,7 @@ function collectCanonicalSessionCandidates( ? resolveDeliveryProvenCanonicalSessionKey(storedKey, entry) : resolveAgentMainSessionKey({ cfg: params.cfg, agentId: target.agentId }), entry, + canonicalOwnerSessionKey, rawEntryJson, sessionKey, storedKey, @@ -134,96 +137,85 @@ function collectCanonicalSessionCandidates( }; }), ); - const canonicalKeysByStoredKey = new Map>(); - const addCanonicalMapping = (mappingKey: string, canonicalKey: string) => { - const mapped = canonicalKeysByStoredKey.get(mappingKey) ?? new Set(); - mapped.add(canonicalKey); - canonicalKeysByStoredKey.set(mappingKey, mapped); - }; - for (const item of inventory) { - const ownerAgentId = parseAgentSessionKey(item.storedKey)?.agentId ?? item.target.agentId; - // Never synthesize folded aliases from a canonical row: the lowercase peer may be a - // distinct case-sensitive session whose row was pruned. Only inventoried keys are proof. - for (const key of [item.sessionKey, item.storedKey]) { - addCanonicalMapping(`${item.target.sqlitePath}\0${ownerAgentId}\0${key}`, item.canonicalKey); - addCanonicalMapping(`*\0${ownerAgentId}\0${key}`, item.canonicalKey); - } - } - return inventory.map(({ canonicalKey, entry, rawEntryJson, sessionKey, target }) => { - const canonicalAgentId = - canonicalKey === "global" || canonicalKey === "unknown" - ? target.agentId - : resolveSessionStoreAgentId(params.cfg, canonicalKey); - const canonicalizeLineageKey = (value: string | undefined) => { - if (!value) { - return undefined; - } - const storedKey = resolveStoredSessionKeyForAgentStore({ - cfg: params.cfg, - agentId: canonicalAgentId, - sessionKey: value, - }); - const ownerAgentId = parseAgentSessionKey(storedKey)?.agentId ?? canonicalAgentId; - for (const key of [value, storedKey]) { - const sameStore = canonicalKeysByStoredKey.get( - `${target.sqlitePath}\0${ownerAgentId}\0${key}`, - ); - if (sameStore?.size === 1) { - return [...sameStore][0]; + const canonicalKeysByStoredKey = applyCanonicalOwnerEvidence(inventory); + return inventory.map( + ({ canonicalKey, canonicalOwnerSessionKey, entry, rawEntryJson, sessionKey, target }) => { + const canonicalAgentId = + canonicalKey === "global" || canonicalKey === "unknown" + ? target.agentId + : resolveSessionStoreAgentId(params.cfg, canonicalKey); + const canonicalizeLineageKey = (value: string | undefined) => { + if (!value) { + return undefined; } - } - for (const key of [value, storedKey]) { - const crossStore = canonicalKeysByStoredKey.get(`*\0${ownerAgentId}\0${key}`); - if (crossStore?.size === 1) { - return [...crossStore][0]; + const storedKey = resolveStoredSessionKeyForAgentStore({ + cfg: params.cfg, + agentId: canonicalAgentId, + sessionKey: value, + }); + const ownerAgentId = parseAgentSessionKey(storedKey)?.agentId ?? canonicalAgentId; + for (const key of [value, storedKey]) { + const sameStore = canonicalKeysByStoredKey.get( + `${target.sqlitePath}\0${ownerAgentId}\0${key}`, + ); + if (sameStore?.size === 1) { + return [...sameStore][0]; + } } - } - return storedKey; - }; - const parentSessionKey = canonicalizeLineageKey(entry.parentSessionKey); - const spawnedBy = canonicalizeLineageKey(entry.spawnedBy); - const forkSourceSessionKey = canonicalizeLineageKey(entry.forkSource?.sessionKey); - const normalizedEntry = { ...entry }; - if (parentSessionKey) { - normalizedEntry.parentSessionKey = parentSessionKey; - } else { - delete normalizedEntry.parentSessionKey; - } - if (spawnedBy) { - normalizedEntry.spawnedBy = spawnedBy; - } else { - delete normalizedEntry.spawnedBy; - } - if (entry.forkSource && forkSourceSessionKey) { - normalizedEntry.forkSource = { - ...entry.forkSource, - sessionKey: forkSourceSessionKey, + for (const key of [value, storedKey]) { + const crossStore = canonicalKeysByStoredKey.get(`*\0${ownerAgentId}\0${key}`); + if (crossStore?.size === 1) { + return [...crossStore][0]; + } + } + return storedKey; }; - } else if (entry.forkSource && entry.forkSource.sessionKey !== undefined) { - // A present but empty-normalized key cannot survive strict runtime validation. Missing - // legacy keys remain untouched so unrelated repair does not erase independent provenance. - const { sessionKey: _invalidSessionKey, ...forkProvenance } = entry.forkSource; - normalizedEntry.forkSource = forkProvenance as typeof entry.forkSource; - } - const lineageRepairRequired = - parentSessionKey !== (entry.parentSessionKey ?? undefined) || - spawnedBy !== (entry.spawnedBy ?? undefined) || - forkSourceSessionKey !== (entry.forkSource?.sessionKey ?? undefined); - const candidate: CanonicalSessionCandidate = { - agentId: target.agentId, - canonicalKey, - entry: normalizedEntry, - expectedEntry: entry, - lineageRepairRequired, - sessionKey, - sqlitePath: target.sqlitePath, - storePath: target.storePath, - }; - if (rawEntryJson !== undefined) { - candidate.rawEntryJson = rawEntryJson; - } - return candidate; - }); + const parentSessionKey = canonicalizeLineageKey(entry.parentSessionKey); + const spawnedBy = canonicalizeLineageKey(entry.spawnedBy); + const forkSourceSessionKey = canonicalizeLineageKey(entry.forkSource?.sessionKey); + const normalizedEntry = { ...entry }; + if (parentSessionKey) { + normalizedEntry.parentSessionKey = parentSessionKey; + } else { + delete normalizedEntry.parentSessionKey; + } + if (spawnedBy) { + normalizedEntry.spawnedBy = spawnedBy; + } else { + delete normalizedEntry.spawnedBy; + } + if (entry.forkSource && forkSourceSessionKey) { + normalizedEntry.forkSource = { + ...entry.forkSource, + sessionKey: forkSourceSessionKey, + }; + } else if (entry.forkSource && entry.forkSource.sessionKey !== undefined) { + // A present but empty-normalized key cannot survive strict runtime validation. Missing + // legacy keys remain untouched so unrelated repair does not erase independent provenance. + const { sessionKey: _invalidSessionKey, ...forkProvenance } = entry.forkSource; + normalizedEntry.forkSource = forkProvenance as typeof entry.forkSource; + } + const lineageRepairRequired = + parentSessionKey !== (entry.parentSessionKey ?? undefined) || + spawnedBy !== (entry.spawnedBy ?? undefined) || + forkSourceSessionKey !== (entry.forkSource?.sessionKey ?? undefined); + const candidate: CanonicalSessionCandidate = { + agentId: target.agentId, + canonicalKey, + entry: normalizedEntry, + expectedEntry: entry, + lineageRepairRequired, + ownerEvidenceOnly: canonicalOwnerSessionKey !== undefined, + sessionKey, + sqlitePath: target.sqlitePath, + storePath: target.storePath, + }; + if (rawEntryJson !== undefined) { + candidate.rawEntryJson = rawEntryJson; + } + return candidate; + }, + ); } function resolveCanonicalDestination(params: { @@ -298,8 +290,9 @@ function selectCanonicalSessionCandidate( env: params.env, sourceAgentId: first.agentId, }); + const metadataCandidates = candidates.filter((candidate) => !candidate.ownerEvidenceOnly); const selected = mergeCanonicalSessionEntryCandidates( - candidates + (metadataCandidates.length > 0 ? metadataCandidates : candidates) .toSorted((left, right) => Buffer.compare( Buffer.from(`${left.sqlitePath}\0${left.sessionKey}`, "utf8"), diff --git a/src/commands/doctor-session-canonical-owner-evidence.ts b/src/commands/doctor-session-canonical-owner-evidence.ts new file mode 100644 index 000000000000..db72b5079716 --- /dev/null +++ b/src/commands/doctor-session-canonical-owner-evidence.ts @@ -0,0 +1,49 @@ +import { parseAgentSessionKey } from "../routing/session-key.js"; + +type CanonicalOwnerEvidenceItem = { + canonicalKey: string; + canonicalOwnerSessionKey?: string; + sessionKey: string; + storedKey: string; + target: { agentId: string; sqlitePath: string }; +}; + +/** Projects transcript-owner evidence through aliases and indexes every proven source key. */ +export function applyCanonicalOwnerEvidence( + inventory: CanonicalOwnerEvidenceItem[], +): Map> { + const bySessionKey = new Map( + inventory.map((item) => [`${item.target.sqlitePath}\0${item.sessionKey}`, item] as const), + ); + const resolveCanonicalKey = ( + item: CanonicalOwnerEvidenceItem, + seen = new Set(), + ): string => { + if (!item.canonicalOwnerSessionKey) { + return item.canonicalKey; + } + const identity = `${item.target.sqlitePath}\0${item.sessionKey}`; + const owner = bySessionKey.get(`${item.target.sqlitePath}\0${item.canonicalOwnerSessionKey}`); + if (!owner || seen.has(identity)) { + return item.canonicalKey; + } + seen.add(identity); + return owner.canonicalOwnerSessionKey ? resolveCanonicalKey(owner, seen) : owner.canonicalKey; + }; + const canonicalKeysByStoredKey = new Map>(); + for (const item of inventory) { + item.canonicalKey = resolveCanonicalKey(item); + const ownerAgentId = parseAgentSessionKey(item.storedKey)?.agentId ?? item.target.agentId; + // Never synthesize folded aliases from a canonical row: the lowercase peer may be a + // distinct case-sensitive session whose row was pruned. Only inventoried keys are proof. + for (const key of [item.sessionKey, item.storedKey]) { + for (const sqlitePath of [item.target.sqlitePath, "*"]) { + const mappingKey = `${sqlitePath}\0${ownerAgentId}\0${key}`; + const mapped = canonicalKeysByStoredKey.get(mappingKey) ?? new Set(); + mapped.add(item.canonicalKey); + canonicalKeysByStoredKey.set(mappingKey, mapped); + } + } + } + return canonicalKeysByStoredKey; +} diff --git a/src/config/sessions/session-accessor.entry.ts b/src/config/sessions/session-accessor.entry.ts index 04f4bd68258c..69fbad8279d4 100644 --- a/src/config/sessions/session-accessor.entry.ts +++ b/src/config/sessions/session-accessor.entry.ts @@ -11,8 +11,8 @@ import { resolveSessionStorePathCore } from "./paths.js"; import { clearPluginOwnedSessionState } from "./plugin-host-cleanup.js"; import { copySqliteSessionOwnedStateForCanonicalRepair as copySessionOwnedStateForCanonicalRepair, - listSqliteSessionGenerationIdsForCanonicalRepair as listSessionGenerationIdsForCanonicalRepair, listSqliteSessionEntriesForCanonicalRepair as listSessionEntriesForCanonicalRepair, + listSqliteSessionGenerationIdsForCanonicalRepair as listSessionGenerationIdsForCanonicalRepair, rehomeSqliteSessionDeliveryReferencesForCanonicalRepair as rehomeSessionDeliveryReferencesForCanonicalRepair, rehomeSqliteSessionDeliveryReferencesForCanonicalRepairBatch as rehomeSessionDeliveryReferencesForCanonicalRepairBatch, } from "./session-accessor.sqlite-canonical-repair.js"; diff --git a/src/config/sessions/session-accessor.sqlite-canonical-inventory.ts b/src/config/sessions/session-accessor.sqlite-canonical-inventory.ts new file mode 100644 index 000000000000..f4628683f7ba --- /dev/null +++ b/src/config/sessions/session-accessor.sqlite-canonical-inventory.ts @@ -0,0 +1,201 @@ +import type { Selectable } from "kysely"; +import { executeSqliteQuerySync } from "../../infra/kysely-sync.js"; +import { withOpenClawAgentDatabaseReadOnly } from "../../state/openclaw-agent-db-readonly.js"; +import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js"; +import { normalizeSessionDeliveryState } from "../../utils/delivery-context.shared.js"; +import type { SessionEntrySummary } from "./session-accessor.sqlite-contract.js"; +import { + getSessionKysely, + resolveSqliteScope, + toDatabaseOptions, +} from "./session-accessor.sqlite-scope.js"; +import { parseSessionEntryJson } from "./session-accessor.sqlite-status.js"; +import type { SessionEntryListScope } from "./session-accessor.types.js"; +import { projectCanonicalSessionEntryShape } from "./store-entry-shape.js"; +import type { SessionEntry } from "./types.js"; + +type CanonicalRepairRow = Selectable & { + current_agent_harness_id: string | null; + current_chat_type: string | null; + current_ended_at: number | null; + current_model: string | null; + current_model_provider: string | null; + current_previous_session_id: string | null; + current_started_at: number | null; + current_window_owner_session_key: string | null; + delivery_account_id: string | null; + delivery_channel: string | null; + delivery_target: string | null; + delivery_thread_id: string | null; +}; + +/** Doctor inventory hydrates rejected legacy blobs from promoted node/window columns. */ +function hydrateCanonicalRepairEntry(row: CanonicalRepairRow): SessionEntry { + let record: Record = {}; + try { + const parsed = JSON.parse(row.entry_json) as unknown; + if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) { + record = parsed as Record; + } + } catch { + // Doctor owns malformed legacy repair; promoted identity columns keep the row reachable. + } + const createdActor = row.created_actor_type + ? { + type: row.created_actor_type, + ...(row.created_actor_id ? { id: row.created_actor_id } : {}), + } + : undefined; + const forkSource = + row.fork_source_session_key && row.fork_source_session_id + ? { + sessionKey: row.fork_source_session_key, + sessionId: row.fork_source_session_id, + ...(row.fork_source_entry_id ? { entryId: row.fork_source_entry_id } : {}), + } + : undefined; + const delivery = + row.delivery_channel && row.delivery_target + ? normalizeSessionDeliveryState({ + context: { + channel: row.delivery_channel, + to: row.delivery_target, + ...(row.delivery_account_id ? { accountId: row.delivery_account_id } : {}), + ...(row.delivery_thread_id ? { threadId: row.delivery_thread_id } : {}), + }, + }) + : undefined; + return projectCanonicalSessionEntryShape({ + ...record, + ...(row.status ? { status: row.status } : {}), + ...(row.current_started_at !== null ? { startedAt: row.current_started_at } : {}), + ...(row.current_ended_at !== null ? { endedAt: row.current_ended_at } : {}), + ...(row.current_chat_type ? { chatType: row.current_chat_type } : {}), + ...(row.current_model_provider ? { modelProvider: row.current_model_provider } : {}), + ...(row.current_model ? { model: row.current_model } : {}), + ...(row.current_previous_session_id + ? { previousSessionId: row.current_previous_session_id } + : {}), + ...(row.current_agent_harness_id ? { agentHarnessId: row.current_agent_harness_id } : {}), + ...(delivery ? { delivery } : {}), + ...(row.created_at !== null ? { createdAt: row.created_at } : {}), + ...(row.created_via ? { createdVia: row.created_via } : {}), + ...(createdActor ? { createdActor } : {}), + ...(row.spawned_by ? { spawnedBy: row.spawned_by } : {}), + ...(row.parent_session_key && row.parent_session_key !== row.spawned_by + ? { parentSessionKey: row.parent_session_key } + : {}), + ...(forkSource ? { forkSource } : {}), + ...(row.label ? { label: row.label } : {}), + ...(row.display_name ? { displayName: row.display_name } : {}), + ...(row.category ? { category: row.category } : {}), + ...(row.pinned_at !== null ? { pinnedAt: row.pinned_at } : {}), + ...(row.archived_at !== null ? { archivedAt: row.archived_at } : {}), + ...(row.last_read_at !== null ? { lastReadAt: row.last_read_at } : {}), + ...(row.last_interaction_at !== null ? { lastInteractionAt: row.last_interaction_at } : {}), + ...(row.last_activity_at !== null ? { lastActivityAt: row.last_activity_at } : {}), + // The canonical parser rejected this blob, so duplicate or malformed identity fields are + // untrusted. Promoted columns remain the durable transcript identity for doctor repair. + sessionId: row.current_session_id, + updatedAt: row.updated_at, + }); +} + +export function listSqliteSessionEntriesWithCanonicalOwnerEvidence( + scope: SessionEntryListScope = {}, +): Array { + const resolved = resolveSqliteScope({ ...scope, sessionKey: "" }); + const databaseOptions = toDatabaseOptions(resolved); + const result = withOpenClawAgentDatabaseReadOnly((database) => { + const db = getSessionKysely(database.db); + const rows = executeSqliteQuerySync( + database.db, + db + .selectFrom("session_nodes") + .leftJoin("session_windows as current_window", (join) => + join + .onRef("current_window.session_id", "=", "session_nodes.current_session_id") + .onRef("current_window.session_key", "=", "session_nodes.session_key"), + ) + .leftJoin( + "session_windows as current_window_owner", + "current_window_owner.session_id", + "session_nodes.current_session_id", + ) + .leftJoin( + "conversations as current_conversation", + "current_conversation.conversation_id", + "current_window.primary_conversation_id", + ) + .selectAll("session_nodes") + .select([ + "current_window_owner.session_key as current_window_owner_session_key", + "current_window.started_at as current_started_at", + "current_window.ended_at as current_ended_at", + "current_window.chat_type as current_chat_type", + "current_window.model_provider as current_model_provider", + "current_window.model as current_model", + "current_window.previous_session_id as current_previous_session_id", + "current_window.agent_harness_id as current_agent_harness_id", + "current_conversation.channel as delivery_channel", + "current_conversation.account_id as delivery_account_id", + "current_conversation.delivery_target", + "current_conversation.thread_id as delivery_thread_id", + ]), + ).rows; + const persistedEntries = new Map( + rows.map((row) => [row.session_key, parseSessionEntryJson(row)] as const), + ); + const validSessionKeysById = new Map(); + for (const row of rows) { + if (row.entry_valid !== 1 || !persistedEntries.get(row.session_key)) { + continue; + } + const keys = validSessionKeysById.get(row.current_session_id) ?? []; + keys.push(row.session_key); + validSessionKeysById.set(row.current_session_id, keys); + } + return rows.flatMap((row) => { + const isEmptyWindowOwner = + row.entry_json === "{}" && row.current_window_owner_session_key === row.session_key; + const competingValidKeys = (validSessionKeysById.get(row.current_session_id) ?? []).filter( + (sessionKey) => sessionKey !== row.session_key, + ); + const canonicalOwnerSessionKey = isEmptyWindowOwner + ? competingValidKeys.length === 1 + ? competingValidKeys[0] + : undefined + : row.entry_json === "{}" && + row.current_window_owner_session_key && + persistedEntries.has(row.current_window_owner_session_key) + ? row.current_window_owner_session_key + : undefined; + // Exact {} plus an unambiguous competing owner is corruption evidence. Without that + // evidence, an owned empty row remains the durable retained-history tombstone. + if (isEmptyWindowOwner && !canonicalOwnerSessionKey) { + return []; + } + const persistedEntry = persistedEntries.get(row.session_key); + const entry = persistedEntry ?? hydrateCanonicalRepairEntry(row); + const lineageProjectionMismatch = Boolean( + persistedEntry && + ((row.parent_session_key ?? undefined) !== + (persistedEntry.parentSessionKey ?? persistedEntry.spawnedBy ?? undefined) || + (row.spawned_by ?? undefined) !== (persistedEntry.spawnedBy ?? undefined) || + (row.fork_source_session_key ?? undefined) !== + (persistedEntry.forkSource?.sessionKey ?? undefined)), + ); + const rawCompareRequired = + row.entry_valid !== 1 || !persistedEntry || lineageProjectionMismatch; + return [ + { + sessionKey: row.session_key, + entry, + ...(canonicalOwnerSessionKey ? { canonicalOwnerSessionKey } : {}), + ...(rawCompareRequired ? { rawEntryJson: row.entry_json } : {}), + }, + ]; + }); + }, databaseOptions); + return result.found ? result.value : []; +} diff --git a/src/config/sessions/session-accessor.sqlite-canonical-repair.ts b/src/config/sessions/session-accessor.sqlite-canonical-repair.ts index c6e29bcd21cf..e694c4aea4df 100644 --- a/src/config/sessions/session-accessor.sqlite-canonical-repair.ts +++ b/src/config/sessions/session-accessor.sqlite-canonical-repair.ts @@ -1,16 +1,13 @@ import { uniqueStrings } from "@openclaw/normalization-core/string-normalization"; -import type { Selectable } from "kysely"; import { executeSqliteQuerySync, executeSqliteQueryTakeFirstSync, } from "../../infra/kysely-sync.js"; -import { withOpenClawAgentDatabaseReadOnly } from "../../state/openclaw-agent-db-readonly.js"; -import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js"; import { openOpenClawAgentDatabase, type OpenClawAgentDatabase, } from "../../state/openclaw-agent-db.js"; -import { normalizeSessionDeliveryState } from "../../utils/delivery-context.shared.js"; +import { listSqliteSessionEntriesWithCanonicalOwnerEvidence } from "./session-accessor.sqlite-canonical-inventory.js"; import type { SessionEntrySummary } from "./session-accessor.sqlite-contract.js"; import { publishSessionEntryCacheInvalidation } from "./session-accessor.sqlite-entry-cache.js"; import { readSessionGenerationIdsForKeys } from "./session-accessor.sqlite-lifecycle-state.js"; @@ -21,7 +18,6 @@ import { import { collectSessionStateIdsForEntry } from "./session-accessor.sqlite-references.js"; import { getSessionKysely, - resolveSqliteScope, resolveSqliteStoreScope, toDatabaseOptions, } from "./session-accessor.sqlite-scope.js"; @@ -33,12 +29,17 @@ import { deleteSessionTranscriptIndexInTransaction, reconcileSessionTranscriptIndexInTransaction, } from "./session-transcript-index.js"; -import { projectCanonicalSessionEntryShape } from "./store-entry-shape.js"; import { normalizeStoreSessionKey } from "./store-entry.js"; import type { SessionEntry } from "./types.js"; // Doctor-only cross-store transfer. Runtime readers never reconcile aliases. +export function listSqliteSessionEntriesForCanonicalRepair( + scope: SessionEntryListScope = {}, +): Array { + return listSqliteSessionEntriesWithCanonicalOwnerEvidence(scope); +} + function resolveSqliteCanonicalRepairLookupKeys( canonicalKey: string, storedKeys: readonly string[], @@ -200,158 +201,6 @@ export function rehomeSqliteSessionDeliveryReferencesForCanonicalRepairBatch( } } -type CanonicalRepairRow = Selectable & { - current_agent_harness_id: string | null; - current_chat_type: string | null; - current_ended_at: number | null; - current_model: string | null; - current_model_provider: string | null; - current_previous_session_id: string | null; - current_started_at: number | null; - current_window_id: string | null; - delivery_account_id: string | null; - delivery_channel: string | null; - delivery_target: string | null; - delivery_thread_id: string | null; -}; - -/** Doctor inventory hydrates rejected legacy blobs from promoted node/window columns. */ -function hydrateCanonicalRepairEntry(row: CanonicalRepairRow): SessionEntry { - let record: Record = {}; - try { - const parsed = JSON.parse(row.entry_json) as unknown; - if (parsed && typeof parsed === "object" && !Array.isArray(parsed)) { - record = parsed as Record; - } - } catch { - // Doctor owns malformed legacy repair; promoted identity columns keep the row reachable. - } - const createdActor = row.created_actor_type - ? { - type: row.created_actor_type, - ...(row.created_actor_id ? { id: row.created_actor_id } : {}), - } - : undefined; - const forkSource = - row.fork_source_session_key && row.fork_source_session_id - ? { - sessionKey: row.fork_source_session_key, - sessionId: row.fork_source_session_id, - ...(row.fork_source_entry_id ? { entryId: row.fork_source_entry_id } : {}), - } - : undefined; - const delivery = - row.delivery_channel && row.delivery_target - ? normalizeSessionDeliveryState({ - context: { - channel: row.delivery_channel, - to: row.delivery_target, - ...(row.delivery_account_id ? { accountId: row.delivery_account_id } : {}), - ...(row.delivery_thread_id ? { threadId: row.delivery_thread_id } : {}), - }, - }) - : undefined; - return projectCanonicalSessionEntryShape({ - ...record, - ...(row.status ? { status: row.status } : {}), - ...(row.current_started_at !== null ? { startedAt: row.current_started_at } : {}), - ...(row.current_ended_at !== null ? { endedAt: row.current_ended_at } : {}), - ...(row.current_chat_type ? { chatType: row.current_chat_type } : {}), - ...(row.current_model_provider ? { modelProvider: row.current_model_provider } : {}), - ...(row.current_model ? { model: row.current_model } : {}), - ...(row.current_previous_session_id - ? { previousSessionId: row.current_previous_session_id } - : {}), - ...(row.current_agent_harness_id ? { agentHarnessId: row.current_agent_harness_id } : {}), - ...(delivery ? { delivery } : {}), - ...(row.created_at !== null ? { createdAt: row.created_at } : {}), - ...(row.created_via ? { createdVia: row.created_via } : {}), - ...(createdActor ? { createdActor } : {}), - ...(row.spawned_by ? { spawnedBy: row.spawned_by } : {}), - ...(row.parent_session_key && row.parent_session_key !== row.spawned_by - ? { parentSessionKey: row.parent_session_key } - : {}), - ...(forkSource ? { forkSource } : {}), - ...(row.label ? { label: row.label } : {}), - ...(row.display_name ? { displayName: row.display_name } : {}), - ...(row.category ? { category: row.category } : {}), - ...(row.pinned_at !== null ? { pinnedAt: row.pinned_at } : {}), - ...(row.archived_at !== null ? { archivedAt: row.archived_at } : {}), - ...(row.last_read_at !== null ? { lastReadAt: row.last_read_at } : {}), - ...(row.last_interaction_at !== null ? { lastInteractionAt: row.last_interaction_at } : {}), - ...(row.last_activity_at !== null ? { lastActivityAt: row.last_activity_at } : {}), - // The canonical parser rejected this blob, so duplicate or malformed identity fields are - // untrusted. Promoted columns remain the durable transcript identity for doctor repair. - sessionId: row.current_session_id, - updatedAt: row.updated_at, - }); -} - -export function listSqliteSessionEntriesForCanonicalRepair( - scope: SessionEntryListScope = {}, -): Array { - const resolved = resolveSqliteScope({ ...scope, sessionKey: "" }); - const databaseOptions = toDatabaseOptions(resolved); - const result = withOpenClawAgentDatabaseReadOnly((database) => { - const db = getSessionKysely(database.db); - return executeSqliteQuerySync( - database.db, - db - .selectFrom("session_nodes") - .leftJoin("session_windows as current_window", (join) => - join - .onRef("current_window.session_id", "=", "session_nodes.current_session_id") - .onRef("current_window.session_key", "=", "session_nodes.session_key"), - ) - .leftJoin( - "conversations as current_conversation", - "current_conversation.conversation_id", - "current_window.primary_conversation_id", - ) - .selectAll("session_nodes") - .select([ - "current_window.session_id as current_window_id", - "current_window.started_at as current_started_at", - "current_window.ended_at as current_ended_at", - "current_window.chat_type as current_chat_type", - "current_window.model_provider as current_model_provider", - "current_window.model as current_model", - "current_window.previous_session_id as current_previous_session_id", - "current_window.agent_harness_id as current_agent_harness_id", - "current_conversation.channel as delivery_channel", - "current_conversation.account_id as delivery_account_id", - "current_conversation.delivery_target", - "current_conversation.thread_id as delivery_thread_id", - ]), - ).rows.flatMap((row) => { - // Exact {} plus an owned window is the durable retained-history tombstone. - if (row.entry_json === "{}" && row.current_window_id === row.current_session_id) { - return []; - } - const persistedEntry = parseSessionEntryJson(row); - const entry = persistedEntry ?? hydrateCanonicalRepairEntry(row); - const lineageProjectionMismatch = Boolean( - persistedEntry && - ((row.parent_session_key ?? undefined) !== - (persistedEntry.parentSessionKey ?? persistedEntry.spawnedBy ?? undefined) || - (row.spawned_by ?? undefined) !== (persistedEntry.spawnedBy ?? undefined) || - (row.fork_source_session_key ?? undefined) !== - (persistedEntry.forkSource?.sessionKey ?? undefined)), - ); - const rawCompareRequired = - row.entry_valid !== 1 || !persistedEntry || lineageProjectionMismatch; - return [ - { - sessionKey: row.session_key, - entry, - ...(rawCompareRequired ? { rawEntryJson: row.entry_json } : {}), - }, - ]; - }); - }, databaseOptions); - return result.found ? result.value : []; -} - function copySqliteSessionOwnedStateForRepair(params: { canonicalKey: string; destination: OpenClawAgentDatabase; diff --git a/src/config/sessions/session-accessor.sqlite-data-version.test.ts b/src/config/sessions/session-accessor.sqlite-data-version.test.ts index 920033a580e6..0dd50f830ed2 100644 --- a/src/config/sessions/session-accessor.sqlite-data-version.test.ts +++ b/src/config/sessions/session-accessor.sqlite-data-version.test.ts @@ -653,6 +653,39 @@ describe("SQLite session entry cache", () => { expect(after.entries.get(scope.sessionKey)).not.toBe(before.entries.get(scope.sessionKey)); }); + it("rejects a transcript write after its persisted owner changes", async () => { + const scope = createSessionScope("transcript-owner-conflict"); + const sessionId = "owned-transcript-session"; + await upsertSessionEntryCore(scope, { sessionId, updatedAt: 1 }); + + expect(() => + runOpenClawAgentWriteTransaction((database) => { + ensureTranscriptSessionRoot( + database, + { + agentId: scope.agentId, + env: scope.env, + sessionId, + sessionKey: "agent:main:stale-owner", + }, + 2, + ); + }, scope), + ).toThrow("resolve the transcript target again before retrying"); + + const database = openOpenClawAgentDatabase(scope); + expect( + database.db + .prepare("SELECT session_key, entry_valid FROM session_nodes ORDER BY session_key") + .all(), + ).toEqual([{ session_key: scope.sessionKey, entry_valid: 1 }]); + expect( + database.db + .prepare("SELECT session_key FROM session_windows WHERE session_id = ?") + .get(sessionId), + ).toEqual({ session_key: scope.sessionKey }); + }); + it("bypasses the cache in a transaction and reuses the persisted snapshot after rollback", async () => { const scope = createSessionScope("transaction-rollback"); await upsertSessionEntryCore(scope, { label: "before", sessionId: "rollback", updatedAt: 1 }); diff --git a/src/config/sessions/session-accessor.sqlite-transcript-state.ts b/src/config/sessions/session-accessor.sqlite-transcript-state.ts index 10e2e4e67ad8..c2bd15ce1c5b 100644 --- a/src/config/sessions/session-accessor.sqlite-transcript-state.ts +++ b/src/config/sessions/session-accessor.sqlite-transcript-state.ts @@ -82,10 +82,22 @@ export function ensureTranscriptSessionRoot( updatedAt: number, options: { allowStoredAlias?: boolean } = {}, ): void { + const db = getSessionKysely(database.db); if (!options.allowStoredAlias) { assertCanonicalSqliteSessionKeysCurrent(database); assertCanonicalSessionKeyWriteMatchesDatabase(database, scope.sessionKey); - const db = getSessionKysely(database.db); + const persistedSessionKey = executeSqliteQueryTakeFirstSync( + database.db, + db + .selectFrom("session_windows") + .select("session_key") + .where("session_id", "=", scope.sessionId), + )?.session_key; + if (persistedSessionKey && persistedSessionKey !== scope.sessionKey) { + throw new Error( + `Transcript session ${scope.sessionId} is owned by ${persistedSessionKey}, not ${scope.sessionKey}; resolve the transcript target again before retrying.`, + ); + } const lookupKeys = uniqueStrings([ scope.sessionKey, ...foldedSessionKeyAliasCandidates(normalizeStoreSessionKey(scope.sessionKey)), @@ -147,7 +159,6 @@ export function ensureTranscriptSessionRoot( } } } - const db = getSessionKysely(database.db); const insertedNode = executeSqliteQuerySync( database.db, db @@ -186,7 +197,6 @@ export function ensureTranscriptSessionRoot( }) .onConflict((conflict) => conflict.column("session_id").doUpdateSet({ - session_key: scope.sessionKey, updated_at: updatedAt, }), ), diff --git a/src/config/sessions/session-accessor.test.ts b/src/config/sessions/session-accessor.test.ts index 6e16e1e6f982..3bed313ef62b 100644 --- a/src/config/sessions/session-accessor.test.ts +++ b/src/config/sessions/session-accessor.test.ts @@ -4201,6 +4201,46 @@ describe("session accessor seam", () => { expect(loadSessionEntry(scope)).not.toHaveProperty("sessionFile"); }); + it("keeps the persisted transcript owner when a caller supplies a stale session key", async () => { + const sessionId = "session-canonical-owner"; + const canonicalScope = { + agentId: "main", + sessionId, + sessionKey: "agent:main:main", + storePath, + }; + await upsertSessionEntryCore(canonicalScope, { sessionId, updatedAt: 10 }); + + const target = await resolveSessionTranscriptRuntimeTarget({ + ...canonicalScope, + sessionKey: "agent:main:telegram:default:direct:fixture-peer", + }); + await appendTranscriptEvent(target, { + id: "canonical-owner-event", + timestamp: new Date(20).toISOString(), + type: "metadata", + }); + + const database = openOpenClawAgentDatabase({ + agentId: "main", + path: expectDefined( + resolveSqliteTargetFromSessionStorePath(storePath, { agentId: "main" }).path, + "session database path", + ), + }); + expect( + database.db + .prepare("SELECT session_key, entry_valid FROM session_nodes ORDER BY session_key") + .all(), + ).toEqual([{ session_key: canonicalScope.sessionKey, entry_valid: 1 }]); + expect( + database.db + .prepare("SELECT session_key FROM session_windows WHERE session_id = ?") + .get(sessionId), + ).toEqual({ session_key: canonicalScope.sessionKey }); + expect(target.sessionKey).toBe(canonicalScope.sessionKey); + }); + it("drops imported legacy session transcript paths from canonical rows", async () => { const sessionKey = "agent:main:main"; await importSqliteSessionRows({ diff --git a/src/config/sessions/session-accessor.transcript-target.ts b/src/config/sessions/session-accessor.transcript-target.ts index 6ad24ea252b3..35c4e9a5157b 100644 --- a/src/config/sessions/session-accessor.transcript-target.ts +++ b/src/config/sessions/session-accessor.transcript-target.ts @@ -3,6 +3,7 @@ import { resolveOpenClawAgentSqlitePath } from "../../state/openclaw-agent-db.js import { getRuntimeConfig } from "../io.js"; import { resolveSessionStorePathCore } from "./paths.js"; import { resolveSessionEntrySelection } from "./session-accessor.entry.js"; +import { resolveSessionKeyBySessionId } from "./session-accessor.sqlite-entry.js"; import { resolveSqliteTranscriptScope, toDatabaseOptions, @@ -22,7 +23,10 @@ type SessionTranscriptRuntimeContext = { }; function resolveRuntimeContext( - scope: Pick, + scope: Pick< + SessionTranscriptRuntimeScope, + "agentId" | "env" | "sessionId" | "sessionKey" | "storePath" + >, ): SessionTranscriptRuntimeContext { const agentId = scope.agentId ?? resolveAgentIdFromSessionKey(scope.sessionKey); if (!agentId) { @@ -37,15 +41,27 @@ function resolveRuntimeContext( sessionKey: scope.sessionKey, storePath: configuredStorePath, }); - const resolved = resolveSessionEntrySelection({ + const persistedSessionKey = resolveSessionKeyBySessionId({ agentId, ...(scope.env ? { env: scope.env } : {}), - sessionKey: scope.sessionKey, + sessionId: scope.sessionId, storePath, }); + const sessionKey = + persistedSessionKey ?? + resolveSessionEntrySelection( + { + agentId, + ...(scope.env ? { env: scope.env } : {}), + sessionKey: scope.sessionKey, + storePath, + }, + { readOnly: true }, + )?.normalizedKey ?? + scope.sessionKey; return { agentId, - sessionKey: resolved?.normalizedKey ?? scope.sessionKey, + sessionKey, storePath, }; } diff --git a/src/config/sessions/session-accessor.transcript-turn.ts b/src/config/sessions/session-accessor.transcript-turn.ts index 950c464e5b45..a63dc6b942ec 100644 --- a/src/config/sessions/session-accessor.transcript-turn.ts +++ b/src/config/sessions/session-accessor.transcript-turn.ts @@ -21,6 +21,7 @@ import { } from "./session-accessor.sqlite-transcript-sequences.js"; import { redactTranscriptMessageForStorage } from "./session-accessor.sqlite-transcript-store.js"; import { appendExpectedSessionTranscriptTurn } from "./session-accessor.sqlite-transcript-write.js"; +import { resolveSessionTranscriptRuntimeTarget } from "./session-accessor.transcript-target.js"; import { appendTranscriptMessage, emitTranscriptUpdate } from "./session-accessor.transcript.js"; import type { SessionTranscriptWriteScope, @@ -170,7 +171,13 @@ export async function persistSessionTranscriptTurn( target.sessionId ) { return await persistExpectedSessionTranscriptTurn( - { ...scope, storePath: target.storePath }, + { + ...scope, + agentId: target.agentId, + sessionId: target.sessionId, + sessionKey: target.sessionKey, + storePath: target.storePath, + }, { ...options, expectedSessionId: target.sessionId, @@ -397,27 +404,32 @@ async function resolveTranscriptTurnTarget( agentId, env: scope.env, }); + const runtimeTarget = scope.sessionStore + ? undefined + : await resolveSessionTranscriptRuntimeTarget({ + agentId, + ...(scope.env ? { env: scope.env } : {}), + sessionId: scope.sessionId, + sessionKey, + storePath, + }); + const resolvedSessionKey = runtimeTarget?.sessionKey ?? sessionKey; const resolved = scope.sessionStore - ? resolveSessionEntryFromStore({ store: scope.sessionStore, sessionKey }) - : resolveSessionEntrySelection( - { - agentId, - ...(scope.env ? { env: scope.env } : {}), - sessionKey, - storePath, - }, - { readOnly: true }, - ); + ? resolveSessionEntryFromStore({ store: scope.sessionStore, sessionKey: resolvedSessionKey }) + : undefined; // Mirrors can represent either durable Gateway state or memory-only internal // sessions. Classify that provenance without materializing SQLite state. - const persistedEntry = scope.sessionStore - ? loadSessionEntryReadOnly({ ...scope, agentId, sessionKey, storePath }) - : resolved?.existing; + const persistedEntry = loadSessionEntryReadOnly({ + ...scope, + agentId, + sessionKey: resolvedSessionKey, + storePath, + }); const sessionEntry = resolved?.existing ?? scope.sessionEntry ?? persistedEntry; return { agentId, sessionId: scope.sessionId, - sessionKey: resolved?.normalizedKey ?? sessionKey, + sessionKey: runtimeTarget?.sessionKey ?? resolved?.normalizedKey ?? resolvedSessionKey, storePath, sessionEntry, entryFromPersistedStore: persistedEntry != null, diff --git a/src/gateway/server-methods/chat.directive-tags.test.ts b/src/gateway/server-methods/chat.directive-tags.test.ts index 92f154c60c17..7ca7971e72c8 100644 --- a/src/gateway/server-methods/chat.directive-tags.test.ts +++ b/src/gateway/server-methods/chat.directive-tags.test.ts @@ -117,6 +117,7 @@ const mockState = vi.hoisted(() => ({ onAfterAgentRunStart: null as (() => void) | null, agentRunId: "run-agent-1", sessionEntry: {} as Record, + sessionIdsByKey: new Map(), sessionMissing: false, loadSessionEntryCalls: [] as Array<{ rawKey: string; opts?: { agentId?: string } }>, lastDispatchCtx: undefined as MsgContext | undefined, @@ -236,7 +237,7 @@ vi.mock("../session-utils.js", async () => { const entry = mockState.sessionMissing ? undefined : { - sessionId: mockState.sessionId, + sessionId: mockState.sessionIdsByKey.get(rawKey) ?? mockState.sessionId, sessionFile: mockState.transcriptPath, ...mockState.sessionEntry, }; @@ -1302,6 +1303,7 @@ describe("chat directive tag stripping for non-streaming final payloads", () => mockState.onAfterAgentRunStart = null; mockState.agentRunId = "run-agent-1"; mockState.sessionEntry = {}; + mockState.sessionIdsByKey.clear(); mockState.sessionMissing = false; mockState.loadSessionEntryCalls = []; mockState.lastDispatchCtx = undefined; @@ -2452,13 +2454,23 @@ describe("chat directive tag stripping for non-streaming final payloads", () => it("persists non-agent plugin-bound replies in the binding-owned session", async () => { await createTranscriptFixture("openclaw-chat-send-plugin-binding-history-"); const targetSessionKey = "plugin-binding:codex:history123"; + const targetSessionId = "plugin-binding-history-session"; + await replaceSessionEntry( + { + agentId: "main", + sessionKey: `agent:main:${targetSessionKey}`, + storePath: mockState.storePath, + }, + { sessionId: targetSessionId, updatedAt: Date.now() }, + ); + mockState.sessionIdsByKey.set(targetSessionKey, targetSessionId); mockState.finalPayload = setReplyPayloadMetadata( { text: "bound history reply" }, { sourceReplyTranscriptMirror: { sessionKey: targetSessionKey, agentId: "main", - expectedSessionId: mockState.sessionId, + expectedSessionId: targetSessionId, }, }, ); diff --git a/src/gateway/server-methods/chat.inject.parentid.test.ts b/src/gateway/server-methods/chat.inject.parentid.test.ts index 2b1ab65f82ba..32e063490a2c 100644 --- a/src/gateway/server-methods/chat.inject.parentid.test.ts +++ b/src/gateway/server-methods/chat.inject.parentid.test.ts @@ -139,7 +139,7 @@ describe("gateway chat.inject transcript writes", () => { } }); - it("emits and returns the redacted injected assistant message", async () => { + it("emits a redacted injected message through its persisted transcript owner", async () => { const fixture = await createSqliteTranscriptFixture({ prefix: "openclaw-chat-inject-redact-", sessionId: "sess-redact", @@ -161,7 +161,7 @@ describe("gateway chat.inject transcript writes", () => { expect(appended.ok).toBe(true); expect(JSON.stringify(appended.message)).not.toContain(fakeApiKey); expect(updates).toHaveLength(1); - expect(updates[0]).toMatchObject({ sessionKey: "global", agentId: "main" }); + expect(updates[0]).toMatchObject({ sessionKey: "agent:main:main", agentId: "main" }); const last = (await readLastTranscriptRecord(fixture)) as { message?: unknown }; expect(JSON.stringify(last.message)).not.toContain(fakeApiKey);