From a19a12577982b9bb34bf4634e42cee91600a68e7 Mon Sep 17 00:00:00 2001 From: Nivesh353 Date: Tue, 28 Jul 2026 12:12:09 +0530 Subject: [PATCH 1/2] refactor: run SkillFlows through core's executeFlow MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The step loop moved into @open-gitagent/gitagent so the CLI, SDK and scheduler can run flows too (open-gitagent/gitagent#89). Voice now supplies only the parts that are genuinely browser- and chat-specific. Deleted: the step loop, the __approval_gate__ branch, prompt building, {input} substitution and context threading. Those now live in core. Kept and wired in as callbacks: - runStep — spawn query(), stream tool calls to the browser - onProgress — flowProgressToBrowser() maps core's events onto the WebSocket messages the UI already understood - requestApproval — the existing Telegram/WhatsApp sender Approval gates no longer auto-approve when no channel is connected. A gate guards a risky step; continuing because nobody could be reached silently removes the protection it exists to provide. The browser now gets a specific message for each outcome — no channel configured, timed out, or explicitly declined — instead of the same bare "denied" line, and the 5-minute timeout is no longer silent. A gate that omits `channel` now resolves to whichever channel is actually connected, rather than assuming Telegram. Hand-written flows previously failed on a WhatsApp-only setup. The @flow trigger is anchored to the start of the message, so an address like bob@daily-report.com no longer fires a flow. Schedules can run flows: runFlow is passed to the scheduler. --- src/server.ts | 170 ++++++++++++++++++++++++++++++++------------------ 1 file changed, 108 insertions(+), 62 deletions(-) diff --git a/src/server.ts b/src/server.ts index aec73b2..dbcb0e8 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"; @@ -795,6 +799,19 @@ export async function startVoiceServer(opts: VoiceServerOptions): Promise<() => 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"; + } + async function sendApprovalRequest(channel: string, message: string): Promise { // Send message via the chosen channel if (channel === "telegram" && telegramToken && lastTelegramChatId) { @@ -807,7 +824,20 @@ 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) @@ -816,7 +846,15 @@ export async function startVoiceServer(opts: VoiceServerOptions): Promise<() => const timeout = setTimeout(() => { if (pendingApproval?.resolve === resolve) { pendingApproval = null; - resolve(false); // Timeout = deny + // 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; @@ -827,69 +865,71 @@ export async function startVoiceServer(opts: VoiceServerOptions): Promise<() => }); } - 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 +991,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 +3178,16 @@ 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. + const flowMatch = msg.text.match(/^@([a-z0-9]+(?:-[a-z0-9]+)*)\b\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}` }); }); From 87e461784bb8f737e13270fdc6d2ec3c00487b01 Mon Sep 17 00:00:00 2001 From: Nivesh353 Date: Tue, 28 Jul 2026 14:17:59 +0530 Subject: [PATCH 2/2] fix: queue approval gates and repair the 5-minute timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pendingApproval was a single slot set unconditionally, so a second gate (a cron flow and a browser @flow can now overlap) overwrote the first and a YES resolved the wrong one. Gates now queue: a second gate does not message anyone until the first is answered. The timeout never fired at all — it compared pendingApproval.resolve against the closure's resolve, but the slot holds a wrapper, so the check was always false. Match on a per-gate id instead. --- src/server.ts | 80 ++++++++++++++++++++++++++++++++++++--------------- 1 file changed, 57 insertions(+), 23 deletions(-) diff --git a/src/server.ts b/src/server.ts index dbcb0e8..689a744 100644 --- a/src/server.ts +++ b/src/server.ts @@ -781,19 +781,25 @@ 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; @@ -812,7 +818,25 @@ export async function startVoiceServer(opts: VoiceServerOptions): Promise<() => 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`, { @@ -840,27 +864,34 @@ export async function startVoiceServer(opts: VoiceServerOptions): Promise<() => 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; - // 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); - } + // 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); + }, }; }); } @@ -3180,8 +3211,11 @@ a{color:#58a6ff;} // 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. - const flowMatch = msg.text.match(/^@([a-z0-9]+(?:-[a-z0-9]+)*)\b\s*([\s\S]*)$/); + // "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);