Skip to content

Commit f296dc4

Browse files
Pick-catclaudesteipete
authored
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 <[email protected]> * 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 <[email protected]> * 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 <[email protected]> * 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 <[email protected]> Co-authored-by: Peter Steinberger <[email protected]>
1 parent 6acc027 commit f296dc4

18 files changed

Lines changed: 668 additions & 62 deletions

‎src/agents/embedded-agent-runner.guard.test.ts‎

Lines changed: 119 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,23 @@
11
// Covers session-manager guard behavior for tool-result pairing and transcript
22
// redaction.
3+
import { readFileSync } from "node:fs";
34
import type { AgentMessage } from "openclaw/plugin-sdk/agent-core";
45
import { SessionManager } from "openclaw/plugin-sdk/agent-sessions";
5-
import { describe, expect, it } from "vitest";
6+
import {
7+
initializeGlobalHookRunner,
8+
resetGlobalHookRunner,
9+
} from "openclaw/plugin-sdk/hook-runtime";
10+
import { createMockPluginRegistry } from "openclaw/plugin-sdk/plugin-test-runtime";
11+
import { afterEach, describe, expect, it } from "vitest";
12+
import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js";
613
import type { OpenClawConfig } from "../config/types.openclaw.js";
14+
import {
15+
attachRuntimeUserTurnTranscriptContext,
16+
createUserTurnTranscriptRecorder,
17+
} from "../sessions/user-turn-transcript.js";
718
import { guardSessionManager } from "./session-tool-result-guard-wrapper.js";
819
import { sanitizeToolUseResultPairing } from "./session-transcript-repair.js";
20+
import { makeAgentAssistantMessage } from "./test-helpers/agent-message-fixtures.js";
921

1022
function assistantToolCall(id: string): AgentMessage {
1123
return {
@@ -15,6 +27,12 @@ function assistantToolCall(id: string): AgentMessage {
1527
}
1628

1729
describe("guardSessionManager integration", () => {
30+
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
31+
32+
afterEach(() => {
33+
resetGlobalHookRunner();
34+
});
35+
1836
it("persists synthetic toolResult before subsequent assistant message", () => {
1937
// Providers require every assistant tool call to be followed by a result
2038
// before the next assistant turn.
@@ -140,6 +158,106 @@ describe("guardSessionManager integration", () => {
140158
expect(messages[1]).toEqual({ role: "user", content: "follow-up" });
141159
});
142160

161+
it("lets a write hook remove sender identity while preserving auth state", () => {
162+
initializeGlobalHookRunner(
163+
createMockPluginRegistry([
164+
{
165+
hookName: "before_message_write",
166+
handler: () => ({
167+
message: {
168+
role: "user",
169+
content: "[redacted by hook]",
170+
timestamp: 124,
171+
__openclaw: { hookOwned: true },
172+
} as AgentMessage,
173+
}),
174+
},
175+
]),
176+
);
177+
const sm = guardSessionManager(SessionManager.inMemory(), {
178+
preparedUserTurnMessage: {
179+
role: "user",
180+
content: "private group prompt",
181+
timestamp: 123,
182+
__openclaw: {
183+
senderIsOwner: true,
184+
senderId: "secret-user",
185+
senderName: "secret-name",
186+
},
187+
} as Extract<AgentMessage, { role: "user" }>,
188+
});
189+
190+
sm.appendMessage({ role: "user", content: "runtime prompt", timestamp: 125 });
191+
192+
const message = sm.getEntries().find((entry) => entry.type === "message") as
193+
| { message?: AgentMessage }
194+
| undefined;
195+
expect(message?.message).toMatchObject({
196+
role: "user",
197+
content: "[redacted by hook]",
198+
__openclaw: {
199+
hookOwned: true,
200+
senderIsOwner: true,
201+
},
202+
});
203+
expect(JSON.stringify(message?.message)).not.toContain("secret-user");
204+
expect(JSON.stringify(message?.message)).not.toContain("secret-name");
205+
});
206+
207+
it("commits queued group sender metadata to JSONL and completes its recorder", () => {
208+
const dir = tempDirs.make("openclaw-queued-group-turn-");
209+
const sessionManager = SessionManager.create(dir, dir);
210+
const sessionFile = sessionManager.getSessionFile();
211+
if (!sessionFile) {
212+
throw new Error("expected file-backed session manager");
213+
}
214+
const recorder = createUserTurnTranscriptRecorder({
215+
input: {
216+
text: "visible group prompt",
217+
sender: { id: "user-42", name: "Ada", username: "ada42" },
218+
},
219+
target: { transcriptPath: sessionFile },
220+
});
221+
const preparedMessage = recorder.message;
222+
if (!preparedMessage) {
223+
throw new Error("expected prepared group turn");
224+
}
225+
const sm = guardSessionManager(sessionManager, {
226+
inputProvenance: { kind: "inter_session", sourceTool: "sessions_send" },
227+
});
228+
const runtimeMessage = attachRuntimeUserTurnTranscriptContext(
229+
{
230+
role: "user",
231+
content: [{ type: "text", text: "runtime group prompt" }],
232+
timestamp: 456,
233+
},
234+
{ message: preparedMessage, recorder },
235+
);
236+
237+
sm.appendMessage(runtimeMessage);
238+
sm.appendMessage(
239+
makeAgentAssistantMessage({
240+
content: [{ type: "text", text: "acknowledged" }],
241+
}),
242+
);
243+
244+
const entries = readFileSync(sessionFile, "utf8")
245+
.trim()
246+
.split("\n")
247+
.map((line) => JSON.parse(line) as { type: string; message?: AgentMessage });
248+
expect(entries.find((entry) => entry.message?.role === "user")?.message).toMatchObject({
249+
role: "user",
250+
content: "visible group prompt",
251+
__openclaw: {
252+
senderId: "user-42",
253+
senderName: "Ada",
254+
senderUsername: "ada42",
255+
},
256+
provenance: { kind: "inter_session", sourceTool: "sessions_send" },
257+
});
258+
expect(recorder.hasPersisted()).toBe(true);
259+
});
260+
143261
it("does not consume prepared user persistence for before-agent-run blocked messages", () => {
144262
// Blocked messages are audit records, not the actual user turn that should
145263
// receive prepared media metadata.

‎src/agents/embedded-agent-runner/run-state.ts‎

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import {
77
listActiveReplyRunSessionKeys,
88
listActiveReplyRunSessionIds,
99
resolveActiveReplyRunSessionId,
10+
type ReplyBackendQueueMessageOptions,
1011
} from "../../auto-reply/reply/reply-run-registry.js";
1112
import { resolveGlobalSingleton } from "../../shared/global-singleton.js";
1213

@@ -30,13 +31,7 @@ export type EmbeddedAgentQueueHandle = {
3031
sourceReplyDeliveryMode?: SourceReplyDeliveryMode;
3132
};
3233

33-
export type EmbeddedAgentQueueMessageOptions = {
34-
steeringMode?: "all";
35-
debounceMs?: number;
36-
deliveryTimeoutMs?: number;
37-
waitForTranscriptCommit?: boolean;
38-
sourceReplyDeliveryMode?: SourceReplyDeliveryMode;
39-
};
34+
export type EmbeddedAgentQueueMessageOptions = ReplyBackendQueueMessageOptions;
4035

4136
export type ActiveEmbeddedRunSnapshot = {
4237
transcriptLeafId: string | null;

‎src/agents/embedded-agent-runner/run/attempt.queue-message.test.ts‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,32 @@
11
// Coverage for queued steering message commit and cancellation behavior.
22
import { describe, expect, it, vi } from "vitest";
3+
import { createUserTurnTranscriptRecorder } from "../../../sessions/user-turn-transcript.js";
34
import {
45
cancelQueuedSteeringMessage,
6+
steerActiveSessionWithOptionalDeliveryWait,
57
steerAndWaitForTranscriptCommit,
68
type EmbeddedAgentActiveSessionSteerTarget,
79
} from "./attempt.queue-message.js";
810

911
describe("embedded OpenClaw queued steering cancellation", () => {
12+
it("forwards prepared transcript context with a queued steering message", async () => {
13+
const steer = vi.fn(async () => undefined);
14+
const recorder = createUserTurnTranscriptRecorder({
15+
input: { text: "visible prompt", sender: { id: "user-42" } },
16+
target: { transcriptPath: "/tmp/unused-session.jsonl" },
17+
});
18+
const activeSession: EmbeddedAgentActiveSessionSteerTarget = {
19+
steer,
20+
subscribe: () => () => {},
21+
};
22+
23+
await steerActiveSessionWithOptionalDeliveryWait(activeSession, "runtime prompt", {
24+
userTurnTranscriptRecorder: recorder,
25+
});
26+
27+
expect(steer).toHaveBeenCalledWith("runtime prompt", undefined, recorder);
28+
});
29+
1030
it("waits for the queued user message_end transcript boundary", async () => {
1131
// A queued steer is only durable once the user message_end event lands in
1232
// the active transcript.

‎src/agents/embedded-agent-runner/run/attempt.queue-message.ts‎

Lines changed: 24 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
* Steers active embedded sessions and waits for transcript commits when needed.
33
*/
44
import { toErrorObject } from "../../../infra/errors.js";
5+
import type { UserTurnTranscriptRecorder } from "../../../sessions/user-turn-transcript.types.js";
56
import { log } from "../logger.js";
67

78
/**
@@ -11,7 +12,11 @@ import { log } from "../logger.js";
1112
export type EmbeddedAgentActiveSessionSteerTarget = {
1213
agent?: unknown;
1314
getSteeringMessages?(): readonly string[];
14-
steer(text: string): Promise<void>;
15+
steer(
16+
text: string,
17+
images?: undefined,
18+
userTurnTranscriptRecorder?: UserTurnTranscriptRecorder,
19+
): Promise<void>;
1520
subscribe(listener: (event: unknown) => void): () => void;
1621
};
1722

@@ -126,6 +131,7 @@ export async function steerAndWaitForTranscriptCommit(
126131
activeSession: EmbeddedAgentActiveSessionSteerTarget,
127132
text: string,
128133
timeoutMs: number,
134+
userTurnTranscriptRecorder?: UserTurnTranscriptRecorder,
129135
): Promise<void> {
130136
await new Promise<void>((resolve, reject) => {
131137
let settled = false;
@@ -206,7 +212,10 @@ export async function steerAndWaitForTranscriptCommit(
206212
scheduleTerminalCancellation();
207213
}
208214
});
209-
activeSession.steer(text).catch((err: unknown) => {
215+
const steer = userTurnTranscriptRecorder
216+
? activeSession.steer(text, undefined, userTurnTranscriptRecorder)
217+
: activeSession.steer(text);
218+
steer.catch((err: unknown) => {
210219
finish(err);
211220
});
212221
});
@@ -219,15 +228,26 @@ export async function steerAndWaitForTranscriptCommit(
219228
export async function steerActiveSessionWithOptionalDeliveryWait(
220229
activeSession: EmbeddedAgentActiveSessionSteerTarget,
221230
text: string,
222-
options: { deliveryTimeoutMs?: number; waitForTranscriptCommit?: boolean } | undefined,
231+
options:
232+
| {
233+
deliveryTimeoutMs?: number;
234+
waitForTranscriptCommit?: boolean;
235+
userTurnTranscriptRecorder?: UserTurnTranscriptRecorder;
236+
}
237+
| undefined,
223238
): Promise<void> {
224239
if (options?.waitForTranscriptCommit !== true) {
225-
await activeSession.steer(text);
240+
if (options?.userTurnTranscriptRecorder) {
241+
await activeSession.steer(text, undefined, options.userTurnTranscriptRecorder);
242+
} else {
243+
await activeSession.steer(text);
244+
}
226245
return;
227246
}
228247
await steerAndWaitForTranscriptCommit(
229248
activeSession,
230249
text,
231250
options.deliveryTimeoutMs ?? DEFAULT_QUEUE_TRANSCRIPT_COMMIT_TIMEOUT_MS,
251+
options.userTurnTranscriptRecorder,
232252
);
233253
}

‎src/agents/embedded-agent-runner/runs.test.ts‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ import {
1616
resetDiagnosticSessionStateForTest,
1717
} from "../../logging/diagnostic-session-state.js";
1818
import { diagnosticLogger } from "../../logging/diagnostic.js";
19+
import { createUserTurnTranscriptRecorder } from "../../sessions/user-turn-transcript.js";
1920
import { MAX_TIMER_TIMEOUT_MS } from "../../shared/number-coercion.js";
2021
import {
2122
testing,
@@ -594,11 +595,15 @@ describe("embedded-agent runner run registry", () => {
594595
queueMessage,
595596
});
596597
operation.setPhase("running");
598+
const recorder = createUserTurnTranscriptRecorder({
599+
input: { text: "visible group prompt", sender: { id: "user-42" } },
600+
target: { transcriptPath: "/tmp/unused-session.jsonl" },
601+
});
597602

598603
const outcome = await queueEmbeddedAgentMessageWithOutcomeAsync(
599604
"session-reply-run",
600605
"completion from child",
601-
{ waitForTranscriptCommit: true },
606+
{ waitForTranscriptCommit: true, userTurnTranscriptRecorder: recorder },
602607
);
603608

604609
expect(outcome.queued).toBe(true);
@@ -613,7 +618,10 @@ describe("embedded-agent runner run registry", () => {
613618
});
614619
expect(outcome.enqueuedAtMs).toEqual(expect.any(Number));
615620
expect(outcome.deliveredAtMs).toBeUndefined();
616-
expect(queueMessage).toHaveBeenCalledWith("completion from child");
621+
expect(queueMessage).toHaveBeenCalledWith("completion from child", {
622+
waitForTranscriptCommit: true,
623+
userTurnTranscriptRecorder: recorder,
624+
});
617625
});
618626

619627
it("force-clears an aborted run that does not drain", async () => {

‎src/agents/embedded-agent-runner/runs.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -441,7 +441,7 @@ function prepareEmbeddedAgentQueueMessage(
441441
): PreparedEmbeddedAgentQueueMessage {
442442
const handle = ACTIVE_EMBEDDED_RUNS.get(sessionId);
443443
if (!handle) {
444-
const queuedReplyRunMessage = queueReplyRunMessage(sessionId, text);
444+
const queuedReplyRunMessage = queueReplyRunMessage(sessionId, text, options);
445445
if (queuedReplyRunMessage) {
446446
logMessageQueued({ sessionId, source: "embedded-agent-runner" });
447447
return {

0 commit comments

Comments
 (0)