Skip to content

Commit c275415

Browse files
openclaw-clownfish[bot]agent-merkava
openclaw-clownfish[bot]
and
agent-merkava
authored
fix(heartbeat): route outbound mirror to isolated session key (#92807)
* fix(heartbeat): route outbound mirror to isolated session key Co-authored-by: Merkava <[email protected]> * fix(clownfish): address review for ghcrawl-143801-autonomous-smoke (1) Co-authored-by: Merkava <[email protected]> * fix(clownfish): address review for ghcrawl-143801-autonomous-smoke (1) Co-authored-by: Merkava <[email protected]> --------- Co-authored-by: openclaw-clownfish[bot] <280122609+openclaw-clownfish[bot]@users.noreply.github.com> Co-authored-by: Merkava <[email protected]>
1 parent 5b21384 commit c275415

4 files changed

Lines changed: 236 additions & 2 deletions

File tree

‎src/infra/heartbeat-runner.isolated-session-mirror.test.ts‎

Lines changed: 178 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,178 @@
1+
// Covers isolated heartbeat outbound session routing and base-session bookkeeping.
2+
import fs from "node:fs/promises";
3+
import path from "node:path";
4+
import { afterEach, describe, expect, it, vi } from "vitest";
5+
import type { OpenClawConfig } from "../config/config.js";
6+
import { resolveMainSessionKey } from "../config/sessions.js";
7+
import { runHeartbeatOnce } from "./heartbeat-runner.js";
8+
import { installHeartbeatRunnerTestRuntime } from "./heartbeat-runner.test-harness.js";
9+
import { seedSessionStore, withTempHeartbeatSandbox } from "./heartbeat-runner.test-utils.js";
10+
11+
const deliverOutboundPayloadsInternal = vi.hoisted(() =>
12+
vi.fn().mockResolvedValue([{ channel: "whatsapp", messageId: "msg-1" }]),
13+
);
14+
15+
vi.mock("./outbound/deliver.js", () => ({
16+
deliverOutboundPayloads: deliverOutboundPayloadsInternal,
17+
deliverOutboundPayloadsInternal,
18+
}));
19+
20+
installHeartbeatRunnerTestRuntime();
21+
22+
afterEach(() => {
23+
deliverOutboundPayloadsInternal.mockClear();
24+
});
25+
26+
type DeliveryRequest = {
27+
channel?: string;
28+
to?: string;
29+
session?: {
30+
key?: string;
31+
policyKey?: string;
32+
};
33+
};
34+
35+
function latestDeliveryRequest(): DeliveryRequest {
36+
const [request] = deliverOutboundPayloadsInternal.mock.calls.at(-1) ?? [];
37+
if (!request || typeof request !== "object") {
38+
throw new Error("expected heartbeat delivery request");
39+
}
40+
return request as DeliveryRequest;
41+
}
42+
43+
function makeIsolatedLastTargetConfig(tmpDir: string, storePath: string): OpenClawConfig {
44+
return {
45+
agents: {
46+
list: [{ id: "main", default: true }],
47+
defaults: {
48+
workspace: tmpDir,
49+
heartbeat: {
50+
every: "5m",
51+
target: "last",
52+
isolatedSession: true,
53+
},
54+
},
55+
},
56+
channels: { whatsapp: { allowFrom: ["*"] } },
57+
session: { store: storePath },
58+
};
59+
}
60+
61+
describe("runHeartbeatOnce - isolated heartbeat outbound session mirror", () => {
62+
it("uses the isolated run key for outbound delivery while base session owns target and state", async () => {
63+
await withTempHeartbeatSandbox(async ({ tmpDir, storePath, replySpy }) => {
64+
const cfg = makeIsolatedLastTargetConfig(tmpDir, storePath);
65+
const baseSessionKey = resolveMainSessionKey(cfg);
66+
const isolatedSessionKey = `${baseSessionKey}:heartbeat`;
67+
const nowMs = Date.now();
68+
await fs.writeFile(
69+
path.join(tmpDir, "HEARTBEAT.md"),
70+
`tasks:
71+
- name: check-in
72+
interval: 5m
73+
prompt: "Check whether the user needs a status update."
74+
`,
75+
"utf-8",
76+
);
77+
await seedSessionStore(storePath, baseSessionKey, {
78+
sessionId: "base-session",
79+
updatedAt: nowMs - 1_000,
80+
lastChannel: "whatsapp",
81+
lastProvider: "whatsapp",
82+
lastTo: "+15551234567",
83+
});
84+
replySpy.mockResolvedValueOnce({ text: "Status needs attention." });
85+
86+
const result = await runHeartbeatOnce({
87+
cfg,
88+
deps: {
89+
getReplyFromConfig: replySpy,
90+
getQueueSize: () => 0,
91+
nowMs: () => nowMs,
92+
},
93+
});
94+
95+
expect(result.status).toBe("ran");
96+
expect(replySpy.mock.calls[0]?.[0]).toMatchObject({
97+
SessionKey: isolatedSessionKey,
98+
});
99+
const deliveryRequest = latestDeliveryRequest();
100+
expect(deliveryRequest).toMatchObject({
101+
channel: "whatsapp",
102+
to: "+15551234567",
103+
session: {
104+
key: isolatedSessionKey,
105+
policyKey: baseSessionKey,
106+
},
107+
});
108+
109+
const store = JSON.parse(await fs.readFile(storePath, "utf-8")) as Record<
110+
string,
111+
{
112+
heartbeatTaskState?: Record<string, number>;
113+
lastHeartbeatText?: string;
114+
lastHeartbeatSentAt?: number;
115+
heartbeatIsolatedBaseSessionKey?: string;
116+
}
117+
>;
118+
expect(store[baseSessionKey]).toMatchObject({
119+
heartbeatTaskState: { "check-in": nowMs },
120+
lastHeartbeatText: "Status needs attention.",
121+
lastHeartbeatSentAt: nowMs,
122+
});
123+
expect(store[isolatedSessionKey]).toMatchObject({
124+
heartbeatIsolatedBaseSessionKey: baseSessionKey,
125+
});
126+
expect(store[isolatedSessionKey]?.heartbeatTaskState).toBeUndefined();
127+
expect(store[isolatedSessionKey]?.lastHeartbeatText).toBeUndefined();
128+
});
129+
});
130+
131+
it("keeps the base policy key when wake re-entry starts from the isolated key", async () => {
132+
await withTempHeartbeatSandbox(async ({ tmpDir, storePath, replySpy }) => {
133+
const cfg = makeIsolatedLastTargetConfig(tmpDir, storePath);
134+
const baseSessionKey = resolveMainSessionKey(cfg);
135+
const isolatedSessionKey = `${baseSessionKey}:heartbeat`;
136+
const nowMs = Date.now();
137+
138+
await fs.writeFile(
139+
storePath,
140+
JSON.stringify({
141+
[isolatedSessionKey]: {
142+
sessionId: "isolated-session",
143+
updatedAt: nowMs - 1_000,
144+
lastChannel: "whatsapp",
145+
lastProvider: "whatsapp",
146+
lastTo: "+15551234567",
147+
heartbeatIsolatedBaseSessionKey: baseSessionKey,
148+
},
149+
}),
150+
"utf-8",
151+
);
152+
replySpy.mockResolvedValueOnce({ text: "Wake result needs attention." });
153+
154+
const result = await runHeartbeatOnce({
155+
cfg,
156+
sessionKey: isolatedSessionKey,
157+
deps: {
158+
getReplyFromConfig: replySpy,
159+
getQueueSize: () => 0,
160+
nowMs: () => nowMs,
161+
},
162+
});
163+
164+
expect(result.status).toBe("ran");
165+
expect(replySpy.mock.calls[0]?.[0]).toMatchObject({
166+
SessionKey: isolatedSessionKey,
167+
});
168+
expect(latestDeliveryRequest()).toMatchObject({
169+
channel: "whatsapp",
170+
to: "+15551234567",
171+
session: {
172+
key: isolatedSessionKey,
173+
policyKey: baseSessionKey,
174+
},
175+
});
176+
});
177+
});
178+
});

‎src/infra/heartbeat-runner.ts‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1559,6 +1559,7 @@ export async function runHeartbeatOnce(opts: {
15591559
}
15601560

15611561
let runSessionKey = sessionKey;
1562+
let outboundPolicySessionKey: string | undefined;
15621563
if (useIsolatedSession) {
15631564
const configuredSession = resolveHeartbeatSession(cfg, agentId, heartbeat);
15641565
// Collapse only the repeated `:heartbeat` suffixes introduced by wake-triggered
@@ -1628,6 +1629,7 @@ export async function runHeartbeatOnce(opts: {
16281629
}
16291630
}
16301631
runSessionKey = isolatedSessionKey;
1632+
outboundPolicySessionKey = isolatedBaseSessionKey;
16311633
}
16321634
// Update task last run times AFTER successful heartbeat completion
16331635
const updateTaskTimestamps = async () => {
@@ -1707,7 +1709,8 @@ export async function runHeartbeatOnce(opts: {
17071709
const outboundSession = buildOutboundSessionContext({
17081710
cfg,
17091711
agentId,
1710-
sessionKey,
1712+
sessionKey: runSessionKey,
1713+
policySessionKey: outboundPolicySessionKey,
17111714
});
17121715
const canAttemptHeartbeatOk = Boolean(
17131716
!hasDueCommitments && visibility.showOk && delivery.channel !== "none" && delivery.to,

‎src/infra/outbound/deliver.test.ts‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1192,6 +1192,59 @@ describe("deliverOutboundPayloads", () => {
11921192
resolveMediaAccessSpy.mockRestore();
11931193
});
11941194

1195+
it("uses the base policy key for isolated heartbeat group media read denies", async () => {
1196+
const resolveMediaAccessSpy = vi.spyOn(
1197+
mediaCapabilityModule,
1198+
"resolveAgentScopedOutboundMediaAccess",
1199+
);
1200+
const sendMatrix = vi.fn().mockResolvedValue({ messageId: "m1", roomId: "!room:example" });
1201+
1202+
await deliverOutboundPayloads({
1203+
cfg: {
1204+
tools: {
1205+
allow: ["read"],
1206+
},
1207+
channels: {
1208+
matrix: {
1209+
groups: {
1210+
ops: {
1211+
toolsBySender: {
1212+
"id:attacker": {
1213+
deny: ["read"],
1214+
},
1215+
},
1216+
},
1217+
},
1218+
},
1219+
} as OpenClawConfig["channels"],
1220+
},
1221+
channel: "matrix",
1222+
to: "!room:example",
1223+
payloads: [{ text: "heartbeat media", mediaUrl: "file:///tmp/policy.png" }],
1224+
deps: { matrix: sendMatrix },
1225+
session: {
1226+
key: "agent:main:matrix:group:ops:heartbeat",
1227+
policyKey: "agent:main:matrix:group:ops",
1228+
requesterSenderId: "attacker",
1229+
},
1230+
});
1231+
1232+
const [mediaAccessOptions] = requireMockCall(resolveMediaAccessSpy, "media access") as [
1233+
{
1234+
requesterSenderId?: unknown;
1235+
sessionKey?: unknown;
1236+
},
1237+
];
1238+
expect(mediaAccessOptions?.sessionKey).toBe("agent:main:matrix:group:ops");
1239+
expect(mediaAccessOptions?.requesterSenderId).toBe("attacker");
1240+
const sendOptions = requireMatrixSendCall(sendMatrix)[2] as Record<string, unknown>;
1241+
expect(sendOptions.mediaReadFile).toBeUndefined();
1242+
expect((sendOptions.mediaLocalRoots as readonly string[] | undefined) ?? []).not.toContain(
1243+
"/tmp",
1244+
);
1245+
resolveMediaAccessSpy.mockRestore();
1246+
});
1247+
11951248
it("forwards all sender fields to media access for non-id policy matching", async () => {
11961249
const resolveMediaAccessSpy = vi.spyOn(
11971250
mediaCapabilityModule,

‎src/infra/outbound/deliver.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1424,7 +1424,7 @@ async function deliverOutboundPayloadsCore(
14241424
agentId: params.session?.agentId ?? params.mirror?.agentId,
14251425
mediaSources,
14261426
mediaAccess: params.mediaAccess,
1427-
sessionKey: params.session?.key,
1427+
sessionKey: params.session?.policyKey ?? params.session?.key,
14281428
messageProvider: params.session?.key ? undefined : channel,
14291429
accountId: params.session?.requesterAccountId ?? accountId,
14301430
requesterSenderId: params.session?.requesterSenderId,

0 commit comments

Comments
 (0)