diff --git a/src/server.ts b/src/server.ts index aec73b2..689a744 100644 --- a/src/server.ts +++ b/src/server.ts @@ -17,6 +17,10 @@ import { appendMessage, loadHistory, deleteHistory, summarizeHistory } from "@op import { getVoiceContext, getAgentContext } from "@open-gitagent/gitagent"; import { discoverSkills } from "@open-gitagent/gitagent"; import { discoverWorkflows, loadFlowDefinition, saveFlowDefinition, deleteFlowDefinition } from "@open-gitagent/gitagent"; +// The step loop lives in core so the CLI, scheduler and SDK run flows the same +// way; voice supplies the browser/Telegram side effects. +import { executeFlow as runFlowSteps } from "@open-gitagent/gitagent"; +import type { FlowEvent } from "@open-gitagent/gitagent"; import { discoverSchedules, saveSchedule, deleteSchedule, updateScheduleMeta } from "@open-gitagent/gitagent"; import { startScheduler, stopScheduler, reloadSchedules, executeScheduledJob } from "@open-gitagent/gitagent"; import cron from "node-cron"; @@ -777,25 +781,62 @@ export async function startVoiceServer(opts: VoiceServerOptions): Promise<() => // ── SkillFlow execution ───────────────────────────────────────────── // ── Approval gate state ──────────────────────────────────────────── - let pendingApproval: { resolve: (approved: boolean) => void } | null = null; + // Only one gate may be awaiting an answer at a time: a bare "YES" arriving + // over Telegram carries no indication of which flow it belongs to, so a + // second gate queues behind the first rather than overwriting it. `id` + // identifies the occupant so a superseded timeout can't resolve someone + // else's gate. + let pendingApproval: { id: number; resolve: (approved: boolean) => void } | null = null; + let approvalSeq = 0; + let approvalQueue: Promise = Promise.resolve(); function handleApprovalReply(text: string): boolean { if (!pendingApproval) return false; const lower = text.trim().toLowerCase(); + // resolve() clears pendingApproval itself, so there's no double-resolve. if (["yes", "approve", "continue", "ok", "go", "y", "proceed"].includes(lower)) { pendingApproval.resolve(true); - pendingApproval = null; return true; } if (["no", "deny", "stop", "cancel", "abort", "n", "reject"].includes(lower)) { pendingApproval.resolve(false); - pendingApproval = null; return true; } return false; } + /** + * Which channel to ask on when the gate didn't name one. Mirrors what the + * flow builder does at authoring time (prefer Telegram, else WhatsApp) — + * but resolved at run time, so a hand-written gate still reaches whichever + * channel happens to be connected. Falls back to "telegram" purely so the + * "not connected" message names something when neither is up. + */ + function defaultApprovalChannel(): string { + if (telegramToken && lastTelegramChatId) return "telegram"; + if (whatsappSock && whatsappConnected && lastWhatsAppJid) return "whatsapp"; + return "telegram"; + } + + /** + * Ask a human to approve, one gate at a time. Concurrent flows are real now + * that schedules can run them — a cron job and a browser @flow can both + * reach a gate — so this serialises them: a second gate doesn't message + * anyone until the first has been answered, timed out, or given up. + */ async function sendApprovalRequest(channel: string, message: string): Promise { + const priorGate = approvalQueue; + let releaseSlot!: () => void; + approvalQueue = new Promise((r) => { releaseSlot = r; }); + await priorGate; + try { + return await askForApproval(channel, message); + } finally { + releaseSlot(); + } + } + + async function askForApproval(channel: string, message: string): Promise { // Send message via the chosen channel if (channel === "telegram" && telegramToken && lastTelegramChatId) { await fetch(`https://api.telegram.org/bot${telegramToken}/sendMessage`, { @@ -807,89 +848,119 @@ export async function startVoiceServer(opts: VoiceServerOptions): Promise<() => const sent = await whatsappSock.sendMessage(lastWhatsAppJid, { text: message }); if (sent?.key?.id) whatsappSentIds.add(sent.key.id); } else { - return true; // No channel available — auto-approve + // No channel to ask on. A gate guards a risky step, so with nobody + // reachable the safe answer is no — previously this auto-approved, + // which silently removed the protection the gate existed to provide. + // + // Say so in the chat as well as the log: from the browser, "denied" + // is indistinguishable from the user having replied NO, and this + // case is a setup problem with a specific fix. + console.log(dim(`[flow] approval gate could not be delivered — no ${channel} channel connected; denying`)); + broadcastToBrowsers({ + type: "transcript", + role: "assistant", + text: `⚠️ This flow has an approval gate, but no ${channel === "whatsapp" ? "WhatsApp" : "Telegram"} connection is available to ask on. Connect one in the Communication tab (and message the bot once so it knows where to reach you), or remove the gate from the flow. Stopping here.`, + }); + return false; } - // Wait for reply (timeout after 5 minutes) + // Wait for reply (timeout after 5 minutes). return new Promise((resolve) => { - pendingApproval = { resolve }; + const id = ++approvalSeq; + const timeout = setTimeout(() => { - if (pendingApproval?.resolve === resolve) { - pendingApproval = null; - resolve(false); // Timeout = deny - } + // Compare by id: the slot's `resolve` is the wrapper below, never + // this closure's `resolve`, so identity on the function itself is + // always false and the timeout would never fire. + if (pendingApproval?.id !== id) return; + pendingApproval = null; + // Timeout = deny. Distinguish it from an explicit NO, which + // otherwise produces the identical "denied" line in chat. + console.log(dim(`[flow] approval gate timed out after 5 minutes on ${channel}; denying`)); + broadcastToBrowsers({ + type: "transcript", + role: "assistant", + text: `⏱ No approval reply on ${channel} within 5 minutes — treating as "no" and stopping the flow.`, + }); + resolve(false); }, 5 * 60 * 1000); - const origResolve = resolve; - pendingApproval.resolve = (val: boolean) => { - clearTimeout(timeout); - origResolve(val); + + pendingApproval = { + id, + resolve: (val: boolean) => { + clearTimeout(timeout); + pendingApproval = null; + resolve(val); + }, }; }); } - async function executeFlow(flowName: string, userContext: string, sendToBrowser: (msg: ServerMessage) => void) { - const flowPath = join(resolve(opts.agentDir), "workflows", flowName + ".yaml"); - const flow = await loadFlowDefinition(flowPath); - - sendToBrowser({ type: "transcript", role: "assistant", - text: `Running flow: ${flow.name} (${flow.steps.length} steps)` }); - - let runningContext = userContext; - - for (let i = 0; i < flow.steps.length; i++) { - const step = flow.steps[i]; - - // ── Approval gate step ── - if (step.skill === "__approval_gate__") { - const channel = step.channel || "telegram"; - const customMsg = step.prompt || ""; - const approvalMsg = customMsg - ? `⏸ Approval Required: ${customMsg}\n\nReply YES to continue or NO to cancel.` - : `⏸ Flow "${flow.name}" paused at step ${i + 1}/${flow.steps.length}.\n\nCompleted so far:\n${runningContext.slice(0, 500)}\n\nReply YES to continue or NO to cancel.`; - - sendToBrowser({ type: "transcript", role: "assistant", - text: `⏸ Waiting for approval via ${channel}...` }); - - const approved = await sendApprovalRequest(channel, approvalMsg); - - if (!approved) { + // Translates core flow events into the browser's message stream. Kept + // separate so the scheduler can run flows with a different sink. + function flowProgressToBrowser(sendToBrowser: (msg: ServerMessage) => void) { + return (e: FlowEvent) => { + switch (e.type) { + case "flow_start": sendToBrowser({ type: "transcript", role: "assistant", - text: `Flow "${flow.name}" was denied at approval gate (step ${i + 1}).` }); - return; - } - sendToBrowser({ type: "transcript", role: "assistant", - text: `✓ Approval received — continuing flow.` }); - runningContext += `\n\n[Step ${i + 1}: approval gate]: Approved via ${channel}`; - continue; + text: `Running flow: ${e.flow} (${e.totalSteps} steps)` }); + break; + case "step_start": + sendToBrowser({ type: "agent_working" as any, query: `Step ${e.index + 1}/${e.total}: ${e.skill}` } as any); + break; + case "step_done": + sendToBrowser({ type: "agent_done" as any, result: `Step ${e.index + 1} complete` } as any); + break; + case "approval_requested": + sendToBrowser({ type: "transcript", role: "assistant", + text: `⏸ Waiting for approval via ${e.channel || "telegram"}...` }); + break; + case "approval_resolved": + if (e.approved) { + sendToBrowser({ type: "transcript", role: "assistant", text: `✓ Approval received — continuing flow.` }); + } + break; + case "flow_aborted": + sendToBrowser({ type: "transcript", role: "assistant", text: e.reason }); + break; + case "flow_done": + sendToBrowser({ type: "transcript", role: "assistant", text: `Flow "${e.flow}" completed.` }); + break; } + }; + } - sendToBrowser({ type: "agent_working" as any, query: `Step ${i + 1}/${flow.steps.length}: ${step.skill}` } as any); - - const prompt = `Use the skill "${step.skill}" (load it with /skill:${step.skill}). -${step.prompt.replace(/\{input\}/g, userContext)} - -Context from previous steps: -${runningContext}`; - - const result = query({ - prompt, - dir: opts.agentDir, - model: opts.model, - env: opts.env, - }); + async function executeFlow(flowName: string, userContext: string, sendToBrowser: (msg: ServerMessage) => void): Promise { + const flowPath = join(resolve(opts.agentDir), "workflows", flowName + ".yaml"); + const flow = await loadFlowDefinition(flowPath); - let stepOutput = ""; - for await (const msg of result) { - if (msg.type === "assistant" && msg.content) stepOutput += msg.content; - if (msg.type === "tool_use") sendToBrowser({ type: "tool_call", toolName: msg.toolName, args: msg.args } as any); - if (msg.type === "tool_result") sendToBrowser({ type: "tool_result", toolName: msg.toolName, content: msg.content, isError: msg.isError } as any); - } + const result = await runFlowSteps(flow, userContext, { + runStep: async (prompt) => { + const stream = query({ + prompt, + dir: opts.agentDir, + model: opts.model, + env: opts.env, + }); - runningContext += `\n\n[Step ${i + 1} result (${step.skill})]: ${stepOutput}`; - sendToBrowser({ type: "agent_done" as any, result: `Step ${i + 1} complete` } as any); - } + let stepOutput = ""; + for await (const msg of stream) { + if (msg.type === "assistant" && msg.content) stepOutput += msg.content; + if (msg.type === "tool_use") sendToBrowser({ type: "tool_call", toolName: msg.toolName, args: msg.args } as any); + if (msg.type === "tool_result") sendToBrowser({ type: "tool_result", toolName: msg.toolName, content: msg.content, isError: msg.isError } as any); + } + return stepOutput; + }, + // Core calls (message, channel); voice's sender takes (channel, message). + // A gate authored by hand may omit the channel, so fall back to + // whichever one is actually connected rather than assuming Telegram. + requestApproval: (message, channel) => sendApprovalRequest(channel || defaultApprovalChannel(), message), + onProgress: flowProgressToBrowser(sendToBrowser), + }); - sendToBrowser({ type: "transcript", role: "assistant", text: `Flow "${flow.name}" completed.` }); + return result.completed + ? result.steps.map((s) => `[${s.skill}] ${s.output}`).join("\n\n") + : result.abortReason || `Flow "${flowName}" did not complete.`; } // ── File API helpers ──────────────────────────────────────────────── @@ -951,6 +1022,10 @@ ${runningContext}`; model: opts.model, env: opts.env, runPrompt: headlessHandler, + // Schedules can now run a SkillFlow instead of a bare prompt. Approval + // gates still route through Telegram/WhatsApp, so an unattended job + // pauses for a real human rather than waving itself through. + runFlow: (flowName: string, input: string) => executeFlow(flowName, input, scheduleSendToBrowser), broadcastToBrowsers, appendToHistory: (msg: any) => appendMessage(opts.agentDir, activeBranch, msg), }; @@ -3134,14 +3209,19 @@ a{color:#58a6ff;} if (msg.type === "text") { appendMessage(opts.agentDir, activeBranch, { type: "transcript", role: "user", text: msg.text }); - // Detect @flow-name triggers - const flowMatch = msg.text.match(/@([a-z0-9]+(?:-[a-z0-9]+)*)/); + // Detect @flow-name triggers. Anchored to the start of the + // message so an address or mention mid-sentence — say + // "email bob@daily-report.com" — can't fire a flow. The name + // must also be followed by whitespace or end-of-message: a + // bare \b would let "@daily-report.com" fire the flow with + // input ".com". + const flowMatch = msg.text.match(/^@([a-z0-9]+(?:-[a-z0-9]+)*)(?=\s|$)\s*([\s\S]*)$/); if (flowMatch) { try { const workflows = await discoverWorkflows(agentRoot); const flow = workflows.find((f) => f.name === flowMatch[1] && f.type === "flow"); if (flow) { - const userContext = msg.text.replace(/@[a-z0-9-]+/, "").trim(); + const userContext = flowMatch[2].trim(); executeFlow(flow.name, userContext, sendToBrowser).catch((err) => { sendToBrowser({ type: "transcript", role: "assistant", text: `Flow error: ${err.message}` }); });