Skip to content

refactor: run SkillFlows through core's executeFlow #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Nivesh353 wants to merge 2 commits into
base: main
Choose a base branch
from
Open

refactor: run SkillFlows through core's executeFlow #3

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a19a125
refactor: run SkillFlows through core's executeFlow
Nivesh353 Jul 28, 2026
87e4617
fix: queue approval gates and repair the 5-minute timeout
Nivesh353 Jul 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
228 changes: 154 additions & 74 deletions src/server.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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<void> = 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<boolean> {
const priorGate = approvalQueue;
let releaseSlot!: () => void;
approvalQueue = new Promise<void>((r) => { releaseSlot = r; });
await priorGate;

@shreyas-lyzr shreyas-lyzr Jul 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Minor: approvalQueue is overwritten before await priorGate returns. If a third concurrent call arrives while the first gate is still active, the third call captures the second's promise as priorGate and the chain stays correct — but the intent takes a second read to see. A one-line comment here ('each caller chains off its predecessor') would help the next person.

try {
return await askForApproval(channel, message);
} finally {
releaseSlot();
}
}

async function askForApproval(channel: string, message: string): Promise<boolean> {
// Send message via the chosen channel
if (channel === "telegram" && telegramToken && lastTelegramChatId) {
await fetch(`https://api.telegram.org/bot${telegramToken}/sendMessage`, {
Expand All @@ -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<boolean>((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;

@shreyas-lyzr shreyas-lyzr Jul 28, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

e.channel || "telegram" will show "telegram" in the browser even when the gate fires on WhatsApp, because e.channel is the value from the flow YAML and may be absent. Fine for now, but worth aligning with defaultApprovalChannel() if the UI message needs to be accurate.

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<string> {
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),
Comment thread
Nivesh353 marked this conversation as resolved.
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 ────────────────────────────────────────────────
Expand Down Expand Up @@ -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),
Comment thread
Nivesh353 marked this conversation as resolved.
broadcastToBrowsers,
appendToHistory: (msg: any) => appendMessage(opts.agentDir, activeBranch, msg),
};
Expand Down Expand Up @@ -3134,14 +3209,19 @@ a{color:#58a6ff;}</style></head>
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}` });
});
Expand Down