From ce602ad5a98fe409b7ba93c0bdece9f41824afe3 Mon Sep 17 00:00:00 2001 From: Patryk Lewczuk Date: Sun, 24 May 2026 02:44:30 +0200 Subject: [PATCH 1/2] perf(runner): collapse heartbeat + events hot paths to single RPCs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The /api/runner/heartbeat and /api/runner/runs/:id/events endpoints were firing 4 and ~30-60 sequential Postgres statements per call respectively. Under multi-runner load this saturated the PgBouncer pool, producing the bimodal 200ms / 10-16s tail latency we hit on dev (and would hit harder in prod). Both paths now do a single round-trip: * runner_heartbeat(p_runner_id, p_status) — one CTE pipeline: bump last_heartbeat_at/status + return cancel_job_ids/pause_run_ids. Replaces 4 sequential statements. * ingest_runner_events(p_run_id, p_workspace_id, p_events jsonb) — one set-based insert opens agent_runs rows from step-start, one upsert closes them from step-end (needs the new agent_runs_run_step_iter_uniq constraint), one insert appends every event to agent_run_events, one atomic UPDATE rolls up tokens_used + current_step_id on workflow_runs. The tokens_used roll-up is now `tokens_used = tokens_used + delta`, fixing a read-modify-write race on concurrent batches. Also drops the duplicate touch_runner_heartbeat from /api/runner/jobs — the daemon's dedicated heartbeat is the single source of truth now. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../gui/src/app/api/runner/heartbeat/route.ts | 47 ++-- packages/gui/src/app/api/runner/jobs/route.ts | 8 +- .../api/runner/runs/[runId]/events/route.ts | 98 ++------- .../supabase/migrations/0020_runner_perf.sql | 201 ++++++++++++++++++ 4 files changed, 238 insertions(+), 116 deletions(-) create mode 100644 packages/gui/supabase/migrations/0020_runner_perf.sql diff --git a/packages/gui/src/app/api/runner/heartbeat/route.ts b/packages/gui/src/app/api/runner/heartbeat/route.ts index fb0ed58..4524346 100644 --- a/packages/gui/src/app/api/runner/heartbeat/route.ts +++ b/packages/gui/src/app/api/runner/heartbeat/route.ts @@ -10,12 +10,21 @@ interface HeartbeatBody { currentJobIds?: string[]; } +interface HeartbeatRpcRow { + cancel_job_ids: string[] | null; + pause_run_ids: string[] | null; +} + /** * POST /api/runner/heartbeat { status?, currentJobIds? } * * Refreshes the runner's `last_heartbeat_at`/`status` and tells it which of its * jobs/runs the operator has asked to cancel/pause (the runner has no other * channel for that). + * + * Implemented as a single `runner_heartbeat` RPC (migration 0020) — previously + * this was four sequential statements, which caused the bimodal 200ms / 10-16s + * tail latency we hit at multi-runner scale. */ export async function POST(req: Request) { const auth = await authRunner(req); @@ -26,34 +35,16 @@ export async function POST(req: Request) { try { body = await req.json(); } catch { /* empty body is fine */ } const status: RunnerStatus = body.status ?? 'online'; - await admin.from('runners').update({ last_heartbeat_at: new Date().toISOString(), status, updated_at: new Date().toISOString() }).eq('id', runner.id); - - // Jobs this runner holds that have been cancelled. - const { data: cancelledJobs } = await admin - .from('jobs') - .select('id') - .eq('claimed_by_runner', runner.id) - .in('status', ['cancelled']); - const cancelJobIds = (cancelledJobs ?? []).map((j) => j.id); - - // Workflow runs driven by this runner's jobs with pause_requested set. - let pauseRunIds: string[] = []; - { - const { data: activeJobs } = await admin - .from('jobs') - .select('id') - .eq('claimed_by_runner', runner.id) - .in('status', ['claimed', 'running']); - const jobIds = (activeJobs ?? []).map((j) => j.id); - if (jobIds.length > 0) { - const { data: runs } = await admin - .from('workflow_runs') - .select('id') - .in('job_id', jobIds) - .eq('pause_requested', true); - pauseRunIds = (runs ?? []).map((r) => r.id); - } - } + const { data, error } = await admin.rpc('runner_heartbeat', { + p_runner_id: runner.id, + p_status: status, + }); + if (error) return NextResponse.json({ error: error.message }, { status: 500 }); + + // PostgREST returns `setof record` functions as an array of rows. + const row = (Array.isArray(data) ? data[0] : data) as HeartbeatRpcRow | null | undefined; + const cancelJobIds = row?.cancel_job_ids ?? []; + const pauseRunIds = row?.pause_run_ids ?? []; return NextResponse.json({ ok: true, cancelJobIds, pauseRunIds }); } diff --git a/packages/gui/src/app/api/runner/jobs/route.ts b/packages/gui/src/app/api/runner/jobs/route.ts index 0d6e4df..51cccec 100644 --- a/packages/gui/src/app/api/runner/jobs/route.ts +++ b/packages/gui/src/app/api/runner/jobs/route.ts @@ -28,11 +28,9 @@ export async function GET(req: Request) { if (auth instanceof NextResponse) return auth; const { runner, admin } = auth; - // Heartbeat-on-poll (best-effort). - await admin.rpc('touch_runner_heartbeat', { p_runner_id: runner.id, p_status: 'online' }).then( - () => {}, - () => admin.from('runners').update({ last_heartbeat_at: new Date().toISOString(), status: 'online' }).eq('id', runner.id), - ); + // No heartbeat-on-poll: the daemon's dedicated `/api/runner/heartbeat` (now + // a single RPC, migration 0020) is the single source of truth. The previous + // double-write here was pure overhead per claim tick. const url = new URL(req.url); const requested = (url.searchParams.get('backends') ?? '').split(',').map((s) => s.trim()).filter(Boolean); diff --git a/packages/gui/src/app/api/runner/runs/[runId]/events/route.ts b/packages/gui/src/app/api/runner/runs/[runId]/events/route.ts index a667170..de5a2c4 100644 --- a/packages/gui/src/app/api/runner/runs/[runId]/events/route.ts +++ b/packages/gui/src/app/api/runner/runs/[runId]/events/route.ts @@ -1,12 +1,10 @@ import { NextResponse } from 'next/server'; import { authRunner, runnerScopesWorkspace } from '../../../_auth'; -import type { Database, AgentRunEventType, AgentRunStatus, AgentRunStepKind } from '@/lib/supabase/types'; +import type { AgentRunEventType } from '@/lib/supabase/types'; export const dynamic = 'force-dynamic'; export const runtime = 'nodejs'; -type AgentRunsRow = Database['public']['Tables']['agent_runs']['Row']; - interface IncomingEvent { type: AgentRunEventType | string; payload?: unknown; @@ -27,10 +25,13 @@ interface IncomingEvent { /** * POST /api/runner/runs/:runId/events { events: IncomingEvent[] } * - * The runner batches events here. For step lifecycle events we upsert an - * `agent_runs` row (insert on `step-start`, update on `step-end`) and advance - * `workflow_runs.current_step_id`; every event is also recorded in - * `agent_run_events`. `tokensUsed` accumulates into `workflow_runs.tokens_used`. + * The runner batches events here. The entire batch is processed by a single + * `ingest_runner_events` RPC (migration 0020): one set-based insert opens + * agent_runs rows from step-start, one upsert closes them from step-end, one + * insert appends every event to agent_run_events, and one atomic UPDATE rolls + * up `tokens_used` + `current_step_id` on workflow_runs. Previously this loop + * fired ~30-60 sequential statements per 25-event batch, which was the main + * driver of the 10-16s tail latency under multi-runner load. */ export async function POST(req: Request, { params }: { params: Promise<{ runId: string }> }) { const auth = await authRunner(req); @@ -45,83 +46,14 @@ export async function POST(req: Request, { params }: { params: Promise<{ runId: let body: { events?: IncomingEvent[] }; try { body = await req.json(); } catch { return NextResponse.json({ error: 'invalid json' }, { status: 400 }); } const events = Array.isArray(body.events) ? body.events : []; + if (events.length === 0) return NextResponse.json({ ok: true }); - // Resolve a step's agent_runs row id within this batch (step-start creates it). - const stepRunIds = new Map(); // `${stepId}#${iteration}` -> agent_runs.id - const stepKey = (e: IncomingEvent) => `${e.stepId}#${e.iteration ?? 1}`; - let tokenDelta = 0; - let lastStepId: string | null = null; - - for (const e of events) { - if (typeof e.tokensUsed === 'number') tokenDelta += e.tokensUsed; - - let agentRunId: string | null = e.agentRunId ?? null; - - if (e.stepId && e.type === 'step-start') { - const { data, error } = await admin - .from('agent_runs') - .insert({ - workspace_id: run.workspace_id, - workflow_run_id: runId, - step_id: e.stepId, - iteration: e.iteration ?? 1, - kind: (e.kind ?? null) as AgentRunStepKind | null, - backend: e.backend ?? null, - model: e.model ?? null, - status: 'running', - started_at: e.startedAt ?? new Date().toISOString(), - }) - .select('id') - .single(); - if (!error && data) { stepRunIds.set(stepKey(e), data.id); agentRunId = data.id; } - lastStepId = e.stepId; - await admin.from('workflow_runs').update({ current_step_id: e.stepId }).eq('id', runId); - } else if (e.stepId && e.type === 'step-end') { - const existing = stepRunIds.get(stepKey(e)); - const dbStatus: AgentRunStatus = - e.status === 'succeeded' ? 'succeeded' : e.status === 'skipped' ? 'skipped' : e.status === 'running' ? 'running' : 'failed'; - if (existing) { - await admin.from('agent_runs').update({ - status: dbStatus, finished_at: e.finishedAt ?? new Date().toISOString(), - tokens_used: e.tokensUsed ?? 0, summary: e.summary ?? null, error: e.error ?? null, - kind: (e.kind ?? undefined) as AgentRunsRow['kind'] | undefined, - }).eq('id', existing); - agentRunId = existing; - } else { - // No matching step-start in this/an earlier batch — insert a closed row. - const { data } = await admin.from('agent_runs').insert({ - workspace_id: run.workspace_id, workflow_run_id: runId, step_id: e.stepId, iteration: e.iteration ?? 1, - kind: (e.kind ?? null) as AgentRunStepKind | null, backend: e.backend ?? null, model: e.model ?? null, - status: dbStatus, started_at: e.startedAt ?? new Date().toISOString(), finished_at: e.finishedAt ?? new Date().toISOString(), - tokens_used: e.tokensUsed ?? 0, summary: e.summary ?? null, error: e.error ?? null, - }).select('id').single(); - agentRunId = data?.id ?? null; - } - lastStepId = e.stepId; - } - - await admin.from('agent_run_events').insert({ - workspace_id: run.workspace_id, - workflow_run_id: runId, - agent_run_id: agentRunId, - type: (isKnownEventType(e.type) ? e.type : 'note') as AgentRunEventType, - payload: (e.payload ?? {}) as Database['public']['Tables']['agent_run_events']['Row']['payload'], - }); - } - - if (tokenDelta > 0) { - const { data: cur } = await admin.from('workflow_runs').select('tokens_used').eq('id', runId).single(); - await admin.from('workflow_runs').update({ tokens_used: (cur?.tokens_used ?? 0) + tokenDelta }).eq('id', runId); - } - if (lastStepId) { - // (current_step_id already set on step-start; this keeps it fresh for batches - // that only carried step-end events.) - await admin.from('workflow_runs').update({ current_step_id: lastStepId }).eq('id', runId); - } + const { error } = await admin.rpc('ingest_runner_events', { + p_run_id: runId, + p_workspace_id: run.workspace_id, + p_events: events, + }); + if (error) return NextResponse.json({ error: error.message }, { status: 500 }); return NextResponse.json({ ok: true }); } - -function isKnownEventType(t: string): t is AgentRunEventType { - return ['lifecycle', 'agent-text', 'tool-call', 'tool-result', 'note', 'step-start', 'step-end'].includes(t); -} diff --git a/packages/gui/supabase/migrations/0020_runner_perf.sql b/packages/gui/supabase/migrations/0020_runner_perf.sql new file mode 100644 index 0000000..b30d088 --- /dev/null +++ b/packages/gui/supabase/migrations/0020_runner_perf.sql @@ -0,0 +1,201 @@ +-- Runner hot-path collapse: 4-query heartbeat → 1 RPC, N+2-query event ingest +-- → 1 RPC. Production symptom was bimodal 200ms / 10-16s tail latency on +-- /api/runner/heartbeat and /api/runner/runs/:id/events under multi-runner +-- load — caused by PgBouncer pool starvation from chatty per-request +-- pipelines (heartbeat = 4 sequential statements, a 25-event batch = 30-60 +-- sequential statements). Both endpoints now do a single RPC round-trip. + +-- ─── agent_runs unique key ───────────────────────────────────────────── +-- Required by ingest_runner_events' `on conflict` upsert: a step's +-- (workflow_run_id, step_id, iteration) is logically unique already, the +-- constraint just makes it enforceable so the RPC can be set-based. +alter table agent_runs + add constraint agent_runs_run_step_iter_uniq + unique (workflow_run_id, step_id, iteration); + +-- ─── runner_heartbeat ────────────────────────────────────────────────── +-- Replaces the per-heartbeat sequence: +-- UPDATE runners … +-- SELECT cancelled jobs claimed by this runner +-- SELECT this runner's active jobs +-- SELECT workflow_runs.pause_requested for those jobs +-- with a single CTE pipeline that runs in one round-trip and returns +-- (cancel_job_ids, pause_run_ids). +create or replace function runner_heartbeat( + p_runner_id uuid, + p_status text default 'online' +) returns table (cancel_job_ids uuid[], pause_run_ids uuid[]) +language sql +as $$ + with hb as ( + update runners + set last_heartbeat_at = now(), + status = coalesce(p_status, status), + updated_at = now() + where id = p_runner_id + returning id + ), + -- Single pass over `jobs` partitioned by status: cancellations + the set of + -- active jobs whose workflow_runs we'll check for pause_requested. + runner_jobs as ( + select id, status + from jobs + where claimed_by_runner = p_runner_id + and status in ('claimed', 'running', 'cancelled') + ), + cancels as ( + select coalesce(array_agg(id), '{}'::uuid[]) as ids + from runner_jobs + where status = 'cancelled' + ), + pauses as ( + select coalesce(array_agg(wr.id), '{}'::uuid[]) as ids + from workflow_runs wr + join runner_jobs rj on rj.id = wr.job_id + where wr.pause_requested = true + and rj.status in ('claimed', 'running') + ) + select cancels.ids, pauses.ids + from hb, cancels, pauses; +$$; + +-- ─── ingest_runner_events ────────────────────────────────────────────── +-- Replaces the per-batch loop in /api/runner/runs/:id/events. A 25-event +-- batch previously generated ~30-60 statements (insert agent_runs per +-- step-start, update agent_runs per step-end, insert agent_run_events per +-- event, select-then-update workflow_runs.tokens_used). This function runs +-- the whole batch in three set-based statements + one workflow_runs update, +-- all in a single round-trip, and makes the token rollup atomic +-- (`tokens_used = tokens_used + delta`). +-- +-- The event payload shape mirrors `RunnerEvent` (packages/runner/src/runner-client.ts): +-- { type, payload, stepId?, iteration?, kind?, backend?, model?, status?, +-- summary?, error?, tokensUsed?, startedAt?, finishedAt? } +-- +-- Caller guarantees: p_run_id is a real workflow_runs.id whose workspace +-- equals p_workspace_id (the route does this check before calling). +create or replace function ingest_runner_events( + p_run_id uuid, + p_workspace_id uuid, + p_events jsonb +) returns void +language plpgsql +as $$ +declare + v_token_delta int; + v_last_step text; +begin + if p_events is null or jsonb_typeof(p_events) <> 'array' or jsonb_array_length(p_events) = 0 then + return; + end if; + + -- 1. step-start: open an agent_runs row per step. Idempotent on the unique + -- key so a re-delivered batch doesn't double-insert; the running row + -- stays in place until step-end (#2) closes it. + insert into agent_runs ( + workspace_id, workflow_run_id, step_id, iteration, kind, backend, model, + status, started_at + ) + select + p_workspace_id, + p_run_id, + e->>'stepId', + coalesce((e->>'iteration')::int, 1), + e->>'kind', + e->>'backend', + e->>'model', + 'running', + coalesce((e->>'startedAt')::timestamptz, now()) + from jsonb_array_elements(p_events) as e + where e->>'type' = 'step-start' + and e->>'stepId' is not null + on conflict (workflow_run_id, step_id, iteration) do nothing; + + -- 2. step-end: close the matching row (or insert a closed row if step-start + -- was never delivered for it — the route's previous fallback path). + insert into agent_runs ( + workspace_id, workflow_run_id, step_id, iteration, kind, backend, model, + status, started_at, finished_at, tokens_used, summary, error + ) + select + p_workspace_id, + p_run_id, + e->>'stepId', + coalesce((e->>'iteration')::int, 1), + e->>'kind', + e->>'backend', + e->>'model', + case e->>'status' + when 'succeeded' then 'succeeded' + when 'skipped' then 'skipped' + when 'running' then 'running' + else 'failed' + end, + coalesce((e->>'startedAt')::timestamptz, now()), + coalesce((e->>'finishedAt')::timestamptz, now()), + coalesce((e->>'tokensUsed')::int, 0), + e->>'summary', + e->>'error' + from jsonb_array_elements(p_events) as e + where e->>'type' = 'step-end' + and e->>'stepId' is not null + on conflict (workflow_run_id, step_id, iteration) do update set + status = excluded.status, + finished_at = excluded.finished_at, + tokens_used = excluded.tokens_used, + summary = excluded.summary, + error = excluded.error, + kind = coalesce(agent_runs.kind, excluded.kind); + + -- 3. agent_run_events: one insert for the whole batch. agent_run_id is + -- resolved via join (NULL for events with no stepId, e.g. lifecycle). + -- `with ordinality` + ORDER BY preserves the runner's emission order so + -- the auto-assigned identity values are stable for the cockpit's + -- chronological view. + insert into agent_run_events (workspace_id, workflow_run_id, agent_run_id, type, payload) + select + p_workspace_id, + p_run_id, + ar.id, + case + when ev.e->>'type' in ('lifecycle','agent-text','tool-call','tool-result','note','step-start','step-end') + then ev.e->>'type' + else 'note' + end, + coalesce(ev.e->'payload', '{}'::jsonb) + from jsonb_array_elements(p_events) with ordinality as ev(e, ord) + left join agent_runs ar + on ar.workflow_run_id = p_run_id + and ar.step_id = ev.e->>'stepId' + and ar.iteration = coalesce((ev.e->>'iteration')::int, 1) + order by ev.ord; + + -- 4. workflow_runs roll-up. Tokens go through `tokens_used = tokens_used + + -- delta` so concurrent batches can't lose increments. current_step_id + -- is updated to the last step touched in this batch. + select + coalesce(sum( (e->>'tokensUsed')::int ) filter (where (e->>'tokensUsed') is not null), 0), + ( + select e2->>'stepId' + from jsonb_array_elements(p_events) with ordinality as t2(e2, ord) + where e2->>'stepId' is not null + order by t2.ord desc + limit 1 + ) + into v_token_delta, v_last_step + from jsonb_array_elements(p_events) as e; + + if v_token_delta > 0 or v_last_step is not null then + update workflow_runs + set tokens_used = tokens_used + coalesce(v_token_delta, 0), + current_step_id = coalesce(v_last_step, current_step_id) + where id = p_run_id; + end if; +end; +$$; + +-- Runner API routes call these via the service-role key (bypasses RLS); the +-- function bodies are the trust boundary, not who may invoke them. Same +-- pattern as 0010_runner_api.sql. +grant execute on function runner_heartbeat(uuid, text) to anon, authenticated, service_role; +grant execute on function ingest_runner_events(uuid, uuid, jsonb) to anon, authenticated, service_role; From dd099b67a5f28b838c44f0a8ec77a826d503c04d Mon Sep 17 00:00:00 2001 From: Patryk Lewczuk Date: Sun, 24 May 2026 21:58:29 +0200 Subject: [PATCH 2/2] feat(flows): user-defined skill chains end-to-end MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lands the in-flight "flows" feature — a UI for declarative skill chains (name + ordered { skill, argsTemplate, stopChainIfContains } steps) that run through the existing workflow engine and surface in the cockpit. Includes: - migration 0021_flows: flows table + auto-triggers - core/workflows/flow-runner: turns a FlowRow into a Workflow; threads {{input}} / {{previousTaskId}} / {{previousPullRequest*}} into each step's user prompt; parses PR_URL= / PR_NUMBER= markers between steps - core/workflows: new unparsedCommentSection callback fires on the onNoParse success path so flow steps actually populate the living comment (fixes "(no sections yet)" / "Done." bug observed on the first end-to-end run on issue #2034) - gui/app/workflows: /workflows page + flow card editor + starter templates dropdown; the autofix template is decomposed into 5 focused steps (verify-in-repo → root-cause → fix → open-pr → review) so each step gets its own token budget instead of one mega-step blowing past the 2.5M cap - gui/app/issues: "Run flow for issue" modal entry point - webhook + dispatch + runner: route incoming flow runs through the engine - docs: open-mercato ephemeral integration guide for downstream verify Co-Authored-By: Claude Opus 4.7 (1M context) --- cezar-ephemeral-guide.md | 267 +++++ packages/core/src/index.ts | 27 + packages/core/src/workflows/flow-runner.ts | 342 ++++++ .../core/src/workflows/workflow-engine.ts | 4 + packages/core/src/workflows/workflow.ts | 15 +- .../gui/src/app/api/github/webhook/route.ts | 147 ++- packages/gui/src/app/api/runner/jobs/route.ts | 52 +- .../gui/src/app/issues/issue-row-menu.tsx | 16 + .../app/issues/run-flow-for-issue-modal.tsx | 192 +++ packages/gui/src/app/workflows/actions.ts | 511 ++++++++ packages/gui/src/app/workflows/flow-card.tsx | 1054 +++++++++++++++++ packages/gui/src/app/workflows/page.tsx | 22 + packages/gui/src/app/workflows/templates.ts | 144 +++ .../src/app/workflows/workflows-client.tsx | 320 +++++ packages/gui/src/components/sidebar.tsx | 2 + packages/gui/src/lib/execute-workflow-job.ts | 54 +- packages/gui/src/lib/persist-workflow-run.ts | 7 +- .../gui/src/lib/scheduler/run-dispatch.ts | 4 +- packages/gui/src/lib/supabase/types.ts | 25 +- .../gui/supabase/migrations/0021_flows.sql | 69 ++ packages/runner/src/execute-job-locally.ts | 36 +- packages/runner/src/runner-client.ts | 6 +- 22 files changed, 3285 insertions(+), 31 deletions(-) create mode 100644 cezar-ephemeral-guide.md create mode 100644 packages/core/src/workflows/flow-runner.ts create mode 100644 packages/gui/src/app/issues/run-flow-for-issue-modal.tsx create mode 100644 packages/gui/src/app/workflows/actions.ts create mode 100644 packages/gui/src/app/workflows/flow-card.tsx create mode 100644 packages/gui/src/app/workflows/page.tsx create mode 100644 packages/gui/src/app/workflows/templates.ts create mode 100644 packages/gui/src/app/workflows/workflows-client.tsx create mode 100644 packages/gui/supabase/migrations/0021_flows.sql diff --git a/cezar-ephemeral-guide.md b/cezar-ephemeral-guide.md new file mode 100644 index 0000000..309e4d6 --- /dev/null +++ b/cezar-ephemeral-guide.md @@ -0,0 +1,267 @@ +# Open-Mercato Ephemeral Environments — Integration Guide for Cezar + +Open-mercato ships two flavors of ephemeral runtime that fit cezar's "spin up an isolated instance per agent run, apply a fix, verify, tear down" loop. Pick by where you want the stack to live. + +## TL;DR — Which to pick + +| Use case | Runtime | How | +|---|---|---| +| Run a fix branch *inside* the open-mercato repo checkout, fast, with HMR, isolated DB | **Local dev ephemeral** | `yarn dev:ephemeral` | +| Test the agent's fix as an end-user against a built production-like image, throwaway, no host coupling | **Docker preview stack** | `yarn docker:ephemeral` | +| Verify the fix automatically with Playwright (recommended for cezar's autoverify step) | **Integration ephemeral** (built on top of the local dev flavor) | `yarn test:integration:ephemeral` | +| Want a Node-level API rather than CLI orchestration | **Programmatic** | Import from `@open-mercato/cli` → `packages/cli/src/lib/testing/integration.ts` | + +All three keep their state in well-known JSON files that cezar can read to discover `baseUrl`, port, and DB URL — so you do not need to parse stdout. + +--- + +## 1. Local dev ephemeral — `yarn dev:ephemeral` + +Implementation: `scripts/dev-ephemeral.ts` (~1200 lines). + +What it does, per invocation: +1. Provisions a throwaway Postgres via `docker run` (image `postgres:16`, container `open-mercato-dev-ephemeral--`, database `om_dev_ephemeral__`, random host port, `--rm`). +2. Runs `yarn initialize -- --reinstall` against that DB (migrations, generated artifacts, seed defaults). +3. Boots a Next.js dev server on a free port (default search range starts at 5000). +4. Waits for `GET /backend/login` to respond (180 s startup budget, 1 s poll, 1.5 s probe timeout). +5. Registers the running instance into `.ai/dev-ephemeral-envs.json`. + +### CLI flags + +- `--classic` — disable the splash UI, plain Next.js dev output +- `--verbose` — raw passthrough of all child-process logs + +### State file (always check this first) + +`.ai/dev-ephemeral-envs.json`: + +```json +{ + "version": 1, + "instances": [ + { + "id": ":", + "pid": 12345, + "port": 5000, + "baseUrl": "http://127.0.0.1:5000", + "backendUrl": "http://127.0.0.1:5000/backend", + "cwd": "/path/to/open-mercato", + "startedAt": "2026-05-23T10:00:00.000Z", + "postgresContainerId": "abc123…", + "postgresPort": 54321, + "databaseUrlRedacted": "postgres://***@127.0.0.1:54321/om_dev_ephemeral_..." + } + ] +} +``` + +Multiple instances are supported concurrently — each gets a unique port, container, and DB name. + +On startup the script also prunes stale entries (dead PIDs or unresponsive endpoints), so leaked rows from a previous SIGKILL self-heal on the next launch. + +### Environment variables + +| Var | Default | Effect | +|---|---|---| +| `DEV_EPHEMERAL_PREFERRED_PORT` | — | Try this app port first; fall back to free port | +| `DEV_EPHEMERAL_POSTGRES_IMAGE` | `postgres:16` | Override Postgres image | +| `DEV_EPHEMERAL_POSTGRES_USER` | `postgres` | DB user | +| `DEV_EPHEMERAL_POSTGRES_PASSWORD` | `postgres` | DB password | +| `OM_DEV_SPLASH_PORT` | `auto` | Splash UI port; `disabled`/`false`/`off` to skip | +| `OM_DEV_AUTO_OPEN` | unset | Set to `0` to suppress browser auto-open | +| `CI` | unset | When `true`, suppresses splash auto-open | +| `OM_DEV_CLASSIC` | unset | Same as `--classic` | +| `MERCATO_DEV_OUTPUT` | unset | Set `verbose` for verbose output | + +### Readiness signaling + +No dedicated `/api/health` endpoint exists. Two equivalent ready signals: + +1. **HTTP probe (recommended for cezar):** `GET http://127.0.0.1:/backend/login` returns any 2xx/3xx → ready. +2. **Splash status:** `GET http://127.0.0.1:/status` returns `{ ready, readyUrl, loginUrl, failed }`. + +### Lifecycle + +- **Start:** `yarn dev:ephemeral` (foreground, attaches stdio). +- **Stop:** send `SIGINT` or `SIGTERM` to the script PID. Both signals are forwarded; the Postgres container is removed in the shutdown handler. +- **No `yarn dev:ephemeral:down`** — there's nothing to call out-of-band. Cleanup is process-scoped. +- **Crash safety:** the Postgres container is started with `--rm`, so Docker auto-deletes it when the container exits even if the parent process is `SIGKILL`'d. The orphan row in the state file is cleaned by the next ephemeral startup's pruner. + +### Auth bootstrap (already done for you) + +`yarn initialize -- --reinstall` (called automatically inside the ephemeral flow) seeds default users: + +| Role | Email | Password | +|---|---|---| +| Superadmin | `superadmin@acme.com` (override via `OM_INIT_SUPERADMIN_EMAIL`) | `secret` | +| Admin | `admin@acme.com` | `secret` | +| Employee | `employee@acme.com` | `secret` | + +These come from `packages/core/src/helpers/integration/auth.ts:DEFAULT_CREDENTIALS`. No manual `auth:create-user` step required. + +--- + +## 2. Integration ephemeral — `yarn test:integration:ephemeral` + +This is the same dev-ephemeral runtime, wrapped by a richer harness that adds build caching, environment reuse, locking, and Playwright orchestration. **For cezar's auto-verify step this is usually the right entry point — it gives you the same instance the integration tests run against.** + +Implementation: `packages/cli/src/lib/testing/integration.ts` (~3200 lines). + +### CLI flavors + +| Command | Use | +|---|---| +| `yarn test:integration:ephemeral` | Spins up ephemeral env, runs the full Playwright suite, tears down | +| `yarn test:integration:ephemeral:start` | **Spins up ephemeral env and leaves it running** — what cezar should call when it wants a long-lived target to throw verification at | +| `yarn test:integration:ephemeral:interactive` | Same as `:start` but with a menu (human use) | +| `yarn test:integration:ephemeral:start:verbose` | Verbose variant of `:start` | + +### State file + +`.ai/qa/ephemeral-env.json` (different path from the dev flow above): + +```json +{ + "status": "running", + "baseUrl": "http://127.0.0.1:5001", + "port": 5001, + "databaseUrl": "postgresql://...", + "queueBaseDir": "/path/to/apps/mercato/.mercato/queue", + "source": "", + "captureScreenshots": false, + "startedAt": "2026-05-23T10:00:00.000Z" +} +``` + +- Default port: **5001** (falls back to a free port if busy). +- File is **cleared automatically** when the env stops, so its presence is a reliable "is one running?" check. +- Lock file: `.ai/qa/ephemeral-env.lock` (5 min timeout, 500 ms poll) — prevents race when two callers try to spawn at once. + +### Environment variables (integration flavor) + +| Var | Default | Effect | +|---|---|---| +| `OM_INTEGRATION_APP_READY_TIMEOUT_SECONDS` | `90` | App-ready probe timeout | +| `OM_INTEGRATION_BUILD_CACHE_TTL_SECONDS` | `600` | Reuse build artifacts within this window | +| `ENABLE_CRUD_API_CACHE` | unset | The npm scripts set this to `true` | +| `OM_INTEGRATION_TEST` | set by harness | Signals integration mode to the app | + +### Programmatic API (recommended for cezar's runner) + +`packages/cli/src/lib/testing/integration.ts` exports a Node-callable surface. **You can import this from `@open-mercato/cli` and orchestrate ephemerals without shelling out.** The relevant exports: + +```typescript +export type EphemeralEnvironmentHandle = { + baseUrl: string // e.g. "http://127.0.0.1:5001" + port: number + databaseUrl: string + commandEnvironment: NodeJS.ProcessEnv // env to inherit into your verification subprocess + ownedByCurrentProcess: boolean // true if you can call stop() + stop: () => Promise +} + +// Primary entry points +export async function startEphemeralEnvironment(opts: EphemeralRuntimeOptions): Promise +export async function tryReuseExistingEnvironment(opts: EphemeralRuntimeOptions): Promise + +// State helpers +export async function readEphemeralEnvironmentState(): Promise +export async function writeEphemeralEnvironmentState(input): Promise +export async function clearEphemeralEnvironmentState(): Promise + +// Coordination +export async function acquireEphemeralRuntimeLock(opts): Promise +export async function shouldReuseBuildArtifacts(ttlSeconds: number): Promise +``` + +`EphemeralRuntimeOptions`: + +```typescript +type EphemeralRuntimeOptions = { + verbose: boolean + captureScreenshots: boolean + logPrefix: string + forceRebuild?: boolean + reuseExisting?: boolean + requiredExistingSource?: string // pin to a specific source fingerprint + environmentOverrides?: NodeJS.ProcessEnv +} +``` + +### Reusable test helpers (for cezar's verify step) + +Published from `@open-mercato/core/helpers/integration/*` — usable from any Node process, including from cezar's own repo (it does not need to live inside open-mercato): + +| Import | What it gives you | +|---|---| +| `…/auth` | `login`, `DEFAULT_CREDENTIALS` | +| `…/api` | `getAuthToken`, `apiRequest` (token-authenticated API calls) | +| `…/crmFixtures` | `createCompanyFixture`, `createPersonFixture`, `deleteEntityIfExists`, `readJsonSafe` | +| `…/catalogFixtures`, `…/salesFixtures`, `…/authFixtures`, `…/dictionariesFixtures` | Domain-specific fixture lifecycle | +| `…/queue` | `drainIntegrationQueue(queueName)` — runs pending jobs deterministically | + +Barrel: `@open-mercato/core/testing/integration` re-exports the common subset. + +--- + +## 3. Docker preview stack — `yarn docker:ephemeral` + +For verifying a fix against a **built production-like image** without involving the host Node toolchain at all. + +Implementation: `docker/preview/Dockerfile` + `docker/preview/preview-entrypoint.sh` + `docker-compose.preview.yaml`. + +- Base image: `node:24-alpine`, multi-stage build. +- Entrypoint internally calls `yarn test:integration:ephemeral:start` — so what runs inside the container is the same integration ephemeral as flavor 2. +- Port mapping: host **5000** → container 5001. +- Fresh DB on **every** `up` (no persistent volume). +- Mounts `/var/run/docker.sock` so the in-container ephemeral can spawn its Postgres on the host's Docker. + +Commands: + +```bash +yarn docker:ephemeral # build + start, blocks attached to logs +yarn docker:ephemeral:down # full teardown +``` + +Discovery from the host: the app is unconditionally at `http://localhost:5000`. There is no state file written to the host filesystem for this flavor — port is fixed in the compose file. + +--- + +## Recommended integration shape for cezar + +Per agent run: + +1. **Reserve the runtime.** Call `tryReuseExistingEnvironment({ reuseExisting: true, requiredExistingSource: , … })`. If it returns a handle, you've got a warm env matching your source — use it. If null, fall through. +2. **Start fresh.** `startEphemeralEnvironment({ … })`. Acquire `acquireEphemeralRuntimeLock` first so two concurrent agents don't race on the same checkout. +3. **Apply the fix** in a worktree on the same checkout the ephemeral env was built from (the env auto-reloads via Turbopack on edits). +4. **Verify** by: + - Hitting `${handle.baseUrl}/backend/login` to confirm reachable. + - Logging in via `@open-mercato/core/helpers/integration/auth` → `login('admin')` or `getAuthToken('admin')`. + - Running module-scoped Playwright specs or your own targeted API calls. +5. **Stop** with `await handle.stop()` *if `handle.ownedByCurrentProcess === true`*. Otherwise leave the env running for the next agent run. + +For parallel agent runs on the **same host**, use flavor 1 (`yarn dev:ephemeral`) — every invocation gets its own port, DB, container, and entry in `.ai/dev-ephemeral-envs.json`. For agent runs on **separate hosts/CI runners**, flavor 3 (`yarn docker:ephemeral`) is cleaner because it has zero host coupling. + +--- + +## Gotchas to surface to the cezar team + +- **Node 24 required** by both flavors. The CLI bin asserts this and exits non-zero on older runtimes. +- **Docker daemon required** by all three flavors (yes, even local dev-ephemeral — Postgres runs in a container). +- **First-run cost.** Cold start is ~1–3 minutes (install + build + generate + DB init + dev boot). The integration flavor caches build artifacts for `OM_INTEGRATION_BUILD_CACHE_TTL_SECONDS` (default 600 s) so subsequent starts in the same checkout are much faster — set this higher for an agent fleet. +- **There is no `/api/health`.** Probe `/backend/login` or the splash `/status`. +- **No SIGKILL safety story for the state file** — the orphan row will sit in `.ai/dev-ephemeral-envs.json` until the next launch, which prunes it. Cezar workers that SIGKILL their own ephemeral should call `clearEphemeralEnvironmentState()` (integration flavor) or accept the next-launch self-heal (dev flavor). +- **Default seed users are static across runs.** If a fix depends on auth edge cases, cezar's verify step can create scoped users via `createUserFixture` from `@open-mercato/core/helpers/integration/authFixtures`. + +--- + +## Authoritative source files + +If the cezar team wants to read the implementations directly: + +- `scripts/dev-ephemeral.ts` — dev flavor entrypoint +- `packages/cli/src/lib/testing/integration.ts` — integration flavor, programmatic API, exported types +- `packages/cli/src/lib/testing/runtime-utils.ts` — port/host/probe helpers +- `docker/preview/Dockerfile`, `docker/preview/preview-entrypoint.sh`, `docker-compose.preview.yaml` — preview container +- `.ai/qa/AGENTS.md` — internal QA workflow doc (the closest existing prose to this guide) +- `packages/core/src/helpers/integration/*` — npm-published test helpers diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 3fa11bb..b771093 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -196,3 +196,30 @@ export { type CiFollowupBlackboard, type CiFollowupSeed, } from './workflows/definitions/ci-followup.workflow.js'; + +// Worktree helpers — exported so the GUI's flow runner can prep a worktree +// before calling `runWorkflow` (same as the autofix path does internally). +export { + createWorktree, + commitAll, + getDiffAgainstBase, + fetchRemoteBranch, + type WorktreeHandle, +} from './actions/autofix/worktree.js'; + +// Flow runner — runtime for the simple "flows" feature (workspace-defined +// chains of skill steps). Used both by the GUI dispatcher (cron-based) and +// the self-hosted runner daemon. +export { + runFlow, + extractPrMarkers, + renderTemplate, + effectiveStepNotes, + composeStepSystemPrompt, + FLOW_STEP_SCAFFOLDING, + DEFAULT_STEP_NOTES, + FLOW_STEP_SYSTEM_PROMPT, + type FlowStep, + type FlowRow, + type RunFlowParams, +} from './workflows/flow-runner.js'; diff --git a/packages/core/src/workflows/flow-runner.ts b/packages/core/src/workflows/flow-runner.ts new file mode 100644 index 0000000..fcdd569 --- /dev/null +++ b/packages/core/src/workflows/flow-runner.ts @@ -0,0 +1,342 @@ +import { z } from 'zod'; +import type { AgentEvent as RunnerAgentEvent } from '../agents/agent-runner.js'; +import type { Config } from '../config/config.model.js'; +import type { GitHubService } from '../services/github.service.js'; +import type { IssueStore } from '../store/store.js'; +import { discoverSkills } from '../skills/skill-catalog.js'; +import { createWorktree } from '../actions/autofix/worktree.js'; +import { + agentStep, + type AgentRunRecord, + type Workflow, + type WorkflowRunResult, + type WorkflowStep, +} from './workflow.js'; +import { runWorkflow } from './workflow-engine.js'; + +/** + * Runtime for the simple "flows" UI (migration 0021). A flow is a named chain + * of `{ skill, argsTemplate }` steps. This module turns one into a multi-step + * `Workflow` and runs it through the existing engine. + * + * - The skill body (from `/.ai/skills/.md`) is appended to the + * step's system prompt via `resolveStepConfig` (existing path). + * - The args template is rendered against a fixed scope: + * {{input}} — passed by the caller (typically the issue number) + * {{previousTaskId}} — the previous step's `agent_runs.id` + * {{previousPullRequestUrl}} — `PR_URL=…` parsed from the previous agent's text + * {{previousPullRequestNumber}} — `PR_NUMBER=…` parsed from the same + * - PR markers are a soft contract: skills that create PRs should emit + * `PR_URL=https://github.com/...` and `PR_NUMBER=42` in their final text + * so the next step can reference them. + */ + +export interface FlowStep { + skill: string; + argsTemplate: string; + /** + * When set, if the step's text output contains this substring the chain + * stops cleanly (status succeeded, reason "stopped at step N…"). Simple + * short-circuit for skills that legitimately decide "no action needed". + */ + stopChainIfContains?: string; + /** + * Extra system-prompt content for this step, prepended to the skill body. + * When unset (or empty/whitespace-only), `DEFAULT_STEP_NOTES` is used — + * the marker contract reminder so the chain works even if the skill body + * doesn't itself document `PR_URL=` / `PR_NUMBER=`. + */ + systemNotes?: string; +} + +export interface FlowRow { + name: string; + steps: FlowStep[]; +} + +export interface RunFlowParams { + flow: FlowRow; + input: string; + store: IssueStore; + config: Config; + github: GitHubService; + issueNumber: number; + onEvent: (msg: string) => void; + onAgentEvent: (e: { type: string; [k: string]: unknown }) => void; + onRunRecord: (r: AgentRunRecord) => void; + pauseRequested?: () => boolean | Promise; + cancelRequested?: () => boolean | Promise; +} + +interface FlowBlackboard extends Record { + input: string; + previousTaskId: string; + previousPullRequestUrl: string; + previousPullRequestNumber: string; +} + +export async function runFlow(params: RunFlowParams): Promise> { + if (params.flow.steps.length === 0) { + throw new Error(`flow '${params.flow.name}' has no steps`); + } + + // Discover skills so resolveStepConfig (called by the engine) can find each + // step's skill body and append it to the system prompt. + const repoRoot = params.config.autofix?.repoRoot; + const skills = repoRoot + ? await discoverSkills(repoRoot, params.config.autofix?.skillsDir ?? '.ai/skills').catch((err: Error) => { + params.onEvent(`[flow] skill discovery failed: ${err.message}`); + return []; + }) + : []; + + // Closure-captured state the engine updates between steps. We need this so + // each step's `buildUserPrompt` sees the freshest `previousTaskId` (which + // the engine doesn't expose through the step context directly). + let lastAgentRunId = ''; + + const workflow: Workflow = { + id: `flow:${params.flow.name}`, + title: params.flow.name, + commentTargetOrder: ['issue', 'pr'], + initialBlackboard: () => ({ + input: params.input, + previousTaskId: '', + previousPullRequestUrl: '', + previousPullRequestNumber: '', + }), + steps: params.flow.steps.map((step, idx) => makeAgentStepForFlowStep(step, idx, params.config)), + }; + + return runWorkflow(workflow, { + store: params.store, + config: params.config, + github: params.github, + issueNumber: params.issueNumber, + apply: true, + skills, + // Bindings: per-step, set the skillName so resolveStepConfig appends the + // skill body to the system prompt. The engine reads these from `bindings`. + bindings: params.flow.steps.map((step, idx) => ({ + stepId: stepIdFor(idx), + skillName: step.skill, + backend: null, + model: null, + extraTools: [], + })), + prepareWorktree: repoRoot + ? async () => { + const wt = await createWorktree({ + repoRoot, + branch: (params.config.autofix?.branchPrefix ?? 'autofix/cezar-issue-') + params.issueNumber, + baseBranch: params.config.autofix?.baseBranch ?? 'main', + remote: params.config.autofix?.remote ?? 'origin', + fetchRemote: params.config.autofix?.fetchBeforeAttempt ?? true, + onWarn: (m: string) => params.onEvent(m), + }); + return { path: wt.path, dispose: wt.dispose }; + } + : undefined, + onEvent: params.onEvent, + onAgentEvent: (e: RunnerAgentEvent) => params.onAgentEvent(e as unknown as { type: string; [k: string]: unknown }), + onRunRecord: (r: AgentRunRecord) => { + lastAgentRunId = r.id; + params.onRunRecord(r); + }, + pauseRequested: params.pauseRequested, + cancelRequested: params.cancelRequested, + }); + + // ─── Helpers ────────────────────────────────────────────────────────────── + + function stepIdFor(idx: number): string { + return `step-${idx + 1}`; + } + + function makeAgentStepForFlowStep( + step: FlowStep, + idx: number, + cfg: Config, + ): WorkflowStep { + const autofix = cfg.autofix; + const builtinTools = autofix?.allowedTools ?? ['Read', 'Edit', 'Write', 'Grep', 'Glob', 'Bash']; + const bashAllowlist = autofix?.bashAllowlist; + const maxTurns = autofix?.maxTurns?.fixer ?? 30; + const model = autofix?.models?.fixer ?? 'claude-sonnet-4-6'; + + return agentStep({ + id: stepIdFor(idx), + kind: 'agent', + builtinSkillId: step.skill, + builtinSystemPrompt: composeStepSystemPrompt(step), + builtinModel: model, + builtinTools, + bashAllowlist, + maxTurns, + cwdRequired: true, + // Free-form output. We don't enforce JSON; we parse PR_URL/PR_NUMBER + // markers from the text in onNoParse below. + responseSchema: z.never() as unknown as z.ZodSchema, + buildUserPrompt: (ctx) => { + // Refresh `previousTaskId` from the closure right before render — it's + // updated by `onRunRecord` between steps. + const scope: FlowBlackboard = { + ...ctx.blackboard, + previousTaskId: lastAgentRunId, + }; + return renderTemplate(step.argsTemplate, scope); + }, + // If the agent emitted JSON that happened to validate, we still want + // to chain — but `z.never()` ensures this branch never fires. + onResult: () => ({ kind: 'continue' as const }), + onNoParse: (rawText, _ctx) => { + const { url, number } = extractPrMarkers(rawText); + const patch: Partial = { + previousPullRequestUrl: url ?? '', + previousPullRequestNumber: number != null ? String(number) : '', + }; + // Stop-chain: cleanly end the workflow when the agent's output + // contains the configured marker. We rely on `skip-run` rather than + // a hard fail so the cockpit shows it as `succeeded (with reason)`. + if (step.stopChainIfContains && rawText.includes(step.stopChainIfContains)) { + return { + kind: 'skip-run', + reason: `stopped at step ${idx + 1}: output matched "${step.stopChainIfContains}"`, + }; + } + return { kind: 'continue', blackboardPatch: patch }; + }, + // Render the agent's actual output into the living comment. The flow + // runtime never enforces a JSON schema, so the engine takes the + // `unparsedCommentSection` path (not `commentSection`). + unparsedCommentSection: (rawText) => ({ + heading: flowStepHeading(idx, step, rawText), + body: flowStepBody(rawText), + }), + failCommentSection: (reason) => ({ + heading: `Step ${idx + 1} — \`${step.skill}\` (failed)`, + body: reason, + }), + }); + } +} + +// ─── Living-comment rendering helpers ────────────────────────────────────── + +const STOP_MARKER_RE = /^NO_ACTION_NEEDED\b/m; +const MAX_BODY_CHARS = 1500; + +function flowStepHeading(idx: number, step: FlowStep, rawText: string): string { + const base = `Step ${idx + 1} — \`${step.skill}\``; + if (step.stopChainIfContains && rawText.includes(step.stopChainIfContains)) { + return `${base} (stopped: \`${step.stopChainIfContains}\`)`; + } + if (STOP_MARKER_RE.test(rawText)) return `${base} (stopped: \`NO_ACTION_NEEDED\`)`; + const { url, number } = extractPrMarkers(rawText); + if (url && number != null) return `${base} → PR [#${number}](${url})`; + return base; +} + +/** + * Render the most useful tail of the agent's output. Most skills end with a + * structured summary block (Status/Files changed/Summary/…); the last ~1500 + * chars reliably capture that without flooding the GitHub comment with full + * tool-call transcripts. + */ +function flowStepBody(rawText: string): string { + const trimmed = (rawText ?? '').trim(); + if (!trimmed) return '_(no output)_'; + if (trimmed.length <= MAX_BODY_CHARS) return trimmed; + const tail = trimmed.slice(-MAX_BODY_CHARS).trimStart(); + return `_… (truncated; showing last ${MAX_BODY_CHARS} chars)_\n\n${tail}`; +} + +// ─── PR marker extraction ────────────────────────────────────────────────── + +/** + * Extracts `PR_URL=...` and `PR_NUMBER=...` markers from a step's text output. + * Soft contract: skills that create PRs should print these lines (anywhere in + * the response) so the next step can reference them via `{{previousPullRequest*}}`. + * Also tries common URL patterns as a fallback. + */ +export function extractPrMarkers(text: string): { url: string | null; number: number | null } { + let url: string | null = null; + let number: number | null = null; + + const urlMatch = text.match(/PR_URL\s*=\s*(\S+)/); + if (urlMatch) url = urlMatch[1]; + const numMatch = text.match(/PR_NUMBER\s*=\s*(\d+)/); + if (numMatch) number = Number(numMatch[1]); + + // Fallback: scrape any GitHub PR URL. + if (!url) { + const fall = text.match(/https?:\/\/[\w./-]*github\.com\/[\w-]+\/[\w.-]+\/pull\/(\d+)/); + if (fall) { + url = fall[0]; + if (number == null) number = Number(fall[1]); + } + } + return { url, number }; +} + +// ─── Template rendering ──────────────────────────────────────────────────── + +const TEMPLATE_RE = /\{\{\s*([\w.]+)\s*\}\}/g; + +/** Render `{{path}}` references against a flat scope. Missing → empty string. */ +export function renderTemplate(tpl: string, scope: Record): string { + return tpl.replace(TEMPLATE_RE, (_, path: string) => { + const v = lookupDotPath(scope, path); + if (v == null) return ''; + if (typeof v === 'string') return v; + if (typeof v === 'number' || typeof v === 'boolean') return String(v); + return JSON.stringify(v); + }); +} + +function lookupDotPath(obj: unknown, path: string): unknown { + const parts = path.split('.'); + let cur: unknown = obj; + for (const part of parts) { + if (cur == null || typeof cur !== 'object') return undefined; + cur = (cur as Record)[part]; + } + return cur; +} + +// ─── Per-step scaffolding system prompt ──────────────────────────────────── + +/** + * Minimal scaffolding shipped to every flow step. Generic — describes that + * this is one step of a workflow and that the skill body holds the actual + * instructions. The marker contract lives in `DEFAULT_STEP_NOTES` so users + * can edit/override it per step (see `step.systemNotes`). + */ +export const FLOW_STEP_SCAFFOLDING = `You are an agent executing one step of a user-defined workflow. Follow the SKILL instructions appended below to complete the task described in the user message. + +When the task is done, respond with a short plain-text summary (no JSON).`; + +/** + * Default notes prepended to the skill body when `step.systemNotes` is unset. + * Documents the marker contract that lets the next step reference the PR via + * `{{previousPullRequestUrl}}` / `{{previousPullRequestNumber}}` templates. + * Users override this per step via the pencil/notes panel in the editor. + */ +export const DEFAULT_STEP_NOTES = `If you open a pull request, end your response with two lines on separate lines: + PR_URL= + PR_NUMBER= +so the next step in the workflow can reference it via {{previousPullRequestUrl}} / {{previousPullRequestNumber}}.`; + +/** Back-compat alias — kept temporarily for any out-of-tree code. */ +export const FLOW_STEP_SYSTEM_PROMPT = FLOW_STEP_SCAFFOLDING + '\n\n' + DEFAULT_STEP_NOTES; + +/** What the scaffolding+notes actually compose to at run time. */ +export function effectiveStepNotes(step: FlowStep): string { + const trimmed = (step.systemNotes ?? '').trim(); + return trimmed.length > 0 ? step.systemNotes! : DEFAULT_STEP_NOTES; +} + +/** The system prompt the engine actually sends (before the skill body is appended). */ +export function composeStepSystemPrompt(step: FlowStep): string { + return `${FLOW_STEP_SCAFFOLDING}\n\n${effectiveStepNotes(step)}`; +} diff --git a/packages/core/src/workflows/workflow-engine.ts b/packages/core/src/workflows/workflow-engine.ts index f2f3c80..3619d1d 100644 --- a/packages/core/src/workflows/workflow-engine.ts +++ b/packages/core/src/workflows/workflow-engine.ts @@ -777,6 +777,10 @@ export class WorkflowEngine { if (outcome.kind === 'fail') record.error = outcome.reason; if (step.failCommentSection && outcome.kind === 'fail') { await args.living.appendSection(step.id, step.failCommentSection(outcome.reason, stepCtx)); + } else if (step.unparsedCommentSection && outcome.kind !== 'fail') { + // Flows runtime path: agent emitted free text and onNoParse said go. + // Surface what the agent actually produced in the living comment. + await args.living.appendSection(step.id, step.unparsedCommentSection(result.text, stepCtx)); } return { outcome, record }; } diff --git a/packages/core/src/workflows/workflow.ts b/packages/core/src/workflows/workflow.ts index 8ee329f..81b00b2 100644 --- a/packages/core/src/workflows/workflow.ts +++ b/packages/core/src/workflows/workflow.ts @@ -137,6 +137,13 @@ export interface AgentStepDef extends BaseStepDef { onNoParse?: (rawText: string, ctx: WorkflowStepContext) => StepOutcome; /** Render this step's section of the living comment from the parsed output. */ commentSection?: (parsed: T, ctx: WorkflowStepContext) => CommentSection; + /** + * Render this step's section when the agent emitted free text (no JSON parse) + * and `onNoParse` returned a success outcome (`continue` / `skip-run`). Used + * by the flows runtime, whose steps don't enforce a response schema. Built-in + * workflow steps that always produce parseable JSON can ignore this. + */ + unparsedCommentSection?: (rawText: string, ctx: WorkflowStepContext) => CommentSection; /** Render a section when the step failed (no parse / outcome=fail). */ failCommentSection?: (reason: string, ctx: WorkflowStepContext) => CommentSection; /** True ⇒ the engine refuses to run this step without a `worktreePath`. */ @@ -245,7 +252,13 @@ export interface WorkflowLoop { } export interface Workflow { - id: 'triage' | 'autofix' | 'ci-followup'; + /** + * Workflow identifier. The three built-ins are `'triage' | 'autofix' | 'ci-followup'`; + * custom workflows compiled from `WorkflowDescriptor` (see `./compiler.ts`) carry + * their own string ids. The engine treats this as opaque except for the autofix + * unified-session gate, which compares against the literal `'autofix'`. + */ + id: string; title: string; /** Pre-PR steps edit the issue comment; once a PR exists, post-PR steps edit a PR comment (docs §3.6). */ commentTargetOrder: CommentTarget[]; diff --git a/packages/gui/src/app/api/github/webhook/route.ts b/packages/gui/src/app/api/github/webhook/route.ts index dd54ff2..40f16fc 100644 --- a/packages/gui/src/app/api/github/webhook/route.ts +++ b/packages/gui/src/app/api/github/webhook/route.ts @@ -100,6 +100,8 @@ interface WebhookPayload { changes?: { title?: unknown; body?: unknown }; issue?: WebhookIssue; pull_request?: WebhookPullRequest; + /** Present on `issues.labeled` — the label that was just added. */ + label?: { name?: string }; repository?: { name: string; owner: { login: string } }; installation?: { id: number }; check_run?: { @@ -133,10 +135,13 @@ const TRIAGE_ACTIONS = new Set(['opened', 'reopened']); async function handleIssues(admin: SupabaseAdmin, payload: WebhookPayload): Promise { const action = payload.action ?? ''; - const relevant = + const isTriageRelevant = TRIAGE_ACTIONS.has(action) || (action === 'edited' && !!payload.changes && ('title' in payload.changes || 'body' in payload.changes)); - if (!relevant) return NextResponse.json({ ok: true, ignored: `issues.${action}` }); + const isFlowRelevant = action === 'opened' || action === 'labeled'; + if (!isTriageRelevant && !isFlowRelevant) { + return NextResponse.json({ ok: true, ignored: `issues.${action}` }); + } const issue = payload.issue; const repo = payload.repository; @@ -146,43 +151,147 @@ async function handleIssues(admin: SupabaseAdmin, payload: WebhookPayload): Prom if (workspaces.length === 0) return NextResponse.json({ ok: true, ignored: 'no matching workspace' }); const repoSlug = `${repo.owner.login}/${repo.name}`; - let enqueued = 0; + // GitHub puts the just-added label on `payload.label` for issues.labeled. + const justLabeled = + action === 'labeled' && payload.label && typeof payload.label.name === 'string' + ? payload.label.name + : null; + + let triageEnqueued = 0; + let flowsEnqueued = 0; for (const ws of workspaces) { - if (!ws.auto_triage_enabled) continue; + // Upsert the issue once per workspace either way so the issues table stays + // current even for flow-only triggers. try { await upsertIssueFromWebhook(admin, ws.id, issue); } catch (err) { console.error(`[github-webhook] issue upsert failed for ws ${ws.id}:`, err instanceof Error ? err.message : err); continue; } - // dedupe: skip if a triage job for this issue is already in flight. + + // Triage path (existing behavior). + if (isTriageRelevant && ws.auto_triage_enabled) { + const { data: open } = await admin + .from('jobs') + .select('id') + .eq('workspace_id', ws.id) + .eq('kind', 'triage') + .eq('issue_number', issue.number) + .in('status', ['queued', 'claimed', 'running']) + .limit(1); + if (!open || open.length === 0) { + const { error } = await admin.from('jobs').insert({ + workspace_id: ws.id, + repo: repoSlug, + kind: 'triage', + issue_number: issue.number, + pr_number: null, + priority: 5, + status: 'queued', + max_attempts: 1, + payload: { trigger: 'webhook', action }, + }); + if (error) { + console.error(`[github-webhook] triage enqueue failed for ws ${ws.id}:`, error.message); + } else { + triageEnqueued++; + } + } + } + + // Flow auto-trigger path. Match flows whose `triggers` include + // `issue.opened` (on opened) or `issue.labeled` with the right label. + if (isFlowRelevant) { + flowsEnqueued += await enqueueFlowsForIssueEvent(admin, { + workspaceId: ws.id, + repoSlug, + issueNumber: issue.number, + action, + labelName: justLabeled, + }); + } + } + return NextResponse.json({ + ok: true, + triageEnqueued, + flowsEnqueued, + workspaces: workspaces.length, + }); +} + +/** + * For a `issues.opened` / `issues.labeled` event, find flows in `workspaceId` + * whose `triggers` match and enqueue one `kind='flow'` job per match. Skips + * flows that already have an in-flight job for this issue (dedupe). + */ +async function enqueueFlowsForIssueEvent( + admin: SupabaseAdmin, + args: { + workspaceId: string; + repoSlug: string; + issueNumber: number; + action: string; + labelName: string | null; + }, +): Promise { + const { data: flows } = await admin + .from('flows') + .select('id, name, triggers, paused') + .eq('workspace_id', args.workspaceId) + .eq('paused', false); + if (!flows || flows.length === 0) return 0; + + let enqueued = 0; + for (const flow of flows) { + const triggers = Array.isArray(flow.triggers) ? (flow.triggers as Array>) : []; + const matches = triggers.some((t) => { + if (!t || typeof t !== 'object') return false; + if (args.action === 'opened' && t.kind === 'issue.opened') return true; + if (args.action === 'labeled' && t.kind === 'issue.labeled') { + return typeof t.label === 'string' && t.label === args.labelName; + } + return false; + }); + if (!matches) continue; + + // Dedupe: don't double-enqueue a flow for the same issue while one's + // already in flight (e.g. label added, then issue edited later). const { data: open } = await admin .from('jobs') - .select('id') - .eq('workspace_id', ws.id) - .eq('kind', 'triage') - .eq('issue_number', issue.number) - .in('status', ['queued', 'claimed', 'running']) - .limit(1); - if (open && open.length > 0) continue; + .select('id, payload') + .eq('workspace_id', args.workspaceId) + .eq('kind', 'flow') + .eq('issue_number', args.issueNumber) + .in('status', ['queued', 'claimed', 'running']); + const alreadyForThisFlow = (open ?? []).some((row) => { + const p = (row.payload ?? {}) as { flowId?: string }; + return p.flowId === flow.id; + }); + if (alreadyForThisFlow) continue; + const { error } = await admin.from('jobs').insert({ - workspace_id: ws.id, - repo: repoSlug, - kind: 'triage', - issue_number: issue.number, + workspace_id: args.workspaceId, + repo: args.repoSlug, + kind: 'flow', + issue_number: args.issueNumber, pr_number: null, priority: 5, status: 'queued', max_attempts: 1, - payload: { trigger: 'webhook', action }, + payload: { + trigger: 'webhook', + action: args.action, + flowId: flow.id, + flowInput: String(args.issueNumber), + }, }); if (error) { - console.error(`[github-webhook] triage enqueue failed for ws ${ws.id}:`, error.message); + console.error(`[github-webhook] flow '${flow.name}' enqueue failed:`, error.message); continue; } enqueued++; } - return NextResponse.json({ ok: true, enqueued, workspaces: workspaces.length }); + return enqueued; } // ─── check_run ────────────────────────────────────────────────────────────── diff --git a/packages/gui/src/app/api/runner/jobs/route.ts b/packages/gui/src/app/api/runner/jobs/route.ts index 51cccec..e09255f 100644 --- a/packages/gui/src/app/api/runner/jobs/route.ts +++ b/packages/gui/src/app/api/runner/jobs/route.ts @@ -83,6 +83,33 @@ export async function GET(req: Request) { ? ((job.payload as { ciFollowup?: { issueNumber?: number } })?.ciFollowup?.issueNumber ?? job.issue_number) : job.issue_number; + // ── flow lookup ── For `kind='flow'` jobs the runner needs the flow's + // name + steps so it can build a Workflow and call runFlow. Load the + // row by `payload.flowId` here so the runner doesn't need DB access. + // `workflow_runs.workflow` becomes `flow:` so the cockpit labels + // runner-driven flow runs the same way cron-driven ones look. + type FlowPayload = { flowId?: string; flowInput?: string }; + let flowOut: { name: string; steps: import('@cezar/core').FlowStep[]; input: string } | null = null; + let workflowLabel: string = job.kind; + if (job.kind === 'flow') { + const payload = (job.payload ?? {}) as FlowPayload; + if (!payload.flowId) throw new Error('flow job missing payload.flowId'); + const { data: flowRow, error: fErr } = await admin + .from('flows') + .select('name, steps, paused') + .eq('id', payload.flowId) + .eq('workspace_id', job.workspace_id) + .single(); + if (fErr || !flowRow) throw new Error(`flow ${payload.flowId} not found: ${fErr?.message ?? 'no row'}`); + const parsedSteps = parseFlowSteps(flowRow.steps); + flowOut = { + name: flowRow.name, + steps: parsedSteps, + input: payload.flowInput ?? (runIssueNumber != null ? String(runIssueNumber) : ''), + }; + workflowLabel = `flow:${flowRow.name}`; + } + // ── workflow_runs row ── const repoSlug = owner && repo ? `${owner}/${repo}` : job.repo; const { data: runRow, error: runErr } = await admin @@ -90,7 +117,7 @@ export async function GET(req: Request) { .insert({ workspace_id: job.workspace_id, job_id: job.id, - workflow: job.kind, + workflow: workflowLabel, repo: repoSlug, issue_number: runIssueNumber ?? null, pr_number: job.pr_number ?? null, @@ -125,6 +152,7 @@ export async function GET(req: Request) { githubToken, store: storeSnapshot, ciFollowupSeed, + flow: flowOut, }); } catch (err) { await releaseJob().catch(() => {}); @@ -134,6 +162,28 @@ export async function GET(req: Request) { } } +/** + * Defensive parser for the `flows.steps` JSONB column — keeps the runner-API + * contract clean even if a row was inserted by an older client. Mirrors the + * GUI-side `parseSteps` in `app/workflows/actions.ts`. + */ +function parseFlowSteps(raw: unknown): import('@cezar/core').FlowStep[] { + if (!Array.isArray(raw)) return []; + return raw + .map((s) => { + if (!s || typeof s !== 'object') return null; + const obj = s as Record; + const step: import('@cezar/core').FlowStep = { + skill: typeof obj.skill === 'string' ? obj.skill : '', + argsTemplate: typeof obj.argsTemplate === 'string' ? obj.argsTemplate : '', + }; + if (typeof obj.stopChainIfContains === 'string' && obj.stopChainIfContains) step.stopChainIfContains = obj.stopChainIfContains; + if (typeof obj.systemNotes === 'string' && obj.systemNotes) step.systemNotes = obj.systemNotes; + return step; + }) + .filter((x): x is import('@cezar/core').FlowStep => x !== null); +} + /** Mirrors the per-workspace token lookup in the crons (`execute-workflow-job.ts`). */ async function resolveWorkspaceToken( workspaceId: string, diff --git a/packages/gui/src/app/issues/issue-row-menu.tsx b/packages/gui/src/app/issues/issue-row-menu.tsx index 92fe6bb..a094101 100644 --- a/packages/gui/src/app/issues/issue-row-menu.tsx +++ b/packages/gui/src/app/issues/issue-row-menu.tsx @@ -6,6 +6,7 @@ import { MoreVerticalIcon } from '@/components/icons'; import { RowMenuPortal } from '@/components/row-menu-portal'; import { startAutofix } from './autofix-actions'; import { RunActionForIssueModal } from './run-action-for-issue-modal'; +import { RunFlowForIssueModal } from './run-flow-for-issue-modal'; export interface IssueRowMenuProps { issueNumber: number; @@ -35,6 +36,7 @@ export function IssueRowMenu({ }: IssueRowMenuProps) { const [open, setOpen] = useState(false); const [runActionOpen, setRunActionOpen] = useState(false); + const [runFlowOpen, setRunFlowOpen] = useState(false); const [pending, startTransition] = useTransition(); const triggerRef = useRef(null); const popoverRef = useRef(null); @@ -99,6 +101,13 @@ export function IssueRowMenu({ disabled: readOnly, group: 1, }, + { + id: 'run-flow', + label: 'Run flow…', + onSelect: () => setRunFlowOpen(true), + disabled: readOnly, + group: 1, + }, { id: 'open', label: 'Open on GitHub', @@ -208,6 +217,13 @@ export function IssueRowMenu({ onClose={() => setRunActionOpen(false)} /> )} + {runFlowOpen && ( + setRunFlowOpen(false)} + /> + )} ); } diff --git a/packages/gui/src/app/issues/run-flow-for-issue-modal.tsx b/packages/gui/src/app/issues/run-flow-for-issue-modal.tsx new file mode 100644 index 0000000..535cd2d --- /dev/null +++ b/packages/gui/src/app/issues/run-flow-for-issue-modal.tsx @@ -0,0 +1,192 @@ +'use client'; + +import { useEffect, useRef, useState, useTransition } from 'react'; +import { useRouter } from 'next/navigation'; +import { cn } from '@/components/ui/cn'; +import { listFlows, runFlowOnIssue, type FlowSummary } from '@/app/workflows/actions'; + +export interface RunFlowForIssueModalProps { + issueNumber: number; + issueTitle: string; + onClose: () => void; +} + +/** + * Companion to `RunActionForIssueModal` but for flows (migration 0021). The + * user picks a saved flow and we queue it as a `kind='flow'` job pinned to + * this issue. On success we redirect to /cockpit so the user sees the run + * appear — matching `startAutofix`'s "redirect to /cockpit" pattern. + */ +export function RunFlowForIssueModal({ issueNumber, issueTitle, onClose }: RunFlowForIssueModalProps) { + const router = useRouter(); + const dialogRef = useRef(null); + const lastFocused = useRef(null); + const [flows, setFlows] = useState([]); + const [loading, setLoading] = useState(true); + const [selectedId, setSelectedId] = useState(''); + const [error, setError] = useState(null); + const [pending, startTransition] = useTransition(); + + useEffect(() => { + lastFocused.current = document.activeElement as HTMLElement | null; + const node = dialogRef.current; + const focusables = node?.querySelectorAll( + 'button, [href], input, select, textarea, [tabindex]:not([tabindex="-1"])', + ); + focusables?.[0]?.focus(); + + function onKey(e: KeyboardEvent) { + if (e.key === 'Escape') { + e.stopPropagation(); + onClose(); + return; + } + if (e.key !== 'Tab') return; + const items = Array.from( + node?.querySelectorAll( + 'button:not([disabled]), [href], input:not([disabled]), select:not([disabled]), textarea:not([disabled]), [tabindex]:not([tabindex="-1"])', + ) ?? [], + ); + if (items.length === 0) return; + const first = items[0]; + const last = items[items.length - 1]; + if (e.shiftKey && document.activeElement === first) { + e.preventDefault(); + last.focus(); + } else if (!e.shiftKey && document.activeElement === last) { + e.preventDefault(); + first.focus(); + } + } + document.addEventListener('keydown', onKey); + return () => { + document.removeEventListener('keydown', onKey); + lastFocused.current?.focus?.(); + }; + }, [onClose]); + + useEffect(() => { + let cancelled = false; + (async () => { + const rows = await listFlows(); + if (cancelled) return; + // Hide paused flows + ones with no steps — can't run them anyway. + const runnable = rows.filter((f) => !f.paused && f.steps.length > 0); + setFlows(runnable); + setLoading(false); + if (runnable.length > 0) setSelectedId(runnable[0].id); + })().catch(() => setLoading(false)); + return () => { + cancelled = true; + }; + }, []); + + function handleRun() { + setError(null); + if (!selectedId) { + setError('Pick a flow'); + return; + } + startTransition(async () => { + const r = await runFlowOnIssue({ flowId: selectedId, issueNumber }); + if (!r.ok) { + setError(r.error ?? 'Run failed'); + return; + } + onClose(); + router.push('/cockpit'); + }); + } + + const selected = flows.find((f) => f.id === selectedId); + + return ( +
{ + if (e.target === e.currentTarget) onClose(); + }} + className="fixed inset-0 z-50 flex items-center justify-center bg-black/50 p-4" + > +
+

+ Run flow on #{issueNumber} +

+

{issueTitle}

+ +
+ + {loading ? ( +

Loading flows…

+ ) : flows.length === 0 ? ( +

+ No runnable flows. Visit{' '} + /workflows{' '} + to create one. +

+ ) : ( + + )} + {selected && selected.steps.length > 0 && ( +
+
Steps
+
    + {selected.steps.map((s, i) => ( +
  1. + {s.skill || '(no skill)'}{' '} + → {s.argsTemplate} +
  2. + ))} +
+
+ )} +
+ + {error && ( +
+ {error} +
+ )} + +
+ + +
+
+
+ ); +} diff --git a/packages/gui/src/app/workflows/actions.ts b/packages/gui/src/app/workflows/actions.ts new file mode 100644 index 0000000..fea69aa --- /dev/null +++ b/packages/gui/src/app/workflows/actions.ts @@ -0,0 +1,511 @@ +'use server'; + +import { revalidatePath } from 'next/cache'; +import { + discoverSkills, + renderTemplate, + FLOW_STEP_SCAFFOLDING, + DEFAULT_STEP_NOTES, + effectiveStepNotes, +} from '@cezar/core'; +import { createSupabaseAdminClient } from '@/lib/supabase/server'; +import { getActiveWorkspace } from '@/lib/workspace'; +import { loadWorkspaceConfig } from '@/lib/load-workspace-config'; +import { ensureRepoClone } from '@/lib/repo-clone'; +import type { Database } from '@/lib/supabase/types'; + +/** + * Server actions backing the `/workflows` page — CRUD for the simple `flows` + * table (migration 0021) + the auto-triggers (migration 0022) + the + * "Run on issue #" entry point + a server-side prompt preview helper. + */ + +export type ActionResult = (T & { ok: true }) | { ok: false; error: string }; + +export interface FlowStep { + skill: string; + argsTemplate: string; + /** + * When set, after the step runs the engine checks if its text output + * contains this substring; if it does, the chain stops cleanly with status + * `succeeded` and reason "stopped at step N". A simple short-circuit for + * skills that legitimately decide "no action needed". + */ + stopChainIfContains?: string; + /** + * Extra system-prompt content prepended to the skill body. When unset + * (or whitespace-only), a built-in default explaining the `PR_URL=` / + * `PR_NUMBER=` marker contract is used. Set to override per step. + */ + systemNotes?: string; +} + +export type FlowTrigger = + | { kind: 'issue.opened' } + | { kind: 'issue.labeled'; label: string }; + +export interface SkillSummary { + name: string; + /** First line of the skill's frontmatter `description` field; empty when none. */ + description: string; +} + +export interface FlowStepOutcome { + stepId: string; + status: string; + iteration: number; + /** Best-effort: error, summary, or tail of the agent's text output. */ + output: string; +} + +export interface FlowRecentRun { + id: string; + status: string; + issueNumber: number | null; + startedAt: string; +} + +export interface FlowStats { + /** Total runs in the last 7 days. */ + totalLast7d: number; + /** Runs whose status is 'succeeded' in the last 7 days. */ + succeededLast7d: number; + /** Average cost-weighted tokens per run over the last 7 days (0 when no runs). */ + avgTokens: number; +} + +export interface FlowSummary { + id: string; + name: string; + steps: FlowStep[]; + triggers: FlowTrigger[]; + paused: boolean; + updatedAt: string; + /** Most recent workflow_runs row whose `workflow` matches `flow:` — with per-step output snippets. */ + lastRun?: { + id: string; + status: string; + issueNumber: number | null; + startedAt: string; + finishedAt: string | null; + reason: string | null; + steps: FlowStepOutcome[]; + }; + /** Lightweight pills for the at-a-glance history strip (up to 5, newest first). */ + recentRuns: FlowRecentRun[]; + /** 7-day stats — success rate + avg tokens. */ + stats: FlowStats; +} + +// ─── Read ──────────────────────────────────────────────────────────────────── + +export async function listFlows(): Promise { + const workspace = await getActiveWorkspace(); + if (!workspace) return []; + const supabase = createSupabaseAdminClient(); + + const { data: rows, error } = await supabase + .from('flows') + .select('id, name, steps, triggers, paused, updated_at') + .eq('workspace_id', workspace.id) + .order('updated_at', { ascending: false }); + if (error || !rows) return []; + + const workflowKeys = rows.map((r) => `flow:${r.name}`); + const lastRuns = new Map(); + const recentByWorkflow = new Map(); + const statsByWorkflow = new Map(); + const sevenDaysAgo = new Date(Date.now() - 7 * 24 * 60 * 60 * 1000).toISOString(); + if (workflowKeys.length > 0) { + const { data: runs } = await supabase + .from('workflow_runs') + .select('id, workflow, status, issue_number, started_at, finished_at, reason, tokens_used') + .eq('workspace_id', workspace.id) + .in('workflow', workflowKeys) + .order('started_at', { ascending: false }); + if (runs) { + // Group runs by workflow (descending), and tally stats over the 7-day window. + const seenLatest = new Set(); + const latestRunIds: string[] = []; // for the per-step output query (only the most-recent per workflow) + for (const run of runs) { + const key = run.workflow; + const list = recentByWorkflow.get(key) ?? []; + if (list.length < 5) list.push({ + id: run.id, + status: run.status, + issueNumber: run.issue_number, + startedAt: run.started_at, + }); + recentByWorkflow.set(key, list); + if (!seenLatest.has(key)) { + seenLatest.add(key); + latestRunIds.push(run.id); + } + if (run.started_at >= sevenDaysAgo) { + const cur = statsByWorkflow.get(key) ?? { totalLast7d: 0, succeededLast7d: 0, avgTokens: 0 }; + cur.totalLast7d += 1; + if (run.status === 'succeeded') cur.succeededLast7d += 1; + // running average — we'll finalize by dividing once at the end. + cur.avgTokens += run.tokens_used ?? 0; + statsByWorkflow.set(key, cur); + } + } + for (const [, s] of statsByWorkflow) { + s.avgTokens = s.totalLast7d > 0 ? Math.round(s.avgTokens / s.totalLast7d) : 0; + } + + // Per-step output snippet — only fetched for the most-recent run per + // workflow (everything else is "open in cockpit"). Keeps this query tight. + const stepRowsByRun = new Map>(); + const stepIdsForEventQuery: string[] = []; + if (latestRunIds.length > 0) { + const { data: ar } = await supabase + .from('agent_runs') + .select('id, workflow_run_id, step_id, status, iteration, started_at, summary, error') + .in('workflow_run_id', latestRunIds) + .order('started_at', { ascending: true }); + for (const row of ar ?? []) { + const list = stepRowsByRun.get(row.workflow_run_id) ?? []; + list.push({ id: row.id, stepId: row.step_id, status: row.status, iteration: row.iteration, summary: row.summary, error: row.error }); + stepRowsByRun.set(row.workflow_run_id, list); + stepIdsForEventQuery.push(row.id); + } + } + + const textByAgentRun = new Map(); + if (stepIdsForEventQuery.length > 0) { + const { data: events } = await supabase + .from('agent_run_events') + .select('agent_run_id, payload, type, created_at') + .in('agent_run_id', stepIdsForEventQuery) + .eq('type', 'agent-text') + .order('created_at', { ascending: true }); + for (const ev of events ?? []) { + const payload = (ev.payload ?? {}) as { text?: unknown }; + if (typeof payload.text !== 'string' || !ev.agent_run_id) continue; + textByAgentRun.set(ev.agent_run_id, (textByAgentRun.get(ev.agent_run_id) ?? '') + payload.text); + } + } + + for (const run of runs) { + if (lastRuns.has(run.workflow)) continue; + const steps = (stepRowsByRun.get(run.id) ?? []).map((s) => { + let output = ''; + if (s.status === 'failed' && s.error) output = s.error; + else if (s.status === 'skipped' && s.summary) output = s.summary; + else { + const text = textByAgentRun.get(s.id) ?? ''; + output = text.length > 240 ? '…' + text.slice(-240) : text; + } + return { stepId: s.stepId, status: s.status, iteration: s.iteration, output }; + }); + lastRuns.set(run.workflow, { + id: run.id, + status: run.status, + issueNumber: run.issue_number, + startedAt: run.started_at, + finishedAt: run.finished_at, + reason: run.reason, + steps, + }); + } + } + } + + return rows.map((row) => { + const key = `flow:${row.name}`; + return { + id: row.id, + name: row.name, + steps: parseSteps(row.steps), + triggers: parseTriggers(row.triggers), + paused: row.paused, + updatedAt: row.updated_at, + lastRun: lastRuns.get(key), + recentRuns: recentByWorkflow.get(key) ?? [], + stats: statsByWorkflow.get(key) ?? { totalLast7d: 0, succeededLast7d: 0, avgTokens: 0 }, + }; + }); +} + +function parseSteps(raw: unknown): FlowStep[] { + if (!Array.isArray(raw)) return []; + return raw + .map((s) => { + if (!s || typeof s !== 'object') return null; + const obj = s as Record; + const step: FlowStep = { + skill: typeof obj.skill === 'string' ? obj.skill : '', + argsTemplate: typeof obj.argsTemplate === 'string' ? obj.argsTemplate : '', + }; + if (typeof obj.stopChainIfContains === 'string' && obj.stopChainIfContains) { + step.stopChainIfContains = obj.stopChainIfContains; + } + if (typeof obj.systemNotes === 'string' && obj.systemNotes) { + step.systemNotes = obj.systemNotes; + } + return step; + }) + .filter((x): x is FlowStep => x !== null); +} + +function parseTriggers(raw: unknown): FlowTrigger[] { + if (!Array.isArray(raw)) return []; + return raw + .map((t): FlowTrigger | null => { + if (!t || typeof t !== 'object') return null; + const obj = t as Record; + if (obj.kind === 'issue.opened') return { kind: 'issue.opened' }; + if (obj.kind === 'issue.labeled' && typeof obj.label === 'string') { + return { kind: 'issue.labeled', label: obj.label }; + } + return null; + }) + .filter((x): x is FlowTrigger => x !== null); +} + +// ─── Write ────────────────────────────────────────────────────────────────── + +export async function upsertFlow(params: { + id?: string; + name: string; + steps: FlowStep[]; + triggers: FlowTrigger[]; + paused?: boolean; +}): Promise> { + const workspace = await getActiveWorkspace(); + if (!workspace) return { ok: false, error: 'no active workspace' }; + if (workspace.role === 'viewer') return { ok: false, error: 'viewers cannot edit flows' }; + if (!params.name.trim()) return { ok: false, error: 'name is required' }; + + const supabase = createSupabaseAdminClient(); + const row: Database['public']['Tables']['flows']['Insert'] = { + workspace_id: workspace.id, + name: params.name.trim(), + steps: params.steps as unknown as Database['public']['Tables']['flows']['Insert']['steps'], + triggers: params.triggers as unknown as Database['public']['Tables']['flows']['Insert']['triggers'], + }; + if (params.paused !== undefined) row.paused = params.paused; + + if (params.id) { + const { error } = await supabase + .from('flows') + .update({ ...row, updated_at: new Date().toISOString() }) + .eq('id', params.id) + .eq('workspace_id', workspace.id); + if (error) return { ok: false, error: error.message }; + revalidatePath('/workflows'); + return { ok: true, id: params.id }; + } + + const { data, error } = await supabase + .from('flows') + .insert(row) + .select('id') + .single(); + if (error || !data) return { ok: false, error: error?.message ?? 'insert failed' }; + revalidatePath('/workflows'); + return { ok: true, id: data.id }; +} + +export async function setFlowPaused(id: string, paused: boolean): Promise { + const workspace = await getActiveWorkspace(); + if (!workspace) return { ok: false, error: 'no active workspace' }; + if (workspace.role === 'viewer') return { ok: false, error: 'viewers cannot pause flows' }; + const supabase = createSupabaseAdminClient(); + const { error } = await supabase + .from('flows') + .update({ paused, updated_at: new Date().toISOString() }) + .eq('id', id) + .eq('workspace_id', workspace.id); + if (error) return { ok: false, error: error.message }; + revalidatePath('/workflows'); + return { ok: true }; +} + +export async function deleteFlow(id: string): Promise { + const workspace = await getActiveWorkspace(); + if (!workspace) return { ok: false, error: 'no active workspace' }; + if (workspace.role === 'viewer') return { ok: false, error: 'viewers cannot delete flows' }; + const supabase = createSupabaseAdminClient(); + const { error } = await supabase + .from('flows') + .delete() + .eq('id', id) + .eq('workspace_id', workspace.id); + if (error) return { ok: false, error: error.message }; + revalidatePath('/workflows'); + return { ok: true }; +} + +export async function runFlowOnIssue(params: { + flowId: string; + issueNumber: number; +}): Promise> { + const workspace = await getActiveWorkspace(); + if (!workspace) return { ok: false, error: 'no active workspace' }; + if (workspace.role === 'viewer') return { ok: false, error: 'viewers cannot start runs' }; + if (!Number.isInteger(params.issueNumber) || params.issueNumber <= 0) { + return { ok: false, error: 'issue number must be a positive integer' }; + } + + const supabase = createSupabaseAdminClient(); + const { data: flow } = await supabase + .from('flows') + .select('id') + .eq('id', params.flowId) + .eq('workspace_id', workspace.id) + .single(); + if (!flow) return { ok: false, error: 'flow not found' }; + + const repo = workspace.repoOwner && workspace.repoName + ? `${workspace.repoOwner}/${workspace.repoName}` + : null; + + const { data: job, error } = await supabase + .from('jobs') + .insert({ + workspace_id: workspace.id, + repo, + kind: 'flow', + issue_number: params.issueNumber, + pr_number: null, + priority: 10, + status: 'queued', + max_attempts: 1, + payload: { + trigger: 'manual', + flowId: params.flowId, + flowInput: String(params.issueNumber), + }, + }) + .select('id') + .single(); + if (error || !job) return { ok: false, error: error?.message ?? 'enqueue failed' }; + + revalidatePath('/workflows'); + revalidatePath('/cockpit'); + return { ok: true, jobId: job.id }; +} + +// ─── Skill catalog with descriptions ──────────────────────────────────────── + +export async function listAvailableSkills(): Promise { + const workspace = await getActiveWorkspace(); + if (!workspace) return []; + const supabase = createSupabaseAdminClient(); + const { data } = await supabase + .from('repo_skills') + .select('skills') + .eq('workspace_id', workspace.id); + if (!data) return []; + const byName = new Map(); + for (const row of data) { + const arr = (row.skills as Array> | null) ?? []; + for (const s of arr) { + if (!s || typeof s.name !== 'string' || !s.name.trim()) continue; + const description = + typeof s.description === 'string' + ? s.description.split('\n')[0].trim().slice(0, 240) + : ''; + // First write wins (repo_skills is one row per repo; for multi-repo + // workspaces, skills across repos with the same name are de-duped here). + if (!byName.has(s.name)) byName.set(s.name, { name: s.name, description }); + } + } + return Array.from(byName.values()).sort((a, b) => a.name.localeCompare(b.name)); +} + +// ─── Prompt preview ───────────────────────────────────────────────────────── + +export interface PreviewResult { + stepNumber: number; + skill: string; + /** Generic flow-step scaffolding (constant). */ + scaffoldingSystemPrompt: string; + /** Effective step notes — either the user's override or DEFAULT_STEP_NOTES. */ + stepNotes: string; + /** True when `stepNotes` came from the built-in default (user hasn't overridden). */ + stepNotesIsDefault: boolean; + /** The default text — useful for the editor's placeholder. */ + defaultStepNotes: string; + /** Just the skill body — UI appends this to scaffolding for the "what the agent sees" view. */ + skillBody: string; + /** Empty when the skill couldn't be found in the workspace's repo. */ + skillBodyMissing: boolean; + userPrompt: string; +} + +export async function renderStepPreview(params: { + flowId: string; + stepIdx: number; + sampleInput: string; + samplePrevTaskId: string; + samplePrevPrUrl: string; + samplePrevPrNumber: string; +}): Promise> { + const workspace = await getActiveWorkspace(); + if (!workspace) return { ok: false, error: 'no active workspace' }; + const supabase = createSupabaseAdminClient(); + + const { data: flow, error } = await supabase + .from('flows') + .select('steps') + .eq('id', params.flowId) + .eq('workspace_id', workspace.id) + .single(); + if (error || !flow) return { ok: false, error: 'flow not found' }; + + const steps = parseSteps(flow.steps); + const step = steps[params.stepIdx]; + if (!step) return { ok: false, error: `step ${params.stepIdx + 1} not found` }; + + const userPrompt = renderTemplate(step.argsTemplate, { + input: params.sampleInput, + previousTaskId: params.samplePrevTaskId, + previousPullRequestUrl: params.samplePrevPrUrl, + previousPullRequestNumber: params.samplePrevPrNumber, + }); + + // Try to read the skill body from the workspace's cloned repo. Best-effort — + // when the clone isn't materialized yet, we still return the rendered user + // prompt so the UI shows the args part of the preview. + let skillBody = ''; + let skillBodyMissing = true; + try { + const config = await loadWorkspaceConfig(workspace.id, supabase, {}); + if (!config.autofix.repoRoot && config.github.owner && config.github.repo) { + config.autofix.repoRoot = await ensureRepoClone( + config.github.owner, + config.github.repo, + config.github.token, + config.autofix.baseBranch, + ); + } + if (config.autofix.repoRoot) { + const skills = await discoverSkills(config.autofix.repoRoot, config.autofix.skillsDir ?? '.ai/skills'); + const hit = skills.find((s) => s.name === step.skill); + if (hit) { + skillBody = hit.body; + skillBodyMissing = false; + } + } + } catch { + // swallow — surface as `skillBodyMissing` and let the UI degrade gracefully. + } + + const trimmed = (step.systemNotes ?? '').trim(); + const stepNotes = effectiveStepNotes(step); + return { + ok: true, + stepNumber: params.stepIdx + 1, + skill: step.skill, + scaffoldingSystemPrompt: FLOW_STEP_SCAFFOLDING, + stepNotes, + stepNotesIsDefault: trimmed.length === 0, + defaultStepNotes: DEFAULT_STEP_NOTES, + skillBody, + skillBodyMissing, + userPrompt, + }; +} diff --git a/packages/gui/src/app/workflows/flow-card.tsx b/packages/gui/src/app/workflows/flow-card.tsx new file mode 100644 index 0000000..dd8e891 --- /dev/null +++ b/packages/gui/src/app/workflows/flow-card.tsx @@ -0,0 +1,1054 @@ +'use client'; + +import { useEffect, useMemo, useRef, useState, useTransition, type KeyboardEvent } from 'react'; +import { + renderStepPreview, + type FlowStep, + type FlowStepOutcome, + type FlowSummary, + type FlowTrigger, + type PreviewResult, +} from './actions'; + +/** + * Mirrors `DEFAULT_STEP_NOTES` in `@/lib/flow-runner` — kept here as a literal + * so the client bundle doesn't pull node-only deps. Update both together. + */ +const DEFAULT_STEP_NOTES_TEXT = `If you open a pull request, end your response with two lines on separate lines: + PR_URL= + PR_NUMBER= +so the next step in the workflow can reference it via {{previousPullRequestUrl}} / {{previousPullRequestNumber}}.`; + +export interface SkillOption { + name: string; + description: string; +} + +export interface FlowCardProps { + flow: FlowSummary; + canWrite: boolean; + availableSkills: SkillOption[]; + onNameChange: (name: string) => void; + onStepChange: (idx: number, patch: Partial) => void; + onAddStep: () => void; + onRemoveStep: (idx: number) => void; + onMoveStepDown: (idx: number) => void; + onTriggersChange: (triggers: FlowTrigger[]) => void; + onPausedChange: (paused: boolean) => void; + onDelete: () => void; + onRun: (issue: number) => void; + onOpenRun: () => void; +} + +const TEMPLATE_VARS = [ + { name: 'input', label: '{{input}}', hint: 'value passed when the flow runs' }, + { name: 'previousTaskId', label: '{{previousTaskId}}', hint: 'previous step\'s agent_runs.id' }, + { name: 'previousPullRequestUrl', label: '{{previousPullRequestUrl}}', hint: 'PR url from previous step (if any)' }, + { name: 'previousPullRequestNumber', label: '{{previousPullRequestNumber}}', hint: 'PR number from previous step (if any)' }, +] as const; + +const PR_CREATING_PATTERN = /(fix|open|create|pr|pull|merge|raise)/i; + +export function FlowCard(props: FlowCardProps) { + const { flow } = props; + + return ( +
+ {/* Name + pause toggle + flow actions */} +
+
+ + props.onNameChange(e.target.value)} + disabled={!props.canWrite} + className="mt-1 w-full rounded-md border border-border bg-bg-subtle px-3 py-2 text-sm text-fg focus:border-accent/50 focus:outline-none" + placeholder="fix github issue" + /> +
+ {props.canWrite && ( + <> + + + + + )} +
+ + {/* Stats + history strip */} + {!flow.id.startsWith('__new__') && (flow.recentRuns.length > 0 || flow.stats.totalLast7d > 0) && ( +
+ {flow.stats.totalLast7d > 0 && ( + + ✓ {flow.stats.succeededLast7d}/{flow.stats.totalLast7d}{' '} + (7d) + + )} + {flow.stats.avgTokens > 0 && ( + + ~{formatTokens(flow.stats.avgTokens)} tokens/run + + )} + {flow.recentRuns.length > 0 && ( + + recent + {flow.recentRuns.map((r) => ( + + ))} + + )} + {flow.paused && ( + + paused + + )} +
+ )} + + {/* Triggers panel */} + + + {/* Steps */} +
+ {flow.steps.length === 0 ? ( +

+ No steps yet. {props.canWrite && 'Click "+ Step" to add one.'} +

+ ) : ( +
+
+
Step
+
Skill
+
Args template
+
+
+
+ {flow.steps.map((step, idx) => ( + s.stepId === `step-${idx + 1}`)} + onChange={(patch) => props.onStepChange(idx, patch)} + onRemove={() => props.onRemoveStep(idx)} + onMoveDown={() => props.onMoveStepDown(idx)} + /> + ))} +
+ )} +
+ + {/* Run on issue + most-recent run summary */} + {!flow.id.startsWith('__new__') && ( +
+ {props.canWrite ? ( + + ) : ( + Viewers can't start runs. + )} + {flow.lastRun && ( +
+ + + {flow.lastRun.issueNumber != null ? `#${flow.lastRun.issueNumber}` : ''} + {' · '} + {new Date(flow.lastRun.startedAt).toLocaleString()} + + {flow.lastRun.reason && ( + + · {flow.lastRun.reason.slice(0, 80)} + {flow.lastRun.reason.length > 80 ? '…' : ''} + + )} + +
+ )} +
+ )} +
+ ); +} + +// ─── Pause toggle ─────────────────────────────────────────────────────────── + +function PauseToggle({ paused, onChange }: { paused: boolean; onChange: (v: boolean) => void }) { + return ( + + ); +} + +// ─── Step notes (pencil disclosure) ──────────────────────────────────────── + +function StepNotesPanel({ + value, + disabled, + onChange, +}: { + value: string; + disabled: boolean; + onChange: (v: string) => void; +}) { + const overriding = value.trim().length > 0; + return ( +
+
+ Step notes — extra system-prompt content + + {overriding ? 'overriding default' : 'using default'} + +
+