Merge 9f84b8c266 into 72fea5e305
This commit is contained in:
commit
7ebb9aaf6c
@ -253,6 +253,27 @@ export const telegramPlugin: ChannelPlugin<ResolvedTelegramAccount> = {
|
||||
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<Array<{ text: string; callback_data: string }>>;
|
||||
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;
|
||||
|
||||
@ -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<MsgContext>) {
|
||||
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"]);
|
||||
});
|
||||
});
|
||||
|
||||
@ -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<string, "allow-once" | "allow-always" | "deny"> = {
|
||||
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 <id> allow-once|allow-always|deny" };
|
||||
@ -54,6 +63,44 @@ function parseApproveCommand(raw: string): ParsedApproveCommand | null {
|
||||
return { ok: false, error: "Usage: /approve <id> 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 <batch-id> allow-once|deny" };
|
||||
}
|
||||
const tokens = rest.split(/\s+/).filter(Boolean);
|
||||
if (tokens.length < 2) {
|
||||
return { ok: false, error: "Usage: /approve-batch <batch-id> allow-once|deny" };
|
||||
}
|
||||
|
||||
const batchId = tokens[0];
|
||||
const decisionRaw = tokens[1].toLowerCase();
|
||||
|
||||
// Batch only supports allow-once and deny
|
||||
const batchDecisions: Record<string, "allow-once" | "deny"> = {
|
||||
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 <batch-id> allow-once|deny" };
|
||||
}
|
||||
|
||||
return { ok: true, batchId, decision };
|
||||
}
|
||||
|
||||
function buildResolvedByLabel(params: Parameters<CommandHandler>[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 || "<unknown>"}`,
|
||||
);
|
||||
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.`,
|
||||
},
|
||||
};
|
||||
};
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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();
|
||||
|
||||
@ -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<string, BatchEntry>();
|
||||
|
||||
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<void>;
|
||||
handleResolved: (resolved: ExecApprovalResolved) => Promise<void>;
|
||||
@ -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<Array<{ text: string; callback_data: string }>>;
|
||||
};
|
||||
|
||||
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 <id> 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 <id> 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<Array<{ text: string; callback_data: string }>>;
|
||||
|
||||
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<string, PendingApproval>();
|
||||
const pendingBatches = new Map<string, PendingBatch>();
|
||||
|
||||
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<string>();
|
||||
@ -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 };
|
||||
|
||||
@ -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");
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@ -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;
|
||||
}
|
||||
|
||||
@ -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;
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user