From f296dc4e68e4cc7e3b98e476a2b1512a692d12ab Mon Sep 17 00:00:00 2001 From: pick-cat Date: Mon, 6 Jul 2026 12:32:19 +0800 Subject: [PATCH] fix(sessions): persist sender metadata in user turn transcript JSONL (#90552) * fix(sessions): persist sender metadata in user turn transcript JSONL Thread senderId/senderName/senderUsername/senderE164 from the channel inbound context into the persisted user-turn transcript message so that group chat session JSONL records include __openclaw sender identity. Ref #90531 * fix(sessions): exclude senderE164 from persisted transcript for privacy Remove phone-number field from the persisted __openclaw sender envelope, keeping only senderId, senderName, and senderUsername. Privacy-sensitive E.164 metadata can be added back by maintainers if needed. Ref #90531 * fix(infra): spread base fields in applyExecPolicyLayer return values The `as TBase & ExecPolicyLayer` casts failed because the returned objects did not spread `...base`, losing generic TBase fields. Co-Authored-By: Claude Opus 4.7 * fix(infra): use ...base spread instead of ...baseWithoutMode The baseWithoutMode destructuring produces Omit which is not assignable to TBase & ExecPolicyLayer when the as-cast is removed. Co-Authored-By: Claude Opus 4.7 * fix(infra): preserve baseWithoutMode in second applyExecPolicyLayer branch The second branch (security/ask override without a mode change) must exclude base.mode from the spread so that mode is not leaked into results when only security or ask fields are being overridden. Without this fix, the spread ...base carries mode through to the returned object, breaking callers that expect applyExecPolicyLayer to clear stale mode when applying explicit security/ask policy fields. Co-Authored-By: Claude Opus 4.7 * fix(sessions): scope persisted sender attribution * fix(sessions): preserve sender metadata through hooks * refactor(sessions): keep sender metadata path lean * fix(sessions): preserve sender metadata in runtime writes * fix(sessions): preserve queued sender attribution * test(sessions): use complete message fixtures * refactor(sessions): rely on narrowed user message type * test(sessions): use shared temp directory helper * test(sessions): align sender metadata assertions * fix(sessions): honor sender metadata redaction hooks * test(agents): use automatic temp cleanup * test(sessions): cover queued turn provenance * test(auto-reply): expect room sender metadata * refactor(sessions): isolate queued transcript context --------- Co-authored-by: Claude Opus 4.7 Co-authored-by: Peter Steinberger --- .../embedded-agent-runner.guard.test.ts | 120 +++++++++++++++++- src/agents/embedded-agent-runner/run-state.ts | 9 +- .../run/attempt.queue-message.test.ts | 20 +++ .../run/attempt.queue-message.ts | 28 +++- src/agents/embedded-agent-runner/runs.test.ts | 12 +- src/agents/embedded-agent-runner/runs.ts | 2 +- .../session-tool-result-guard-wrapper.ts | 39 +++++- src/agents/sessions/agent-session.ts | 39 +++++- src/agents/sessions/sdk.test.ts | 37 ++++++ .../agent-runner.runreplyagent.e2e.test.ts | 56 +++++++- src/auto-reply/reply/agent-runner.ts | 3 + .../reply/get-reply-run.media-only.test.ts | 53 +++++++- src/auto-reply/reply/get-reply-run.ts | 33 +++-- src/auto-reply/reply/reply-run-registry.ts | 22 +++- .../user-turn-transcript-runtime-context.ts | 69 ++++++++++ src/sessions/user-turn-transcript.test.ts | 98 +++++++++++++- src/sessions/user-turn-transcript.ts | 88 ++++++++++--- src/sessions/user-turn-transcript.types.ts | 2 + 18 files changed, 668 insertions(+), 62 deletions(-) create mode 100644 src/sessions/user-turn-transcript-runtime-context.ts diff --git a/src/agents/embedded-agent-runner.guard.test.ts b/src/agents/embedded-agent-runner.guard.test.ts index 33c04be058dc..e11205b89137 100644 --- a/src/agents/embedded-agent-runner.guard.test.ts +++ b/src/agents/embedded-agent-runner.guard.test.ts @@ -1,11 +1,23 @@ // Covers session-manager guard behavior for tool-result pairing and transcript // redaction. +import { readFileSync } from "node:fs"; import type { AgentMessage } from "openclaw/plugin-sdk/agent-core"; import { SessionManager } from "openclaw/plugin-sdk/agent-sessions"; -import { describe, expect, it } from "vitest"; +import { + initializeGlobalHookRunner, + resetGlobalHookRunner, +} from "openclaw/plugin-sdk/hook-runtime"; +import { createMockPluginRegistry } from "openclaw/plugin-sdk/plugin-test-runtime"; +import { afterEach, describe, expect, it } from "vitest"; +import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; +import { + attachRuntimeUserTurnTranscriptContext, + createUserTurnTranscriptRecorder, +} from "../sessions/user-turn-transcript.js"; import { guardSessionManager } from "./session-tool-result-guard-wrapper.js"; import { sanitizeToolUseResultPairing } from "./session-transcript-repair.js"; +import { makeAgentAssistantMessage } from "./test-helpers/agent-message-fixtures.js"; function assistantToolCall(id: string): AgentMessage { return { @@ -15,6 +27,12 @@ function assistantToolCall(id: string): AgentMessage { } describe("guardSessionManager integration", () => { + const tempDirs = useAutoCleanupTempDirTracker(afterEach); + + afterEach(() => { + resetGlobalHookRunner(); + }); + it("persists synthetic toolResult before subsequent assistant message", () => { // Providers require every assistant tool call to be followed by a result // before the next assistant turn. @@ -140,6 +158,106 @@ describe("guardSessionManager integration", () => { expect(messages[1]).toEqual({ role: "user", content: "follow-up" }); }); + it("lets a write hook remove sender identity while preserving auth state", () => { + initializeGlobalHookRunner( + createMockPluginRegistry([ + { + hookName: "before_message_write", + handler: () => ({ + message: { + role: "user", + content: "[redacted by hook]", + timestamp: 124, + __openclaw: { hookOwned: true }, + } as AgentMessage, + }), + }, + ]), + ); + const sm = guardSessionManager(SessionManager.inMemory(), { + preparedUserTurnMessage: { + role: "user", + content: "private group prompt", + timestamp: 123, + __openclaw: { + senderIsOwner: true, + senderId: "secret-user", + senderName: "secret-name", + }, + } as Extract, + }); + + sm.appendMessage({ role: "user", content: "runtime prompt", timestamp: 125 }); + + const message = sm.getEntries().find((entry) => entry.type === "message") as + | { message?: AgentMessage } + | undefined; + expect(message?.message).toMatchObject({ + role: "user", + content: "[redacted by hook]", + __openclaw: { + hookOwned: true, + senderIsOwner: true, + }, + }); + expect(JSON.stringify(message?.message)).not.toContain("secret-user"); + expect(JSON.stringify(message?.message)).not.toContain("secret-name"); + }); + + it("commits queued group sender metadata to JSONL and completes its recorder", () => { + const dir = tempDirs.make("openclaw-queued-group-turn-"); + const sessionManager = SessionManager.create(dir, dir); + const sessionFile = sessionManager.getSessionFile(); + if (!sessionFile) { + throw new Error("expected file-backed session manager"); + } + const recorder = createUserTurnTranscriptRecorder({ + input: { + text: "visible group prompt", + sender: { id: "user-42", name: "Ada", username: "ada42" }, + }, + target: { transcriptPath: sessionFile }, + }); + const preparedMessage = recorder.message; + if (!preparedMessage) { + throw new Error("expected prepared group turn"); + } + const sm = guardSessionManager(sessionManager, { + inputProvenance: { kind: "inter_session", sourceTool: "sessions_send" }, + }); + const runtimeMessage = attachRuntimeUserTurnTranscriptContext( + { + role: "user", + content: [{ type: "text", text: "runtime group prompt" }], + timestamp: 456, + }, + { message: preparedMessage, recorder }, + ); + + sm.appendMessage(runtimeMessage); + sm.appendMessage( + makeAgentAssistantMessage({ + content: [{ type: "text", text: "acknowledged" }], + }), + ); + + const entries = readFileSync(sessionFile, "utf8") + .trim() + .split("\n") + .map((line) => JSON.parse(line) as { type: string; message?: AgentMessage }); + expect(entries.find((entry) => entry.message?.role === "user")?.message).toMatchObject({ + role: "user", + content: "visible group prompt", + __openclaw: { + senderId: "user-42", + senderName: "Ada", + senderUsername: "ada42", + }, + provenance: { kind: "inter_session", sourceTool: "sessions_send" }, + }); + expect(recorder.hasPersisted()).toBe(true); + }); + it("does not consume prepared user persistence for before-agent-run blocked messages", () => { // Blocked messages are audit records, not the actual user turn that should // receive prepared media metadata. diff --git a/src/agents/embedded-agent-runner/run-state.ts b/src/agents/embedded-agent-runner/run-state.ts index 1f21b544db6f..60ef60ae4ad3 100644 --- a/src/agents/embedded-agent-runner/run-state.ts +++ b/src/agents/embedded-agent-runner/run-state.ts @@ -7,6 +7,7 @@ import { listActiveReplyRunSessionKeys, listActiveReplyRunSessionIds, resolveActiveReplyRunSessionId, + type ReplyBackendQueueMessageOptions, } from "../../auto-reply/reply/reply-run-registry.js"; import { resolveGlobalSingleton } from "../../shared/global-singleton.js"; @@ -30,13 +31,7 @@ export type EmbeddedAgentQueueHandle = { sourceReplyDeliveryMode?: SourceReplyDeliveryMode; }; -export type EmbeddedAgentQueueMessageOptions = { - steeringMode?: "all"; - debounceMs?: number; - deliveryTimeoutMs?: number; - waitForTranscriptCommit?: boolean; - sourceReplyDeliveryMode?: SourceReplyDeliveryMode; -}; +export type EmbeddedAgentQueueMessageOptions = ReplyBackendQueueMessageOptions; export type ActiveEmbeddedRunSnapshot = { transcriptLeafId: string | null; diff --git a/src/agents/embedded-agent-runner/run/attempt.queue-message.test.ts b/src/agents/embedded-agent-runner/run/attempt.queue-message.test.ts index 1131f0793a92..a564985654a2 100644 --- a/src/agents/embedded-agent-runner/run/attempt.queue-message.test.ts +++ b/src/agents/embedded-agent-runner/run/attempt.queue-message.test.ts @@ -1,12 +1,32 @@ // Coverage for queued steering message commit and cancellation behavior. import { describe, expect, it, vi } from "vitest"; +import { createUserTurnTranscriptRecorder } from "../../../sessions/user-turn-transcript.js"; import { cancelQueuedSteeringMessage, + steerActiveSessionWithOptionalDeliveryWait, steerAndWaitForTranscriptCommit, type EmbeddedAgentActiveSessionSteerTarget, } from "./attempt.queue-message.js"; describe("embedded OpenClaw queued steering cancellation", () => { + it("forwards prepared transcript context with a queued steering message", async () => { + const steer = vi.fn(async () => undefined); + const recorder = createUserTurnTranscriptRecorder({ + input: { text: "visible prompt", sender: { id: "user-42" } }, + target: { transcriptPath: "/tmp/unused-session.jsonl" }, + }); + const activeSession: EmbeddedAgentActiveSessionSteerTarget = { + steer, + subscribe: () => () => {}, + }; + + await steerActiveSessionWithOptionalDeliveryWait(activeSession, "runtime prompt", { + userTurnTranscriptRecorder: recorder, + }); + + expect(steer).toHaveBeenCalledWith("runtime prompt", undefined, recorder); + }); + it("waits for the queued user message_end transcript boundary", async () => { // A queued steer is only durable once the user message_end event lands in // the active transcript. diff --git a/src/agents/embedded-agent-runner/run/attempt.queue-message.ts b/src/agents/embedded-agent-runner/run/attempt.queue-message.ts index fcd9dd93260f..60b481de9f99 100644 --- a/src/agents/embedded-agent-runner/run/attempt.queue-message.ts +++ b/src/agents/embedded-agent-runner/run/attempt.queue-message.ts @@ -2,6 +2,7 @@ * Steers active embedded sessions and waits for transcript commits when needed. */ import { toErrorObject } from "../../../infra/errors.js"; +import type { UserTurnTranscriptRecorder } from "../../../sessions/user-turn-transcript.types.js"; import { log } from "../logger.js"; /** @@ -11,7 +12,11 @@ import { log } from "../logger.js"; export type EmbeddedAgentActiveSessionSteerTarget = { agent?: unknown; getSteeringMessages?(): readonly string[]; - steer(text: string): Promise; + steer( + text: string, + images?: undefined, + userTurnTranscriptRecorder?: UserTurnTranscriptRecorder, + ): Promise; subscribe(listener: (event: unknown) => void): () => void; }; @@ -126,6 +131,7 @@ export async function steerAndWaitForTranscriptCommit( activeSession: EmbeddedAgentActiveSessionSteerTarget, text: string, timeoutMs: number, + userTurnTranscriptRecorder?: UserTurnTranscriptRecorder, ): Promise { await new Promise((resolve, reject) => { let settled = false; @@ -206,7 +212,10 @@ export async function steerAndWaitForTranscriptCommit( scheduleTerminalCancellation(); } }); - activeSession.steer(text).catch((err: unknown) => { + const steer = userTurnTranscriptRecorder + ? activeSession.steer(text, undefined, userTurnTranscriptRecorder) + : activeSession.steer(text); + steer.catch((err: unknown) => { finish(err); }); }); @@ -219,15 +228,26 @@ export async function steerAndWaitForTranscriptCommit( export async function steerActiveSessionWithOptionalDeliveryWait( activeSession: EmbeddedAgentActiveSessionSteerTarget, text: string, - options: { deliveryTimeoutMs?: number; waitForTranscriptCommit?: boolean } | undefined, + options: + | { + deliveryTimeoutMs?: number; + waitForTranscriptCommit?: boolean; + userTurnTranscriptRecorder?: UserTurnTranscriptRecorder; + } + | undefined, ): Promise { if (options?.waitForTranscriptCommit !== true) { - await activeSession.steer(text); + if (options?.userTurnTranscriptRecorder) { + await activeSession.steer(text, undefined, options.userTurnTranscriptRecorder); + } else { + await activeSession.steer(text); + } return; } await steerAndWaitForTranscriptCommit( activeSession, text, options.deliveryTimeoutMs ?? DEFAULT_QUEUE_TRANSCRIPT_COMMIT_TIMEOUT_MS, + options.userTurnTranscriptRecorder, ); } diff --git a/src/agents/embedded-agent-runner/runs.test.ts b/src/agents/embedded-agent-runner/runs.test.ts index cdfdf77814bc..c0129d54a197 100644 --- a/src/agents/embedded-agent-runner/runs.test.ts +++ b/src/agents/embedded-agent-runner/runs.test.ts @@ -16,6 +16,7 @@ import { resetDiagnosticSessionStateForTest, } from "../../logging/diagnostic-session-state.js"; import { diagnosticLogger } from "../../logging/diagnostic.js"; +import { createUserTurnTranscriptRecorder } from "../../sessions/user-turn-transcript.js"; import { MAX_TIMER_TIMEOUT_MS } from "../../shared/number-coercion.js"; import { testing, @@ -594,11 +595,15 @@ describe("embedded-agent runner run registry", () => { queueMessage, }); operation.setPhase("running"); + const recorder = createUserTurnTranscriptRecorder({ + input: { text: "visible group prompt", sender: { id: "user-42" } }, + target: { transcriptPath: "/tmp/unused-session.jsonl" }, + }); const outcome = await queueEmbeddedAgentMessageWithOutcomeAsync( "session-reply-run", "completion from child", - { waitForTranscriptCommit: true }, + { waitForTranscriptCommit: true, userTurnTranscriptRecorder: recorder }, ); expect(outcome.queued).toBe(true); @@ -613,7 +618,10 @@ describe("embedded-agent runner run registry", () => { }); expect(outcome.enqueuedAtMs).toEqual(expect.any(Number)); expect(outcome.deliveredAtMs).toBeUndefined(); - expect(queueMessage).toHaveBeenCalledWith("completion from child"); + expect(queueMessage).toHaveBeenCalledWith("completion from child", { + waitForTranscriptCommit: true, + userTurnTranscriptRecorder: recorder, + }); }); it("force-clears an aborted run that does not drain", async () => { diff --git a/src/agents/embedded-agent-runner/runs.ts b/src/agents/embedded-agent-runner/runs.ts index fe28aa2a8af7..96b3d4ec06a9 100644 --- a/src/agents/embedded-agent-runner/runs.ts +++ b/src/agents/embedded-agent-runner/runs.ts @@ -441,7 +441,7 @@ function prepareEmbeddedAgentQueueMessage( ): PreparedEmbeddedAgentQueueMessage { const handle = ACTIVE_EMBEDDED_RUNS.get(sessionId); if (!handle) { - const queuedReplyRunMessage = queueReplyRunMessage(sessionId, text); + const queuedReplyRunMessage = queueReplyRunMessage(sessionId, text, options); if (queuedReplyRunMessage) { logMessageQueued({ sessionId, source: "embedded-agent-runner" }); return { diff --git a/src/agents/session-tool-result-guard-wrapper.ts b/src/agents/session-tool-result-guard-wrapper.ts index c802fa4914a0..17e3c82dde20 100644 --- a/src/agents/session-tool-result-guard-wrapper.ts +++ b/src/agents/session-tool-result-guard-wrapper.ts @@ -9,9 +9,16 @@ import { applyInputProvenanceToUserMessage, type InputProvenance, } from "../sessions/input-provenance.js"; +import { + attachRuntimeUserTurnTranscriptRecorder, + takeRuntimeUserTurnTranscriptContext, + takeRuntimeUserTurnTranscriptRecorder, +} from "../sessions/user-turn-transcript-runtime-context.js"; import { mergePreparedUserTurnMessageForRuntime, + restorePreparedUserTurnOperationalMetaForRuntime, type PersistedUserTurnMessage, + type UserTurnTranscriptRecorder, } from "../sessions/user-turn-transcript.js"; import { resolveLiveToolResultMaxChars } from "./embedded-agent-runner/tool-result-truncation.js"; import type { AgentMessage } from "./runtime/index.js"; @@ -65,6 +72,7 @@ export function guardSessionManager( const hookRunner = getGlobalHookRunner(); let pendingPreparedUserTurnMessage = opts?.preparedUserTurnMessage; + let queuedUserTurnTranscriptRecorder: UserTurnTranscriptRecorder | undefined; const beforeMessageWrite = (event: { message: AgentMessage }) => { let message = event.message; let changed = false; @@ -74,10 +82,15 @@ export function guardSessionManager( sessionKey: opts?.sessionKey, }); if (result?.block) { + queuedUserTurnTranscriptRecorder?.markBlocked(); + queuedUserTurnTranscriptRecorder = undefined; return result; } if (result?.message) { - message = result.message; + message = restorePreparedUserTurnOperationalMetaForRuntime({ + runtimeMessage: result.message, + ...(event.message.role === "user" ? { preparedMessage: event.message } : {}), + }); changed = true; } } @@ -86,6 +99,14 @@ export function guardSessionManager( message = redacted; changed = true; } + if (message.role !== "user" && queuedUserTurnTranscriptRecorder) { + queuedUserTurnTranscriptRecorder.markBlocked(); + queuedUserTurnTranscriptRecorder = undefined; + } + if (message.role === "user" && queuedUserTurnTranscriptRecorder) { + message = attachRuntimeUserTurnTranscriptRecorder(message, queuedUserTurnTranscriptRecorder); + queuedUserTurnTranscriptRecorder = undefined; + } return changed ? { message } : undefined; }; @@ -116,14 +137,20 @@ export function guardSessionManager( sessionKey: opts?.sessionKey, agentId: opts?.agentId, transformMessageForPersistence: (message) => { + queuedUserTurnTranscriptRecorder = undefined; const withProvenance = applyInputProvenanceToUserMessage(message, opts?.inputProvenance); - const prepared = pendingPreparedUserTurnMessage; + const runtimeContext = takeRuntimeUserTurnTranscriptContext(message); + const prepared = runtimeContext?.message ?? pendingPreparedUserTurnMessage; const merged = mergePreparedUserTurnMessageForRuntime({ runtimeMessage: withProvenance, ...(prepared ? { preparedMessage: prepared } : {}), }); if (merged !== withProvenance) { - pendingPreparedUserTurnMessage = undefined; + if (runtimeContext) { + queuedUserTurnTranscriptRecorder = runtimeContext.recorder; + } else { + pendingPreparedUserTurnMessage = undefined; + } } return merged; }, @@ -146,7 +173,11 @@ export function guardSessionManager( suppressAssistantErrorPersistence: opts?.suppressAssistantErrorPersistence, onMessagePersisted: opts?.onMessagePersisted, withCompactionPersistence: opts?.withCompactionPersistence, - onUserMessagePersisted: opts?.onUserMessagePersisted, + onUserMessagePersisted: async (message) => { + const recorder = takeRuntimeUserTurnTranscriptRecorder(message); + recorder?.markRuntimePersisted(message); + await opts?.onUserMessagePersisted?.(message); + }, onUserMessageBlocked: opts?.onUserMessageBlocked, onAssistantErrorMessagePersisted: opts?.onAssistantErrorMessagePersisted, }); diff --git a/src/agents/sessions/agent-session.ts b/src/agents/sessions/agent-session.ts index ffd9bdfbcc98..9ac96fc3708f 100644 --- a/src/agents/sessions/agent-session.ts +++ b/src/agents/sessions/agent-session.ts @@ -34,6 +34,11 @@ import type { TextContent, } from "../../llm/types.js"; import { isRetryableAssistantError } from "../../llm/utils/retry.js"; +import { attachRuntimeUserTurnTranscriptContext } from "../../sessions/user-turn-transcript-runtime-context.js"; +import type { + PersistedUserTurnMessage, + UserTurnTranscriptRecorder, +} from "../../sessions/user-turn-transcript.types.js"; import type { Agent, AgentEvent, @@ -1361,9 +1366,14 @@ export class AgentSession { * before the next LLM call. * Expands skill commands and prompt templates. Errors on extension commands. * @param images Optional image attachments to include with the message + * @param userTurnTranscriptRecorder Prepared channel fields for transcript-only persistence * @throws Error if text is an extension command */ - async steer(text: string, images?: ImageContent[]): Promise { + async steer( + text: string, + images?: ImageContent[], + userTurnTranscriptRecorder?: UserTurnTranscriptRecorder, + ): Promise { // Check for extension commands (cannot be queued) if (text.startsWith("/")) { this.throwIfExtensionCommand(text); @@ -1373,7 +1383,14 @@ export class AgentSession { let expandedText = this.expandSkillCommand(text); expandedText = expandPromptTemplate(expandedText, [...this.promptTemplates]); - await this.queueSteer(expandedText, images); + const preparedMessage = await userTurnTranscriptRecorder?.resolveMessage(); + await this.queueSteer( + expandedText, + images, + preparedMessage && userTurnTranscriptRecorder + ? { message: preparedMessage, recorder: userTurnTranscriptRecorder } + : undefined, + ); } /** @@ -1399,18 +1416,30 @@ export class AgentSession { /** * Internal: Queue a steering message (already expanded, no extension command check). */ - private async queueSteer(text: string, images?: ImageContent[]): Promise { + private async queueSteer( + text: string, + images?: ImageContent[], + transcriptContext?: { + message: PersistedUserTurnMessage; + recorder: UserTurnTranscriptRecorder; + }, + ): Promise { this.steeringMessages.push(text); this.emitQueueUpdate(); const content: (TextContent | ImageContent)[] = [{ type: "text", text }]; if (images) { content.push(...images); } - this.agent.steer({ + const runtimeMessage = { role: "user", content, timestamp: Date.now(), - }); + } satisfies PersistedUserTurnMessage; + this.agent.steer( + transcriptContext + ? attachRuntimeUserTurnTranscriptContext(runtimeMessage, transcriptContext) + : runtimeMessage, + ); } /** diff --git a/src/agents/sessions/sdk.test.ts b/src/agents/sessions/sdk.test.ts index 9db3edb16d8d..e00274cce00d 100644 --- a/src/agents/sessions/sdk.test.ts +++ b/src/agents/sessions/sdk.test.ts @@ -3,6 +3,10 @@ import { Type } from "typebox"; import { beforeEach, describe, expect, it, vi } from "vitest"; import type { Context, Model, SimpleStreamOptions } from "../../llm/types.js"; +import { + createUserTurnTranscriptRecorder, + takeRuntimeUserTurnTranscriptContext, +} from "../../sessions/user-turn-transcript.js"; const thinkingMocks = vi.hoisted(() => ({ resolveThinkingDefaultForModel: vi.fn(() => "medium"), @@ -188,6 +192,39 @@ describe("AgentSession getLastAssistantText", () => { }); }); +describe("AgentSession queued user turns", () => { + it("carries prepared transcript context on the exact steered message", async () => { + const session = await createSessionFromManager(SessionManager.inMemory()); + const recorder = createUserTurnTranscriptRecorder({ + input: { + text: "visible group prompt", + sender: { id: "user-42", name: "Ada" }, + }, + target: { transcriptPath: "/tmp/unused-session.jsonl" }, + }); + const steer = vi.spyOn(session.agent, "steer").mockImplementation(() => undefined); + + await session.steer("runtime group prompt", undefined, recorder); + + const runtimeMessage = steer.mock.calls[0]?.[0]; + expect(runtimeMessage).toMatchObject({ + role: "user", + content: [{ type: "text", text: "runtime group prompt" }], + }); + if (!runtimeMessage) { + throw new Error("expected queued runtime message"); + } + expect(takeRuntimeUserTurnTranscriptContext(runtimeMessage)).toMatchObject({ + message: { + role: "user", + content: "visible group prompt", + __openclaw: { senderId: "user-42", senderName: "Ada" }, + }, + recorder, + }); + }); +}); + describe("createAgentSession attribution headers", () => { it("tolerates Bedrock models that do not expose baseUrl", async () => { const options = await createSessionAndStreamModel( diff --git a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts index b1a921bfda2b..61191c8e7c84 100644 --- a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts +++ b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts @@ -10,6 +10,7 @@ import { HEARTBEAT_RUN_SCOPE, type ReplyOptionsWithHeartbeatRunScope, } from "../../infra/heartbeat-run-scope.js"; +import { createUserTurnTranscriptRecorder } from "../../sessions/user-turn-transcript.js"; import type { TemplateContext } from "../templating.js"; import type { GetReplyOptions } from "../types.js"; import { @@ -135,8 +136,27 @@ vi.mock("../../agents/embedded-agent.js", () => ({ })); vi.mock("../../agents/embedded-agent-runner/runs.js", () => ({ - queueEmbeddedAgentMessage: (sessionId: string, prompt: string, options: unknown) => - state.queueEmbeddedAgentMessageMock(sessionId, prompt, options), + formatEmbeddedAgentQueueFailureSummary: () => "test queue rejection", + queueEmbeddedAgentMessageWithOutcomeAsync: async ( + sessionId: string, + prompt: string, + options: unknown, + ) => + state.queueEmbeddedAgentMessageMock(sessionId, prompt, options) + ? { + queued: true, + sessionId, + target: "embedded_run", + gatewayHealth: "live", + enqueuedAtMs: Date.now(), + } + : { + queued: false, + sessionId, + reason: "no_active_run", + target: "none", + gatewayHealth: "live", + }, })); vi.mock("./queue.js", () => ({ @@ -234,6 +254,7 @@ function createMinimalRun(params?: { } as unknown as FollowupRun; return { + followupRun, typing, opts, run: async () => { @@ -267,6 +288,37 @@ function createMinimalRun(params?: { }; } +describe("runReplyAgent active steering", () => { + it("carries the prepared user-turn recorder into the embedded queue", async () => { + state.queueEmbeddedAgentMessageMock.mockReturnValueOnce(true); + const recorder = createUserTurnTranscriptRecorder({ + input: { + text: "visible group prompt", + sender: { id: "user-42", name: "Ada" }, + }, + target: { transcriptPath: "/tmp/unused-session.jsonl" }, + }); + const { followupRun, run } = createMinimalRun({ + isActive: true, + isStreaming: true, + shouldSteer: true, + resolvedQueueMode: "steer", + }); + followupRun.userTurnTranscriptRecorder = recorder; + + await expect(run()).resolves.toBeUndefined(); + + expect(state.queueEmbeddedAgentMessageMock).toHaveBeenCalledWith( + "session", + "hello", + expect.objectContaining({ + steeringMode: "all", + userTurnTranscriptRecorder: recorder, + }), + ); + }); +}); + describe("runReplyAgent heartbeat followup guard", () => { it("drops heartbeat runs when reply-lane admission finds an active owner", async () => { const runState: ReplyOperationRunState = {}; diff --git a/src/auto-reply/reply/agent-runner.ts b/src/auto-reply/reply/agent-runner.ts index c7db86e97e42..35173699aa0a 100644 --- a/src/auto-reply/reply/agent-runner.ts +++ b/src/auto-reply/reply/agent-runner.ts @@ -1269,6 +1269,9 @@ export async function runReplyAgent(params: { { steeringMode: "all", ...(resolvedQueue.debounceMs !== undefined ? { debounceMs: resolvedQueue.debounceMs } : {}), + ...(followupRun.userTurnTranscriptRecorder + ? { userTurnTranscriptRecorder: followupRun.userTurnTranscriptRecorder } + : {}), }, ); if (steerOutcome.queued) { diff --git a/src/auto-reply/reply/get-reply-run.media-only.test.ts b/src/auto-reply/reply/get-reply-run.media-only.test.ts index a84663420786..b0eb0134ee26 100644 --- a/src/auto-reply/reply/get-reply-run.media-only.test.ts +++ b/src/auto-reply/reply/get-reply-run.media-only.test.ts @@ -1170,6 +1170,57 @@ describe("runPreparedReply media-only handling", () => { expect(call.followupRun.userTurnTranscriptRecorder?.message).not.toHaveProperty("MediaPaths"); }); + it.each([ + ["group", true], + ["channel", true], + ["direct", false], + ] as const)("persists sender attribution for %s turns only", async (chatType, shouldPersist) => { + await runPreparedReply( + baseParams({ + ctx: { + Body: "hello", + RawBody: "hello", + CommandBody: "hello", + OriginatingChannel: "telegram", + OriginatingTo: "chat-1", + ChatType: chatType, + }, + sessionCtx: { + Body: "hello", + BodyStripped: "hello", + Provider: "telegram", + OriginatingChannel: "telegram", + OriginatingTo: "chat-1", + ChatType: chatType, + SenderId: "user-42", + SenderName: "Ada", + SenderUsername: "ada", + }, + sessionEntry: { + sessionId: "session-1", + updatedAt: 1, + chatType, + channel: "telegram", + } as SessionEntry, + }), + ); + + const message = requireRunReplyAgentCall().followupRun.userTurnTranscriptRecorder?.message; + if (shouldPersist) { + expect(message).toMatchObject({ + __openclaw: { + senderId: "user-42", + senderName: "Ada", + senderUsername: "ada", + }, + }); + } else { + expect(message).not.toHaveProperty("__openclaw.senderId"); + expect(message).not.toHaveProperty("__openclaw.senderName"); + expect(message).not.toHaveProperty("__openclaw.senderUsername"); + } + }); + it("normalizes second-based inbound timestamps before preparing user turns", async () => { await runPreparedReply( baseParams({ @@ -2212,7 +2263,7 @@ describe("runPreparedReply media-only handling", () => { role: "user", content: "#35676 Keśava: No wtf", timestamp: expect.any(Number), - __openclaw: { senderIsOwner: false }, + __openclaw: { senderIsOwner: false, senderName: "Keśava" }, }); call?.followupRun.userTurnTranscriptRecorder?.markRuntimePersisted({ role: "user", diff --git a/src/auto-reply/reply/get-reply-run.ts b/src/auto-reply/reply/get-reply-run.ts index fab3e4bcd842..ab415566b7f2 100644 --- a/src/auto-reply/reply/get-reply-run.ts +++ b/src/auto-reply/reply/get-reply-run.ts @@ -1305,6 +1305,19 @@ export async function runPreparedReply( opts?.queuedFollowupLifecycle || inboundEventKind === "room_event" ? (opts?.queuedFollowupAbortSignal ?? opts?.abortSignal) : undefined; + const replyRoute = resolveEffectiveReplyRoute({ + ctx: { + Provider: ctx.Provider ?? sessionCtx.Provider, + Surface: ctx.Surface ?? sessionCtx.Surface, + OriginatingChannel: ctx.OriginatingChannel ?? sessionCtx.OriginatingChannel, + OriginatingTo: ctx.OriginatingTo ?? sessionCtx.OriginatingTo, + AccountId: ctx.AccountId ?? sessionCtx.AccountId, + InputProvenance: ctx.InputProvenance ?? sessionCtx.InputProvenance, + ChatType: ctx.ChatType ?? sessionCtx.ChatType, + }, + entry: preparedSessionState.sessionEntry, + }); + const persistGroupSender = replyRoute.chatType === "group" || replyRoute.chatType === "channel"; const userTurnMediaForPersistence = buildPersistedUserTurnMediaInputsFromFields(ctx); const inputProvenance = ctx.InputProvenance ?? sessionCtx.InputProvenance; const userTurnTimestamp = normalizeMessageTimestampMs(ctx.Timestamp); @@ -1329,6 +1342,14 @@ export async function runPreparedReply( // is identical whether this turn is sent as the current turn or // replayed as history. See: https://github.com/openclaw/openclaw/issues/3658 ...(userTurnTimestamp ? { timestamp: userTurnTimestamp } : {}), + // Direct transcripts keep their existing identity-storage boundary. + sender: persistGroupSender + ? { + id: normalizeOptionalString(sessionCtx.SenderId), + name: normalizeOptionalString(sessionCtx.SenderName), + username: normalizeOptionalString(sessionCtx.SenderUsername), + } + : undefined, } : undefined; const userTurnTranscriptRecorder = @@ -1359,18 +1380,6 @@ export async function runPreparedReply( : undefined, }) : undefined); - const replyRoute = resolveEffectiveReplyRoute({ - ctx: { - Provider: ctx.Provider ?? sessionCtx.Provider, - Surface: ctx.Surface ?? sessionCtx.Surface, - OriginatingChannel: ctx.OriginatingChannel ?? sessionCtx.OriginatingChannel, - OriginatingTo: ctx.OriginatingTo ?? sessionCtx.OriginatingTo, - AccountId: ctx.AccountId ?? sessionCtx.AccountId, - InputProvenance: ctx.InputProvenance ?? sessionCtx.InputProvenance, - ChatType: ctx.ChatType ?? sessionCtx.ChatType, - }, - entry: preparedSessionState.sessionEntry, - }); const messageProvider = resolveOriginMessageProvider({ originatingChannel: replyRoute.channel, // Prefer Provider over Surface for fallback channel identity. diff --git a/src/auto-reply/reply/reply-run-registry.ts b/src/auto-reply/reply/reply-run-registry.ts index b13b5199aff9..e7ae2eef028b 100644 --- a/src/auto-reply/reply/reply-run-registry.ts +++ b/src/auto-reply/reply/reply-run-registry.ts @@ -9,8 +9,10 @@ import { markDiagnosticEmbeddedRunEnded, markDiagnosticEmbeddedRunStarted, } from "../../logging/diagnostic-run-activity.js"; +import type { UserTurnTranscriptRecorder } from "../../sessions/user-turn-transcript.types.js"; import { resolveGlobalSingleton } from "../../shared/global-singleton.js"; import { resolveTimerTimeoutMs } from "../../shared/number-coercion.js"; +import type { SourceReplyDeliveryMode } from "../get-reply-options.types.js"; import type { ReplyFollowupAdmissionBarrierTimeoutPolicy } from "./reply-dispatcher.types.js"; export type ReplyRunKey = string; @@ -19,13 +21,23 @@ export type ReplyBackendKind = "embedded" | "cli"; export type ReplyBackendCancelReason = "user_abort" | "restart" | "superseded"; +export type ReplyBackendQueueMessageOptions = { + steeringMode?: "all"; + debounceMs?: number; + deliveryTimeoutMs?: number; + waitForTranscriptCommit?: boolean; + sourceReplyDeliveryMode?: SourceReplyDeliveryMode; + /** Prepared channel turn to merge only at transcript persistence. */ + userTurnTranscriptRecorder?: UserTurnTranscriptRecorder; +}; + export type ReplyBackendHandle = { readonly kind: ReplyBackendKind; cancel(reason?: ReplyBackendCancelReason): void; isStreaming(): boolean; isStopped?: () => boolean; isAbortable?: () => boolean; - queueMessage?: (text: string) => Promise; + queueMessage?: (text: string, options?: ReplyBackendQueueMessageOptions) => Promise; /** * Compatibility-only hook so legacy "abort compacting runs" paths can still * find embedded runs that are compacting during the main run phase. @@ -811,7 +823,11 @@ export function isReplyRunStreamingForSessionId(sessionId: string): boolean { return getAttachedBackend(operation)?.isStreaming() ?? false; } -export function queueReplyRunMessage(sessionId: string, text: string): boolean { +export function queueReplyRunMessage( + sessionId: string, + text: string, + options?: ReplyBackendQueueMessageOptions, +): boolean { const operation = resolveReplyRunForCurrentSessionId(sessionId); const backend = operation ? getAttachedBackend(operation) : undefined; if (!operation || operation.phase !== "running" || !backend?.queueMessage) { @@ -820,7 +836,7 @@ export function queueReplyRunMessage(sessionId: string, text: string): boolean { if (!isReplyBackendMessageInjectable(backend)) { return false; } - void backend.queueMessage(text); + void (options ? backend.queueMessage(text, options) : backend.queueMessage(text)); return true; } diff --git a/src/sessions/user-turn-transcript-runtime-context.ts b/src/sessions/user-turn-transcript-runtime-context.ts new file mode 100644 index 000000000000..e2fda63c8458 --- /dev/null +++ b/src/sessions/user-turn-transcript-runtime-context.ts @@ -0,0 +1,69 @@ +// Transient user-turn transcript context carried through runtime queues. +import type { AgentMessage } from "../../packages/agent-core/src/types.js"; +import type { + PersistedUserTurnMessage, + UserTurnTranscriptRecorder, +} from "./user-turn-transcript.types.js"; + +const RUNTIME_USER_TURN_TRANSCRIPT_CONTEXT = Symbol.for( + "openclaw.runtimeUserTurnTranscriptContext", +); +const RUNTIME_USER_TURN_TRANSCRIPT_RECORDER = Symbol.for( + "openclaw.runtimeUserTurnTranscriptRecorder", +); + +export type RuntimeUserTurnTranscriptContext = { + message: PersistedUserTurnMessage; + recorder: UserTurnTranscriptRecorder; +}; + +/** Carries transcript-only fields with a queued runtime message without exposing them to the model. */ +export function attachRuntimeUserTurnTranscriptContext( + runtimeMessage: PersistedUserTurnMessage, + context: RuntimeUserTurnTranscriptContext, +): PersistedUserTurnMessage { + Object.defineProperty(runtimeMessage, RUNTIME_USER_TURN_TRANSCRIPT_CONTEXT, { + configurable: true, + value: context, + }); + return runtimeMessage; +} + +/** Consumes the transient queued-turn context before the message is serialized. */ +export function takeRuntimeUserTurnTranscriptContext( + runtimeMessage: AgentMessage, +): RuntimeUserTurnTranscriptContext | undefined { + const record = runtimeMessage as unknown as Record; + const context = record[RUNTIME_USER_TURN_TRANSCRIPT_CONTEXT] as + | RuntimeUserTurnTranscriptContext + | undefined; + if (context) { + delete record[RUNTIME_USER_TURN_TRANSCRIPT_CONTEXT]; + } + return context; +} + +/** Keeps the queued recorder attached to the exact final message until persistence succeeds. */ +export function attachRuntimeUserTurnTranscriptRecorder( + runtimeMessage: AgentMessage, + recorder: UserTurnTranscriptRecorder, +): AgentMessage { + Object.defineProperty(runtimeMessage, RUNTIME_USER_TURN_TRANSCRIPT_RECORDER, { + configurable: true, + value: recorder, + }); + return runtimeMessage; +} + +export function takeRuntimeUserTurnTranscriptRecorder( + runtimeMessage: AgentMessage, +): UserTurnTranscriptRecorder | undefined { + const record = runtimeMessage as unknown as Record; + const recorder = record[RUNTIME_USER_TURN_TRANSCRIPT_RECORDER] as + | UserTurnTranscriptRecorder + | undefined; + if (recorder) { + delete record[RUNTIME_USER_TURN_TRANSCRIPT_RECORDER]; + } + return recorder; +} diff --git a/src/sessions/user-turn-transcript.test.ts b/src/sessions/user-turn-transcript.test.ts index 06eea2204997..009d0abd88ba 100644 --- a/src/sessions/user-turn-transcript.test.ts +++ b/src/sessions/user-turn-transcript.test.ts @@ -201,6 +201,33 @@ describe("user turn transcript persistence", () => { }); }); + it("preserves runtime metadata when adding prepared sender attribution", () => { + const recorder = createUserTurnTranscriptRecorder({ + input: { + text: "group prompt", + sender: { id: "user-42", name: "Ada" }, + }, + target: { transcriptPath: "/tmp/session.jsonl" }, + }); + + expect( + mergePreparedUserTurnMessageForRuntime({ + runtimeMessage: castAgentMessage({ + role: "user", + content: "runtime prompt", + __openclaw: { mirrorIdentity: "run-1:prompt" }, + }), + preparedMessage: recorder.message, + }), + ).toMatchObject({ + __openclaw: { + mirrorIdentity: "run-1:prompt", + senderId: "user-42", + senderName: "Ada", + }, + }); + }); + it("does not replace blocked before_agent_run user markers", () => { const recorder = createUserTurnTranscriptRecorder({ input: { text: "raw prompt" }, @@ -320,6 +347,67 @@ describe("user turn transcript persistence", () => { ]); }); + it("persists sender metadata as __openclaw envelope", async () => { + const dir = createTempDir("openclaw-user-turn-append-sender-"); + const transcriptPath = path.join(dir, "session.jsonl"); + + const appended = await appendUserTurnTranscriptMessage({ + transcriptPath, + sessionId: "session-1", + sessionKey: "main", + cwd: dir, + input: { + text: "hello from group", + sender: { + id: "8489979671", + name: "Ram Shenoy", + username: "ram_s", + }, + }, + updateMode: "none", + }); + + expect(appended?.message).toMatchObject({ + role: "user", + content: "hello from group", + __openclaw: { + senderId: "8489979671", + senderName: "Ram Shenoy", + senderUsername: "ram_s", + }, + }); + expect(readTranscriptMessages(transcriptPath)).toEqual([ + expect.objectContaining({ + role: "user", + content: "hello from group", + __openclaw: { + senderId: "8489979671", + senderName: "Ram Shenoy", + senderUsername: "ram_s", + }, + }), + ]); + }); + + it("omits __openclaw when no sender metadata is provided", async () => { + const dir = createTempDir("openclaw-user-turn-append-nosender-"); + const transcriptPath = path.join(dir, "session.jsonl"); + + const appended = await appendUserTurnTranscriptMessage({ + transcriptPath, + sessionId: "session-1", + sessionKey: "main", + cwd: dir, + input: { + text: "hello without sender", + sender: { id: "", name: null }, + }, + updateMode: "none", + }); + + expect(appended?.message).not.toHaveProperty("__openclaw"); + }); + it("uses inline update mode by default", async () => { const dir = createTempDir("openclaw-user-turn-append-inline-"); const transcriptPath = path.join(dir, "session.jsonl"); @@ -394,7 +482,7 @@ describe("user turn transcript persistence", () => { ]); }); - it("preserves idempotency keys when before_message_write replaces a user turn", async () => { + it("preserves transcript metadata when before_message_write replaces a user turn", async () => { let hookCalls = 0; const provenance = { kind: "inter_session" as const, @@ -411,6 +499,7 @@ describe("user turn transcript persistence", () => { message: castAgentMessage({ role: "user", content: "[redacted by hook]", + __openclaw: { hookOwned: true }, }), }; }, @@ -427,6 +516,7 @@ describe("user turn transcript persistence", () => { idempotencyKey: "chat-run-1:user", senderIsOwner: true, provenance, + sender: { id: "user-42", name: "Ada" }, }, beforeMessageWrite: runAgentHarnessBeforeMessageWriteHook, }); @@ -437,6 +527,7 @@ describe("user turn transcript persistence", () => { idempotencyKey: "chat-run-1:user", senderIsOwner: true, provenance, + sender: { id: "user-42", name: "Ada" }, }, beforeMessageWrite: runAgentHarnessBeforeMessageWriteHook, }); @@ -446,8 +537,11 @@ describe("user turn transcript persistence", () => { role: "user", content: "[redacted by hook]", idempotencyKey: "chat-run-1:user", - __openclaw: { senderIsOwner: true }, provenance, + __openclaw: { + hookOwned: true, + senderIsOwner: true, + }, }), ]); expect(hookCalls).toBe(1); diff --git a/src/sessions/user-turn-transcript.ts b/src/sessions/user-turn-transcript.ts index 7287f265c22c..15f8d8ee9dd0 100644 --- a/src/sessions/user-turn-transcript.ts +++ b/src/sessions/user-turn-transcript.ts @@ -24,6 +24,13 @@ export type { UserTurnInput, UserTurnTranscriptRecorder, } from "./user-turn-transcript.types.js"; +export { + attachRuntimeUserTurnTranscriptContext, + attachRuntimeUserTurnTranscriptRecorder, + takeRuntimeUserTurnTranscriptContext, + takeRuntimeUserTurnTranscriptRecorder, + type RuntimeUserTurnTranscriptContext, +} from "./user-turn-transcript-runtime-context.js"; type PersistedUserTurnMediaFields = { MediaPath?: string; @@ -228,6 +235,29 @@ function buildPersistedUserTurnMediaFields( }; } +function buildUserTurnSenderMeta( + sender: UserTurnInput["sender"], +): Record | undefined { + const senderId = normalizeOptionalText(sender?.id); + const senderName = normalizeOptionalText(sender?.name); + const senderUsername = normalizeOptionalText(sender?.username); + if (!senderId && !senderName && !senderUsername) { + return undefined; + } + return { + ...(senderId ? { senderId } : {}), + ...(senderName ? { senderName } : {}), + ...(senderUsername ? { senderUsername } : {}), + }; +} + +function readOpenClawMessageMeta(message: AgentMessage): Record | undefined { + const meta = (message as unknown as Record)["__openclaw"]; + return meta && typeof meta === "object" && !Array.isArray(meta) + ? (meta as Record) + : undefined; +} + function buildPersistedUserTurnMessage(params: UserTurnInput): PersistedUserTurnMessage { const mediaFields = buildPersistedUserTurnMediaFields(params.media); const hasMedia = Boolean(mediaFields.MediaPath); @@ -239,16 +269,18 @@ function buildPersistedUserTurnMessage(params: UserTurnInput): PersistedUserTurn // here would NOT match the bare-current arrival (the gateway no longer stamps // the live turn) — see https://github.com/openclaw/openclaw/issues/3658. const content = text || (hasMedia ? (params.mediaOnlyText ?? "") : ""); - + const senderMeta = buildUserTurnSenderMeta(params.sender); + const openClawMeta = { + ...(params.senderIsOwner === undefined ? {} : { senderIsOwner: params.senderIsOwner }), + ...senderMeta, + }; const message = { role: "user", content, timestamp: params.timestamp ?? Date.now(), ...(params.idempotencyKey ? { idempotencyKey: params.idempotencyKey } : {}), - ...(params.senderIsOwner === undefined - ? {} - : { __openclaw: { senderIsOwner: params.senderIsOwner } }), ...mediaFields, + ...(Object.keys(openClawMeta).length > 0 ? { __openclaw: openClawMeta } : {}), } as PersistedUserTurnMessage; return applyInputProvenanceToUserMessage(message, params.provenance) as PersistedUserTurnMessage; } @@ -301,15 +333,39 @@ export function mergePreparedUserTurnMessageForRuntime(params: { ) { return params.runtimeMessage; } + const runtimeMessage = params.runtimeMessage as unknown as Record; + const preparedMessage = params.preparedMessage as unknown as Record; + const runtimeMeta = readOpenClawMessageMeta(params.runtimeMessage); + const preparedMeta = readOpenClawMessageMeta(params.preparedMessage); return { - ...(params.runtimeMessage as unknown as Record), - ...(params.preparedMessage as unknown as Record), + ...runtimeMessage, + ...preparedMessage, + ...(preparedMeta ? { __openclaw: { ...runtimeMeta, ...preparedMeta } } : {}), ...(userMessageHasImageContent(params.runtimeMessage) ? { content: params.runtimeMessage.content } : {}), } as unknown as AgentMessage; } +/** Restores only auth state that write hooks must not be able to forge or erase. */ +export function restorePreparedUserTurnOperationalMetaForRuntime(params: { + runtimeMessage: AgentMessage; + preparedMessage?: PersistedUserTurnMessage; +}): AgentMessage { + if (!params.preparedMessage || !isUserMessage(params.runtimeMessage)) { + return params.runtimeMessage; + } + const preparedMeta = readOpenClawMessageMeta(params.preparedMessage); + const senderIsOwner = preparedMeta?.senderIsOwner; + if (typeof senderIsOwner !== "boolean") { + return params.runtimeMessage; + } + return { + ...(params.runtimeMessage as unknown as Record), + __openclaw: { ...readOpenClawMessageMeta(params.runtimeMessage), senderIsOwner }, + } as unknown as AgentMessage; +} + /** Applies before-message hooks while preserving user-turn transcript metadata. */ export function preparePersistedUserTurnMessageForTranscriptWrite( message: PersistedUserTurnMessage, @@ -327,6 +383,7 @@ export function preparePersistedUserTurnMessageForTranscriptWrite( const provenance = normalizeInputProvenance( (message as unknown as { provenance?: unknown }).provenance, ); + const senderIsOwner = readOpenClawMessageMeta(message)?.senderIsOwner; const nextMessage = params.beforeMessageWrite({ message, ...(params.agentId ? { agentId: params.agentId } : {}), @@ -338,24 +395,19 @@ export function preparePersistedUserTurnMessageForTranscriptWrite( const nextUserMessage = provenance ? (applyInputProvenanceToUserMessage(nextMessage, provenance) as PersistedUserTurnMessage) : nextMessage; - const originalOpenClaw = (message as unknown as { __openclaw?: unknown })["__openclaw"]; - const senderIsOwner = - originalOpenClaw && typeof originalOpenClaw === "object" - ? (originalOpenClaw as { senderIsOwner?: unknown }).senderIsOwner - : undefined; if (!idempotencyKey && typeof senderIsOwner !== "boolean") { return nextUserMessage; } - const nextRecord = nextUserMessage as unknown as Record; - const nextOpenClaw = - nextRecord["__openclaw"] && typeof nextRecord["__openclaw"] === "object" - ? (nextRecord["__openclaw"] as Record) - : {}; return { - ...nextRecord, + ...(nextUserMessage as unknown as Record), ...(idempotencyKey ? { idempotencyKey } : {}), ...(typeof senderIsOwner === "boolean" - ? { __openclaw: { ...nextOpenClaw, senderIsOwner } } + ? { + __openclaw: { + ...readOpenClawMessageMeta(nextUserMessage), + senderIsOwner, + }, + } : {}), } as unknown as PersistedUserTurnMessage; } diff --git a/src/sessions/user-turn-transcript.types.ts b/src/sessions/user-turn-transcript.types.ts index e468817daa92..bf956e896797 100644 --- a/src/sessions/user-turn-transcript.types.ts +++ b/src/sessions/user-turn-transcript.types.ts @@ -26,6 +26,8 @@ export type UserTurnInput = { senderIsOwner?: boolean; provenance?: InputProvenance; mediaOnlyText?: string; + /** Durable participant attribution. Callers must opt in at the product boundary. */ + sender?: { id?: string | null; name?: string | null; username?: string | null } | null; }; export type UserTurnTranscriptUpdateMode = "inline" | "none";