@@ -33,6 +33,7 @@ vi.mock("../config/config.js", () => ({
33333434import "./test-helpers/fast-openclaw-tools-sessions.js";
3535import { setActivePluginRegistry } from "../plugins/runtime.js";
36+import { testing as embeddedRunsTesting, setActiveEmbeddedRun } from "./pi-embedded-runner/runs.js";
3637import { testing as agentStepTesting } from "./tools/agent-step.js";
3738import { createSessionsHistoryTool } from "./tools/sessions-history-tool.js";
3839import { createSessionsListTool } from "./tools/sessions-list-tool.js";
@@ -224,6 +225,7 @@ function sessionsSendDetails(details: unknown): SessionsSendDetails {
224225describe("sessions tools", () => {
225226beforeEach(() => {
226227callGatewayMock.mockClear();
228+embeddedRunsTesting.resetActiveEmbeddedRuns();
227229loadSessionEntryByKeyMock.mockReset();
228230loadSessionEntryByKeyMock.mockReturnValue(undefined);
229231installMessagingTestRegistry();
@@ -1296,6 +1298,211 @@ describe("sessions tools", () => {
12961298expect(calls.find((call) => call.method === "send")).toBeUndefined();
12971299});
129813001301+it("sessions_send reroutes run-scoped active deliveries when transcript steering is rejected", async () => {
1302+const calls: Array<{ method?: string; params?: unknown }> = [];
1303+const requesterKey = "agent:re-portal:main";
1304+const runScopedCallerKey = "agent:leasing-ops:cron:monthly-utility:run:run-fast";
1305+const durableCallerKey = "agent:leasing-ops:cron:monthly-utility";
1306+const queueMessage = vi.fn(async (_text: string, _options?: unknown) => {
1307+throw new Error("active session ended before queued steering message was committed");
1308+});
1309+setActiveEmbeddedRun(
1310+"caller-active-session",
1311+{
1312+ queueMessage,
1313+isStreaming: () => true,
1314+isCompacting: () => false,
1315+supportsTranscriptCommitWait: true,
1316+abort: () => {},
1317+},
1318+runScopedCallerKey,
1319+);
1320+callGatewayMock.mockImplementation(async (opts: unknown) => {
1321+const request = opts as { method?: string; params?: unknown };
1322+calls.push(request);
1323+if (request.method === "agent") {
1324+return { runId: "fallback-run", status: "accepted", acceptedAt: 2000 };
1325+}
1326+if (request.method === "agent.wait") {
1327+const params = request.params as { runId?: string } | undefined;
1328+return { runId: params?.runId ?? "fallback-run", status: "ok" };
1329+}
1330+if (request.method === "chat.history") {
1331+return { messages: [] };
1332+}
1333+return {};
1334+});
1335+1336+const tool = createOpenClawTools({
1337+agentSessionKey: requesterKey,
1338+agentChannel: "telegram",
1339+config: {
1340+ ...TEST_CONFIG,
1341+session: {
1342+ ...TEST_CONFIG.session,
1343+agentToAgent: { maxPingPongTurns: 0 },
1344+},
1345+},
1346+}).find((candidate) => candidate.name === "sessions_send");
1347+if (!tool) {
1348+throw new Error("missing sessions_send tool");
1349+}
1350+1351+const result = await tool.execute("call-run-scoped-caller", {
1352+sessionKey: runScopedCallerKey,
1353+message: "[TASK-COMPLETE] re-portal occupancy ready",
1354+timeoutSeconds: 0,
1355+});
1356+const details = sessionsSendDetails(result.details);
1357+expect(details.status).toBe("accepted");
1358+expect(details.sessionKey).toBe(runScopedCallerKey);
1359+expect(details.delivery?.status).toBe("pending");
1360+const queuedText = queueMessage.mock.calls[0]?.[0];
1361+expect(queuedText).toContain("[Inter-session message]");
1362+expect(queuedText).toContain("[TASK-COMPLETE] re-portal occupancy ready");
1363+expect(queueMessage).toHaveBeenCalledWith(queuedText, {
1364+steeringMode: "all",
1365+debounceMs: 0,
1366+deliveryTimeoutMs: 30_000,
1367+waitForTranscriptCommit: true,
1368+});
1369+1370+await vi.waitFor(() => {
1371+const fallbackCall = calls.find(
1372+(call) =>
1373+call.method === "agent" &&
1374+(call.params as { sessionKey?: string } | undefined)?.sessionKey === durableCallerKey,
1375+);
1376+expect(fallbackCall).toBeDefined();
1377+});
1378+1379+const agentCalls = calls.filter((call) => call.method === "agent");
1380+expect(
1381+agentCalls.some(
1382+(call) =>
1383+(call.params as { sessionKey?: string } | undefined)?.sessionKey === runScopedCallerKey,
1384+),
1385+).toBe(false);
1386+const fallbackParams = agentCalls.find(
1387+(call) =>
1388+(call.params as { sessionKey?: string } | undefined)?.sessionKey === durableCallerKey,
1389+)?.params as { inputProvenance?: { sourceSessionKey?: string }; message?: string } | undefined;
1390+expect(fallbackParams?.message).toContain("[Inter-session message]");
1391+expect(fallbackParams?.message).toContain("[TASK-COMPLETE] re-portal occupancy ready");
1392+expect(fallbackParams?.inputProvenance?.sourceSessionKey).toBe(requesterKey);
1393+1394+await vi.waitFor(() => {
1395+const waitCall = calls.find(
1396+(call) =>
1397+call.method === "agent.wait" &&
1398+(call.params as { runId?: string } | undefined)?.runId === "fallback-run",
1399+);
1400+expect(waitCall).toBeDefined();
1401+});
1402+await vi.waitFor(() => {
1403+const historyCall = calls.find(
1404+(call) =>
1405+call.method === "chat.history" &&
1406+(call.params as { sessionKey?: string } | undefined)?.sessionKey === durableCallerKey,
1407+);
1408+expect(historyCall).toBeDefined();
1409+});
1410+});
1411+1412+it("sessions_send preserves active delivery when transcript commit wait is unsupported", async () => {
1413+const calls: Array<{ method?: string }> = [];
1414+const runScopedCallerKey = "agent:leasing-ops:cron:monthly-utility:run:run-fast";
1415+const queueMessage = vi.fn(async () => {});
1416+setActiveEmbeddedRun(
1417+"caller-active-session",
1418+{
1419+ queueMessage,
1420+isStreaming: () => true,
1421+isCompacting: () => false,
1422+abort: () => {},
1423+},
1424+runScopedCallerKey,
1425+);
1426+callGatewayMock.mockImplementation(async (opts: unknown) => {
1427+const request = opts as { method?: string };
1428+calls.push(request);
1429+if (request.method === "agent") {
1430+throw new Error("fallback agent should not start");
1431+}
1432+return {};
1433+});
1434+1435+const tool = createOpenClawTools({
1436+agentSessionKey: "agent:re-portal:main",
1437+agentChannel: "telegram",
1438+}).find((candidate) => candidate.name === "sessions_send");
1439+if (!tool) {
1440+throw new Error("missing sessions_send tool");
1441+}
1442+1443+const result = await tool.execute("call-run-scoped-caller", {
1444+sessionKey: runScopedCallerKey,
1445+message: "[TASK-COMPLETE] re-portal occupancy ready",
1446+timeoutSeconds: 0,
1447+});
1448+1449+const details = sessionsSendDetails(result.details);
1450+expect(details.status).toBe("accepted");
1451+expect(details.sessionKey).toBe(runScopedCallerKey);
1452+expect(queueMessage).toHaveBeenCalledOnce();
1453+expect(queueMessage).toHaveBeenCalledWith(expect.stringContaining("[Inter-session message]"), {
1454+steeringMode: "all",
1455+debounceMs: 0,
1456+deliveryTimeoutMs: 30_000,
1457+});
1458+expect(calls.some((call) => call.method === "agent")).toBe(false);
1459+});
1460+1461+it("sessions_send reports run-scoped fallback admission failures", async () => {
1462+const runScopedCallerKey = "agent:leasing-ops:cron:monthly-utility:run:run-fast";
1463+const queueMessage = vi.fn(async () => {
1464+throw new Error("active session ended before queued steering message was committed");
1465+});
1466+setActiveEmbeddedRun(
1467+"caller-active-session",
1468+{
1469+ queueMessage,
1470+isStreaming: () => true,
1471+isCompacting: () => false,
1472+supportsTranscriptCommitWait: true,
1473+abort: () => {},
1474+},
1475+runScopedCallerKey,
1476+);
1477+callGatewayMock.mockImplementation(async (opts: unknown) => {
1478+const request = opts as { method?: string; params?: unknown };
1479+if (request.method === "agent") {
1480+throw new Error("gateway request timeout for agent");
1481+}
1482+return {};
1483+});
1484+1485+const tool = createOpenClawTools({
1486+agentSessionKey: "agent:re-portal:main",
1487+agentChannel: "telegram",
1488+}).find((candidate) => candidate.name === "sessions_send");
1489+if (!tool) {
1490+throw new Error("missing sessions_send tool");
1491+}
1492+1493+const result = await tool.execute("call-run-scoped-caller", {
1494+sessionKey: runScopedCallerKey,
1495+message: "[TASK-COMPLETE] re-portal occupancy ready",
1496+timeoutSeconds: 0,
1497+});
1498+1499+const details = sessionsSendDetails(result.details);
1500+expect(details.status).toBe("error");
1501+expect(details.sessionKey).toBe(runScopedCallerKey);
1502+expect(details.error).toContain("queue_message_failed reason=runtime_rejected");
1503+expect(details.error).toContain("fallback_failed error=gateway request timeout for agent");
1504+});
1505+12991506it("sessions_send preserves terminal timeouts without starting A2A", async () => {
13001507const calls: Array<{ method?: string; params?: unknown }> = [];
13011508const requesterKey = "agent:main:main";
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。