surface: envelope inbound messages for agent
This commit is contained in:
@@ -150,6 +150,11 @@ In chat, send `/status` to see if the agent is reachable, how much context the s
|
|||||||
- Direct chats now share a canonical session key `main` by default (configurable via `inbound.reply.session.mainKey`). Groups stay isolated as `group:<jid>`.
|
- Direct chats now share a canonical session key `main` by default (configurable via `inbound.reply.session.mainKey`). Groups stay isolated as `group:<jid>`.
|
||||||
- WebChat always attaches to the `main` session and hydrates the full Tau history from `~/.clawdis/sessions/<SessionId>.jsonl`, so desktop view mirrors WhatsApp/Telegram turns.
|
- WebChat always attaches to the `main` session and hydrates the full Tau history from `~/.clawdis/sessions/<SessionId>.jsonl`, so desktop view mirrors WhatsApp/Telegram turns.
|
||||||
- Inbound contexts carry a `Surface` hint (e.g., `whatsapp`, `webchat`, `telegram`) for logging; replies still go back to the originating surface deterministically.
|
- Inbound contexts carry a `Surface` hint (e.g., `whatsapp`, `webchat`, `telegram`) for logging; replies still go back to the originating surface deterministically.
|
||||||
|
- Every inbound message is wrapped for the agent as `[Surface FROM HOST/IP TIMESTAMP] body`:
|
||||||
|
- WhatsApp: `[WhatsApp +15551234567 2025-12-09 12:34] …`
|
||||||
|
- Telegram: `[Telegram telegram:123456789 2025-12-09 12:34] …`
|
||||||
|
- WebChat: `[WebChat my-mac.local 10.0.0.5 2025-12-09 12:34] …`
|
||||||
|
This keeps the model aware of the transport, sender, host, and time without relying on implicit context.
|
||||||
|
|
||||||
## Credits
|
## Credits
|
||||||
|
|
||||||
|
|||||||
23
src/auto-reply/envelope.test.ts
Normal file
23
src/auto-reply/envelope.test.ts
Normal file
@@ -0,0 +1,23 @@
|
|||||||
|
import { describe, expect, it } from "vitest";
|
||||||
|
|
||||||
|
import { formatAgentEnvelope } from "./envelope.js";
|
||||||
|
|
||||||
|
describe("formatAgentEnvelope", () => {
|
||||||
|
it("includes surface, from, ip, host, and timestamp", () => {
|
||||||
|
const ts = Date.UTC(2025, 0, 2, 3, 4); // 2025-01-02T03:04:00Z
|
||||||
|
const body = formatAgentEnvelope({
|
||||||
|
surface: "WebChat",
|
||||||
|
from: "user1",
|
||||||
|
host: "mac-mini",
|
||||||
|
ip: "10.0.0.5",
|
||||||
|
timestamp: ts,
|
||||||
|
body: "hello",
|
||||||
|
});
|
||||||
|
expect(body).toBe("[WebChat user1 mac-mini 10.0.0.5 2025-01-02 03:04] hello");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("handles missing optional fields", () => {
|
||||||
|
const body = formatAgentEnvelope({ surface: "Telegram", body: "hi" });
|
||||||
|
expect(body).toBe("[Telegram] hi");
|
||||||
|
});
|
||||||
|
});
|
||||||
28
src/auto-reply/envelope.ts
Normal file
28
src/auto-reply/envelope.ts
Normal file
@@ -0,0 +1,28 @@
|
|||||||
|
export type AgentEnvelopeParams = {
|
||||||
|
surface: string;
|
||||||
|
from?: string;
|
||||||
|
timestamp?: number | Date;
|
||||||
|
host?: string;
|
||||||
|
ip?: string;
|
||||||
|
body: string;
|
||||||
|
};
|
||||||
|
|
||||||
|
function formatTimestamp(ts?: number | Date): string | undefined {
|
||||||
|
if (!ts) return undefined;
|
||||||
|
const date = ts instanceof Date ? ts : new Date(ts);
|
||||||
|
if (Number.isNaN(date.getTime())) return undefined;
|
||||||
|
// Compact ISO-like format with minutes precision.
|
||||||
|
return date.toISOString().slice(0, 16).replace("T", " ");
|
||||||
|
}
|
||||||
|
|
||||||
|
export function formatAgentEnvelope(params: AgentEnvelopeParams): string {
|
||||||
|
const surface = params.surface?.trim() || "Surface";
|
||||||
|
const parts: string[] = [surface];
|
||||||
|
if (params.from?.trim()) parts.push(params.from.trim());
|
||||||
|
if (params.host?.trim()) parts.push(params.host.trim());
|
||||||
|
if (params.ip?.trim()) parts.push(params.ip.trim());
|
||||||
|
const ts = formatTimestamp(params.timestamp);
|
||||||
|
if (ts) parts.push(ts);
|
||||||
|
const header = `[${parts.join(" ")}]`;
|
||||||
|
return `${header} ${params.body}`;
|
||||||
|
}
|
||||||
@@ -23,7 +23,13 @@ vi.mock("@grammyjs/transformer-throttler", () => ({
|
|||||||
apiThrottler: () => throttlerSpy(),
|
apiThrottler: () => throttlerSpy(),
|
||||||
}));
|
}));
|
||||||
|
|
||||||
|
vi.mock("../auto-reply/reply.js", () => {
|
||||||
|
const replySpy = vi.fn();
|
||||||
|
return { getReplyFromConfig: replySpy, __replySpy: replySpy };
|
||||||
|
});
|
||||||
|
|
||||||
import { createTelegramBot } from "./bot.js";
|
import { createTelegramBot } from "./bot.js";
|
||||||
|
import * as replyModule from "../auto-reply/reply.js";
|
||||||
|
|
||||||
describe("createTelegramBot", () => {
|
describe("createTelegramBot", () => {
|
||||||
it("installs grammY throttler", () => {
|
it("installs grammY throttler", () => {
|
||||||
@@ -31,4 +37,32 @@ describe("createTelegramBot", () => {
|
|||||||
expect(throttlerSpy).toHaveBeenCalledTimes(1);
|
expect(throttlerSpy).toHaveBeenCalledTimes(1);
|
||||||
expect(useSpy).toHaveBeenCalledWith("throttler");
|
expect(useSpy).toHaveBeenCalledWith("throttler");
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("wraps inbound message with Telegram envelope", async () => {
|
||||||
|
onSpy.mockReset();
|
||||||
|
const replySpy = replyModule.__replySpy as unknown as ReturnType<
|
||||||
|
typeof vi.fn
|
||||||
|
>;
|
||||||
|
replySpy.mockReset();
|
||||||
|
|
||||||
|
createTelegramBot({ token: "tok" });
|
||||||
|
expect(onSpy).toHaveBeenCalledWith("message", expect.any(Function));
|
||||||
|
const handler = onSpy.mock.calls[0][1] as (ctx: any) => Promise<void>;
|
||||||
|
|
||||||
|
const message = {
|
||||||
|
chat: { id: 1234, type: "private" },
|
||||||
|
text: "hello world",
|
||||||
|
date: 1736380800, // 2025-01-09T00:00:00Z
|
||||||
|
};
|
||||||
|
await handler({
|
||||||
|
message,
|
||||||
|
me: { username: "clawdis_bot" },
|
||||||
|
getFile: async () => ({ download: async () => new Uint8Array() }),
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(replySpy).toHaveBeenCalledTimes(1);
|
||||||
|
const payload = replySpy.mock.calls[0][0];
|
||||||
|
expect(payload.Body).toMatch(/^\[Telegram telegram:1234 2025-01-09 00:00]/);
|
||||||
|
expect(payload.Body).toContain("hello world");
|
||||||
|
});
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import { Bot, InputFile, webhookCallback } from "grammy";
|
|||||||
import { chunkText } from "../auto-reply/chunk.js";
|
import { chunkText } from "../auto-reply/chunk.js";
|
||||||
import { getReplyFromConfig } from "../auto-reply/reply.js";
|
import { getReplyFromConfig } from "../auto-reply/reply.js";
|
||||||
import type { ReplyPayload } from "../auto-reply/types.js";
|
import type { ReplyPayload } from "../auto-reply/types.js";
|
||||||
|
import { formatAgentEnvelope } from "../auto-reply/envelope.js";
|
||||||
import { loadConfig } from "../config/config.js";
|
import { loadConfig } from "../config/config.js";
|
||||||
import { danger, logVerbose } from "../globals.js";
|
import { danger, logVerbose } from "../globals.js";
|
||||||
import { getChildLogger } from "../logging.js";
|
import { getChildLogger } from "../logging.js";
|
||||||
@@ -98,8 +99,15 @@ export function createTelegramBot(opts: TelegramBotOptions) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
const media = await resolveMedia(ctx, mediaMaxBytes);
|
const media = await resolveMedia(ctx, mediaMaxBytes);
|
||||||
const body = (msg.text ?? msg.caption ?? media?.placeholder ?? "").trim();
|
const rawBody = (msg.text ?? msg.caption ?? media?.placeholder ?? "").trim();
|
||||||
if (!body) return;
|
if (!rawBody) return;
|
||||||
|
|
||||||
|
const body = formatAgentEnvelope({
|
||||||
|
surface: "Telegram",
|
||||||
|
from: isGroup ? `group:${chatId}` : `telegram:${chatId}`,
|
||||||
|
timestamp: msg.date ? msg.date * 1000 : undefined,
|
||||||
|
body: rawBody,
|
||||||
|
});
|
||||||
|
|
||||||
const ctxPayload = {
|
const ctxPayload = {
|
||||||
Body: body,
|
Body: body,
|
||||||
|
|||||||
@@ -680,8 +680,8 @@ describe("web auto-reply", () => {
|
|||||||
|
|
||||||
expect(resolver).toHaveBeenCalledTimes(1);
|
expect(resolver).toHaveBeenCalledTimes(1);
|
||||||
const args = resolver.mock.calls[0][0];
|
const args = resolver.mock.calls[0][0];
|
||||||
expect(args.Body).toContain("[Jan 1 00:00] [clawdis] first");
|
expect(args.Body).toContain("[WhatsApp +1 2025-01-01 00:00] [clawdis] first");
|
||||||
expect(args.Body).toContain("[Jan 1 01:00] [clawdis] second");
|
expect(args.Body).toContain("[WhatsApp +1 2025-01-01 01:00] [clawdis] second");
|
||||||
|
|
||||||
// Max listeners bumped to avoid warnings in multi-instance test runs
|
// Max listeners bumped to avoid warnings in multi-instance test runs
|
||||||
expect(process.getMaxListeners?.()).toBeGreaterThanOrEqual(50);
|
expect(process.getMaxListeners?.()).toBeGreaterThanOrEqual(50);
|
||||||
@@ -1292,7 +1292,8 @@ describe("web auto-reply", () => {
|
|||||||
// The resolver should receive a prefixed body with the configured marker
|
// The resolver should receive a prefixed body with the configured marker
|
||||||
const callArg = resolver.mock.calls[0]?.[0] as { Body?: string };
|
const callArg = resolver.mock.calls[0]?.[0] as { Body?: string };
|
||||||
expect(callArg?.Body).toBeDefined();
|
expect(callArg?.Body).toBeDefined();
|
||||||
expect(callArg?.Body).toBe("[same-phone] hello");
|
expect(callArg?.Body).toContain("[WhatsApp +1555");
|
||||||
|
expect(callArg?.Body).toContain("[same-phone] hello");
|
||||||
resetLoadConfigMock();
|
resetLoadConfigMock();
|
||||||
});
|
});
|
||||||
|
|
||||||
@@ -1324,9 +1325,10 @@ describe("web auto-reply", () => {
|
|||||||
sendMedia: vi.fn(),
|
sendMedia: vi.fn(),
|
||||||
});
|
});
|
||||||
|
|
||||||
// Body should NOT be prefixed
|
// Body should include envelope but not the same-phone prefix
|
||||||
const callArg = resolver.mock.calls[0]?.[0] as { Body?: string };
|
const callArg = resolver.mock.calls[0]?.[0] as { Body?: string };
|
||||||
expect(callArg?.Body).toBe("hello");
|
expect(callArg?.Body).toContain("[WhatsApp +1555");
|
||||||
|
expect(callArg?.Body).toContain("hello");
|
||||||
});
|
});
|
||||||
|
|
||||||
it("applies responsePrefix to regular replies", async () => {
|
it("applies responsePrefix to regular replies", async () => {
|
||||||
|
|||||||
@@ -31,6 +31,8 @@ import {
|
|||||||
sleepWithAbort,
|
sleepWithAbort,
|
||||||
} from "./reconnect.js";
|
} from "./reconnect.js";
|
||||||
import { formatError, getWebAuthAgeMs, readWebSelfId } from "./session.js";
|
import { formatError, getWebAuthAgeMs, readWebSelfId } from "./session.js";
|
||||||
|
import { formatAgentEnvelope } from "../auto-reply/envelope.js";
|
||||||
|
import { formatAgentEnvelope } from "../auto-reply/envelope.js";
|
||||||
|
|
||||||
const WEB_TEXT_LIMIT = 4000;
|
const WEB_TEXT_LIMIT = 4000;
|
||||||
const DEFAULT_GROUP_HISTORY_LIMIT = 50;
|
const DEFAULT_GROUP_HISTORY_LIMIT = 50;
|
||||||
@@ -759,19 +761,6 @@ export async function monitorWebProvider(
|
|||||||
type PendingBatch = { messages: WebInboundMsg[]; timer?: NodeJS.Timeout };
|
type PendingBatch = { messages: WebInboundMsg[]; timer?: NodeJS.Timeout };
|
||||||
const pendingBatches = new Map<string, PendingBatch>();
|
const pendingBatches = new Map<string, PendingBatch>();
|
||||||
|
|
||||||
const formatTimestamp = (ts?: number) => {
|
|
||||||
const tsCfg = cfg.inbound?.timestampPrefix;
|
|
||||||
const tsEnabled = tsCfg !== false; // default true
|
|
||||||
if (!tsEnabled) return "";
|
|
||||||
const tz = typeof tsCfg === "string" ? tsCfg : "UTC";
|
|
||||||
const date = ts ? new Date(ts) : new Date();
|
|
||||||
try {
|
|
||||||
return `[${date.toLocaleDateString("en-US", { month: "short", day: "numeric", timeZone: tz })} ${date.toLocaleTimeString("en-US", { hour: "2-digit", minute: "2-digit", hour12: false, timeZone: tz })}] `;
|
|
||||||
} catch {
|
|
||||||
return `[${date.toISOString().slice(5, 16).replace("T", " ")}] `;
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
const buildLine = (msg: WebInboundMsg) => {
|
const buildLine = (msg: WebInboundMsg) => {
|
||||||
// Build message prefix: explicit config > default based on allowFrom
|
// Build message prefix: explicit config > default based on allowFrom
|
||||||
let messagePrefix = cfg.inbound?.messagePrefix;
|
let messagePrefix = cfg.inbound?.messagePrefix;
|
||||||
@@ -784,7 +773,18 @@ export async function monitorWebProvider(
|
|||||||
msg.chatType === "group"
|
msg.chatType === "group"
|
||||||
? `${msg.senderName ?? msg.senderE164 ?? "Someone"}: `
|
? `${msg.senderName ?? msg.senderE164 ?? "Someone"}: `
|
||||||
: "";
|
: "";
|
||||||
return `${formatTimestamp(msg.timestamp)}${prefixStr}${senderLabel}${msg.body}`;
|
const baseLine = `${prefixStr}${senderLabel}${msg.body}`;
|
||||||
|
|
||||||
|
// Wrap with standardized envelope for the agent.
|
||||||
|
return formatAgentEnvelope({
|
||||||
|
surface: "WhatsApp",
|
||||||
|
from:
|
||||||
|
msg.chatType === "group"
|
||||||
|
? msg.from
|
||||||
|
: msg.from?.replace(/^whatsapp:/, ""),
|
||||||
|
timestamp: msg.timestamp,
|
||||||
|
body: baseLine,
|
||||||
|
});
|
||||||
};
|
};
|
||||||
|
|
||||||
const processBatch = async (conversationId: string) => {
|
const processBatch = async (conversationId: string) => {
|
||||||
@@ -806,9 +806,13 @@ export async function monitorWebProvider(
|
|||||||
history.length > 0 ? history.slice(0, -1) : [];
|
history.length > 0 ? history.slice(0, -1) : [];
|
||||||
if (historyWithoutCurrent.length > 0) {
|
if (historyWithoutCurrent.length > 0) {
|
||||||
const historyText = historyWithoutCurrent
|
const historyText = historyWithoutCurrent
|
||||||
.map(
|
.map((m) =>
|
||||||
(m) =>
|
formatAgentEnvelope({
|
||||||
`${m.sender}: ${m.body}${m.timestamp ? ` [${new Date(m.timestamp).toISOString()}]` : ""}`,
|
surface: "WhatsApp",
|
||||||
|
from: conversationId,
|
||||||
|
timestamp: m.timestamp,
|
||||||
|
body: `${m.sender}: ${m.body}`,
|
||||||
|
}),
|
||||||
)
|
)
|
||||||
.join("\\n");
|
.join("\\n");
|
||||||
combinedBody = `[Chat messages since your last reply - for context]\\n${historyText}\\n\\n[Current message - respond to this]\\n${buildLine(latest)}`;
|
combinedBody = `[Chat messages since your last reply - for context]\\n${historyText}\\n\\n[Current message - respond to this]\\n${buildLine(latest)}`;
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ import path from "node:path";
|
|||||||
import { fileURLToPath } from "node:url";
|
import { fileURLToPath } from "node:url";
|
||||||
import { type WebSocket, WebSocketServer } from "ws";
|
import { type WebSocket, WebSocketServer } from "ws";
|
||||||
import { loadConfig } from "../config/config.js";
|
import { loadConfig } from "../config/config.js";
|
||||||
|
import { formatAgentEnvelope } from "../auto-reply/envelope.js";
|
||||||
import {
|
import {
|
||||||
loadSessionStore,
|
loadSessionStore,
|
||||||
resolveStorePath,
|
resolveStorePath,
|
||||||
@@ -139,6 +140,7 @@ function broadcastAll(payload: unknown) {
|
|||||||
|
|
||||||
async function handleRpc(
|
async function handleRpc(
|
||||||
body: unknown,
|
body: unknown,
|
||||||
|
meta?: { remoteAddress?: string | null; senderHost?: string },
|
||||||
): Promise<{ ok: boolean; payloads?: RpcPayload[]; error?: string }> {
|
): Promise<{ ok: boolean; payloads?: RpcPayload[]; error?: string }> {
|
||||||
const payload = body as {
|
const payload = body as {
|
||||||
text?: unknown;
|
text?: unknown;
|
||||||
@@ -148,8 +150,8 @@ async function handleRpc(
|
|||||||
timeout?: unknown;
|
timeout?: unknown;
|
||||||
};
|
};
|
||||||
|
|
||||||
const text: string = (payload.text ?? "").toString();
|
const textRaw: string = (payload.text ?? "").toString();
|
||||||
if (!text.trim()) return { ok: false, error: "empty text" };
|
if (!textRaw.trim()) return { ok: false, error: "empty text" };
|
||||||
if (!gateway || !gatewayReady) {
|
if (!gateway || !gatewayReady) {
|
||||||
return { ok: false, error: "gateway unavailable" };
|
return { ok: false, error: "gateway unavailable" };
|
||||||
}
|
}
|
||||||
@@ -163,11 +165,20 @@ async function handleRpc(
|
|||||||
|
|
||||||
const idempotencyKey = randomUUID();
|
const idempotencyKey = randomUUID();
|
||||||
try {
|
try {
|
||||||
|
// Wrap user text with surface + host/IP envelope
|
||||||
|
const message = formatAgentEnvelope({
|
||||||
|
surface: "WebChat",
|
||||||
|
from: meta?.senderHost ?? os.hostname(),
|
||||||
|
ip: meta?.remoteAddress ?? undefined,
|
||||||
|
timestamp: Date.now(),
|
||||||
|
body: textRaw.trim(),
|
||||||
|
});
|
||||||
|
|
||||||
// Send agent request; wait for final res (status ok/error)
|
// Send agent request; wait for final res (status ok/error)
|
||||||
const res = (await gateway.request(
|
const res = (await gateway.request(
|
||||||
"agent",
|
"agent",
|
||||||
{
|
{
|
||||||
message: text,
|
message,
|
||||||
thinking,
|
thinking,
|
||||||
deliver,
|
deliver,
|
||||||
to,
|
to,
|
||||||
@@ -254,7 +265,13 @@ export async function startWebChatServer(
|
|||||||
} catch {
|
} catch {
|
||||||
// ignore
|
// ignore
|
||||||
}
|
}
|
||||||
const result = await handleRpc(body);
|
const forwarded =
|
||||||
|
(req.headers["x-forwarded-for"] as string | undefined)?.split(",")[0]?.trim() ??
|
||||||
|
req.socket.remoteAddress;
|
||||||
|
const result = await handleRpc(body, {
|
||||||
|
remoteAddress: forwarded,
|
||||||
|
senderHost: os.hostname(),
|
||||||
|
});
|
||||||
res.setHeader("Content-Type", "application/json");
|
res.setHeader("Content-Type", "application/json");
|
||||||
res.end(JSON.stringify(result));
|
res.end(JSON.stringify(result));
|
||||||
return;
|
return;
|
||||||
|
|||||||
Reference in New Issue
Block a user