mirror of
https://github.com/openclaw/openclaw.git
synced 2026-07-09 06:43:56 +00:00
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 <noreply@anthropic.com> * fix(infra): use ...base spread instead of ...baseWithoutMode The baseWithoutMode destructuring produces Omit<TBase, "mode"> which is not assignable to TBase & ExecPolicyLayer when the as-cast is removed. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> * 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 <noreply@anthropic.com> * 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 <noreply@anthropic.com> Co-authored-by: Peter Steinberger <steipete@gmail.com>
This commit is contained in:
pick-cat
@@ -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<AgentMessage, { role: "user" }>,
|
||||
});
|
||||
|
||||
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.
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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<void>;
|
||||
steer(
|
||||
text: string,
|
||||
images?: undefined,
|
||||
userTurnTranscriptRecorder?: UserTurnTranscriptRecorder,
|
||||
): Promise<void>;
|
||||
subscribe(listener: (event: unknown) => void): () => void;
|
||||
};
|
||||
|
||||
@@ -126,6 +131,7 @@ export async function steerAndWaitForTranscriptCommit(
|
||||
activeSession: EmbeddedAgentActiveSessionSteerTarget,
|
||||
text: string,
|
||||
timeoutMs: number,
|
||||
userTurnTranscriptRecorder?: UserTurnTranscriptRecorder,
|
||||
): Promise<void> {
|
||||
await new Promise<void>((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<void> {
|
||||
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,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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,
|
||||
});
|
||||
|
||||
@@ -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<void> {
|
||||
async steer(
|
||||
text: string,
|
||||
images?: ImageContent[],
|
||||
userTurnTranscriptRecorder?: UserTurnTranscriptRecorder,
|
||||
): Promise<void> {
|
||||
// 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<void> {
|
||||
private async queueSteer(
|
||||
text: string,
|
||||
images?: ImageContent[],
|
||||
transcriptContext?: {
|
||||
message: PersistedUserTurnMessage;
|
||||
recorder: UserTurnTranscriptRecorder;
|
||||
},
|
||||
): Promise<void> {
|
||||
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,
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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 = {};
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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<void>;
|
||||
queueMessage?: (text: string, options?: ReplyBackendQueueMessageOptions) => Promise<void>;
|
||||
/**
|
||||
* 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;
|
||||
}
|
||||
|
||||
|
||||
69
src/sessions/user-turn-transcript-runtime-context.ts
Normal file
69
src/sessions/user-turn-transcript-runtime-context.ts
Normal file
@@ -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<PropertyKey, unknown>;
|
||||
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<PropertyKey, unknown>;
|
||||
const recorder = record[RUNTIME_USER_TURN_TRANSCRIPT_RECORDER] as
|
||||
| UserTurnTranscriptRecorder
|
||||
| undefined;
|
||||
if (recorder) {
|
||||
delete record[RUNTIME_USER_TURN_TRANSCRIPT_RECORDER];
|
||||
}
|
||||
return recorder;
|
||||
}
|
||||
@@ -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);
|
||||
|
||||
@@ -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<string, string> | 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<string, unknown> | undefined {
|
||||
const meta = (message as unknown as Record<string, unknown>)["__openclaw"];
|
||||
return meta && typeof meta === "object" && !Array.isArray(meta)
|
||||
? (meta as Record<string, unknown>)
|
||||
: 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<string, unknown>;
|
||||
const preparedMessage = params.preparedMessage as unknown as Record<string, unknown>;
|
||||
const runtimeMeta = readOpenClawMessageMeta(params.runtimeMessage);
|
||||
const preparedMeta = readOpenClawMessageMeta(params.preparedMessage);
|
||||
return {
|
||||
...(params.runtimeMessage as unknown as Record<string, unknown>),
|
||||
...(params.preparedMessage as unknown as Record<string, unknown>),
|
||||
...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<string, unknown>),
|
||||
__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<string, unknown>;
|
||||
const nextOpenClaw =
|
||||
nextRecord["__openclaw"] && typeof nextRecord["__openclaw"] === "object"
|
||||
? (nextRecord["__openclaw"] as Record<string, unknown>)
|
||||
: {};
|
||||
return {
|
||||
...nextRecord,
|
||||
...(nextUserMessage as unknown as Record<string, unknown>),
|
||||
...(idempotencyKey ? { idempotencyKey } : {}),
|
||||
...(typeof senderIsOwner === "boolean"
|
||||
? { __openclaw: { ...nextOpenClaw, senderIsOwner } }
|
||||
? {
|
||||
__openclaw: {
|
||||
...readOpenClawMessageMeta(nextUserMessage),
|
||||
senderIsOwner,
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
} as unknown as PersistedUserTurnMessage;
|
||||
}
|
||||
|
||||
@@ -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";
|
||||
|
||||
Reference in New Issue
Block a user