diff --git a/crates/manager-core/src/workflow_repo.rs b/crates/manager-core/src/workflow_repo.rs index 9d3f5f1..3f32eb1 100644 --- a/crates/manager-core/src/workflow_repo.rs +++ b/crates/manager-core/src/workflow_repo.rs @@ -989,6 +989,23 @@ pub async fn get_workflow_by_name( row.map(TryInto::try_into).transpose() } +/// Resolve an app-owned workflow by id (app-scoped). Backs the run-detail +/// API's DAG-edge lookup (`depends_on` lives on the definition, not the run). +pub async fn get_workflow_by_id( + pool: &PgPool, + app_id: AppId, + id: WorkflowId, +) -> Result, WorkflowRepoError> { + let row: Option = sqlx::query_as(&format!( + "SELECT {SELECT_COLS} FROM workflows WHERE app_id = $1 AND id = $2" + )) + .bind(app_id.into_inner()) + .bind(id.into_inner()) + .fetch_optional(pool) + .await?; + row.map(TryInto::try_into).transpose() +} + /// Start a nested sub-workflow run and **park** the parent step on it. In one /// token-gated transaction: the parent step (still claimed by the caller) is /// set `running` with `claim_token` cleared and `child_run_id` pointing at a diff --git a/crates/manager-core/src/workflows_api.rs b/crates/manager-core/src/workflows_api.rs index 53ec4e2..01d18c8 100644 --- a/crates/manager-core/src/workflows_api.rs +++ b/crates/manager-core/src/workflows_api.rs @@ -26,8 +26,8 @@ use uuid::Uuid; use crate::app_repo::AppRepository; use crate::authz::{require, AuthzDenied, AuthzError, AuthzRepo, Capability}; use crate::workflow_repo::{ - get_run, get_workflow_by_name, list_run_steps, list_runs_for_workflow, list_workflows_for_app, - start_run, NewWorkflowRun, WorkflowRun, WorkflowRunStep, + get_run, get_workflow_by_id, get_workflow_by_name, list_run_steps, list_runs_for_workflow, + list_workflows_for_app, start_run, NewWorkflowRun, WorkflowRun, WorkflowRunStep, }; use sqlx::PgPool; @@ -103,10 +103,12 @@ struct StepDto { output: Option, error: Option, child_run_id: Option, + /// DAG edges (from the definition) so the dashboard can draw the graph. + depends_on: Vec, } -impl From for StepDto { - fn from(s: WorkflowRunStep) -> Self { +impl StepDto { + fn from_step(s: WorkflowRunStep, depends_on: Vec) -> Self { Self { name: s.step_name, status: s.status.as_str().to_string(), @@ -115,6 +117,7 @@ impl From for StepDto { output: s.output, error: s.error, child_run_id: s.child_run_id.map(WorkflowRunId::into_inner), + depends_on, } } } @@ -219,8 +222,29 @@ async fn get_run_detail( let steps = list_run_steps(&s.pool, run.id) .await .map_err(|e| WorkflowsApiError::Backend(e.to_string()))?; + // Load the definition for the per-step `depends_on` DAG edges (they live on + // the definition, not the run's step rows). + let deps: std::collections::HashMap> = + match get_workflow_by_id(&s.pool, app_id, run.workflow_id) + .await + .map_err(|e| WorkflowsApiError::Backend(e.to_string()))? + { + Some(w) => w + .definition + .steps + .into_iter() + .map(|st| (st.name, st.depends_on)) + .collect(), + None => std::collections::HashMap::new(), + }; let run_dto: RunDto = run.into(); - let step_dtos: Vec = steps.into_iter().map(Into::into).collect(); + let step_dtos: Vec = steps + .into_iter() + .map(|st| { + let d = deps.get(&st.step_name).cloned().unwrap_or_default(); + StepDto::from_step(st, d) + }) + .collect(); Ok(Json(json!({ "run": run_dto, "steps": step_dtos }))) } diff --git a/dashboard/src/lib/AppTabBar.svelte b/dashboard/src/lib/AppTabBar.svelte index dedda28..4748d97 100644 --- a/dashboard/src/lib/AppTabBar.svelte +++ b/dashboard/src/lib/AppTabBar.svelte @@ -33,6 +33,12 @@ { id: 'users', label: 'Users', href: `${base}/apps/${slug}/users`, adminOnly: true }, { id: 'files', label: 'Files', href: `${base}/apps/${slug}/files`, adminOnly: true }, { id: 'queues', label: 'Queues', href: `${base}/apps/${slug}/queues`, adminOnly: true }, + { + id: 'workflows', + label: 'Workflows', + href: `${base}/apps/${slug}/workflows`, + adminOnly: true + }, { id: 'dead-letters', label: 'Dead letters', diff --git a/dashboard/src/lib/api.ts b/dashboard/src/lib/api.ts index c2cb631..7dbdc04 100644 --- a/dashboard/src/lib/api.ts +++ b/dashboard/src/lib/api.ts @@ -444,6 +444,57 @@ export interface CreateQueueTriggerInput { retry_base_ms?: number; } +// v1.2 Workflows — read/run types (admin workflows API). +export interface WorkflowSummary { + name: string; + enabled: boolean; + steps: number; +} + +export type WorkflowRunStatus = + | 'pending' + | 'running' + | 'succeeded' + | 'failed' + | 'canceled'; + +export interface WorkflowRun { + id: string; + workflow_id?: string; + status: WorkflowRunStatus; + error?: string | null; + workflow_depth: number; + created_at: string; + input?: unknown; + output?: unknown; + started_at?: string | null; + finished_at?: string | null; +} + +export type WorkflowStepStatus = + | 'pending' + | 'ready' + | 'running' + | 'succeeded' + | 'failed' + | 'skipped'; + +export interface WorkflowRunStep { + name: string; + status: WorkflowStepStatus; + attempt: number; + max_attempts: number; + error?: string | null; + child_run_id?: string | null; + depends_on: string[]; + output?: unknown; +} + +export interface WorkflowRunDetail { + run: WorkflowRun; + steps: WorkflowRunStep[]; +} + // v1.1.9 — read-only queue inspection types (admin queues API). export interface QueueSummary { queue_name: string; @@ -1214,6 +1265,26 @@ export const api = { ) }, + workflows: { + list: (idOrSlug: string) => + adminRequest( + `/api/v1/admin/apps/${encodeURIComponent(idOrSlug)}/workflows` + ), + runs: (idOrSlug: string, name: string) => + adminRequest( + `/api/v1/admin/apps/${encodeURIComponent(idOrSlug)}/workflows/${encodeURIComponent(name)}/runs` + ), + start: (idOrSlug: string, name: string, input: unknown) => + adminRequest( + `/api/v1/admin/apps/${encodeURIComponent(idOrSlug)}/workflows/${encodeURIComponent(name)}/runs`, + { method: 'POST', body: JSON.stringify({ input }) } + ), + run: (idOrSlug: string, runId: string) => + adminRequest( + `/api/v1/admin/apps/${encodeURIComponent(idOrSlug)}/workflow-runs/${encodeURIComponent(runId)}` + ) + }, + topics: { list: (idOrSlug: string) => adminRequest<{ topics: Topic[] }>( diff --git a/dashboard/src/routes/apps/[slug]/+layout.svelte b/dashboard/src/routes/apps/[slug]/+layout.svelte index a89fa41..5ed1424 100644 --- a/dashboard/src/routes/apps/[slug]/+layout.svelte +++ b/dashboard/src/routes/apps/[slug]/+layout.svelte @@ -70,6 +70,8 @@ return 'files'; case 'queues': return 'queues'; + case 'workflows': + return 'workflows'; case 'dead-letters': return 'dead-letters'; default: diff --git a/dashboard/src/routes/apps/[slug]/workflows/+page.svelte b/dashboard/src/routes/apps/[slug]/workflows/+page.svelte new file mode 100644 index 0000000..09789a9 --- /dev/null +++ b/dashboard/src/routes/apps/[slug]/workflows/+page.svelte @@ -0,0 +1,433 @@ + + + + Workflows — {appName} — PiCloud + + +
+

Workflows

+

+ Durable DAG workflows (v1.2). Definitions are authored declaratively + ([[workflows]] in picloud.toml); start runs and + inspect their per-step progress here. +

+
+ +
+
+ +{#if error} +

{error}

+{/if} + +{#if loading && workflows.length === 0} +

Loading…

+{:else if workflows.length === 0} +

+ No workflows in this app yet. Author one with a [[workflows]] + block in your manifest and pic apply. +

+{:else} + + + + + + {#each workflows as w (w.name)} + + + + + + + {/each} + +
WorkflowStepsEnabled
{w.name}{w.steps}{w.enabled ? 'yes' : 'no'} + +
+{/if} + +{#if selectedWf} +
+
+

Runs — {selectedWf}

+ +
+ {#if runs.length === 0} +

No runs yet.

+ {:else} + + + + + + {#each runs as r (r.id)} + + + + + + + + {/each} + +
RunStatusDepthCreated
{short(r.id)}{r.status}{r.workflow_depth}{r.created_at} + +
+ {/if} +
+{/if} + +{#if detail} +
+

+ Run {short(detail.run.id)} + {detail.run.status} +

+ {#if detail.run.error} +

{detail.run.error}

+ {/if} + +

DAG

+
+ + {#each layout.edges as e} + + {/each} + + + + + + {#each layout.nodes as n (n.step.name)} + + + + {n.step.name} + + + {/each} + +
+ +

Steps

+ + + + + + {#each detail.steps as s (s.name)} + + + + + + + + {/each} + +
StepStatusAttemptDepends onError
{s.name}{s.status}{s.attempt}/{s.max_attempts}{s.depends_on.join(', ')}{s.error ?? ''}
+
+{/if} + + diff --git a/dashboard/tests/e2e/navigation/tabs.spec.ts b/dashboard/tests/e2e/navigation/tabs.spec.ts index 526f45c..012b893 100644 --- a/dashboard/tests/e2e/navigation/tabs.spec.ts +++ b/dashboard/tests/e2e/navigation/tabs.spec.ts @@ -105,6 +105,7 @@ const TABS: Array<{ subpath: string; label: string }> = [ { subpath: '', label: 'main' }, { subpath: '/queues', label: 'queues' }, { subpath: '/queues/never-enqueued', label: 'queue drilldown' }, + { subpath: '/workflows', label: 'workflows' }, { subpath: '/files', label: 'files' }, { subpath: '/dead-letters', label: 'dead-letters' }, { subpath: '/users', label: 'app users' }, @@ -154,4 +155,21 @@ test.describe('per-app tab navigation accepts slug URLs', () => { // DOM element. await expect(page.getByTestId('queues-empty-state')).toBeVisible(); }); + + test('workflows tab renders its empty state (v1.2)', async ({ page, uniqueSlug }) => { + const slug = uniqueSlug('wf'); + cleanup.app(slug); + const api = await adminApi(); + try { + const res = await api.post('/api/v1/admin/apps', { + data: { slug, name: slug } + }); + expect(res.ok()).toBe(true); + } finally { + await api.dispose(); + } + await expectTabLoadsCleanly(page, `/admin/apps/${slug}/workflows`); + // A fresh app has no workflows → the empty-state anchors the smoke test. + await expect(page.getByTestId('workflows-empty-state')).toBeVisible(); + }); });