From 7c034f486204b54162a1fa5932a771d3f9792158 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sun, 9 Aug 2026 03:15:00 -0700 Subject: [PATCH] FN-8888: emit approval mail for triage pauses Route triage approval pauses through the shared structural-mail helper. - Pass the runtime message store into triage processing. - Emit idempotent, fail-soft approval mail when triage pauses for approval. - Cover triage approval-mail delivery and publish the feature changeset. Files changed: .changeset/fn-8888-triage-approval-mail.md | 7 ++ .../src/__tests__/approval-mail-emission.test.ts | 101 +++++++++++++++++++++ packages/engine/src/runtimes/in-process-runtime.ts | 1 + packages/engine/src/triage.ts | 14 +++ 4 files changed, 123 insertions(+) Fusion-Task-Id: FN-8888 Fusion-Task-Lineage: f0970c59-abca-4081-b213-c9aec0833416 Co-authored-by: Fusion (runfusion.ai) --- .changeset/fn-8888-triage-approval-mail.md | 7 ++ .../__tests__/approval-mail-emission.test.ts | 101 ++++++++++++++++++ .../engine/src/runtimes/in-process-runtime.ts | 1 + packages/engine/src/triage.ts | 14 +++ 4 files changed, 123 insertions(+) create mode 100644 .changeset/fn-8888-triage-approval-mail.md diff --git a/.changeset/fn-8888-triage-approval-mail.md b/.changeset/fn-8888-triage-approval-mail.md new file mode 100644 index 0000000000..b0afa4bef8 --- /dev/null +++ b/.changeset/fn-8888-triage-approval-mail.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": minor +--- + +summary: Approvals raised while planning a task now appear in the mailbox. +category: feature +dev: Reuses emitApprovalMail through an optional TriageProcessor message-store option. diff --git a/packages/engine/src/__tests__/approval-mail-emission.test.ts b/packages/engine/src/__tests__/approval-mail-emission.test.ts index ce5a03717c..fa7a75434f 100644 --- a/packages/engine/src/__tests__/approval-mail-emission.test.ts +++ b/packages/engine/src/__tests__/approval-mail-emission.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it, vi } from "vitest"; import { DASHBOARD_USER_ID } from "@fusion/core"; import { TaskExecutor } from "../executor.js"; import { HeartbeatMonitor } from "../agent-heartbeat.js"; +import { TriageProcessor } from "../triage.js"; import { emitApprovalMail } from "../agents/approval-mail.js"; function createMessageStore(options: { reject?: boolean } = {}) { @@ -134,4 +135,104 @@ describe("approval-mail emission", () => { expect(taskStore.pauseTask).toHaveBeenCalledOnce(); expect(agentStore.updateAgentState).toHaveBeenCalledOnce(); }); + + /* + FNXC:StructuralMail 2026-08-09-09:57: + Triage was the one FN-8870-deferred approval-pause closure. Cover delivery, re-raise dedupe, + fail-soft persistence, optional wiring, and fallback actor identity so planning pauses cannot become + mailbox-invisible while preserving their existing task and agent pause behavior. + */ + describe("triage action gate", () => { + function createTriageHarness(options: { messageStore?: ReturnType; agent?: typeof heartbeatAgent | null } = {}) { + const taskStore = createExecutorStore(); + const agentStore = { + updateAgentState: vi.fn().mockResolvedValue(undefined), + updateAgent: vi.fn().mockResolvedValue(undefined), + }; + const processor = new TriageProcessor(taskStore as never, "/tmp/test", { + ...(options.messageStore ? { messageStore: options.messageStore as never } : {}), + agentStore: agentStore as never, + }); + const taskId = "FN-triage"; + const context = (processor as any).buildActionGateContext(taskId, "run-1", options.agent === undefined ? heartbeatAgent : options.agent); + return { taskStore, agentStore, context, taskId }; + } + + it("writes one reference-only mailbox item for the triage approval pause", async () => { + const messageStore = createMessageStore(); + const { context, taskId } = createTriageHarness({ messageStore }); + + await context.pauseForApproval({ approvalRequestId: "triage-action", decision: actionDecision }); + await new Promise((resolve) => setImmediate(resolve)); + + expect(messageStore.sendMessageOnce).toHaveBeenCalledTimes(1); + expect(messageStore.sendMessageOnce).toHaveBeenCalledWith( + expect.objectContaining({ + fromId: "system", + toId: DASHBOARD_USER_ID, + type: "system", + metadata: { mailKind: "approval", approvalRequestId: "triage-action", taskId }, + }), + "approval-mail:triage-action", + ); + const message = messageStore.persisted.get("approval-mail:triage-action") as { metadata: Record }; + expect(message.metadata).not.toHaveProperty("status"); + expect(message.metadata).not.toHaveProperty("decision"); + }); + + it("keeps a re-raised triage approval to one durable mailbox row", async () => { + const messageStore = createMessageStore(); + const { context } = createTriageHarness({ messageStore }); + + await context.pauseForApproval({ approvalRequestId: "triage-repeat", decision: actionDecision }); + await context.pauseForApproval({ approvalRequestId: "triage-repeat", decision: actionDecision }); + await new Promise((resolve) => setImmediate(resolve)); + + expect(messageStore.sendMessageOnce).toHaveBeenCalledTimes(2); + expect(messageStore.persisted).toHaveLength(1); + }); + + it("keeps the triage pause and agent updates intact when mailbox persistence rejects", async () => { + const messageStore = createMessageStore({ reject: true }); + const { context, taskStore, agentStore } = createTriageHarness({ messageStore }); + + await expect(context.pauseForApproval({ approvalRequestId: "triage-reject", decision: actionDecision })).resolves.toBeUndefined(); + await new Promise((resolve) => setImmediate(resolve)); + + expect(taskStore.pauseTask).toHaveBeenCalledOnce(); + expect(taskStore.logEntry).toHaveBeenCalledOnce(); + expect(agentStore.updateAgentState).toHaveBeenCalledOnce(); + expect(agentStore.updateAgentState).toHaveBeenCalledWith("agent-1", "paused"); + expect(agentStore.updateAgent).toHaveBeenCalledOnce(); + expect(agentStore.updateAgent).toHaveBeenCalledWith("agent-1", { pauseReason: "awaiting-approval" }); + expect(messageStore.sendMessageOnce).toHaveBeenCalledOnce(); + }); + + it("keeps the triage pause usable without a message store", async () => { + const { context, taskStore, agentStore } = createTriageHarness(); + + await expect(context.pauseForApproval({ approvalRequestId: "triage-no-mail", decision: actionDecision })).resolves.toBeUndefined(); + + expect(taskStore.pauseTask).toHaveBeenCalledOnce(); + expect(taskStore.logEntry).toHaveBeenCalledOnce(); + expect(agentStore.updateAgentState).toHaveBeenCalledOnce(); + expect(agentStore.updateAgentState).toHaveBeenCalledWith("agent-1", "paused"); + expect(agentStore.updateAgent).toHaveBeenCalledOnce(); + expect(agentStore.updateAgent).toHaveBeenCalledWith("agent-1", { pauseReason: "awaiting-approval" }); + }); + + it("emits triage approval mail with fallback identity when no agent is assigned", async () => { + const messageStore = createMessageStore(); + const { context, agentStore } = createTriageHarness({ messageStore, agent: null }); + + await context.pauseForApproval({ approvalRequestId: "triage-null-agent", decision: actionDecision }); + await new Promise((resolve) => setImmediate(resolve)); + + expect(messageStore.persisted.get("approval-mail:triage-null-agent")).toEqual(expect.objectContaining({ + content: expect.stringContaining("Triage planner FN-triage"), + })); + expect(agentStore.updateAgentState).not.toHaveBeenCalled(); + expect(agentStore.updateAgent).not.toHaveBeenCalled(); + }); + }); }); diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index 2b6f344676..bd6f70b502 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -1624,6 +1624,7 @@ export class InProcessRuntime stuckTaskDetector: this.stuckTaskDetector, usageLimitPauser: this.usageLimitPauser, agentStore: this.agentStore, + messageStore: this.messageStore, pluginRunner: this.pluginRunner, // FNXC:NodeWorktreeIsolation 2026-07-25-22:10: planning acquires (or reuses) the task's own // worktree through the executor's acquisition path, so no lane runs in the shared checkout. diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index 2461f02fac..6f6a3b1faa 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -165,6 +165,7 @@ import { type AgentSemaphore, } from "./concurrency/concurrency.js"; import { AgentLogger } from "./agents/agent-logger.js"; +import { emitApprovalMail } from "./agents/approval-mail.js"; import { acquireActiveSessionPath, activeSessionRegistry } from "./agents/active-session-registry.js"; import { resolveAgentInstructions, @@ -270,6 +271,12 @@ export interface TriageProcessorOptions { onAgentText?: (taskId: string, delta: string) => void; /** AgentStore for resolving per-agent custom instructions. */ agentStore?: import("@fusion/core").AgentStore; + /* + FNXC:StructuralMail 2026-08-09-09:57: + The triage approval gate receives this store solely to deliver its mailbox item. It remains optional so + existing callers and tests stay compatible; without it, approval mail is a no-op rather than a failure. + */ + messageStore?: import("@fusion/core").MessageStore; /** Plugin runner for runtime selection. When provided, enables plugin runtime lookup. */ pluginRunner?: import("./plugins/plugin-runner.js").PluginRunner; /* @@ -530,6 +537,13 @@ export class TriageProcessor { await this.options.agentStore.updateAgentState(agent.id, "paused"); await this.options.agentStore.updateAgent(agent.id, { pauseReason: "awaiting-approval" }); } + /* + FNXC:StructuralMail 2026-08-09-09:57: + FN-8870 deliberately left triage out of its approval-mail coverage. The shared helper owns + `approval-mail:` idempotency and fail-soft behavior; do not recreate either here. + Without a message store it is a silent no-op, preserving the approval-pause path. + */ + void emitApprovalMail({ messageStore: this.options.messageStore, approvalRequestId, toolName: decision.toolName, taskId, agentId: agent?.id ?? actorId, agentName: agent?.name ?? actorName }); queueMicrotask(() => this.activeSessions.get(taskId)?.dispose()); }, markApprovalCompleted: async (approvalRequestId) => {