diff --git a/packages/dashboard/src/__tests__/routes-approval.test.ts b/packages/dashboard/src/__tests__/routes-approval.test.ts index 236a4b1161..c4bd9d2aee 100644 --- a/packages/dashboard/src/__tests__/routes-approval.test.ts +++ b/packages/dashboard/src/__tests__/routes-approval.test.ts @@ -319,6 +319,9 @@ describe("approval routes", async () => { ); expect(res.status).toBe(200); expect(res.body.status).toBe("denied"); + expect(state.task.paused).toBe(false); + expect(state.pauseTaskCalls).toEqual([{ id: "FN-1", paused: false }]); + expect(updateAgent).toHaveBeenCalledWith("agent-1", { pauseReason: undefined }); }); it("approves provisioning create and records audit", async () => { diff --git a/packages/engine/src/__tests__/agent-action-gate.test.ts b/packages/engine/src/__tests__/agent-action-gate.test.ts index e7d8fdf84b..5d2c5305ca 100644 --- a/packages/engine/src/__tests__/agent-action-gate.test.ts +++ b/packages/engine/src/__tests__/agent-action-gate.test.ts @@ -122,6 +122,75 @@ describe("agent-action-gate", () => { expect(result.resourceType).toBe("command"); }); + it("classifies namespaced MCP tools as governed network API actions", () => { + const decision = evaluateAgentActionGate({ + agentId: "agent-1", + taskId: "MAIN-008", + toolName: "mcp__postiz__integrationlist", + args: {}, + permissionPolicy: approvalPolicy, + }); + + expect(decision).toMatchObject({ + category: "network_api", + disposition: "require-approval", + operation: "mcp__postiz__integrationlist", + resourceType: "research", + }); + }); + + it("executes an approved namespaced MCP operation once and never executes a denied one", async () => { + const execute = vi.fn().mockResolvedValue({ content: [{ type: "text", text: "real-shape-result" }] }); + const tool = { name: "mcp__postiz__integrationlist", label: "List integrations", description: "", parameters: {}, execute }; + const { wrapToolsWithActionGate } = await import("../pi.js"); + const requests = new Map(); + const createApprovalRequest = vi.fn(async (decision: { approvalDedupeKey: string }) => { + const request = { id: "apr-main-008", status: "pending" as const }; + requests.set(decision.approvalDedupeKey, request); + return request; + }); + const findApprovalByDedupeKey = vi.fn(async (key: string) => requests.get(key) ?? null); + const markApprovalCompleted = vi.fn(async (id: string) => { + for (const [key, request] of requests) { + if (request.id === id) requests.set(key, { ...request, status: "completed" }); + } + }); + const context = { + agentId: "agent-main-008", + agentName: "MAIN-008 agent", + isEphemeral: false, + taskId: "MAIN-008", + permissionPolicy: approvalPolicy, + createApprovalRequest, + findApprovalByDedupeKey, + pauseForApproval: vi.fn(), + markApprovalCompleted, + }; + + const initial = wrapToolsWithActionGate([tool as any], context); + const pending = await (initial[0] as any).execute("pending", {}); + expect(pending.isError).toBe(true); + expect(execute).not.toHaveBeenCalled(); + expect(createApprovalRequest).toHaveBeenCalledTimes(1); + + const [dedupeKey, request] = [...requests.entries()][0]!; + requests.set(dedupeKey, { ...request, status: "approved" }); + const resumed = wrapToolsWithActionGate([tool as any], context); + await expect((resumed[0] as any).execute("approved", {})).resolves.toEqual({ + content: [{ type: "text", text: "real-shape-result" }], + }); + expect(execute).toHaveBeenCalledTimes(1); + expect(markApprovalCompleted).toHaveBeenCalledWith("apr-main-008"); + expect(requests.get(dedupeKey)?.status).toBe("completed"); + + execute.mockClear(); + requests.set(dedupeKey, { id: "apr-denied", status: "denied" }); + const denied = wrapToolsWithActionGate([tool as any], context); + const rejection = await (denied[0] as any).execute("denied", {}); + expect(rejection.error).toMatch(/denied/i); + expect(execute).not.toHaveBeenCalled(); + }); + it("classifies explicit network and management tools", () => { expect(evaluateAgentActionGate({ agentId: "a1", toolName: "fn_research_run", args: {}, permissionPolicy: unrestrictedPolicy }).category).toBe("network_api"); expect(evaluateAgentActionGate({ agentId: "a1", toolName: "fn_task_create", args: {}, permissionPolicy: unrestrictedPolicy }).category).toBe("task_agent_mutation"); diff --git a/packages/engine/src/__tests__/executor-project-mcp-resume.test.ts b/packages/engine/src/__tests__/executor-project-mcp-resume.test.ts index 8b06f60f2a..e228c822ae 100644 --- a/packages/engine/src/__tests__/executor-project-mcp-resume.test.ts +++ b/packages/engine/src/__tests__/executor-project-mcp-resume.test.ts @@ -74,6 +74,36 @@ describe("executor project MCP bootstrap and approval resume invariant", () => { ); }); + it("defers an approval decision received during unwind and dispatches exactly one resume", async () => { + const listeners = new Map unknown>(); + const task = { id: "MAIN-008", title: "test", description: "test", column: "in-progress", paused: false, userPaused: false, steps: [], currentStep: 0 }; + const store = { + on: vi.fn((event: string, listener: (...args: any[]) => unknown) => listeners.set(event, listener)), + off: vi.fn(), + getSettings: vi.fn(async () => ({ globalPause: false, enginePaused: false })), + listTasks: vi.fn(async () => []), + getTask: vi.fn(async () => task), + updateTask: vi.fn(async () => task), + logEntry: vi.fn(async () => undefined), + } as any; + const executor = new TaskExecutor(store, "/tmp/project"); + const execute = vi.spyOn(executor, "execute").mockResolvedValue(undefined); + (executor as any).approvalSuspended.add(task.id); + (executor as any).executing.add(task.id); + + await listeners.get("task:updated")?.(task); + await listeners.get("task:updated")?.(task); + expect(execute).not.toHaveBeenCalled(); + expect((executor as any).approvalResumeAfterUnwind.size).toBe(1); + + (executor as any).executing.delete(task.id); + await (executor as any).resumeApprovalAfterUnwindIfNeeded(task.id); + await (executor as any).resumeApprovalAfterUnwindIfNeeded(task.id); + + expect(execute).toHaveBeenCalledTimes(1); + expect((executor as any).approvalSuspended.has(task.id)).toBe(false); + }); + it("closes a client whose connection fails before registration", async () => { const close = vi.fn(async () => undefined); const client = fakeClient(close); diff --git a/packages/engine/src/__tests__/pi-create-fn-agent.test.ts b/packages/engine/src/__tests__/pi-create-fn-agent.test.ts index 964a92d617..fa8bc735b6 100644 --- a/packages/engine/src/__tests__/pi-create-fn-agent.test.ts +++ b/packages/engine/src/__tests__/pi-create-fn-agent.test.ts @@ -957,20 +957,22 @@ describe("wrapToolsWithActionGate", () => { expect(tool.execute).not.toHaveBeenCalled(); }); - it("skips gating wrapper for ephemeral contexts", async () => { + it("applies the status-aware action gate to ephemeral and fallback task workers", async () => { const tool = { name: "write", label: "Write", description: "", parameters: {}, execute: vi.fn().mockResolvedValue({ ok: true }) }; const { wrapToolsWithActionGate } = await import("../pi.js"); const wrapped = wrapToolsWithActionGate([tool as any], { - agentId: "agent-1", - agentName: "Agent", + agentId: "executor-MAIN-008", + agentName: "Fallback task worker", isEphemeral: true, + taskId: "MAIN-008", permissionPolicy: { presetId: "locked-down", rules: lockedDownRules }, createApprovalRequest: vi.fn(), - findApprovalByDedupeKey: vi.fn(), + findApprovalByDedupeKey: vi.fn().mockResolvedValue(null), }); - await (wrapped[0] as any).execute("t1", { path: "a.ts" }); - expect(tool.execute).toHaveBeenCalled(); + const result = await (wrapped[0] as any).execute("t1", { path: "a.ts" }); + expect(result.isError).toBe(true); + expect(tool.execute).not.toHaveBeenCalled(); }); it("governs newly exposed heartbeat network tools by policy instead of withholding them", async () => { @@ -2246,6 +2248,50 @@ describe("createFnAgent", () => { expect(createSessionArgs.customTools.map((tool) => tool.name)).toContain("fn_heartbeat_done"); }); + it("uses the status-aware action gate as the single approval authority when both gate contexts are present", async () => { + const execute = vi.fn().mockResolvedValue({ content: [{ type: "text", text: "ok" }] }); + const permanentCreateApproval = vi.fn(); + const markApprovalCompleted = vi.fn(); + const { createFnAgent } = await import("../pi.js"); + + await createFnAgent({ + cwd: "/tmp", + systemPrompt: "test", + tools: "coding", + customTools: [{ name: "mcp__postiz__integrationlist", label: "List", description: "", parameters: {}, execute } as any], + actionGateContext: { + agentId: "executor-MAIN-008", + agentName: "Fallback worker", + isEphemeral: true, + taskId: "MAIN-008", + permissionPolicy: { + presetId: "approval", + rules: { git_write: "allow", file_write_delete: "allow", command_execution: "allow", network_api: "require-approval", task_agent_mutation: "allow" }, + }, + createApprovalRequest: vi.fn(), + findApprovalByDedupeKey: vi.fn().mockResolvedValue({ id: "apr-main-008", status: "approved" }), + markApprovalCompleted, + }, + permanentAgentGating: { + permissionPolicy: { + presetId: "approval", + rules: { git_write: "allow", file_write_delete: "allow", command_execution: "allow", network_api: "require-approval", task_agent_mutation: "allow" }, + }, + requester: { actorId: "executor-MAIN-008", actorType: "agent", actorName: "Fallback worker" }, + taskId: "MAIN-008", + createApprovalRequest: permanentCreateApproval, + findPendingApprovalRequest: vi.fn(), + } as any, + }); + + const createSessionArgs = createAgentSessionMock.mock.calls[0]?.[0] as { customTools: Array<{ name: string; execute: (...args: any[]) => Promise }> }; + const mcpTool = createSessionArgs.customTools.find((tool) => tool.name === "mcp__postiz__integrationlist")!; + await expect(mcpTool.execute("call", {})).resolves.toEqual({ content: [{ type: "text", text: "ok" }] }); + expect(execute).toHaveBeenCalledTimes(1); + expect(markApprovalCompleted).toHaveBeenCalledWith("apr-main-008"); + expect(permanentCreateApproval).not.toHaveBeenCalled(); + }); + it("exposes connected MCP tools in readonly sessions only with the explicit opt-in", async () => { const { createFnAgent } = await import("../pi.js"); const close = vi.fn(async () => undefined); diff --git a/packages/engine/src/agent-action-gate.ts b/packages/engine/src/agent-action-gate.ts index 2e6bbca6fb..b2b787e65f 100644 --- a/packages/engine/src/agent-action-gate.ts +++ b/packages/engine/src/agent-action-gate.ts @@ -179,7 +179,14 @@ export function evaluateAgentActionGate(params: { category = "command_execution"; operation = params.toolName; resourceType = "command"; - } else if (NETWORK_API_TOOLS.has(params.toolName)) { + } else if (NETWORK_API_TOOLS.has(params.toolName) || params.toolName.startsWith("mcp__")) { + /* + FNXC:AgentGating 2026-07-12-17:18: + MAIN-008 requires every namespaced project MCP operation to remain inside + the external-action approval boundary. MCP tools are dynamically named and + therefore cannot live in the static tool registry; classify the namespace + as network_api instead of falling through to the exempt default. + */ category = "network_api"; operation = params.toolName; resourceType = "research"; diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 83b12bef32..b5095f567c 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -1723,6 +1723,10 @@ export class TaskExecutor { private executing = new Set(); /** Tasks currently being prepared for unpause resume, before execute() has registered them. */ private resumingUnpaused = new Set(); + /** Tasks whose active session was intentionally suspended by an action gate. */ + private approvalSuspended = new Set(); + /** Approval decisions received while the old execute() lifecycle is still unwinding. */ + private approvalResumeAfterUnwind = new Set(); /** Completed orphan recovery tasks currently running during startup. */ private recoveringCompleted = new Set(); /** @@ -2335,8 +2339,20 @@ export class TaskExecutor { `paused: true` was set with no reason, which self-healing's autoReboundPausedScopeDecay could rebound before the operator ever decided. + + FNXC:ApprovalResume 2026-07-12-17:02: + MAIN-008: record the approval-specific suspension before pauseTask emits its + task:updated event so every abort branch can preserve the in-progress row + for a deterministic fresh resume. Clear the mark if pauseTask fails so a + failed pause does not leave a sticky suspended marker. */ - await this.store.pauseTask(taskId, true, this.getRunContextFor(taskId), { pausedByAgentId: actorId, pausedReason: AWAITING_APPROVAL_PAUSE_REASON }); + this.approvalSuspended.add(taskId); + try { + await this.store.pauseTask(taskId, true, this.getRunContextFor(taskId), { pausedByAgentId: actorId, pausedReason: AWAITING_APPROVAL_PAUSE_REASON }); + } catch (error) { + this.approvalSuspended.delete(taskId); + throw error; + } await this.store.logEntry( taskId, `Approval required for ${decision.toolName}. Request ${approvalRequestId} created; task and agent paused awaiting decision.`, @@ -2481,6 +2497,8 @@ export class TaskExecutor { this.executing.delete(taskId); this.recoveringCompleted.delete(taskId); this.resumingUnpaused.delete(taskId); + this.approvalSuspended.delete(taskId); + this.approvalResumeAfterUnwind.delete(taskId); TaskExecutor.processWideGraphRouting.delete(taskId); executingTaskLock.release(taskId); this.effectiveColumnAgentByTask.delete(taskId); @@ -2812,6 +2830,74 @@ export class TaskExecutor { * prevents new work dispatch — running sessions continue to completion. * Paused tasks are moved back to `todo` rather than marked as `failed`. */ + private async parkApprovalSuspension(taskId: string, surface: string): Promise { + if (!this.approvalSuspended.has(taskId)) return false; + this.clearPausedAborted(taskId); + await this.store.logEntry( + taskId, + `Execution suspended for approval — ${surface} disposed; task remains in progress for decision resume`, + undefined, + this.getRunContextFor(taskId), + ); + executorLog.log(`${taskId}: approval suspension parked after ${surface} disposal`); + return true; + } + + private async dispatchUnpauseResume(task: Task): Promise { + if ( + this.executing.has(task.id) + || this.resumingUnpaused.has(task.id) + || this.recoveringCompleted.has(task.id) + || this.activeSessions.has(task.id) + || this.activeStepExecutors.has(task.id) + || this.activeWorkflowStepSessions.has(task.id) + ) { + return false; + } + + const pauseLabel = await this.getExecutionPauseLabel(); + if (pauseLabel) { + executorLog.log(`Skipping unpause resume for ${task.id} — ${pauseLabel} active`); + return false; + } + + this.approvalSuspended.delete(task.id); + if (this.isTaskWorkComplete(task) && !task.mergeDetails) { + this.recoveringCompleted.add(task.id); + executorLog.log(`${task.id} unpaused with completed work and no session — recovering directly to in-review`); + void this.recoverCompletedTask(task) + .catch((err) => executorLog.error(`Failed to recover completed unpaused task ${task.id}:`, err)) + .finally(() => this.recoveringCompleted.delete(task.id)); + return true; + } + + this.resumingUnpaused.add(task.id); + executorLog.log(`Unpaused ${task.id} in-progress with no session — resuming execution`); + try { + await this.clearResumeFailureState(task); + await this.store.updateTask(task.id, { + resumeLimboCount: 0, + resumeLimboTipSha: null, + resumeLimboStepSignature: null, + }); + await this.store.logEntry(task.id, "Resuming execution after unpause", undefined, this.getRunContextFor(task.id)); + await this.recoverApprovedStepsOnResume(task.id); + } catch (clearErr) { + executorLog.warn(`${task.id} clearResumeFailureState failed during unpause: ${clearErr instanceof Error ? clearErr.message : String(clearErr)}`); + } + this.execute(task) + .catch((err) => executorLog.error(`Failed to resume unpaused ${task.id}:`, err)) + .finally(() => this.resumingUnpaused.delete(task.id)); + return true; + } + + private async resumeApprovalAfterUnwindIfNeeded(taskId: string): Promise { + if (!this.approvalResumeAfterUnwind.delete(taskId)) return false; + const latestTask = await this.store.getTask(taskId); + if (latestTask.paused || latestTask.userPaused || latestTask.column !== "in-progress") return false; + return this.dispatchUnpauseResume(latestTask); + } + private async resolveMcpServers(agentId?: string | null) { /* * FNXC:McpConfig 2026-06-25-22:20: @@ -2911,6 +2997,8 @@ export class TaskExecutor { }); store.on("task:deleted", (task) => { + this.approvalSuspended.delete(task.id); + this.approvalResumeAfterUnwind.delete(task.id); this.trackTaskDisposal( task.id, this.awaitAbortInFlightTaskWork(task.id, "task soft-deleted", { userCanceled: true }), @@ -2947,9 +3035,23 @@ export class TaskExecutor { } // Handle unpause of an in-progress task with no active session. - // This covers orphaned states (e.g., engine restarted while task was - // paused in-progress) where the task needs to resume execution. - // The executing/resuming guards prevent duplicate runs. + // Approval can be decided while the old session is still unwinding; + // remember that edge instead of losing the only task:updated event. + if (!task.paused && task.column === "in-progress" && this.approvalSuspended.has(task.id)) { + if ( + this.executing.has(task.id) + || this.activeSessions.has(task.id) + || this.activeStepExecutors.has(task.id) + || this.activeWorkflowStepSessions.has(task.id) + ) { + this.approvalResumeAfterUnwind.add(task.id); + executorLog.log(`${task.id}: approval decision received during session unwind — deferred one resume`); + return; + } + } + + // This also covers orphaned states (for example, engine restart while + // paused in-progress). dispatchUnpauseResume owns all duplicate guards. if ( !task.paused && task.column === "in-progress" @@ -2957,52 +3059,7 @@ export class TaskExecutor { && !this.activeStepExecutors.has(task.id) && !this.activeWorkflowStepSessions.has(task.id) ) { - if ( - !this.executing.has(task.id) - && !this.resumingUnpaused.has(task.id) - && !this.recoveringCompleted.has(task.id) - ) { - const pauseLabel = await this.getExecutionPauseLabel(); - if (pauseLabel) { - executorLog.log(`Skipping unpause resume for ${task.id} — ${pauseLabel} active`); - return; - } - - if (this.isTaskWorkComplete(task) && !task.mergeDetails) { - this.recoveringCompleted.add(task.id); - executorLog.log(`${task.id} unpaused with completed work and no session — recovering directly to in-review`); - void this.recoverCompletedTask(task) - .catch((err) => - executorLog.error(`Failed to recover completed unpaused task ${task.id}:`, err), - ) - .finally(() => { - this.recoveringCompleted.delete(task.id); - }); - return; - } - - this.resumingUnpaused.add(task.id); - executorLog.log(`Unpaused ${task.id} in-progress with no session — resuming execution`); - try { - await this.clearResumeFailureState(task); - await this.store.updateTask(task.id, { - resumeLimboCount: 0, - resumeLimboTipSha: null, - resumeLimboStepSignature: null, - }); - await this.store.logEntry(task.id, "Resuming execution after unpause", undefined, this.getRunContextFor(task.id)); - await this.recoverApprovedStepsOnResume(task.id); - } catch (clearErr) { - executorLog.warn(`${task.id} clearResumeFailureState failed during unpause: ${clearErr instanceof Error ? clearErr.message : String(clearErr)}`); - } - this.execute(task) - .catch((err) => - executorLog.error(`Failed to resume unpaused ${task.id}:`, err), - ) - .finally(() => { - this.resumingUnpaused.delete(task.id); - }); - } + await this.dispatchUnpauseResume(task); return; } @@ -10102,6 +10159,7 @@ export class TaskExecutor { await this.store.logEntry(task.id, "Execution canceled by user — leaving task in todo"); return; } + if (await this.parkApprovalSuspension(task.id, "step sessions")) return; this.clearPausedAborted(task.id); await this.store.logEntry(task.id, "Execution paused — step sessions terminated, moved to todo", undefined, this.getRunContextFor(task.id)); this.markGraphExecuteSelfRequeued(task.id); @@ -10385,6 +10443,7 @@ export class TaskExecutor { await this.store.logEntry(task.id, "Execution canceled by user — leaving task in todo"); return; } + if (await this.parkApprovalSuspension(task.id, "step session")) return; this.clearPausedAborted(task.id); await this.store.logEntry(task.id, "Execution paused during step-session", undefined, this.getRunContextFor(task.id)); this.markGraphExecuteSelfRequeued(task.id); @@ -11062,6 +11121,10 @@ export class TaskExecutor { await this.store.logEntry(task.id, "Execution canceled by user — leaving task in todo"); return; } + if (await this.parkApprovalSuspension(task.id, "agent session")) { + wasPaused = true; + return; + } this.clearPausedAborted(task.id); wasPaused = true; if (await this.shouldFinalizeCompletedTask(task.id, taskDone)) { @@ -11600,6 +11663,7 @@ export class TaskExecutor { await this.store.logEntry(task.id, "Execution canceled by user — leaving task in todo"); return; } + if (await this.parkApprovalSuspension(task.id, "executor session")) return; this.clearPausedAborted(task.id); const latestTask = await this.store.getTask(task.id); if ( @@ -12342,6 +12406,16 @@ export class TaskExecutor { } } } + + /* + * FNXC:AgentGating 2026-07-12-17:12: + * MAIN-008 closes the approval-decision/unwind race. The dashboard can + * unpause while the original executor still owns its process-wide lock; + * consume that single deferred edge only after every old-session cleanup + * path above has run, then bootstrap one new executor session. A Set plus + * resumingUnpaused makes duplicate task updates idempotent. + */ + await this.resumeApprovalAfterUnwindIfNeeded(task.id); } } diff --git a/packages/engine/src/pi.ts b/packages/engine/src/pi.ts index 7ae4dabe2e..34d1d07614 100644 --- a/packages/engine/src/pi.ts +++ b/packages/engine/src/pi.ts @@ -1904,7 +1904,7 @@ export function wrapToolsWithActionGate( tools: ToolDefinition[], gateContext: AgentActionGateContext | undefined, ): ToolDefinition[] { - if (!gateContext || gateContext.isEphemeral) { + if (!gateContext) { return tools; } @@ -2369,10 +2369,18 @@ export async function createFnAgent(options: AgentOptions): Promise ...allowlistFilteredCustomTools.allowed, ]; const toolsWithRtkRewrite = wrapToolsWithRtkRewrite(toolChainStart); - const toolsWithPermanentGating = wrapToolsWithPermanentAgentGating( - toolsWithRtkRewrite, - options.permanentAgentGating, - ); + /* + * FNXC:AgentGating 2026-07-12-17:22: + * MAIN-008 requires one approval authority per tool call. Executor sessions + * provide the status-aware action gate for permanent, ephemeral, and + * fallback task-worker identities; applying the legacy permanent gate + * inside it would reject the call again after the outer gate consumed an + * approved request. Standalone lanes without actionGateContext retain the + * permanent gate unchanged. + */ + const toolsWithPermanentGating = options.actionGateContext + ? toolsWithRtkRewrite + : wrapToolsWithPermanentAgentGating(toolsWithRtkRewrite, options.permanentAgentGating); const toolsWithActionGate = wrapToolsWithActionGate( toolsWithPermanentGating, options.actionGateContext,