From 7cb86e49db8fbb8243e07c63820b315c5186f529 Mon Sep 17 00:00:00 2001 From: flint Date: Wed, 2 Sep 2026 10:26:48 -0700 Subject: [PATCH] fix(openclaw-tps-mail): dispatcher delivers one signed, idempotent reply per inbound (cli#338) Implemented by Anvil on tps-anvil (local commit 1517218); landed by Flint because the host's GitHub credential and outbound mail relay were both dead. Refs #338. Co-Authored-By: anvil Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01Y7z6Gbs5LKG1RczzmRa98D --- packages/cli/package.json | 1 + plugins/openclaw-tps-mail/package.json | 1 + plugins/openclaw-tps-mail/src/index.ts | 174 ++++++++--- .../openclaw-tps-mail/src/verify-adapter.ts | 2 +- .../test/dispatcher-reply.test.ts | 290 ++++++++++++++++++ .../openclaw-tps-mail/test/startup.test.ts | 2 +- .../test/verify-strict.test.ts | 2 +- 7 files changed, 434 insertions(+), 38 deletions(-) create mode 100644 plugins/openclaw-tps-mail/test/dispatcher-reply.test.ts diff --git a/packages/cli/package.json b/packages/cli/package.json index c531071..ed0a5ba 100644 --- a/packages/cli/package.json +++ b/packages/cli/package.json @@ -10,6 +10,7 @@ "exports": { ".": "./dist/src/index.js", "./lib/signEnvelope": "./dist/src/lib/signEnvelope.js", + "./utils/agent-keys": "./dist/src/utils/agent-keys.js", "./utils/flair-client": "./dist/src/utils/flair-client.js" }, "optionalDependencies": { diff --git a/plugins/openclaw-tps-mail/package.json b/plugins/openclaw-tps-mail/package.json index 23da403..fbc3708 100644 --- a/plugins/openclaw-tps-mail/package.json +++ b/plugins/openclaw-tps-mail/package.json @@ -53,6 +53,7 @@ "node": ">=22" }, "dependencies": { + "@tpsdev-ai/agent": "file:../../packages/agent", "@tpsdev-ai/cli": "file:../../packages/cli" } } diff --git a/plugins/openclaw-tps-mail/src/index.ts b/plugins/openclaw-tps-mail/src/index.ts index 1ee7fd6..8c11c50 100644 --- a/plugins/openclaw-tps-mail/src/index.ts +++ b/plugins/openclaw-tps-mail/src/index.ts @@ -38,8 +38,9 @@ import { randomUUID } from "node:crypto"; import { existsSync, mkdirSync, readdirSync, readFileSync, renameSync, unlinkSync, writeFileSync, watch as fsWatch, type FSWatcher } from "node:fs"; import { homedir } from "node:os"; import { basename, resolve } from "node:path"; -import type { Envelope } from "@tpsdev-ai/cli/lib/signEnvelope"; -import { verifyEnvelope } from "@tpsdev-ai/cli/lib/signEnvelope"; +import type { Envelope, ChainEntry } from "@tpsdev-ai/agent"; +import { signEnvelope, verifyEnvelope } from "@tpsdev-ai/agent"; +import { readAgentPrivateKey } from "@tpsdev-ai/cli/utils/agent-keys"; import { createVerifyClient } from "./verify-adapter.js"; import type { OpenClawPluginApi } from "openclaw/plugin-sdk"; import type { @@ -212,6 +213,88 @@ function deliverOutboundMail( return { path: writeOutboxFile(message), route: "outbox" }; } +/** + * Sign a dispatcher reply as a v1 signed envelope, exactly like `tps mail + * send` (packages/cli/src/commands/mail.ts `maybeSignEnvelopeBody`). Returns + * the JSON-stringified signed envelope, or null when the agent has no + * signing key (caller must warn and write nothing). + */ +function signReplyEnvelope(from: string, to: string, body: string): string | null { + const privkey = readAgentPrivateKey(from); + if (!privkey) return null; + + const now = new Date().toISOString(); + const chain: ChainEntry[] = [ + { + agent: "system", + kind: "human", + timestamp: now, + rationale: "tps-mail dispatcher reply (no inbound chain)", + signature: null, + }, + { + agent: from, + kind: "agent", + timestamp: now, + rationale: `agent ${from} dispatcher reply`, + signature: null, + }, + ]; + + const envelope: Envelope = { + v: 1, + from, + to, + subject: `mail to ${to}`, + body, + messageId: randomUUID(), + timestamp: now, + delegationChain: chain, + }; + + return JSON.stringify(signEnvelope(envelope, { [from]: privkey })); +} + +/** + * Idempotency check (cli#338 requirement 2): has this agent already sent an + * explicit reply to `sender` since the inbound was delivered? Scans the + * sender's new/ and cur/ maildirs for a mail whose `from == fromAgent` and + * `timestamp >= sinceTs` — the signed envelope `tps mail send` writes. + */ +function hasExplicitReply( + mailDir: string, + sender: string, + fromAgent: string, + sinceTs: string, +): boolean { + for (const sub of ["new", "cur"]) { + const dir = resolve(mailDir, sender, sub); + if (!existsSync(dir)) continue; + for (const f of readdirSync(dir)) { + if (!f.endsWith(".json")) continue; + const m = readMailFile(resolve(dir, f)); + if (!m) continue; + if (m.from === fromAgent && m.timestamp >= sinceTs) return true; + } + } + return false; +} + +/** + * Is `to` a local recipient? True when it has a maildir under `mailDir` on + * this host, or when it is bound to this gateway. Local recipients are + * delivered to their maildir — never to ~/.tps/outbox (cli#338 requirement 3). + */ +function isLocalRecipient( + mailDir: string, + cfg: any, + accountId: string, + to: string, +): boolean { + if (existsSync(resolve(mailDir, to))) return true; + return findBoundAgents(cfg, accountId).includes(to); +} + /** * Move a mail file from /new/ to /cur/ with the given state * patch applied (typically `ackedAt` on success or `nackedAt` on failure). @@ -498,21 +581,22 @@ const gateway: ChannelGatewayAdapter = { ? await channelRuntime.reply.finalizeInboundContext(rawMsgCtx) : { ...rawMsgCtx, CommandAuthorized: false }; + // One reply per inbound, at most (cli#338). The dispatcher's deliver + // callback receives blocks in order; we emit only the FINAL message, + // once, and only if the agent did not already send an explicit reply + // via `tps mail send` during the turn. + let delivered = false; try { - // NOTE on the deliver callback: in practice, openclaw agents using - // tool-based replies (e.g., `tps mail send flint "..."`) write their - // responses via their own tool calls rather than emitting through - // the reply dispatcher. The deliver callback below handles the case - // where an agent DOES emit output through the dispatcher — useful - // for future agents that use the reply path instead of tools. For - // tool-using agents (the current K&S / Anvil / Pulse pattern) the - // deliver callback is a no-op and the reply still lands in the - // recipient's inbox via the `tps mail send` tool path. await channelRuntime.reply.dispatchReplyWithBufferedBlockDispatcher({ ctx: msgCtx, cfg, dispatcherOptions: { - deliver: async (payload: any, _info: any) => { + deliver: async (payload: any, info: any) => { + // Requirement 1: only the final message, once. + if (info?.kind !== "final") return; + if (delivered) return; + delivered = true; + const replyText: string = (typeof payload?.text === "string" ? payload.text : "") || (Array.isArray(payload?.content) @@ -524,29 +608,49 @@ const gateway: ChannelGatewayAdapter = { ""; if (!replyText.trim()) return; - const reply: TpsMailBody = { - id: randomUUID(), - from: recipient, - to: msg.from, - body: replyText, - timestamp: new Date().toISOString(), - replyToId: msg.id, - headers: { - "X-TPS-Trust": "agent", - "X-TPS-Surface": CHANNEL_ID, - "X-TPS-InReplyTo": msg.id, - }, - deliveryAttempts: 0, - }; - const { route } = deliverOutboundMail( - cfg as any, - ctx.accountId ?? "default", - account.mailDir, - reply, - ); - log?.info?.( - `tps-mail: reply ${reply.id} from ${recipient} to ${msg.from} (via dispatcher, route=${route})`, - ); + // Requirement 2: idempotent with explicit sends — if the agent + // already ran `tps mail send `, write nothing. + if (hasExplicitReply(account.mailDir, msg.from, recipient, msg.timestamp)) { + log?.info?.( + `tps-mail: explicit reply already sent to ${msg.from}; dispatcher skipping`, + ); + return; + } + + // Requirement 4: sign the reply (Ed25519 + messageId). + const signedBody = signReplyEnvelope(recipient, msg.from, replyText); + if (!signedBody) { + log?.warn?.( + `tps-mail: no signing key for ${recipient}; cannot sign dispatcher reply to ${msg.from}`, + ); + return; + } + + // Requirement 3 + 5: local recipient → maildir (signed); else warn. + if (isLocalRecipient(account.mailDir, cfg as any, ctx.accountId ?? "default", msg.from)) { + const reply: TpsMailBody = { + id: randomUUID(), + from: recipient, + to: msg.from, + body: signedBody, + timestamp: new Date().toISOString(), + replyToId: msg.id, + headers: { + "X-TPS-Trust": "agent", + "X-TPS-Surface": CHANNEL_ID, + "X-TPS-InReplyTo": msg.id, + }, + deliveryAttempts: 0, + }; + writeMailFile(account.mailDir, msg.from, reply); + log?.info?.( + `tps-mail: reply ${reply.id} from ${recipient} to ${msg.from} (via dispatcher, route=local)`, + ); + } else { + log?.warn?.( + `tps-mail: cannot deliver reply to ${msg.from}: no local maildir or binding`, + ); + } }, }, }); diff --git a/plugins/openclaw-tps-mail/src/verify-adapter.ts b/plugins/openclaw-tps-mail/src/verify-adapter.ts index b01f2c5..5ef311f 100644 --- a/plugins/openclaw-tps-mail/src/verify-adapter.ts +++ b/plugins/openclaw-tps-mail/src/verify-adapter.ts @@ -13,7 +13,7 @@ */ import { FlairClient as CliFlairClient, createFlairClient } from "@tpsdev-ai/cli/utils/flair-client"; -import type { FlairClient as VerifyFlairClient } from "@tpsdev-ai/cli/lib/signEnvelope"; +import type { FlairClient as VerifyFlairClient } from "@tpsdev-ai/agent"; import { homedir } from "node:os"; import { join } from "node:path"; diff --git a/plugins/openclaw-tps-mail/test/dispatcher-reply.test.ts b/plugins/openclaw-tps-mail/test/dispatcher-reply.test.ts new file mode 100644 index 0000000..7af29e0 --- /dev/null +++ b/plugins/openclaw-tps-mail/test/dispatcher-reply.test.ts @@ -0,0 +1,290 @@ +/** + * dispatcher-reply.test.ts — cli#338: the dispatcher reply path must deliver + * ONE explicit, signed, idempotent reply per inbound — never a dead-drop + * outbox full of unsigned intermediate blocks. + * + * Three must-fail tests (each RED before the fix): + * 1. local unbound sender → 2 text blocks, no explicit send → exactly ONE + * file in ~/.tps/mail//new/, signed, with messageId; zero in + * ~/.tps/outbox/new/. + * 2. same, but the agent runs `tps mail send ` mid-turn → zero + * dispatcher files (the explicit send is the only delivery). + * 3. recipient with neither maildir nor binding → one warning, zero files. + */ +import { describe, expect, it, beforeEach, afterEach, mock } from "bun:test"; +import { mkdtempSync, rmSync, mkdirSync, writeFileSync, readdirSync, readFileSync, existsSync } from "node:fs"; +import { join, resolve } from "node:path"; +import { tmpdir } from "node:os"; +import * as ed from "@noble/ed25519"; +import { createHash } from "node:crypto"; +import { + signEnvelope, + type Envelope, + type ChainEntry, +} from "@tpsdev-ai/agent"; + +// Wire sha512 for sync sign operations. +import { hashes } from "@noble/ed25519"; +hashes.sha512 = (message: Uint8Array) => { + return new Uint8Array(createHash("sha512").update(message).digest()); +}; + +const FLINT_SEED = Buffer.alloc(32, 0x01); +const ANVIL_SEED = Buffer.alloc(32, 0x02); + +function pubkeyFromSeed(seed: Buffer): Buffer { + return Buffer.from(ed.getPublicKey(new Uint8Array(seed))); +} + +// Import the plugin — default export gives us { register }. +import pluginModule from "../src/index.js"; + +let capturedPlugin: any; +const mockApi: any = { + registerChannel: ({ plugin }: { plugin: any }) => { + capturedPlugin = plugin; + }, + logger: { + info: (..._: any[]) => {}, + warn: (..._: any[]) => {}, + error: (..._: any[]) => {}, + }, +}; +pluginModule.register(mockApi); + +function makeMailEnvelope(body: string, overrides: Partial<{ id: string; from: string; to: string; timestamp: string }> = {}) { + return { + id: overrides.id ?? `msg-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`, + from: overrides.from ?? "flint", + to: overrides.to ?? "anvil", + body, + timestamp: overrides.timestamp ?? new Date().toISOString(), + headers: { "X-TPS-Trust": "agent", "X-TPS-Surface": "tps-mail" }, + deliveryAttempts: 0, + }; +} + +function buildSignedBody(from: string, to: string, body: string): string { + const chain: ChainEntry[] = [ + { agent: "system", kind: "human", timestamp: new Date().toISOString(), rationale: "originates", signature: null }, + { agent: from, kind: "agent", timestamp: new Date().toISOString(), rationale: `agent ${from} dispatches`, signature: null }, + ]; + const env = signEnvelope( + { v: 1, from, to, body, messageId: `msg-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`, timestamp: new Date().toISOString(), delegationChain: chain }, + { [from]: FLINT_SEED }, + ); + return JSON.stringify(env); +} + +function readdirSafe(dir: string): string[] { + try { return readdirSync(dir); } catch { return []; } +} + +describe("openclaw-tps-mail: dispatcher single-reply (cli#338)", () => { + let tempMailDir: string; + let tempKeysDir: string; + let tempHome: string; + let abortController: AbortController; + let origHome: string | undefined; + let origKeysDir: string | undefined; + + beforeEach(() => { + tempMailDir = mkdtempSync(join(tmpdir(), "tps-dispatch-mail-")); + tempKeysDir = mkdtempSync(join(tmpdir(), "tps-dispatch-keys-")); + tempHome = mkdtempSync(join(tmpdir(), "tps-dispatch-home-")); + abortController = new AbortController(); + + // Point the agent's signing key at a hermetic temp dir. + writeFileSync(join(tempKeysDir, "anvil.key"), ANVIL_SEED); + origKeysDir = process.env.TPS_TEST_KEYS_DIR; + process.env.TPS_TEST_KEYS_DIR = tempKeysDir; + + // Point ~/.tps/outbox at a hermetic temp dir so we can assert "zero files". + origHome = process.env.HOME; + process.env.HOME = tempHome; + }); + + afterEach(() => { + abortController.abort(); + if (origHome === undefined) delete process.env.HOME; else process.env.HOME = origHome; + if (origKeysDir === undefined) delete process.env.TPS_TEST_KEYS_DIR; else process.env.TPS_TEST_KEYS_DIR = origKeysDir; + try { rmSync(tempMailDir, { recursive: true, force: true }); } catch { /* best effort */ } + try { rmSync(tempKeysDir, { recursive: true, force: true }); } catch { /* best effort */ } + try { rmSync(tempHome, { recursive: true, force: true }); } catch { /* best effort */ } + }); + + /** + * Start the plugin for a single inbound from `sender` to `agentId`, capture + * the dispatcher's `deliver` callback, and return everything the test needs + * to drive the reply path and assert on the filesystem. + */ + async function startDispatcher( + agentId: string, + sender: string, + opts: { senderHasMaildir?: boolean; warnCalls?: string[] } = {}, + ) { + mock.module("../src/verify-adapter.js", () => ({ + createVerifyClient: async () => ({ + async getAgent(name: string) { + if (name === sender) return { publicKey: pubkeyFromSeed(FLINT_SEED) }; + return null; + }, + }), + })); + + // Local maildir for the sender (unbound) — the "local recipient" case. + if (opts.senderHasMaildir) { + mkdirSync(resolve(tempMailDir, sender, "new"), { recursive: true }); + } + + const newDir = resolve(tempMailDir, agentId, "new"); + mkdirSync(newDir, { recursive: true }); + + const signedBody = buildSignedBody(sender, agentId, "inbound payload"); + const envelope = makeMailEnvelope(signedBody, { from: sender, to: agentId, id: `msg-${Date.now()}` }); + const filename = `2026-05-26T00-00-00-${envelope.id}.json`; + writeFileSync(resolve(newDir, filename), JSON.stringify(envelope, null, 2), "utf-8"); + + let dispatchResolve: (val: any) => void; + const dispatchPromise = new Promise((res) => { dispatchResolve = res; }); + + const warnCalls = opts.warnCalls ?? []; + const channelRuntime = { + routing: { + buildAgentSessionKey: (params: any) => + `agent:${params.agentId}:tps-mail:default:${params.peer.id}`, + }, + reply: { + finalizeInboundContext: async (ctx: any) => ({ ...ctx, CommandAuthorized: false }), + dispatchReplyWithBufferedBlockDispatcher: async ({ ctx, dispatcherOptions }: any) => { + dispatchResolve({ ctx, dispatcherOptions }); + }, + }, + }; + + const cfg = { + bindings: [{ agentId, match: { channel: "tps-mail", accountId: "default" } }], + }; + + const ctx = { + account: { accountId: "default", mailDir: tempMailDir, enabled: true }, + cfg, + log: { + info: () => {}, + warn: (...args: any[]) => { warnCalls.push(args.map(String).join(" ")); }, + error: () => {}, + }, + channelRuntime, + abortSignal: abortController.signal, + }; + + const startPromise = capturedPlugin.gateway.startAccount(ctx); + + const result = await Promise.race([ + dispatchPromise, + new Promise((_, reject) => setTimeout(() => reject(new Error("timed out waiting for dispatch")), 5000)), + ]); + + return { result, startPromise, warnCalls }; + } + + it("delivers exactly ONE signed reply to a local unbound sender, zero to outbox", async () => { + const { result, startPromise, warnCalls } = await startDispatcher("anvil", "flint", { senderHasMaildir: true }); + + const { dispatcherOptions } = result; + + // Two text blocks, no explicit send. + await dispatcherOptions.deliver({ text: "intermediate narration" }, { kind: "block" }); + await dispatcherOptions.deliver({ text: "final verdict" }, { kind: "final" }); + + // Assert: exactly ONE file in flint's local maildir. + const flintNew = resolve(tempMailDir, "flint", "new"); + const files = readdirSafe(flintNew).filter((f) => f.endsWith(".json")); + expect(files.length).toBe(1); + + // Assert: the single file is a signed envelope with a messageId. + const raw = readFileSync(resolve(flintNew, files[0]!), "utf-8"); + const mail = JSON.parse(raw); + expect(mail.from).toBe("anvil"); + expect(mail.to).toBe("flint"); + const env: Envelope = JSON.parse(mail.body); + expect(env.v).toBe(1); + expect(typeof env.signature).toBe("string"); + expect(typeof env.messageId).toBe("string"); + expect(env.from).toBe("anvil"); + expect(env.to).toBe("flint"); + expect(env.body).toBe("final verdict"); + + // Assert: zero files in outbox. + const outboxNew = resolve(tempHome, ".tps", "outbox", "new"); + expect(readdirSafe(outboxNew).filter((f) => f.endsWith(".json")).length).toBe(0); + + abortController.abort(); + try { await startPromise; } catch { /* expected on abort */ } + }); + + it("writes zero dispatcher files when the agent already sent explicitly", async () => { + const { result, startPromise } = await startDispatcher("anvil", "flint", { senderHasMaildir: true }); + + const { dispatcherOptions } = result; + + // Simulate the agent running `tps mail send flint "..."` mid-turn: a signed + // envelope from anvil → flint lands in flint's maildir with a timestamp + // >= the inbound's timestamp. + const explicitBody = buildSignedBody("anvil", "flint", "explicit send"); + const explicitMail = { + id: `msg-explicit-${Date.now()}`, + from: "anvil", + to: "flint", + body: explicitBody, + timestamp: new Date().toISOString(), + headers: { "X-TPS-Trust": "agent" }, + deliveryAttempts: 0, + }; + const flintNew = resolve(tempMailDir, "flint", "new"); + mkdirSync(flintNew, { recursive: true }); + writeFileSync( + resolve(flintNew, `2026-05-26T00-00-01-${explicitMail.id}.json`), + JSON.stringify(explicitMail, null, 2), + "utf-8", + ); + + // Dispatcher emits a final block — but must write NOTHING (idempotent). + await dispatcherOptions.deliver({ text: "final verdict" }, { kind: "final" }); + + // Assert: the ONLY file in flint's maildir is the explicit send (1 file), + // not a second dispatcher file. + const files = readdirSafe(flintNew).filter((f) => f.endsWith(".json")); + expect(files.length).toBe(1); + expect(files[0]).toContain(explicitMail.id); + + // Assert: zero files in outbox. + const outboxNew = resolve(tempHome, ".tps", "outbox", "new"); + expect(readdirSafe(outboxNew).filter((f) => f.endsWith(".json")).length).toBe(0); + + abortController.abort(); + try { await startPromise; } catch { /* expected on abort */ } + }); + + it("warns and writes zero files for a recipient with neither maildir nor binding", async () => { + const warnCalls: string[] = []; + const { result, startPromise } = await startDispatcher("anvil", "flint", { senderHasMaildir: false, warnCalls }); + + const { dispatcherOptions } = result; + + await dispatcherOptions.deliver({ text: "final verdict" }, { kind: "final" }); + + // Assert: one warning naming the recipient. + expect(warnCalls.length).toBeGreaterThanOrEqual(1); + expect(warnCalls.join("\n")).toContain("flint"); + + // Assert: zero files in flint's maildir (doesn't even exist) and zero in outbox. + const flintNew = resolve(tempMailDir, "flint", "new"); + expect(existsSync(flintNew)).toBe(false); + const outboxNew = resolve(tempHome, ".tps", "outbox", "new"); + expect(readdirSafe(outboxNew).filter((f) => f.endsWith(".json")).length).toBe(0); + + abortController.abort(); + try { await startPromise; } catch { /* expected on abort */ } + }); +}); diff --git a/plugins/openclaw-tps-mail/test/startup.test.ts b/plugins/openclaw-tps-mail/test/startup.test.ts index b24cf06..bf666d6 100644 --- a/plugins/openclaw-tps-mail/test/startup.test.ts +++ b/plugins/openclaw-tps-mail/test/startup.test.ts @@ -22,7 +22,7 @@ import { signEnvelope, type Envelope, type ChainEntry, -} from "@tpsdev-ai/cli/lib/signEnvelope"; +} from "@tpsdev-ai/agent"; // Wire sha512 for sync sign operations. import { hashes } from "@noble/ed25519"; diff --git a/plugins/openclaw-tps-mail/test/verify-strict.test.ts b/plugins/openclaw-tps-mail/test/verify-strict.test.ts index 58136f5..8a2a32f 100644 --- a/plugins/openclaw-tps-mail/test/verify-strict.test.ts +++ b/plugins/openclaw-tps-mail/test/verify-strict.test.ts @@ -17,7 +17,7 @@ import { verifyEnvelope, type Envelope, type ChainEntry, -} from "@tpsdev-ai/cli/lib/signEnvelope"; +} from "@tpsdev-ai/agent"; // Wire sha512 for sync sign operations. import { hashes } from "@noble/ed25519";