diff --git a/extensions/telegram/src/channel.ts b/extensions/telegram/src/channel.ts index 76bc37ff8..9ce9d1a28 100644 --- a/extensions/telegram/src/channel.ts +++ b/extensions/telegram/src/channel.ts @@ -253,6 +253,27 @@ export const telegramPlugin: ChannelPlugin = { chunker: (text, limit) => getTelegramRuntime().channel.text.chunkMarkdownText(text, limit), chunkerMode: "markdown", textChunkLimit: 4000, + sendPayload: async (ctx) => { + const send = + ctx.deps?.sendTelegram ?? getTelegramRuntime().channel.telegram.sendMessageTelegram; + const replyToMessageId = parseReplyToMessageId(ctx.replyToId); + const messageThreadId = parseThreadId(ctx.threadId); + // Extract buttons from channelData.telegram.buttons + type TelegramButtons = Array>; + const channelData = ctx.payload.channelData; + const telegram = channelData?.telegram as { buttons?: TelegramButtons } | undefined; + const buttons = telegram?.buttons; + const result = await send(ctx.to, ctx.text, { + verbose: false, + mediaUrl: ctx.mediaUrl, + textMode: "html", + messageThreadId, + replyToMessageId, + accountId: ctx.accountId ?? undefined, + buttons, + }); + return { channel: "telegram", ...result }; + }, sendText: async ({ to, text, accountId, deps, replyToId, threadId }) => { const send = deps?.sendTelegram ?? getTelegramRuntime().channel.telegram.sendMessageTelegram; diff --git a/src/auto-reply/reply/commands-approve.test.ts b/src/auto-reply/reply/commands-approve.test.ts index ec2695372..21b5aa290 100644 --- a/src/auto-reply/reply/commands-approve.test.ts +++ b/src/auto-reply/reply/commands-approve.test.ts @@ -1,15 +1,22 @@ -import { describe, expect, it, vi } from "vitest"; +import { describe, expect, it, vi, beforeEach } from "vitest"; import type { ClawdbotConfig } from "../../config/config.js"; import type { MsgContext } from "../templating.js"; import { buildCommandContext, handleCommands } from "./commands.js"; import { parseInlineDirectives } from "./directive-handling.js"; import { callGateway } from "../../gateway/call.js"; +import * as approvalForwarder from "../../infra/exec-approval-forwarder.js"; vi.mock("../../gateway/call.js", () => ({ callGateway: vi.fn(), })); +vi.mock("../../infra/exec-approval-forwarder.js", () => ({ + getBatchApprovalIds: vi.fn(), + deleteBatch: vi.fn(), + updateBatchApprovalIds: vi.fn(), +})); + function buildParams(commandBody: string, cfg: ClawdbotConfig, ctxOverrides?: Partial) { const ctx = { Body: commandBody, @@ -49,6 +56,10 @@ function buildParams(commandBody: string, cfg: ClawdbotConfig, ctxOverrides?: Pa } describe("/approve command", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + it("rejects invalid usage", async () => { const cfg = { commands: { text: true }, @@ -80,4 +91,110 @@ describe("/approve command", () => { }), ); }); + + it("does not intercept /approve-batch", async () => { + const cfg = { + commands: { text: true }, + channels: { whatsapp: { allowFrom: ["*"] } }, + } as ClawdbotConfig; + const params = buildParams("/approve-batch test-batch allow-once", cfg, { SenderId: "123" }); + + vi.mocked(approvalForwarder.getBatchApprovalIds).mockReturnValue(null); + + const result = await handleCommands(params); + // Should be handled by /approve-batch, not /approve + expect(result.reply?.text).toContain("Batch not found"); + expect(result.reply?.text).not.toContain("Usage: /approve"); + }); +}); + +describe("/approve-batch command", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("rejects invalid usage", async () => { + const cfg = { + commands: { text: true }, + channels: { whatsapp: { allowFrom: ["*"] } }, + } as ClawdbotConfig; + const params = buildParams("/approve-batch", cfg); + const result = await handleCommands(params); + expect(result.shouldContinue).toBe(false); + expect(result.reply?.text).toContain("Usage: /approve-batch"); + }); + + it("rejects unknown batch ID", async () => { + const cfg = { + commands: { text: true }, + channels: { whatsapp: { allowFrom: ["*"] } }, + } as ClawdbotConfig; + const params = buildParams("/approve-batch unknown-batch allow-once", cfg, { SenderId: "123" }); + + vi.mocked(approvalForwarder.getBatchApprovalIds).mockReturnValue(null); + + const result = await handleCommands(params); + expect(result.shouldContinue).toBe(false); + expect(result.reply?.text).toContain("Batch not found or expired"); + }); + + it("approves all commands in batch", async () => { + const cfg = { + commands: { text: true }, + channels: { whatsapp: { allowFrom: ["*"] } }, + } as ClawdbotConfig; + const params = buildParams("/approve-batch batch-123 allow-once", cfg, { SenderId: "123" }); + + vi.mocked(approvalForwarder.getBatchApprovalIds).mockReturnValue(["id1", "id2", "id3"]); + const mockCallGateway = vi.mocked(callGateway); + mockCallGateway.mockResolvedValue({ ok: true }); + + const result = await handleCommands(params); + expect(result.shouldContinue).toBe(false); + expect(result.reply?.text).toContain("3 commands approved"); + expect(mockCallGateway).toHaveBeenCalledTimes(3); + expect(approvalForwarder.deleteBatch).toHaveBeenCalledWith("batch-123"); + expect(approvalForwarder.updateBatchApprovalIds).not.toHaveBeenCalled(); + }); + + it("denies all commands in batch", async () => { + const cfg = { + commands: { text: true }, + channels: { whatsapp: { allowFrom: ["*"] } }, + } as ClawdbotConfig; + const params = buildParams("/approve-batch batch-456 deny", cfg, { SenderId: "123" }); + + vi.mocked(approvalForwarder.getBatchApprovalIds).mockReturnValue(["id1", "id2"]); + const mockCallGateway = vi.mocked(callGateway); + mockCallGateway.mockResolvedValue({ ok: true }); + + const result = await handleCommands(params); + expect(result.shouldContinue).toBe(false); + expect(result.reply?.text).toContain("2 commands approved"); + expect(mockCallGateway).toHaveBeenCalledTimes(2); + expect(approvalForwarder.deleteBatch).toHaveBeenCalledWith("batch-456"); + expect(approvalForwarder.updateBatchApprovalIds).not.toHaveBeenCalled(); + }); + + it("reports partial failures", async () => { + const cfg = { + commands: { text: true }, + channels: { whatsapp: { allowFrom: ["*"] } }, + } as ClawdbotConfig; + const params = buildParams("/approve-batch batch-789 allow-once", cfg, { SenderId: "123" }); + + vi.mocked(approvalForwarder.getBatchApprovalIds).mockReturnValue(["id1", "id2", "id3"]); + const mockCallGateway = vi.mocked(callGateway); + mockCallGateway + .mockResolvedValueOnce({ ok: true }) + .mockRejectedValueOnce(new Error("Failed")) + .mockResolvedValueOnce({ ok: true }); + + const result = await handleCommands(params); + expect(result.shouldContinue).toBe(false); + expect(result.reply?.text).toContain("2 succeeded"); + expect(result.reply?.text).toContain("1 failed"); + expect(approvalForwarder.deleteBatch).not.toHaveBeenCalled(); + expect(approvalForwarder.updateBatchApprovalIds).toHaveBeenCalledWith("batch-789", ["id2"]); + }); }); diff --git a/src/auto-reply/reply/commands-approve.ts b/src/auto-reply/reply/commands-approve.ts index a34e4b31c..8c1ca327c 100644 --- a/src/auto-reply/reply/commands-approve.ts +++ b/src/auto-reply/reply/commands-approve.ts @@ -1,9 +1,15 @@ import { callGateway } from "../../gateway/call.js"; import { GATEWAY_CLIENT_MODES, GATEWAY_CLIENT_NAMES } from "../../utils/message-channel.js"; import { logVerbose } from "../../globals.js"; +import { + getBatchApprovalIds, + deleteBatch, + updateBatchApprovalIds, +} from "../../infra/exec-approval-forwarder.js"; import type { CommandHandler } from "./commands-types.js"; const COMMAND = "/approve"; +const BATCH_COMMAND = "/approve-batch"; const DECISION_ALIASES: Record = { allow: "allow-once", @@ -24,7 +30,10 @@ type ParsedApproveCommand = function parseApproveCommand(raw: string): ParsedApproveCommand | null { const trimmed = raw.trim(); - if (!trimmed.toLowerCase().startsWith(COMMAND)) return null; + const lower = trimmed.toLowerCase(); + // Don't match /approve-batch - it has its own handler + if (!lower.startsWith(COMMAND)) return null; + if (lower.startsWith(BATCH_COMMAND)) return null; const rest = trimmed.slice(COMMAND.length).trim(); if (!rest) { return { ok: false, error: "Usage: /approve allow-once|allow-always|deny" }; @@ -54,6 +63,44 @@ function parseApproveCommand(raw: string): ParsedApproveCommand | null { return { ok: false, error: "Usage: /approve allow-once|allow-always|deny" }; } +type ParsedBatchCommand = + | { ok: true; batchId: string; decision: "allow-once" | "deny" } + | { ok: false; error: string }; + +function parseBatchCommand(raw: string): ParsedBatchCommand | null { + const trimmed = raw.trim(); + if (!trimmed.toLowerCase().startsWith(BATCH_COMMAND)) return null; + const rest = trimmed.slice(BATCH_COMMAND.length).trim(); + if (!rest) { + return { ok: false, error: "Usage: /approve-batch allow-once|deny" }; + } + const tokens = rest.split(/\s+/).filter(Boolean); + if (tokens.length < 2) { + return { ok: false, error: "Usage: /approve-batch allow-once|deny" }; + } + + const batchId = tokens[0]; + const decisionRaw = tokens[1].toLowerCase(); + + // Batch only supports allow-once and deny + const batchDecisions: Record = { + allow: "allow-once", + once: "allow-once", + "allow-once": "allow-once", + allowonce: "allow-once", + deny: "deny", + reject: "deny", + block: "deny", + }; + + const decision = batchDecisions[decisionRaw]; + if (!decision) { + return { ok: false, error: "Usage: /approve-batch allow-once|deny" }; + } + + return { ok: true, batchId, decision }; +} + function buildResolvedByLabel(params: Parameters[0]): string { const channel = params.command.channel; const sender = params.command.senderId ?? "unknown"; @@ -99,3 +146,74 @@ export const handleApproveCommand: CommandHandler = async (params, allowTextComm reply: { text: `✅ Exec approval ${parsed.decision} submitted for ${parsed.id}.` }, }; }; + +export const handleApproveBatchCommand: CommandHandler = async (params, allowTextCommands) => { + if (!allowTextCommands) return null; + const normalized = params.command.commandBodyNormalized; + const parsed = parseBatchCommand(normalized); + if (!parsed) return null; + if (!params.command.isAuthorizedSender) { + logVerbose( + `Ignoring /approve-batch from unauthorized sender: ${params.command.senderId || ""}`, + ); + return { shouldContinue: false }; + } + + if (!parsed.ok) { + return { shouldContinue: false, reply: { text: parsed.error } }; + } + + // Look up the batch + const approvalIds = getBatchApprovalIds(parsed.batchId); + if (!approvalIds || approvalIds.length === 0) { + return { + shouldContinue: false, + reply: { text: `❌ Batch not found or expired: ${parsed.batchId}` }, + }; + } + + const resolvedBy = buildResolvedByLabel(params); + const results: { id: string; success: boolean; error?: string }[] = []; + + // Resolve all approvals in the batch + for (const id of approvalIds) { + try { + await callGateway({ + method: "exec.approval.resolve", + params: { id, decision: parsed.decision }, + clientName: GATEWAY_CLIENT_NAMES.GATEWAY_CLIENT, + clientDisplayName: `Batch approval (${resolvedBy})`, + mode: GATEWAY_CLIENT_MODES.BACKEND, + }); + results.push({ id, success: true }); + } catch (err) { + results.push({ id, success: false, error: String(err) }); + } + } + + const failedIds = results.filter((r) => !r.success).map((r) => r.id); + if (failedIds.length === 0) { + deleteBatch(parsed.batchId); + } else { + updateBatchApprovalIds(parsed.batchId, failedIds); + } + + const succeeded = results.filter((r) => r.success).length; + const failed = results.filter((r) => !r.success).length; + + if (failed === 0) { + return { + shouldContinue: false, + reply: { + text: `✅ Batch ${parsed.decision}: ${succeeded} command${succeeded > 1 ? "s" : ""} approved.`, + }, + }; + } + + return { + shouldContinue: false, + reply: { + text: `⚠️ Batch ${parsed.decision}: ${succeeded} succeeded, ${failed} failed.`, + }, + }; +}; diff --git a/src/auto-reply/reply/commands-core.ts b/src/auto-reply/reply/commands-core.ts index a54f90b2b..adbd209de 100644 --- a/src/auto-reply/reply/commands-core.ts +++ b/src/auto-reply/reply/commands-core.ts @@ -14,7 +14,7 @@ import { handleWhoamiCommand, } from "./commands-info.js"; import { handleAllowlistCommand } from "./commands-allowlist.js"; -import { handleApproveCommand } from "./commands-approve.js"; +import { handleApproveCommand, handleApproveBatchCommand } from "./commands-approve.js"; import { handleSubagentsCommand } from "./commands-subagents.js"; import { handleModelsCommand } from "./commands-models.js"; import { handleTtsCommands } from "./commands-tts.js"; @@ -47,6 +47,7 @@ const HANDLERS: CommandHandler[] = [ handleStatusCommand, handleAllowlistCommand, handleApproveCommand, + handleApproveBatchCommand, handleContextCommand, handleWhoamiCommand, handleSubagentsCommand, diff --git a/src/infra/exec-approval-forwarder.test.ts b/src/infra/exec-approval-forwarder.test.ts index 422e22c48..cd3dfc5b4 100644 --- a/src/infra/exec-approval-forwarder.test.ts +++ b/src/infra/exec-approval-forwarder.test.ts @@ -34,6 +34,8 @@ describe("exec approval forwarder", () => { }); await forwarder.handleRequested(baseRequest); + expect(deliver).toHaveBeenCalledTimes(0); + await vi.advanceTimersByTimeAsync(1500); expect(deliver).toHaveBeenCalledTimes(1); await forwarder.handleResolved({ @@ -69,6 +71,8 @@ describe("exec approval forwarder", () => { }); await forwarder.handleRequested(baseRequest); + expect(deliver).toHaveBeenCalledTimes(0); + await vi.advanceTimersByTimeAsync(1500); expect(deliver).toHaveBeenCalledTimes(1); await vi.runAllTimersAsync(); diff --git a/src/infra/exec-approval-forwarder.ts b/src/infra/exec-approval-forwarder.ts index 2fbf53ae6..475d8b9e4 100644 --- a/src/infra/exec-approval-forwarder.ts +++ b/src/infra/exec-approval-forwarder.ts @@ -14,6 +14,12 @@ import { resolveSessionDeliveryTarget } from "./outbound/targets.js"; const log = createSubsystemLogger("gateway/exec-approvals"); +// Batch approval constants +const BATCH_WINDOW_MS = 1500; // Initial window: collect requests for 1.5 seconds +const BATCH_EXTEND_WINDOW_MS = 10000; // Extended window when session has pending approvals +const BATCH_MAX_WINDOW_MS = 30000; // Maximum time to wait before flushing (30 seconds) +const BATCH_TTL_MS = 5 * 60 * 1000; // Batch IDs expire after 5 minutes + export type ExecApprovalRequest = { id: string; request: { @@ -43,8 +49,63 @@ type PendingApproval = { request: ExecApprovalRequest; targets: ForwardTarget[]; timeoutId: NodeJS.Timeout | null; + batchId?: string; }; +type PendingBatch = { + sessionKey: string; + requests: ExecApprovalRequest[]; + targets: ForwardTarget[]; + flushTimeoutId: NodeJS.Timeout | null; + createdAtMs: number; + lastRequestAtMs: number; +}; + +type BatchEntry = { + approvalIds: string[]; + sessionKey: string; + createdAtMs: number; +}; + +// Global batch registry - maps batch IDs to their approval IDs +const batchRegistry = new Map(); + +function generateBatchId(): string { + return `batch-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`; +} + +function cleanupExpiredBatches(nowMs: number): void { + for (const [batchId, entry] of batchRegistry) { + if (nowMs - entry.createdAtMs > BATCH_TTL_MS) { + batchRegistry.delete(batchId); + } + } +} + +export function getBatchApprovalIds(batchId: string): string[] | null { + // Cleanup expired batches opportunistically + cleanupExpiredBatches(Date.now()); + const entry = batchRegistry.get(batchId); + if (!entry) return null; + return entry.approvalIds; +} + +export function deleteBatch(batchId: string): void { + batchRegistry.delete(batchId); +} + +export function updateBatchApprovalIds(batchId: string, approvalIds: string[]): void { + if (approvalIds.length === 0) { + batchRegistry.delete(batchId); + return; + } + // Cleanup expired batches opportunistically + cleanupExpiredBatches(Date.now()); + const entry = batchRegistry.get(batchId); + if (!entry) return; + entry.approvalIds = approvalIds; +} + export type ExecApprovalForwarder = { handleRequested: (request: ExecApprovalRequest) => Promise; handleResolved: (resolved: ExecApprovalResolved) => Promise; @@ -105,7 +166,12 @@ function buildTargetKey(target: ExecApprovalForwardTarget): string { return [channel, target.to, accountId, threadId].join(":"); } -function buildRequestMessage(request: ExecApprovalRequest, nowMs: number) { +type ApprovalMessage = { + text: string; + buttons: Array>; +}; + +function buildRequestMessage(request: ExecApprovalRequest, nowMs: number): ApprovalMessage { const lines: string[] = ["🔒 Exec approval required", `ID: ${request.id}`]; lines.push(`Command: ${request.request.command}`); if (request.request.cwd) lines.push(`CWD: ${request.request.cwd}`); @@ -115,8 +181,16 @@ function buildRequestMessage(request: ExecApprovalRequest, nowMs: number) { if (request.request.ask) lines.push(`Ask: ${request.request.ask}`); const expiresIn = Math.max(0, Math.round((request.expiresAtMs - nowMs) / 1000)); lines.push(`Expires in: ${expiresIn}s`); - lines.push("Reply with: /approve allow-once|allow-always|deny"); - return lines.join("\n"); + + const buttons = [ + [ + { text: "✅ Allow Once", callback_data: `/approve ${request.id} allow-once` }, + { text: "✅ Always Allow", callback_data: `/approve ${request.id} allow-always` }, + ], + [{ text: "❌ Deny", callback_data: `/approve ${request.id} deny` }], + ]; + + return { text: lines.join("\n"), buttons }; } function decisionLabel(decision: ExecApprovalDecision): string { @@ -135,6 +209,53 @@ function buildExpiredMessage(request: ExecApprovalRequest) { return `⏱️ Exec approval expired. ID: ${request.id}`; } +function buildBatchRequestMessage( + requests: ExecApprovalRequest[], + batchId: string, + nowMs: number, +): ApprovalMessage { + const count = requests.length; + const lines: string[] = [ + `🔒 Exec approval required (${count} command${count > 1 ? "s" : ""})`, + `Batch ID: ${batchId}`, + ]; + lines.push(""); + + // List each command with a number + requests.forEach((req, idx) => { + const cmd = req.request.command; + // Truncate long commands for readability + const displayCmd = cmd.length > 60 ? cmd.slice(0, 57) + "..." : cmd; + lines.push(`${idx + 1}. ${displayCmd} (${req.id})`); + }); + + lines.push(""); + + // Show session info if available + const sessionKey = requests[0]?.request.sessionKey; + if (sessionKey) { + // Truncate long session keys + const displayKey = sessionKey.length > 40 ? "..." + sessionKey.slice(-37) : sessionKey; + lines.push(`Session: ${displayKey}`); + } + + // Use the earliest expiry time + const earliestExpiry = Math.min(...requests.map((r) => r.expiresAtMs)); + const expiresIn = Math.max(0, Math.round((earliestExpiry - nowMs) / 1000)); + lines.push(`Expires in: ${expiresIn}s`); + lines.push(`Approve: /approve-batch ${batchId} allow-once|deny`); + lines.push("Or: /approve allow-once|allow-always|deny"); + + const buttons = [ + [ + { text: `✅ Approve All (${count})`, callback_data: `/approve-batch ${batchId} allow-once` }, + { text: `❌ Deny All`, callback_data: `/approve-batch ${batchId} deny` }, + ], + ]; + + return { text: lines.join("\n"), buttons }; +} + function defaultResolveSessionTarget(params: { cfg: ClawdbotConfig; request: ExecApprovalRequest; @@ -158,10 +279,25 @@ function defaultResolveSessionTarget(params: { }; } +type TelegramButtons = Array>; + +function buildPayloadWithButtons(text: string, buttons?: TelegramButtons) { + if (!buttons?.length) { + log.debug("exec approvals: buildPayloadWithButtons called without buttons"); + return { text }; + } + log.debug(`exec approvals: buildPayloadWithButtons adding ${buttons.length} button rows`); + return { + text, + channelData: { telegram: { buttons } }, + }; +} + async function deliverToTargets(params: { cfg: ClawdbotConfig; targets: ForwardTarget[]; text: string; + buttons?: TelegramButtons; deliver: typeof deliverOutboundPayloads; shouldSend?: () => boolean; }) { @@ -170,13 +306,17 @@ async function deliverToTargets(params: { const channel = normalizeMessageChannel(target.channel) ?? target.channel; if (!isDeliverableMessageChannel(channel)) return; try { + const payload = buildPayloadWithButtons(params.text, params.buttons); + log.info( + `exec approvals: delivering to ${channel}:${target.to} payload=${JSON.stringify(payload).slice(0, 200)}`, + ); await params.deliver({ cfg: params.cfg, channel, to: target.to, accountId: target.accountId, threadId: target.threadId, - payloads: [{ text: params.text }], + payloads: [payload], }); } catch (err) { log.error(`exec approvals: failed to deliver to ${channel}:${target.to}: ${String(err)}`); @@ -193,12 +333,10 @@ export function createExecApprovalForwarder( const nowMs = deps.nowMs ?? Date.now; const resolveSessionTarget = deps.resolveSessionTarget ?? defaultResolveSessionTarget; const pending = new Map(); + const pendingBatches = new Map(); - const handleRequested = async (request: ExecApprovalRequest) => { - const cfg = getConfig(); + const resolveTargets = (cfg: ClawdbotConfig, request: ExecApprovalRequest): ForwardTarget[] => { const config = cfg.approvals?.exec; - if (!shouldForward({ config, request })) return; - const mode = normalizeMode(config?.mode); const targets: ForwardTarget[] = []; const seen = new Set(); @@ -224,8 +362,94 @@ export function createExecApprovalForwarder( } } + return targets; + }; + + const flushBatch = async (sessionKey: string) => { + const batch = pendingBatches.get(sessionKey); + if (!batch) return; + pendingBatches.delete(sessionKey); + + if (batch.flushTimeoutId) clearTimeout(batch.flushTimeoutId); + + const cfg = getConfig(); + const { requests, targets } = batch; + const liveRequests = requests.filter((req) => pending.has(req.id)); + + if (liveRequests.length === 0 || targets.length === 0) return; + + let batchId: string | null = null; + if (liveRequests.length > 1) { + // Generate batch ID and register it + batchId = generateBatchId(); + const approvalIds = liveRequests.map((r) => r.id); + // Cleanup any expired batches before inserting + cleanupExpiredBatches(nowMs()); + batchRegistry.set(batchId, { + approvalIds, + sessionKey, + createdAtMs: nowMs(), + }); + + // Update pending entries with batch ID + for (const req of liveRequests) { + const entry = pending.get(req.id); + if (entry) entry.batchId = batchId; + } + } + + log.info( + `exec approvals: flushing ${ + liveRequests.length > 1 ? `batch ${batchId}` : "single request" + } with ${liveRequests.length} request${liveRequests.length > 1 ? "s" : ""}`, + ); + + // Send single or batch message depending on count + if (liveRequests.length === 1) { + // Single request - use original format with all buttons + const message = buildRequestMessage(liveRequests[0], nowMs()); + await deliverToTargets({ + cfg, + targets, + text: message.text, + buttons: message.buttons, + deliver, + shouldSend: () => pending.has(liveRequests[0].id), + }); + } else { + // Multiple requests - use batch format + if (!batchId) return; + const message = buildBatchRequestMessage(liveRequests, batchId, nowMs()); + await deliverToTargets({ + cfg, + targets, + text: message.text, + buttons: message.buttons, + deliver, + }); + } + }; + + // Count pending (unresolved) approvals for a session + const countPendingForSession = (sessionKey: string): number => { + let count = 0; + for (const entry of pending.values()) { + if (entry.request.request.sessionKey === sessionKey) { + count++; + } + } + return count; + }; + + const handleRequested = async (request: ExecApprovalRequest) => { + const cfg = getConfig(); + const config = cfg.approvals?.exec; + if (!shouldForward({ config, request })) return; + + const targets = resolveTargets(cfg, request); if (targets.length === 0) return; + // Set up expiry timeout for this request const expiresInMs = Math.max(0, request.expiresAtMs - nowMs()); const timeoutId = setTimeout(() => { void (async () => { @@ -241,16 +465,75 @@ export function createExecApprovalForwarder( const pendingEntry: PendingApproval = { request, targets, timeoutId }; pending.set(request.id, pendingEntry); - if (pending.get(request.id) !== pendingEntry) return; + // Batch by session key + const sessionKey = request.request.sessionKey ?? "default"; + const now = nowMs(); + let batch = pendingBatches.get(sessionKey); - const text = buildRequestMessage(request, nowMs()); - await deliverToTargets({ - cfg, - targets, - text, - deliver, - shouldSend: () => pending.get(request.id) === pendingEntry, - }); + // Check if there are already pending (unresolved) approvals for this session + // (excluding the one we just added) + const existingPendingCount = countPendingForSession(sessionKey) - 1; + const hasPendingApprovals = existingPendingCount > 0; + + if (!batch) { + // Start a new batch + // Use extended window if there are already pending approvals for this session + const windowMs = hasPendingApprovals ? BATCH_EXTEND_WINDOW_MS : BATCH_WINDOW_MS; + + batch = { + sessionKey, + requests: [], + targets, + flushTimeoutId: null, + createdAtMs: now, + lastRequestAtMs: now, + }; + pendingBatches.set(sessionKey, batch); + + log.info( + `exec approvals: starting new batch for session, window=${windowMs}ms, hasPending=${hasPendingApprovals}`, + ); + + // Set up flush timeout + batch.flushTimeoutId = setTimeout(() => { + void flushBatch(sessionKey); + }, windowMs); + batch.flushTimeoutId.unref?.(); + } else { + // Batch already exists - extend the window if we haven't hit the max + const batchAge = now - batch.createdAtMs; + const timeRemaining = BATCH_MAX_WINDOW_MS - batchAge; + + if (timeRemaining > 0) { + // Clear existing timeout and set a new one + if (batch.flushTimeoutId) clearTimeout(batch.flushTimeoutId); + + // Use the shorter of: extended window or time remaining until max + const extensionMs = Math.min(BATCH_EXTEND_WINDOW_MS, timeRemaining); + + log.info( + `exec approvals: extending batch window by ${extensionMs}ms (age=${batchAge}ms, requests=${batch.requests.length + 1})`, + ); + + batch.flushTimeoutId = setTimeout(() => { + void flushBatch(sessionKey); + }, extensionMs); + batch.flushTimeoutId.unref?.(); + } else { + log.info(`exec approvals: batch at max window (${BATCH_MAX_WINDOW_MS}ms), will flush soon`); + } + + batch.lastRequestAtMs = now; + } + + batch.requests.push(request); + // Merge targets (in case they differ, though unlikely) + for (const target of targets) { + const key = buildTargetKey(target); + if (!batch.targets.some((t) => buildTargetKey(t) === key)) { + batch.targets.push(target); + } + } }; const handleResolved = async (resolved: ExecApprovalResolved) => { @@ -269,6 +552,10 @@ export function createExecApprovalForwarder( if (entry.timeoutId) clearTimeout(entry.timeoutId); } pending.clear(); + for (const batch of pendingBatches.values()) { + if (batch.flushTimeoutId) clearTimeout(batch.flushTimeoutId); + } + pendingBatches.clear(); }; return { handleRequested, handleResolved, stop }; diff --git a/src/infra/exec-approvals.test.ts b/src/infra/exec-approvals.test.ts index 57abeb695..4b47a80d8 100644 --- a/src/infra/exec-approvals.test.ts +++ b/src/infra/exec-approvals.test.ts @@ -33,7 +33,7 @@ function makeTempDir() { } describe("exec approvals allowlist matching", () => { - it("ignores basename-only patterns", () => { + it("matches basename-only patterns case-insensitively", () => { const resolution = { rawExecutable: "rg", resolvedPath: "/opt/homebrew/bin/rg", @@ -41,7 +41,7 @@ describe("exec approvals allowlist matching", () => { }; const entries: ExecAllowlistEntry[] = [{ pattern: "RG" }]; const match = matchAllowlist(entries, resolution); - expect(match).toBeNull(); + expect(match?.pattern).toBe("RG"); }); it("matches by resolved path with **", () => { @@ -66,7 +66,7 @@ describe("exec approvals allowlist matching", () => { expect(match).toBeNull(); }); - it("requires a resolved path", () => { + it("falls back to rawExecutable when resolvedPath is undefined", () => { const resolution = { rawExecutable: "bin/rg", resolvedPath: undefined, @@ -74,7 +74,7 @@ describe("exec approvals allowlist matching", () => { }; const entries: ExecAllowlistEntry[] = [{ pattern: "bin/rg" }]; const match = matchAllowlist(entries, resolution); - expect(match).toBeNull(); + expect(match?.pattern).toBe("bin/rg"); }); }); diff --git a/src/infra/exec-approvals.ts b/src/infra/exec-approvals.ts index 0830ed89a..c2d06fd99 100644 --- a/src/infra/exec-approvals.ts +++ b/src/infra/exec-approvals.ts @@ -508,14 +508,20 @@ export function matchAllowlist( entries: ExecAllowlistEntry[], resolution: CommandResolution | null, ): ExecAllowlistEntry | null { - if (!entries.length || !resolution?.resolvedPath) return null; + if (!entries.length || !resolution) return null; + const rawExecutable = resolution.rawExecutable; const resolvedPath = resolution.resolvedPath; + const executableName = resolution.executableName; for (const entry of entries) { const pattern = entry.pattern?.trim(); if (!pattern) continue; const hasPath = pattern.includes("/") || pattern.includes("\\") || pattern.includes("~"); - if (!hasPath) continue; - if (matchesPattern(pattern, resolvedPath)) return entry; + if (hasPath) { + const target = resolvedPath ?? rawExecutable; + if (target && matchesPattern(pattern, target)) return entry; + continue; + } + if (executableName && matchesPattern(pattern, executableName)) return entry; } return null; } diff --git a/src/infra/outbound/deliver.ts b/src/infra/outbound/deliver.ts index d246889e9..e957dfa87 100644 --- a/src/infra/outbound/deliver.ts +++ b/src/infra/outbound/deliver.ts @@ -17,6 +17,7 @@ import { sendMessageSignal } from "../../signal/send.js"; import type { sendMessageSlack } from "../../slack/send.js"; import type { sendMessageTelegram } from "../../telegram/send.js"; import type { sendMessageWhatsApp } from "../../web/outbound.js"; +import { createSubsystemLogger } from "../../logging/subsystem.js"; import { appendAssistantMessageToSessionTranscript, resolveMirroredTranscriptText, @@ -28,6 +29,8 @@ import type { OutboundChannel } from "./targets.js"; export type { NormalizedOutboundPayload } from "./payloads.js"; export { normalizeOutboundPayloads } from "./payloads.js"; +const log = createSubsystemLogger("gateway/outbound"); + type SendMatrixMessage = ( to: string, text: string, @@ -321,7 +324,11 @@ export async function deliverOutboundPayloads(params: { try { throwIfAborted(abortSignal); params.onPayload?.(payloadSummary); + log.debug( + `outbound: channel=${channel} hasSendPayload=${!!handler.sendPayload} hasChannelData=${!!payload.channelData}`, + ); if (handler.sendPayload && payload.channelData) { + log.debug(`outbound: using sendPayload for ${channel}`); results.push(await handler.sendPayload(payload)); continue; }