From 3962222863cb36c07a2c19ec48854367cdbffef0 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Mon, 20 Jul 2026 11:26:08 -0700 Subject: [PATCH] FN-8424: route CLI chat replies through inbox mail Route agent replies to the correct CLI or dashboard mailbox with bounded polling deadlines. - add reply-parent routing validation and CLI/dashboard inbox selection - preserve named mailbox conversations while handling per-message reply deadlines - document chat and inbox interfaces and cover deadline and routing regressions Files changed: .changeset/fn-8424-cli-chat-reply-routing.md | 7 + docs/agents.md | 20 +- docs/cli-reference.md | 29 +- packages/cli/src/bin.ts | 15 +- packages/cli/src/commands/__tests__/chat.test.ts | 262 +++++++++--------- .../cli/src/commands/__tests__/message.test.ts | 12 + packages/cli/src/commands/chat.ts | 293 ++++++++++++--------- packages/cli/src/commands/message.ts | 18 +- ...tools-send-message-recipient-validation.test.ts | 86 +++++- packages/engine/src/agent-heartbeat-prompts.ts | 8 +- packages/engine/src/agent-tools.ts | 71 +++-- 11 files changed, 523 insertions(+), 298 deletions(-) Fusion-Task-Id: FN-8424 Fusion-Task-Lineage: 28d0ef88-717e-4f39-8880-64d2fef94706 Co-authored-by: Fusion (runfusion.ai) --- .changeset/fn-8424-cli-chat-reply-routing.md | 7 + docs/agents.md | 20 +- docs/cli-reference.md | 29 +- packages/cli/src/bin.ts | 15 +- .../cli/src/commands/__tests__/chat.test.ts | 270 ++++++++-------- .../src/commands/__tests__/message.test.ts | 12 + packages/cli/src/commands/chat.ts | 293 +++++++++++------- packages/cli/src/commands/message.ts | 18 +- ...-send-message-recipient-validation.test.ts | 86 ++++- .../engine/src/agent-heartbeat-prompts.ts | 8 +- packages/engine/src/agent-tools.ts | 71 ++++- 11 files changed, 527 insertions(+), 302 deletions(-) create mode 100644 .changeset/fn-8424-cli-chat-reply-routing.md diff --git a/.changeset/fn-8424-cli-chat-reply-routing.md b/.changeset/fn-8424-cli-chat-reply-routing.md new file mode 100644 index 0000000000..e20fd4fd89 --- /dev/null +++ b/.changeset/fn-8424-cli-chat-reply-routing.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Return CLI chat replies to the terminal and expose dashboard inbox reads. +category: fix +dev: Reply routing now validates parent ownership, polls with deadline-aware sleeps, tracks interactive pending replies, and supports `fn message inbox --user dashboard`. diff --git a/docs/agents.md b/docs/agents.md index 48316e4bc5..6f1f7d6b1b 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -26,16 +26,17 @@ Use `fn chat` to message an agent from your terminal. ### Synopsis ```bash -fn chat [message…] [--once] [--non-interactive] [--poll-ms ] [--conversation-id ] +fn chat [message…] [--once] [--non-interactive] [--poll-ms ] [--reply-timeout-ms ] [--conversation-id ] ``` ### Behavior -- `fn chat ` opens an interactive **mailbox conversation** REPL. It delivers to the target agent's MessageStore inbox (`fn_read_messages`), not to a dashboard ChatStore/ChatView session or multi-agent room. +- `fn chat ` opens an interactive mailbox-conversation REPL. - Each message is stored as a `user-to-agent` MessageStore message from `cli` with `metadata.wakeRecipient=true`, `metadata.kind="cli-chat"`, and a durable `metadata.conversationId`. -- The default conversation id is `cli-chat:cli:` and is stable for that CLI-user/agent pair in the selected project. Use `--conversation-id ` to name or share a different mailbox thread; the CLI banner prints the active id. -- `fn message send` remains a distinct one-shot mailbox command and does not add CLI-chat conversation metadata. -- Agent replies are polled from your inbox and printed only when they carry that `conversationId` or use `replyTo.messageId` to reference a message already in the thread; reply streaming and dashboard operator-inbox readability are separately tracked by FN-8424 / issue #2363. +- CLI chat is MessageStore mail plus polling, not a token-streaming SSE session. Replies are printed only when they carry the active conversation ID or reply to a known thread message. +- On a direct-message reply, agents must pass `reply_to_message_id` and either set `to_id` to the exact `[from: type:id]` value reported by `fn_read_messages` (including `cli`) or omit it to use the safe parent-sender default. Parent-derived routing is allowed only when the parent was addressed to the replying agent; an explicit `to_id` remains available for intentional forwarding. +- The default conversation ID is `cli-chat:cli:`; use `--conversation-id ` to name or share a different mailbox thread. +- One-shot replies have a deadline independent of `--poll-ms`; polling sleeps are capped at the remaining deadline. The interactive REPL maintains one pending deadline per outbound message, reports and clears an unanswered request, then continues to receive later replies. - Dashboard-created agent chat sessions request the target agent's declared `metadata.skills` plus enabled plugin-contributed skills, so skills such as `ce-debug` are available in chat when the contributing plugin is enabled for the requesting project. Model-only QuickChat sessions request enabled plugin skills, and room responder sessions request the responder agent's skills. - Agent-acting session lanes share the same skill-injection contract as executor sessions: executor, merger, triage, reviewer, heartbeat, step-session, dashboard chat/room responders, CLI agent execution, planning, mission interview, milestone/slice interview, agent-onboarding interview, workflow design, memory dreams/insight extraction, and scheduled cron automation all request agent/fallback skills plus enabled plugin-contributed skills when a plugin runner is available. Utility-only lanes that only summarize/extract/generate JSON (title/PR summaries, memory compaction, subtask breakdown, text refinement, agent generation, PR metadata generation, evaluator/research synthesis, and similar one-shot helpers) intentionally stay exempt to avoid loading skills where no agent-style tool loop can use them. - In dashboard model-loop chat (main chat, QuickChat, and room responders), typing `/skill:{name}` requests that skill for the current AI session and strips the slash token from the prompt sent to the model. Slash and catalog-style names such as `/skill:review/pr`, `/skill:review/pr/SKILL.md`, and `source::skills/review/pr/SKILL.md` resolve to the matching discovered bare skill token across chat and agent session lanes. The requested skill is still subject to the normal enabled/disabled execution-skill filters; CLI-agent-backed PTY chat keeps raw terminal input semantics and does not interpret this command. @@ -63,8 +64,9 @@ For the user-facing gallery and notification UX, see [Artifacts View](./dashboar - `--once` send one message and exit after first reply (or timeout) - `--non-interactive` read full stdin to EOF as message body -- `--poll-ms ` override poll interval in milliseconds (default `1000`, or `FUSION_CHAT_POLL_MS`) -- `--conversation-id ` override the default named mailbox conversation +- `--poll-ms ` override poll interval in milliseconds (default `1000`, or `FUSION_CHAT_POLL_MS`); sleeps never extend past the nearest reply deadline +- `--reply-timeout-ms ` set the per-reply deadline in milliseconds (default `60000`, or `FUSION_CHAT_REPLY_TIMEOUT_MS`) +- `--conversation-id ` override the default named mailbox conversation ID ### Examples @@ -1052,8 +1054,8 @@ Mailbox replies use `message.metadata.replyTo.messageId` as the stable reply lin - `fn_read_messages` includes each message ID in its human-readable output so agents can target a specific message. - When a message has `metadata.replyTo.messageId`, `fn_read_messages` now includes one-level reply-parent context inline (and in structured tool details) so heartbeat/mailbox runs can understand what the message is replying to without expanding full threads. -- `fn_send_message` supports `reply_to_message_id`; when provided, the sent message is stored with `metadata.replyTo.messageId`. -- Heartbeat prompts explicitly instruct agents to include `reply_to_message_id` when replying. +- `fn_send_message` supports `reply_to_message_id`; when provided, the sent message is stored with `metadata.replyTo.messageId`. If that parent was addressed to the sending agent, the tool safely defaults its recipient to the parent's sender; foreign, missing, or misaddressed parents cannot supply a recipient. +- Heartbeat prompts explicitly instruct agents to include `reply_to_message_id` and the exact sender ID when replying. The dashboard mailbox UI also uses the same metadata contract when users click **Reply**, so user and agent replies share one threading model. diff --git a/docs/cli-reference.md b/docs/cli-reference.md index 0476aa67e5..58f80022ce 100644 --- a/docs/cli-reference.md +++ b/docs/cli-reference.md @@ -1031,6 +1031,7 @@ User mailbox operations for sending and managing direct messages with agents. ```bash fn message inbox +fn message inbox --user dashboard fn message outbox fn message send AGENT-001 "Please prioritize FN-222" fn message read MSG-123 @@ -1039,7 +1040,7 @@ fn message delete MSG-123 | Subcommand | Description | |---|---| -| `fn message inbox` | List your inbox messages (newest first, up to 20). | +| `fn message inbox [--user ]` | List the selected user mailbox (newest first, up to 20); defaults to the CLI mailbox. | | `fn message outbox` | List messages you sent (newest first, up to 20). | | `fn message send ` | Send a user→agent message and print the created message ID. | | `fn message read ` | Show one full message by ID and auto-mark it as read if unread. | @@ -1047,7 +1048,8 @@ fn message delete MSG-123 ### Mailbox behavior -- `inbox` header shows unread totals as `Inbox ( unread)`. +- `inbox` defaults to the separate CLI user mailbox (`cli`). Use `fn message inbox --user dashboard` to automate reads of the dashboard operator mailbox (`dashboard`) shown by the UI; the two identities are intentionally not unified. +- `inbox` header shows unread totals as `Inbox ( unread)` or `Dashboard Inbox ( unread)`. - Unread inbox rows are prefixed with `●`; read rows have no dot. - Inbox sender labels use `Agent ` for agent senders and raw user IDs for user senders. - Outbox recipient labels use `Agent ` for agent recipients. @@ -1062,29 +1064,30 @@ fn message delete MSG-123 | Option | Description | |---|---| | `--project ` | Route mailbox operations to a specific registered project (resolved via project context). Supported by all `fn message` subcommands. | +| `--user ` | Select the mailbox for `fn message inbox`; defaults to `cli`. | ### Related command -`fn agent mailbox ` is separate from `fn message`: it inspects an **agent-owned mailbox** (agent inbox view), while `fn message ...` manages the **CLI user mailbox**. +`fn agent mailbox ` is separate from `fn message`: it inspects an **agent-owned mailbox** (agent inbox view), while `fn message ...` manages the CLI or dashboard operator user mailbox. --- ## `fn chat` -Named mailbox conversation loop with a specific agent. It delivers through the agent's MessageStore inbox; it is **not** a dashboard ChatView/ChatStore session or a multi-agent room. +Interactive CLI conversation loop with a specific agent. ```bash -fn chat [message…] [--once] [--non-interactive] [--poll-ms ] [--conversation-id ] +fn chat [message…] [--once] [--non-interactive] [--poll-ms ] [--reply-timeout-ms ] [--conversation-id ] ``` ### Behavior - `fn chat ` starts an interactive mailbox-conversation REPL. - `fn chat ` sends one message and waits for a reply (`--once` implied). -- The default conversation id is `cli-chat:cli:`, stable for the CLI user and target agent in the selected project. The session banner prints it so a later invocation resumes the same named thread. -- Messages are sent as `user-to-agent` records from CLI user `cli` with `metadata.wakeRecipient=true`, `metadata.kind="cli-chat"`, and `metadata.conversationId`. -- `fn message send` remains a one-shot mailbox command and does not add `cli-chat` conversation metadata. -- Replies are polled from the CLI user inbox and printed only when they carry the active `conversationId` or use `replyTo.messageId` to reference a message already in that thread. This mailbox path does not create a dashboard chat session or a multi-agent room. +- Messages are sent as `user-to-agent` records from CLI user `cli` with `metadata.wakeRecipient=true`, `metadata.kind="cli-chat"`, and a durable `metadata.conversationId`. +- This is MessageStore mail plus polling, not token-streaming SSE. Replies are printed only when they carry the active conversation ID or reply to a known thread message. +- Agents replying through `fn_send_message` should pass `reply_to_message_id`; replies default to the original sender only when that parent message was addressed to the replying agent. +- One-shot chat has a reply deadline independent of the polling interval. Interactive chat tracks each outbound message independently: it prints a timeout for an unanswered request, clears that request, and continues the REPL for later messages. ### Options @@ -1092,8 +1095,9 @@ fn chat [message…] [--once] [--non-interactive] [--poll-ms ] [-- |---|---| | `--once` | Send one message and exit after first reply (or timeout). | | `--non-interactive` | Read full stdin to EOF as message body (useful for pipes/scripts). | -| `--poll-ms ` | Poll interval in milliseconds (default `1000`, or `FUSION_CHAT_POLL_MS`). | -| `--conversation-id ` | Override the default mailbox conversation id to name or share a thread. | +| `--poll-ms ` | Poll interval in milliseconds (default `1000`, or `FUSION_CHAT_POLL_MS`). Poll sleeps are capped at the nearest reply deadline. | +| `--reply-timeout-ms ` | Per-reply deadline in milliseconds (default `60000`, or `FUSION_CHAT_REPLY_TIMEOUT_MS`), independent of `--poll-ms`. | +| `--conversation-id ` | Override the default named mailbox conversation ID. | ### Examples @@ -1108,7 +1112,7 @@ fn chat agent-abc123 "status update?" printf "deploy report" | fn chat agent-abc123 --once --non-interactive ``` -> Agent replies require a running engine for the same project (for example `fn dashboard` or `fn serve`). Reply streaming and dashboard-operator inbox readability are tracked separately in FN-8424 / issue #2363. +> Agent replies require a running engine for the same project (for example `fn dashboard` or `fn serve`). > > See [Agents: Interactive CLI Chat](./agents.md#interactive-cli-chat) for agent-oriented details. @@ -1293,6 +1297,7 @@ Subcommands: `search`, `install`. | `--once` | `fn chat` | | `--non-interactive` | `fn chat` | | `--poll-ms` | `fn chat` | +| `--reply-timeout-ms` | `fn chat` | For configuration details used by these commands, see [Settings Reference](./settings-reference.md). diff --git a/packages/cli/src/bin.ts b/packages/cli/src/bin.ts index 5bb649cc65..f4eebb8337 100644 --- a/packages/cli/src/bin.ts +++ b/packages/cli/src/bin.ts @@ -448,13 +448,14 @@ PR: Export Fusion agents to an Agent Companies package directory (agent skills assigned via metadata.skills affect execution-time tools) fn agent mailbox View an agent's mailbox - fn message inbox List inbox messages + fn message inbox [--user ] + List CLI or dashboard operator inbox messages fn message outbox List sent messages fn message send Send a message to an agent fn message read Read a specific message fn message delete Delete a message - fn chat [message…] [--once] [--non-interactive] [--poll-ms ] [--conversation-id ] - Named mailbox conversation; delivers to agent inbox (not a chat room) + fn chat [message…] [--once] [--non-interactive] [--poll-ms ] [--reply-timeout-ms ] [--conversation-id ] + Named mailbox conversation with deadline-bounded inbox replies fn backup --create Create a database backup immediately fn backup --list List all database backups fn backup --restore Restore database from a backup file @@ -2069,7 +2070,12 @@ async function main() { const subcommand = args[1]; switch (subcommand) { case "inbox": { - await runMessageInbox(projectName); + const inboxUser = getFlagValue(args.slice(2), "--user"); + if (inboxUser !== undefined && inboxUser !== "cli" && inboxUser !== "dashboard") { + console.error("Usage: fn message inbox [--user ]"); + process.exit(1); + } + await runMessageInbox(projectName, inboxUser); break; } case "outbox": { @@ -2119,6 +2125,7 @@ async function main() { once: parsed.once, nonInteractive: parsed.nonInteractive, pollIntervalMs: parsed.pollIntervalMs, + replyTimeoutMs: parsed.replyTimeoutMs, conversationId: parsed.conversationId, input, }); diff --git a/packages/cli/src/commands/__tests__/chat.test.ts b/packages/cli/src/commands/__tests__/chat.test.ts index 1dd1bcdc2e..12042708ed 100644 --- a/packages/cli/src/commands/__tests__/chat.test.ts +++ b/packages/cli/src/commands/__tests__/chat.test.ts @@ -1,162 +1,172 @@ import { PassThrough, Readable } from "node:stream"; -import { beforeEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -const { - mockGetAgent, - mockGetConversation, - mockGetInbox, - mockSendMessage, - mockMarkAsRead, - mockClose, - mockCleanup, -} = vi.hoisted(() => ({ - mockGetAgent: vi.fn(), - mockGetConversation: vi.fn(), - mockGetInbox: vi.fn(), - mockSendMessage: vi.fn(), - mockMarkAsRead: vi.fn(), - mockClose: vi.fn(), - mockCleanup: vi.fn(), -})); +const mockGetAgent = vi.fn(); +const mockGetConversation = vi.fn(); +const mockGetInbox = vi.fn(); +const mockSendMessage = vi.fn(); +const mockMarkAsRead = vi.fn(); +const mockClose = vi.fn(); -vi.mock("@fusion/core", () => ({ - AgentStore: vi.fn(function () { - return { init: vi.fn(), getAgent: mockGetAgent, close: mockClose }; - }), - MessageStore: vi.fn(function () { - return { - getConversation: mockGetConversation, - getInbox: mockGetInbox, - sendMessage: mockSendMessage, - markAsRead: mockMarkAsRead, - }; - }), -})); +vi.mock("@fusion/core", () => { + class AgentStore { + init = vi.fn(async () => undefined); + getAgent = mockGetAgent; + close = mockClose; + } + class MessageStore { + getConversation = mockGetConversation; + getInbox = mockGetInbox; + sendMessage = mockSendMessage; + markAsRead = mockMarkAsRead; + } + return { AgentStore, MessageStore, DASHBOARD_USER_ID: "dashboard" }; +}); vi.mock("../../project-context.js", () => ({ - resolveAgentStoreBase: vi.fn().mockResolvedValue({ - rootDir: "/tmp/fusion-cli-chat-test", + resolveAgentStoreBase: vi.fn(async () => ({ + rootDir: "/tmp/chat-test", asyncLayer: {}, - cleanup: mockCleanup, - }), + cleanup: vi.fn(async () => undefined), + })), })); import { runChatInteractive } from "../chat.js"; -import { runMessageSend } from "../message.js"; -function outputBuffer(): { output: PassThrough; read: () => string } { +function outputBuffer() { const output = new PassThrough(); - const chunks: Buffer[] = []; - output.on("data", (chunk: Buffer) => chunks.push(chunk)); - return { output, read: () => Buffer.concat(chunks).toString("utf8") }; + let text = ""; + output.on("data", (chunk: Buffer) => { text += chunk.toString(); }); + return { output, text: () => text }; } -describe("runChatInteractive mailbox conversation", () => { - beforeEach(() => { - vi.clearAllMocks(); - mockGetAgent.mockResolvedValue({ id: "agent-001" }); - mockGetConversation.mockResolvedValue([]); - mockGetInbox.mockResolvedValue([]); - mockSendMessage.mockResolvedValue({ id: "msg-001" }); - mockCleanup.mockResolvedValue(undefined); - }); +function reply(id: string, content: string, replyTo?: string) { + return { + id, + fromId: "agent-a", + fromType: "agent" as const, + toId: "cli", + toType: "user" as const, + content, + type: "agent-to-user" as const, + read: false, + ...(replyTo ? { metadata: { replyTo: { messageId: replyTo } } } : {}), + createdAt: new Date(0).toISOString(), + updatedAt: new Date(0).toISOString(), + }; +} - it("stamps sequential default sends with one stable cli-chat conversation id", async () => { - for (const content of ["first", "second"]) { - await runChatInteractive("agent-001", { - once: true, - nonInteractive: true, - input: Readable.from(content), - output: outputBuffer().output, - replyTimeoutMs: 0, - }); - } +beforeEach(() => { + mockGetAgent.mockResolvedValue({ id: "agent-a" }); + mockGetConversation.mockResolvedValue([]); + mockGetInbox.mockResolvedValue([]); + mockSendMessage.mockImplementation(async () => ({ id: `outbound-${mockSendMessage.mock.calls.length}` })); + mockMarkAsRead.mockResolvedValue(undefined); +}); - expect(mockSendMessage).toHaveBeenCalledTimes(2); - const metadata = mockSendMessage.mock.calls.map(([input]) => input.metadata); - expect(metadata).toEqual([ - { wakeRecipient: true, kind: "cli-chat", conversationId: "cli-chat:cli:agent-001" }, - { wakeRecipient: true, kind: "cli-chat", conversationId: "cli-chat:cli:agent-001" }, - ]); - }); +afterEach(() => { + vi.useRealTimers(); + vi.clearAllMocks(); +}); - it("renders only history from its conversation id or replies to its messages", async () => { - mockGetConversation.mockResolvedValue([ - { id: "other", fromId: "agent-001", fromType: "agent", content: "other thread", type: "agent-to-user", read: false, createdAt: "2026-07-20T00:00:00.000Z", updatedAt: "2026-07-20T00:00:00.000Z", metadata: { conversationId: "other-thread" } }, - { id: "thread-message", fromId: "user:cli", fromType: "user", content: "thread start", type: "user-to-agent", read: true, createdAt: "2026-07-20T00:00:01.000Z", updatedAt: "2026-07-20T00:00:01.000Z", metadata: { conversationId: "custom-thread" } }, - { id: "thread-reply", fromId: "agent-001", fromType: "agent", content: "thread reply", type: "agent-to-user", read: false, createdAt: "2026-07-20T00:00:02.000Z", updatedAt: "2026-07-20T00:00:02.000Z", metadata: { replyTo: { messageId: "thread-message" } } }, - ]); - const buffer = outputBuffer(); +describe("runChatInteractive", () => { + it("prints a reply delivered to the CLI mailbox for a one-shot CLI message", async () => { + const { output, text } = outputBuffer(); + const replies = [reply("reply-1", "board review complete", "outbound-1")]; + mockGetInbox.mockImplementation(async () => replies); - await runChatInteractive("agent-001", { + const code = await runChatInteractive("agent-a", { once: true, nonInteractive: true, - input: Readable.from(""), - output: buffer.output, - conversationId: "custom-thread", - }); - - expect(buffer.read()).toContain("thread start"); - expect(buffer.read()).toContain("thread reply"); - expect(buffer.read()).not.toContain("other thread"); - }); - - it("renders only replies associated with the current conversation and leaves other mail unread", async () => { - mockGetInbox.mockResolvedValue([ - { id: "unrelated", fromId: "agent-001", fromType: "agent", content: "unrelated", type: "agent-to-user", read: false, createdAt: "2026-07-20T00:00:00.000Z", updatedAt: "2026-07-20T00:00:00.000Z" }, - { id: "reply", fromId: "agent-001", fromType: "agent", content: "associated", type: "agent-to-user", read: false, createdAt: "2026-07-20T00:00:01.000Z", updatedAt: "2026-07-20T00:00:01.000Z", metadata: { replyTo: { messageId: "msg-001" } } }, - ]); - const buffer = outputBuffer(); - - await runChatInteractive("agent-001", { - once: true, - nonInteractive: true, - input: Readable.from("hello"), - output: buffer.output, - replyTimeoutMs: 5, + input: Readable.from(["review the board"]), + output, pollIntervalMs: 1, + replyTimeoutMs: 100, }); - expect(buffer.read()).toContain("associated"); - expect(buffer.read()).not.toContain("unrelated"); - expect(mockMarkAsRead).toHaveBeenCalledWith("reply"); - expect(mockMarkAsRead).not.toHaveBeenCalledWith("unrelated"); + expect(code).toBe(0); + expect(mockSendMessage).toHaveBeenCalledWith(expect.objectContaining({ fromId: "cli", toId: "agent-a" })); + expect(text()).toContain("board review complete"); + expect(mockMarkAsRead).toHaveBeenCalledWith("reply-1"); }); - it("uses an explicit conversation id and names inbox delivery in the session banner", async () => { - const buffer = outputBuffer(); - await runChatInteractive("agent-001", { + it("returns at the reply deadline rather than a large poll interval", async () => { + vi.useFakeTimers(); + const { output } = outputBuffer(); + const command = runChatInteractive("agent-a", { once: true, nonInteractive: true, - input: Readable.from("hello"), - output: buffer.output, - conversationId: "custom-thread", - replyTimeoutMs: 0, + input: Readable.from(["ping"]), + output, + pollIntervalMs: 300_000, + replyTimeoutMs: 5_000, + }); + + await vi.advanceTimersByTimeAsync(0); + await vi.advanceTimersByTimeAsync(5_001); + await expect(command).resolves.toBe(0); + }); + + it("expires one interactive pending reply but keeps polling for a later reply", async () => { + vi.useFakeTimers(); + const input = new PassThrough(); + const { output, text } = outputBuffer(); + const replies: ReturnType[] = []; + mockGetInbox.mockImplementation(async () => replies); + + const command = runChatInteractive("agent-a", { + input, + output, + pollIntervalMs: 300_000, + replyTimeoutMs: 5_000, + }); + await vi.advanceTimersByTimeAsync(0); + + input.write("first request\n"); + await vi.advanceTimersByTimeAsync(0); + expect(mockSendMessage).toHaveBeenCalledTimes(1); + + await vi.advanceTimersByTimeAsync(5_001); + expect(text()).toContain("No reply within 5s for: first request"); + + input.write("second request\n"); + await vi.advanceTimersByTimeAsync(0); + expect(mockSendMessage).toHaveBeenCalledTimes(2); + replies.push(reply("reply-2", "second answer", "outbound-2")); + // The only scheduled poll is capped by the second request's own deadline, + // not the 300-second normal interval. + await vi.advanceTimersByTimeAsync(5_001); + + expect(text()).toContain("second answer"); + expect(text()).not.toContain("No reply within 5s for: second request"); + input.end("/exit\n"); + await expect(command).resolves.toBe(0); + }); +}); + +describe("named mailbox conversations", () => { + it("stamps a stable conversation ID and leaves unrelated agent mail unread", async () => { + const { output, text } = outputBuffer(); + mockGetInbox.mockResolvedValue([ + reply("other-thread", "other mailbox traffic"), + reply("thread-reply", "threaded answer", "outbound-1"), + ]); + + await runChatInteractive("agent-a", { + once: true, + nonInteractive: true, + input: Readable.from(["status"]), + output, + pollIntervalMs: 1, + replyTimeoutMs: 100, }); expect(mockSendMessage).toHaveBeenCalledWith(expect.objectContaining({ - metadata: { wakeRecipient: true, kind: "cli-chat", conversationId: "custom-thread" }, + metadata: { wakeRecipient: true, kind: "cli-chat", conversationId: "cli-chat:cli:agent-a" }, })); - expect(buffer.read()).toContain("Mailbox conversation"); - expect(buffer.read()).toContain("agent inbox"); - expect(buffer.read()).toContain("conversation-id: custom-thread"); - }); - - it("explains mailbox delivery and the conversation id in REPL help", async () => { - const buffer = outputBuffer(); - await runChatInteractive("agent-001", { - input: Readable.from("/help\n/exit\n"), - output: buffer.output, - pollIntervalMs: 1, - }); - - expect(buffer.read()).toContain("Mailbox delivery to the agent inbox"); - expect(buffer.read()).toContain("conversation-id: cli-chat:cli:agent-001"); - }); - - it("keeps fn message send as an unstamped one-shot message", async () => { - await runMessageSend("agent-001", "ordinary mail"); - - expect(mockSendMessage).toHaveBeenCalledWith(expect.not.objectContaining({ metadata: expect.anything() })); + expect(text()).toContain("threaded answer"); + expect(text()).not.toContain("other mailbox traffic"); + expect(mockMarkAsRead).toHaveBeenCalledWith("thread-reply"); + expect(mockMarkAsRead).not.toHaveBeenCalledWith("other-thread"); }); }); diff --git a/packages/cli/src/commands/__tests__/message.test.ts b/packages/cli/src/commands/__tests__/message.test.ts index 926ebb2ffe..41e3a0cb9e 100644 --- a/packages/cli/src/commands/__tests__/message.test.ts +++ b/packages/cli/src/commands/__tests__/message.test.ts @@ -32,6 +32,7 @@ vi.mock("@fusion/core", () => { }; return { createDatabase: vi.fn().mockReturnValue(mockDb), + DASHBOARD_USER_ID: "dashboard", MessageStore: makeConstructibleMock(() => ({ getInbox: mockGetInbox, getOutbox: mockGetOutbox, @@ -125,6 +126,17 @@ describe("runMessageInbox", () => { expect(logSpy).toHaveBeenCalledWith(expect.stringContaining("No messages")); }); + it("lists the distinct dashboard operator mailbox when requested", async () => { + mockGetMailbox.mockReturnValue({ unreadCount: 1, ownerId: "dashboard", ownerType: "user" }); + mockGetInbox.mockReturnValue([{ ...mockMessage, toId: "dashboard" }]); + + await runMessageInbox(undefined, "dashboard"); + + expect(mockGetMailbox).toHaveBeenCalledWith("dashboard", "user"); + expect(mockGetInbox).toHaveBeenCalledWith("dashboard", "user", { limit: 20 }); + expect(logSpy).toHaveBeenCalledWith(expect.stringContaining("Dashboard Inbox")); + }); + it("should show unread marker for unread messages", async () => { await runMessageInbox(); diff --git a/packages/cli/src/commands/chat.ts b/packages/cli/src/commands/chat.ts index 9922e2ba6e..d99eaf29f2 100644 --- a/packages/cli/src/commands/chat.ts +++ b/packages/cli/src/commands/chat.ts @@ -6,12 +6,12 @@ import { createInterface } from "node:readline/promises"; const MAX_MESSAGE_LENGTH = 8192; const DEFAULT_POLL_MS = 1000; +const DEFAULT_REPLY_TIMEOUT_MS = 60_000; const HISTORY_LIMIT = 20; /** * FNXC:CliChatConversation 2026-07-20-12:00: - * CLI chats use MessageStore's project-scoped mailbox transport, so the stable - * CLI-user/agent pair is sufficient to resume a named thread within a project. + * CLI chats use a durable MessageStore thread per CLI-user/agent pair. */ export function buildCliChatConversationId(agentId: string, override?: string): string { return override ?? `cli-chat:${CLI_USER_ID}:${agentId}`; @@ -28,50 +28,48 @@ export interface ChatInteractiveOptions { output?: NodeJS.WritableStream; } -export type ChatCliArgs = Pick & { +interface PendingReply { + outboundMessageId: string; + sentAt: number; + deadlineAt: number; + preview: string; +} + +export type ChatCliArgs = Pick & { agentId: string; contentArg: string; }; /** Parse chat-only argv after the `chat` command for dispatch and unit tests. */ export function parseChatCliArgs(args: string[]): ChatCliArgs | { error: string } { - const usage = "Usage: fn chat [message…] [--once] [--non-interactive] [--poll-ms ] [--conversation-id ]"; + const usage = "Usage: fn chat [message…] [--once] [--non-interactive] [--poll-ms ] [--reply-timeout-ms ] [--conversation-id ]"; const agentId = args[0]; if (!agentId) return { error: usage }; - - const pollIdx = args.indexOf("--poll-ms"); - const pollValue = pollIdx === -1 ? undefined : args[pollIdx + 1]; - const pollIntervalMs = pollValue === undefined ? undefined : Number.parseInt(pollValue, 10); - if (pollIdx !== -1 && (!pollValue || pollValue.startsWith("--") || !Number.isFinite(pollIntervalMs) || (pollIntervalMs ?? 0) <= 0)) { - return { error: usage }; - } - + const readPositiveFlag = (flag: string) => { + const index = args.indexOf(flag); + const value = index === -1 ? undefined : args[index + 1]; + const parsed = value === undefined ? undefined : Number.parseInt(value, 10); + return { index, parsed, valid: index === -1 || (!!value && !value.startsWith("--") && Number.isFinite(parsed) && (parsed ?? 0) > 0) }; + }; + const poll = readPositiveFlag("--poll-ms"); + const timeout = readPositiveFlag("--reply-timeout-ms"); + if (!poll.valid || !timeout.valid) return { error: usage }; let conversationId: string | undefined; for (let index = 1; index < args.length; index += 1) { if (args[index] !== "--conversation-id") continue; const value = args[index + 1]; - // FNXC:CliChatConversation 2026-07-20-14:30: Every occurrence must have a value. A first valid flag must not hide a later incomplete flag and silently route mail to the wrong thread. - if (conversationId !== undefined || !value || value.startsWith("--")) { - return { error: usage }; - } + if (conversationId !== undefined || !value || value.startsWith("--")) return { error: usage }; conversationId = value; index += 1; } - - const filteredArgs = args.slice(1).filter((arg, index, values) => { - if (arg === "--once" || arg === "--non-interactive" || arg === "--poll-ms" || arg === "--conversation-id") return false; - if (index > 0 && (values[index - 1] === "--poll-ms" || values[index - 1] === "--conversation-id")) return false; - return true; - }); - const contentArg = filteredArgs.join(" ").trim(); - return { - agentId, - conversationId, - pollIntervalMs: pollIdx === -1 ? undefined : pollIntervalMs, - contentArg, + const flagsWithValues = new Set(["--poll-ms", "--reply-timeout-ms", "--conversation-id"]); + const contentArg = args.slice(1).filter((arg, index, values) => + arg !== "--once" && arg !== "--non-interactive" && !flagsWithValues.has(arg) + && !(index > 0 && flagsWithValues.has(values[index - 1] ?? "")), + ).join(" ").trim(); + return { agentId, conversationId, pollIntervalMs: poll.parsed, replyTimeoutMs: timeout.parsed, contentArg, once: args.includes("--once") || contentArg.length > 0, - nonInteractive: args.includes("--non-interactive") || contentArg.length > 0, - }; + nonInteractive: args.includes("--non-interactive") || contentArg.length > 0 }; } /* @@ -103,11 +101,20 @@ async function createAgentStore(projectName?: string): Promise<{ store: AgentSto } } +function parsePositiveMs(value: number | undefined, fallback: number): number { + return Number.isFinite(value) && (value ?? 0) > 0 ? value! : fallback; +} + function parsePollMs(options: ChatInteractiveOptions): number { const envValue = process.env.FUSION_CHAT_POLL_MS; const envPollMs = envValue ? Number.parseInt(envValue, 10) : Number.NaN; - const candidate = options.pollIntervalMs ?? (Number.isFinite(envPollMs) ? envPollMs : DEFAULT_POLL_MS); - return Number.isFinite(candidate) && candidate > 0 ? candidate : DEFAULT_POLL_MS; + return parsePositiveMs(options.pollIntervalMs ?? envPollMs, DEFAULT_POLL_MS); +} + +function parseReplyTimeoutMs(options: ChatInteractiveOptions): number { + const envValue = process.env.FUSION_CHAT_REPLY_TIMEOUT_MS; + const envTimeoutMs = envValue ? Number.parseInt(envValue, 10) : Number.NaN; + return parsePositiveMs(options.replyTimeoutMs ?? envTimeoutMs, DEFAULT_REPLY_TIMEOUT_MS); } function printMessage(output: NodeJS.WritableStream, message: Message): void { @@ -124,16 +131,13 @@ function printConversationTail(output: NodeJS.WritableStream, messages: Message[ } output.write("\nRecent conversation:\n\n"); - for (const message of messages) { - printMessage(output, message); - } + for (const message of messages) printMessage(output, message); } /** * FNXC:CliChatConversation 2026-07-20-14:30: - * MessageStore queries are participant-wide, not conversation-scoped. A mailbox - * message belongs to this CLI thread only when it carries this id or replies to - * an already-known thread message; unassociated agent mail remains unread. + * Participant queries are not conversation-scoped; only thread-tagged mail or + * replies to known thread messages may be displayed or marked read. */ function collectConversationMessages(messages: Message[], conversationId: string, threadMessageIds = new Set()): Message[] { const includedIds = new Set(); @@ -143,14 +147,10 @@ function collectConversationMessages(messages: Message[], conversationId: string for (const message of messages) { if (includedIds.has(message.id)) continue; const directMatch = message.metadata?.conversationId === conversationId; - const replyMatch = typeof message.metadata?.replyTo?.messageId === "string" - && threadMessageIds.has(message.metadata.replyTo.messageId); + const replyMatch = typeof message.metadata?.replyTo?.messageId === "string" && threadMessageIds.has(message.metadata.replyTo.messageId); if (!threadMessageIds.has(message.id) && !directMatch && !replyMatch) continue; includedIds.add(message.id); - if (!threadMessageIds.has(message.id)) { - threadMessageIds.add(message.id); - changed = true; - } + if (!threadMessageIds.has(message.id)) { threadMessageIds.add(message.id); changed = true; } } } return messages.filter((message) => includedIds.has(message.id)); @@ -160,22 +160,64 @@ function isConversationReply(message: Message, conversationId: string, threadMes return collectConversationMessages([message], conversationId, threadMessageIds).length > 0; } -function sleep(ms: number, signal: AbortSignal): Promise { +function isCliReply(message: Message, agentId: string): boolean { + return message.fromId === agentId + && message.fromType === "agent" + && message.toId === CLI_USER_ID + && message.toType === "user"; +} + +function replyToId(message: Message): string | undefined { + return message.metadata?.replyTo?.messageId; +} + +function findPendingReply(pendingReplies: Map, message: Message): PendingReply | undefined { + const threaded = replyToId(message); + if (threaded) return pendingReplies.get(threaded); + // Replies without metadata retain useful behavior by consuming the oldest open request once. + return [...pendingReplies.values()].sort((a, b) => a.sentAt - b.sentAt)[0]; +} + +async function getChatReplies( + messageStore: Awaited>["store"], + agentId: string, + conversationId: string, + threadMessageIds: Set, +): Promise { + // Conversation lookup sees replies already marked read by another CLI process; inbox keeps the normal unread path cheap. + const [conversation, inbox] = await Promise.all([ + messageStore.getConversation({ id: CLI_USER_ID, type: "user" }, { id: agentId, type: "agent" }, { limit: 50 }), + messageStore.getInbox(CLI_USER_ID, "user", { limit: 50 }), + ]); + const messages = new Map(); + for (const message of [...conversation, ...inbox]) { + if (isCliReply(message, agentId) && isConversationReply(message, conversationId, threadMessageIds)) messages.set(message.id, message); + } + return [...messages.values()].sort((a, b) => a.createdAt.localeCompare(b.createdAt)); +} + +function sleep(ms: number, signal?: AbortSignal, wake?: Promise): Promise { return new Promise((resolve, reject) => { - const timer = setTimeout(resolve, ms); + const timer = setTimeout(done, ms); const onAbort = () => { clearTimeout(timer); reject(new Error("aborted")); }; - if (signal.aborted) { + function done() { + clearTimeout(timer); + signal?.removeEventListener("abort", onAbort); + resolve(); + } + if (signal?.aborted) { onAbort(); return; } - signal.addEventListener("abort", onAbort, { once: true }); + signal?.addEventListener("abort", onAbort, { once: true }); + void wake?.then(done); }); } -async function waitForReply( +export async function waitForReply( messageStore: Awaited>["store"], agentId: string, printedIds: Set, @@ -185,27 +227,48 @@ async function waitForReply( conversationId: string, threadMessageIds: Set, ): Promise { - const started = Date.now(); - while (Date.now() - started < timeoutMs) { - const inbox = await messageStore.getInbox(CLI_USER_ID, "user", { limit: 50 }); - for (const message of inbox.slice().reverse()) { - if (message.fromId !== agentId || message.fromType !== "agent") continue; - if (!isConversationReply(message, conversationId, threadMessageIds)) continue; + const deadlineAt = Date.now() + timeoutMs; + while (true) { + for (const message of await getChatReplies(messageStore, agentId, conversationId, threadMessageIds)) { if (printedIds.has(message.id)) continue; printedIds.add(message.id); printMessage(output, message); await messageStore.markAsRead(message.id); return true; } - await new Promise((resolve) => setTimeout(resolve, pollIntervalMs)); + + const remainingTimeoutMs = Math.max(0, deadlineAt - Date.now()); + if (remainingTimeoutMs === 0) return false; + /* + FNXC:CliChatReplyRouting 2026-07-20-12:00: + A user-selected poll interval must not postpone a one-shot reply timeout. + Bound every sleep to the remaining deadline so --poll-ms 300000 still exits + at the configured reply timeout rather than minutes later. + */ + await sleep(Math.min(pollIntervalMs, remainingTimeoutMs)); } - return false; +} + +function expirePendingReplies(pendingReplies: Map, output: NodeJS.WritableStream, now: number): void { + for (const [id, pending] of pendingReplies) { + if (now < pending.deadlineAt) continue; + pendingReplies.delete(id); + output.write(`No reply within ${Math.ceil((pending.deadlineAt - pending.sentAt) / 1000)}s for: ${pending.preview}\n`); + } +} + +function nearestPendingSleep(pendingReplies: Map, pollIntervalMs: number, now: number): number { + const nearestDeadline = Math.min(...[...pendingReplies.values()].map((pending) => pending.deadlineAt)); + return Number.isFinite(nearestDeadline) + ? Math.min(pollIntervalMs, Math.max(0, nearestDeadline - now)) + : pollIntervalMs; } export async function runChatInteractive(agentId: string, options: ChatInteractiveOptions = {}): Promise { const output = options.output ?? process.stdout; const input = options.input ?? process.stdin; const pollIntervalMs = parsePollMs(options); + const replyTimeoutMs = parseReplyTimeoutMs(options); const conversationId = buildCliChatConversationId(agentId, options.conversationId); const ownedAgentStore = await createAgentStore(options.project); @@ -224,72 +287,79 @@ export async function runChatInteractive(agentId: string, options: ChatInteracti const printedIds = new Set(); const conversation = await messageStore.getConversation( - { id: CLI_USER_ID, type: "user" }, - { id: agentId, type: "agent" }, - ); + { id: CLI_USER_ID, type: "user" }, + { id: agentId, type: "agent" }, + ); const threadMessageIds = new Set(); const tail = collectConversationMessages(conversation, conversationId, threadMessageIds).slice(-HISTORY_LIMIT); for (const message of tail) printedIds.add(message.id); - /* - FNXC:CliChatConversation 2026-07-20-12:00: - The CLI must name MessageStore inbox delivery honestly: this is a resumable - mailbox conversation, not a dashboard ChatView session or multi-agent room. - */ output.write(`Mailbox conversation with Agent ${agentId} — type /exit or Ctrl-C to quit, /help for commands\n`); output.write(`conversation-id: ${conversationId}\n`); output.write("Delivery: agent inbox (fn_read_messages). Not a dashboard chat session or multi-agent room.\n"); output.write("Replies appear when this project's engine is running (fn dashboard or fn serve).\n"); printConversationTail(output, tail); - const runOnce = options.once === true; - if (runOnce) { + if (options.once === true) { const content = await readSingleMessage(input, output, options.nonInteractive); if (!content.trim()) return 0; - if (content.length > MAX_MESSAGE_LENGTH) { console.error(`Message too long; max ${MAX_MESSAGE_LENGTH} chars`); return 0; } - const sentMessage = await messageStore.sendMessage({ + const outbound = await messageStore.sendMessage({ fromId: CLI_USER_ID, fromType: "user", toId: agentId, toType: "agent", content, type: "user-to-agent", - /* - FNXC:CliChatConversation 2026-07-20-12:00: - Keep wake-on-message inbox delivery for durable agents, but stamp every - CLI chat turn so agents can recognize the resumable mailbox thread. - */ metadata: { wakeRecipient: true, kind: "cli-chat", conversationId }, }); - + threadMessageIds.add(outbound.id); output.write(`you → ${agentId}: ${content}\n`); - const timeoutMs = options.replyTimeoutMs ?? Math.max(pollIntervalMs * 10, 30_000); - threadMessageIds.add(sentMessage.id); - const replied = await waitForReply(messageStore, agentId, printedIds, output, pollIntervalMs, timeoutMs, conversationId, threadMessageIds); - if (!replied) { - console.error(`No reply within ${Math.ceil(timeoutMs / 1000)}s`); - } + const replied = await waitForReply(messageStore, agentId, printedIds, output, pollIntervalMs, replyTimeoutMs, conversationId, threadMessageIds); + if (!replied) console.error(`No reply within ${Math.ceil(replyTimeoutMs / 1000)}s`); return 0; } const abortController = new AbortController(); + const pendingReplies = new Map(); + let wakePoller: (() => void) | undefined; const poller = (async () => { while (!abortController.signal.aborted) { - const inbox = await messageStore.getInbox(CLI_USER_ID, "user", { limit: 50 }); - for (const message of inbox.slice().reverse()) { - if (message.fromId !== agentId || message.fromType !== "agent") continue; - if (!isConversationReply(message, conversationId, threadMessageIds)) continue; + for (const message of await getChatReplies(messageStore, agentId, conversationId, threadMessageIds)) { if (printedIds.has(message.id)) continue; printedIds.add(message.id); + const pending = findPendingReply(pendingReplies, message); + /* + FNXC:CliChatReplyRouting 2026-07-20-12:00: + Polling can wake exactly at a pending deadline after a reply was + already persisted. Match replies created by that deadline before + expiring requests, otherwise the terminal falsely prints a timeout + immediately before the reply it has just retrieved. + */ + if (pending && Date.parse(message.createdAt) <= pending.deadlineAt) { + pendingReplies.delete(pending.outboundMessageId); + } printMessage(output, message); await messageStore.markAsRead(message.id); } - await sleep(pollIntervalMs, abortController.signal); + expirePendingReplies(pendingReplies, output, Date.now()); + const delay = nearestPendingSleep(pendingReplies, pollIntervalMs, Date.now()); + let resolveWake: (() => void) | undefined; + const wake = new Promise((resolve) => { resolveWake = resolve; }); + wakePoller = resolveWake; + /* + FNXC:CliChatReplyRouting 2026-07-20-12:00: + Interactive chat owns independent pending deadlines. Wake a normal poll + when a new outbound is registered, then cap its sleep at the nearest + pending deadline; timing out one request only clears that entry and the + REPL continues polling for later messages. + */ + await sleep(delay, abortController.signal, wake); + wakePoller = undefined; } })().catch(() => undefined); @@ -306,13 +376,12 @@ export async function runChatInteractive(agentId: string, options: ChatInteracti if (!line) continue; if (line === "/exit" || line === "/quit") break; if (line === "/help") { - output.write(`Commands: /help, /history, /clear, /exit, /quit. Mailbox delivery to the agent inbox; conversation-id: ${conversationId}\n`); + output.write(`Commands: /help, /history, /clear, /exit, /quit\nMailbox delivery to the agent inbox; conversation-id: ${conversationId}\n`); continue; } if (line === "/history") { const history = collectConversationMessages(await messageStore.getConversation( - { id: CLI_USER_ID, type: "user" }, - { id: agentId, type: "agent" }, + { id: CLI_USER_ID, type: "user" }, { id: agentId, type: "agent" }, ), conversationId, threadMessageIds).slice(-HISTORY_LIMIT); for (const message of history) printedIds.add(message.id); printConversationTail(output, history); @@ -327,21 +396,24 @@ export async function runChatInteractive(agentId: string, options: ChatInteracti continue; } - const sentMessage = await messageStore.sendMessage({ + const outbound = await messageStore.sendMessage({ fromId: CLI_USER_ID, fromType: "user", toId: agentId, toType: "agent", content: line, type: "user-to-agent", - /* - FNXC:CliChatConversation 2026-07-20-12:00: - REPL sends share the same conversation identity as once-mode sends; - MessageStore remains the transport rather than masquerading as a room. - */ metadata: { wakeRecipient: true, kind: "cli-chat", conversationId }, }); - threadMessageIds.add(sentMessage.id); + threadMessageIds.add(outbound.id); + const sentAt = Date.now(); + pendingReplies.set(outbound.id, { + outboundMessageId: outbound.id, + sentAt, + deadlineAt: sentAt + replyTimeoutMs, + preview: line.length > 80 ? `${line.slice(0, 80)}…` : line, + }); + wakePoller?.(); output.write(`you → ${agentId}: ${line}\n`); } @@ -355,21 +427,9 @@ export async function runChatInteractive(agentId: string, options: ChatInteracti } finally { /* FNXC:PostgresCliLifecycle 2026-07-14-22:55: Chat owns three independently-failing resources. Always attempt AgentStore, message database, and borrowed project teardown; report all cleanup failures without discarding an earlier command failure. */ const cleanupFailures: unknown[] = []; - try { - agentStore.close(); - } catch (error) { - cleanupFailures.push(error); - } - try { - await messageOwner?.db.close(); - } catch (error) { - cleanupFailures.push(error); - } - try { - await ownedAgentStore.cleanup(); - } catch (error) { - cleanupFailures.push(error); - } + try { agentStore.close(); } catch (error) { cleanupFailures.push(error); } + try { await messageOwner?.db.close(); } catch (error) { cleanupFailures.push(error); } + try { await ownedAgentStore.cleanup(); } catch (error) { cleanupFailures.push(error); } if (cleanupFailures.length > 0) { // eslint-disable-next-line no-unsafe-finally -- cleanup must aggregate with, rather than silently lose, the active command failure. throw new AggregateError( @@ -380,19 +440,12 @@ export async function runChatInteractive(agentId: string, options: ChatInteracti } } -async function readSingleMessage( - input: NodeJS.ReadableStream, - output: NodeJS.WritableStream, - nonInteractive?: boolean, -): Promise { +async function readSingleMessage(input: NodeJS.ReadableStream, output: NodeJS.WritableStream, nonInteractive?: boolean): Promise { if (nonInteractive) { const chunks: Buffer[] = []; - for await (const chunk of input) { - chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk))); - } + for await (const chunk of input) chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk))); return Buffer.concat(chunks).toString("utf8").trimEnd(); } - const rl = createInterface({ input, output }); try { return await rl.question(""); diff --git a/packages/cli/src/commands/message.ts b/packages/cli/src/commands/message.ts index 338d09a405..2b73ec29cd 100644 --- a/packages/cli/src/commands/message.ts +++ b/packages/cli/src/commands/message.ts @@ -1,4 +1,4 @@ -import { MessageStore } from "@fusion/core"; +import { DASHBOARD_USER_ID, MessageStore } from "@fusion/core"; import type { ParticipantType } from "@fusion/core"; import { resolveAgentStoreBase } from "../project-context.js"; @@ -23,16 +23,22 @@ export async function createMessageStore(projectName?: string): Promise<{ store: export const CLI_USER_ID = "cli"; /** - * List inbox messages. + * List inbox messages for the CLI or dashboard operator mailbox. + * + * FNXC:CliChatReplyRouting 2026-07-20-12:00: + * Fusion deliberately keeps CLI (`cli`) and dashboard (`dashboard`) user + * mailboxes distinct. Default to CLI mail for backwards compatibility, while + * allowing automation to inspect the dashboard mailbox where legacy replies + * may have landed. */ -export async function runMessageInbox(projectName?: string): Promise { +export async function runMessageInbox(projectName?: string, ownerId = CLI_USER_ID): Promise { const { store, db } = await createMessageStore(projectName); try { - const mailbox = await store.getMailbox(CLI_USER_ID, "user"); - const messages = await store.getInbox(CLI_USER_ID, "user", { limit: 20 }); + const mailbox = await store.getMailbox(ownerId, "user"); + const messages = await store.getInbox(ownerId, "user", { limit: 20 }); console.log(); - console.log(` 📬 Inbox (${mailbox.unreadCount} unread)`); + console.log(` 📬 ${ownerId === DASHBOARD_USER_ID ? "Dashboard Inbox" : "Inbox"} (${mailbox.unreadCount} unread)`); console.log(); if (messages.length === 0) { diff --git a/packages/engine/src/__tests__/agent-tools-send-message-recipient-validation.test.ts b/packages/engine/src/__tests__/agent-tools-send-message-recipient-validation.test.ts index fad1d8ed5a..fe60550bc4 100644 --- a/packages/engine/src/__tests__/agent-tools-send-message-recipient-validation.test.ts +++ b/packages/engine/src/__tests__/agent-tools-send-message-recipient-validation.test.ts @@ -8,15 +8,16 @@ function firstText(result: { content: Array<{ type: string; text?: string }> }): return first?.type === "text" ? (first.text ?? "") : ""; } -function createMessageStoreHarness() { +function createMessageStoreHarness(parent?: Record | null) { const wakeSpy = vi.fn(); + const getMessage = vi.fn(async () => parent ?? null); const sendMessage = vi.fn(async (input: Record) => { if (input.toType === "agent") { await wakeSpy(input); } return { id: "msg-1" }; }); - return { messageStore: { sendMessage }, sendMessage, wakeSpy }; + return { messageStore: { getMessage, sendMessage }, getMessage, sendMessage, wakeSpy }; } async function executeSend( @@ -75,6 +76,87 @@ describe("createSendMessageTool recipient validation", () => { expect(firstText(await executeSend(legacyTool, { to_id: "agent-b", content: "hello", reply_to_message_id: " " }) as never)).toBe("ERROR: reply_to_message_id must be a non-empty string"); }); + it("routes an owned CLI parent reply to the CLI user mailbox", async () => { + const parent = { + id: "parent-cli", + fromId: "cli", + fromType: "user", + toId: "agent-a", + toType: "agent", + }; + const { messageStore, sendMessage } = createMessageStoreHarness(parent); + const tool = createSendMessageTool(messageStore as never, "agent-a"); + + // Heartbeat guidance intentionally names the sender explicitly; it must still + // preserve the parent user's type rather than treating `cli` as an agent. + const result = await executeSend(tool, { content: "received", reply_to_message_id: "parent-cli", to_id: "cli" }); + + expect(firstText(result as never)).toContain("Message sent to cli"); + expect(sendMessage).toHaveBeenCalledWith(expect.objectContaining({ + toId: "cli", + toType: "user", + type: "agent-to-user", + metadata: { replyTo: { messageId: "parent-cli" } }, + })); + }); + + it("routes an owned dashboard parent reply to the dashboard mailbox", async () => { + const parent = { + id: "parent-dashboard", + fromId: "dashboard", + fromType: "user", + toId: "agent-a", + toType: "agent", + }; + const { messageStore, sendMessage } = createMessageStoreHarness(parent); + const tool = createSendMessageTool(messageStore as never, "agent-a"); + + await executeSend(tool, { content: "received", reply_to_message_id: "parent-dashboard" }); + + expect(sendMessage).toHaveBeenCalledWith(expect.objectContaining({ + toId: "dashboard", + toType: "user", + type: "agent-to-user", + })); + }); + + it("allows an explicit alternate recipient without inheriting a foreign parent", async () => { + const foreignParent = { + id: "parent-foreign", + fromId: "cli", + fromType: "user", + toId: "agent-b", + toType: "agent", + }; + const { messageStore, sendMessage } = createMessageStoreHarness(foreignParent); + const tool = createSendMessageTool(messageStore as never, "agent-a"); + + await executeSend(tool, { content: "forward", reply_to_message_id: "parent-foreign", to_id: "agent-c" }); + + // A different explicit ID is a forward: without type it keeps the legacy + // agent-to-agent default instead of inheriting the foreign parent's user type. + expect(sendMessage).toHaveBeenCalledWith(expect.objectContaining({ toId: "agent-c", toType: "agent", type: "agent-to-agent" })); + }); + + it("rejects foreign or missing parents when no explicit recipient is supplied", async () => { + const foreignParent = { + id: "parent-foreign", + fromId: "cli", + fromType: "user", + toId: "agent-b", + toType: "agent", + }; + const foreignHarness = createMessageStoreHarness(foreignParent); + const foreignTool = createSendMessageTool(foreignHarness.messageStore as never, "agent-a"); + const missingHarness = createMessageStoreHarness(); + const missingTool = createSendMessageTool(missingHarness.messageStore as never, "agent-a"); + + expect(firstText(await executeSend(foreignTool, { content: "nope", reply_to_message_id: "parent-foreign" }) as never)).toMatch(/^ERROR: reply_to_message_id/); + expect(foreignHarness.sendMessage).not.toHaveBeenCalled(); + expect(firstText(await executeSend(missingTool, { content: "nope", reply_to_message_id: "missing" }) as never)).toMatch(/^ERROR: reply_to_message_id/); + expect(missingHarness.sendMessage).not.toHaveBeenCalled(); + }); + it("does not report delivery when recipient validation is unavailable", async () => { const failedLookupHarness = createMessageStoreHarness(); const failedLookupTool = createSendMessageTool(failedLookupHarness.messageStore as never, "agent-a", { diff --git a/packages/engine/src/agent-heartbeat-prompts.ts b/packages/engine/src/agent-heartbeat-prompts.ts index 343cb16155..cb1a3d9985 100644 --- a/packages/engine/src/agent-heartbeat-prompts.ts +++ b/packages/engine/src/agent-heartbeat-prompts.ts @@ -107,7 +107,7 @@ When you are woken by an incoming message (source includes "wake-on-message"), y 2. For each message, classify it: informational, question, request, or escalation. 3. Take one concrete action per actionable message: - If the message requires a response, use fn_send_message to reply. - - When replying, include 'reply_to_message_id' with the original message ID from fn_read_messages output. + - When replying, include 'reply_to_message_id' and set 'to_id' to the exact [from: type:id] ID from fn_read_messages (including cli). Omit 'to_id' only to use the safe reply-to-parent default for a message addressed to you. - If the message is informational, acknowledge it by logging with fn_task_log. - If the message requests net-new work, first check whether an open task already covers it; idle/no-task heartbeats may create only with approved Feature → Slice → Milestone → Mission lineage. - If ownership is clear and an agent is available, delegate only approved mission-linked work using fn_delegate_task. @@ -123,7 +123,7 @@ Example flow: When sending messages: - Be concise and clear about what you need or what you've done. -- Use 'reply_to_message_id' when replying so threaded conversations stay linked. +- Use 'reply_to_message_id' when replying so threaded conversations stay linked, and use the exact sender ID reported by fn_read_messages (including cli). - Include relevant context (task IDs, file paths) in metadata when applicable. - Use agent-to-agent for inter-agent communication.`; @@ -209,7 +209,7 @@ When you are woken by an incoming message (source includes "wake-on-message"), y 1. If fn_read_messages is available, use it to check your inbox for unread messages. 2. Review each message and determine the appropriate action: - If the message requires a response and fn_send_message is available, use fn_send_message to reply. - - When replying, include 'reply_to_message_id' with the original message ID from fn_read_messages output. + - When replying, include 'reply_to_message_id' and set 'to_id' to the exact [from: type:id] ID from fn_read_messages (including cli). Omit 'to_id' only to use the safe reply-to-parent default for a message addressed to you. - If the message is informational, acknowledge it and respond via fn_send_message when appropriate. - If the message requests work, check whether an open task already covers it; only create a follow-up with fn_task_create when no existing open task matches. - If the request has a clear owner and fn_delegate_task is available, delegate it directly. @@ -221,7 +221,7 @@ Example flow: When sending messages: - Be concise and clear about what you need or what you've done. -- Use 'reply_to_message_id' when replying so threaded conversations stay linked. +- Use 'reply_to_message_id' when replying so threaded conversations stay linked, and use the exact sender ID reported by fn_read_messages (including cli). - Include relevant context (task IDs, file paths) in metadata when applicable. - Use agent-to-agent for inter-agent communication.`; diff --git a/packages/engine/src/agent-tools.ts b/packages/engine/src/agent-tools.ts index 85c3471396..7b9d8f9cee 100644 --- a/packages/engine/src/agent-tools.ts +++ b/packages/engine/src/agent-tools.ts @@ -450,14 +450,14 @@ export const deleteAgentParams = Type.Object({ }); export const sendMessageParams = Type.Object({ - to_id: Type.String({ description: "Recipient ID (agent ID or user ID, depending on message type)" }), + to_id: Type.Optional(Type.String({ description: "Recipient ID. When replying, omit to deliver to the parent sender; otherwise provide the exact ID from fn_read_messages." })), content: Type.String({ description: "Message body (1-2000 characters)" }), type: Type.Optional(Type.Union([ Type.Literal("agent-to-agent"), Type.Literal("agent-to-user"), - ], { description: "Message type (defaults to 'agent-to-agent')" })), + ], { description: "Message type. Required for explicit non-dashboard user recipients; inferred from a valid reply parent when omitted." })), reply_to_message_id: Type.Optional( - Type.String({ description: "Optional ID of the message you are replying to (use IDs from fn_read_messages output)" }), + Type.String({ description: "Optional ID of the message you are replying to. Parent-based recipient inference is allowed only when that message was addressed to you." }), ), }); @@ -4697,8 +4697,9 @@ export function createSendMessageTool( label: "Send Message", description: "Send a message to another agent or user. The recipient will be woken if they have " + - "`messageResponseMode: 'immediate'` configured. When replying to an existing message, " + - "include `reply_to_message_id` to preserve threading.", + "`messageResponseMode: 'immediate'` configured. When replying, include `reply_to_message_id`; omit " + + "`to_id` to reply to that message's sender only when the parent was addressed to you. Otherwise provide " + + "the exact recipient ID and appropriate type explicitly.", parameters: sendMessageParams, // eslint-disable-next-line @typescript-eslint/no-explicit-any execute: async (_id: string, params: Static, _signal?: any, _onUpdate?: any, _ctx?: any) => { @@ -4717,13 +4718,6 @@ export function createSendMessageTool( } try { - const inferredDashboardRecipient = normalizeMessageParticipant(params.to_id, "user"); - const messageType = params.type - ?? (inferredDashboardRecipient.id === DASHBOARD_USER_ID ? "agent-to-user" : "agent-to-agent"); - const recipientType: "user" | "agent" = messageType === "agent-to-user" ? "user" : "agent"; - const recipient = recipientType === "user" - ? normalizeMessageParticipant(params.to_id, recipientType) - : { id: params.to_id, type: recipientType }; const replyToMessageId = params.reply_to_message_id?.trim(); if (params.reply_to_message_id !== undefined && !replyToMessageId) { @@ -4733,6 +4727,53 @@ export function createSendMessageTool( }; } + /* + FNXC:CliChatReplyRouting 2026-07-20-12:00: + CLI mail belongs to `cli`, while dashboard mail belongs to `dashboard`. + FN-8424 requires a reply to a message addressed to this agent to default + to that parent's sender, preserving both mailbox identities. A foreign, + missing, or non-agent-addressed parent must never supply routing data: + agents may still intentionally name an explicit recipient, but cannot + launder a recipient through another agent's reply thread. + */ + const parent = replyToMessageId ? await messageStore.getMessage(replyToMessageId) : undefined; + const parentWasAddressedToSender = parent != null + && normalizeMessageParticipant(parent.toId, parent.toType).id === fromAgentId + && parent.toType === "agent"; + if (replyToMessageId && !parentWasAddressedToSender && !params.to_id?.trim()) { + return { + content: [{ type: "text" as const, text: "ERROR: reply_to_message_id does not reference a message addressed to this agent; provide an explicit to_id to send intentionally" }], + details: {}, + }; + } + + const parentRecipient = parentWasAddressedToSender && parent + ? normalizeMessageParticipant(parent.fromId, parent.fromType) + : undefined; + const explicitRecipientId = params.to_id?.trim(); + const recipientId = explicitRecipientId ?? parentRecipient?.id; + // FNXC:CliChatReplyRouting 2026-07-20-12:00: An explicit recipient that names the valid parent sender remains a reply, so it inherits that sender's participant type (notably `cli` -> user). A different explicit ID is an intentional forward and retains the legacy agent-to-agent default unless its type is stated. + const explicitTargetsParent = explicitRecipientId != null + && parentRecipient != null + && normalizeMessageParticipant(explicitRecipientId, parentRecipient.type).id === parentRecipient.id; + const recipientParticipantType = !explicitRecipientId || explicitTargetsParent + ? parentRecipient?.type + : undefined; + if (!recipientId) { + return { + content: [{ type: "text" as const, text: "ERROR: to_id is required unless replying to a message addressed to this agent" }], + details: {}, + }; + } + + const inferredDashboardRecipient = normalizeMessageParticipant(recipientId, "user"); + const messageType = params.type + ?? (recipientParticipantType === "user" || inferredDashboardRecipient.id === DASHBOARD_USER_ID ? "agent-to-user" : "agent-to-agent"); + const recipientType: "user" | "agent" = messageType === "agent-to-user" ? "user" : "agent"; + const recipient = recipientType === "user" + ? normalizeMessageParticipant(recipientId, recipientType) + : { id: recipientId, type: recipientType }; + /* FNXC:AgentMessaging 2026-07-28-12:10: Agent-to-agent sends must reject missing recipients rather than store an unread, undeliverable message and report false delivery success. Use async getAgent instead of getCachedAgent because the synchronous cache always returns null in PostgreSQL mode. A lookup failure is validation-unavailable and must block the send; only a successful lookup may establish delivery confidence. @@ -4743,13 +4784,13 @@ export function createSendMessageTool( resolvedRecipient = await options.agentStore.getAgent(recipient.id); } catch { return { - content: [{ type: "text" as const, text: `ERROR: Recipient agent '${params.to_id}' could not be validated — message not sent` }], + content: [{ type: "text" as const, text: `ERROR: Recipient agent '${recipient.id}' could not be validated — message not sent` }], details: {}, }; } if (resolvedRecipient == null) { return { - content: [{ type: "text" as const, text: `ERROR: Recipient agent '${params.to_id}' does not exist — message not sent` }], + content: [{ type: "text" as const, text: `ERROR: Recipient agent '${recipient.id}' does not exist — message not sent` }], details: {}, }; } @@ -4788,7 +4829,7 @@ export function createSendMessageTool( return { content: [{ type: "text" as const, - text: `Message sent to ${recipient.id === DASHBOARD_USER_ID ? DASHBOARD_USER_ID : params.to_id} (ID: ${result.value.id})`, + text: `Message sent to ${recipient.id} (ID: ${result.value.id})`, }], details: { messageId: result.value.id }, };