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";