feat: implement article flow with parallel steps and retry logic

- Added task functions for fetching, summarizing, extracting keywords, and publishing
- Updated flow definition to run summarize and extract keywords in parallel
- Included retry mechanism in summarize function via attemptNumber parameter
- Simplified flow configuration with maxAttempts and default settings
- Improved code clarity and structure for Phase 2 article processing flow

Author: jumski

.tool-versions

@@ -0,0 +1,2 @@
+nodejs 24.11.0
+pnpm 10.20.0

PHASE_2_ARTICLE_FLOW.md

@@ -5,6 +5,7 @@
 **Goal:** Implement 4-step article processing flow with state management. Establish data flow foundation for observability features.
 
 **Success Criteria:**
+
 - ✅ Article processing flow with 4 steps (fetch, summarize, extractKeywords, publish)
 - ✅ Simulated retry on summarize step
 - ✅ pgflow State Store manages run state
@@ -25,12 +26,27 @@ cd ../..
 ```
 
 This installs:
-- `@xyflow/svelte` - DAG visualization
-- `shiki` - Syntax highlighting
+
+- `@xyflow/svelte` - DAG visualization (used in Phase 3)
+- `shiki` - Syntax highlighting (used in Phase 4+)
+
+**Note:** Installing all frontend dependencies now simplifies later phases and avoids switching contexts when building UI components.
+
+---
+
+### 2. Copy Logo Assets
+
+Copy pgflow logos for header component (used in Phase 6):
+
+```bash
+cp pkgs/website/src/assets/pgflow-logo-*.svg apps/demo/static/
+```
+
+This copies both light and dark variants of the pgflow logo.
 
 ---
 
-### 2. Create Article Flow Worker
+### 3. Create Article Flow Worker
 
 Create new Edge Function for the article flow:
 
@@ -43,55 +59,75 @@ This creates `apps/demo/supabase/functions/article_flow_worker/` directory.
 
 ---
 
-### 3. Create Article Processing Flow
+### 4. Create Article Processing Flow
 
 Create `apps/demo/supabase/functions/article_flow_worker/article_flow.ts` with 4 steps:
 
 ```typescript
 import { Flow } from '@pgflow/dsl';
 
+// Task functions (implementations can be simple for Phase 2)
+async function fetchArticle(url: string) {
+  const response = await fetch(`https://r.jina.ai/${url}`);
+  const content = await response.text();
+  return { content, title: 'Article Title' };
+}
+
+function summarizeArticle(content: string, attemptNumber: number) {
+  // Simulate failure on first attempt for retry demo
+  if (attemptNumber === 1) {
+    throw new Error('Simulated failure for retry demo');
+  }
+  return { summary: `Summary of article`, sentiment: 'positive' };
+}
+
+function extractKeywords(content: string) {
+  return { keywords: ['keyword1', 'keyword2', 'keyword3'] };
+}
+
+function publishArticle(data: { summary: any; keywords: any }) {
+  return {
+    articleId: crypto.randomUUID(),
+    publishedAt: new Date().toISOString(),
+    ...data,
+  };
+}
+
+// Flow definition - clean and minimal
 export default new Flow<{ url: string }>({
   slug: 'article_flow',
   maxAttempts: 3,
-  baseDelay: 1,
-  timeout: 60
 })
-  .step({ slug: 'fetch_article' }, async (input) => {
-    // Call r.jina.ai API
-    const response = await fetch(`https://r.jina.ai/${input.run.url}`);
-    const content = await response.text();
-    return { content, title: 'Article Title' };
-  })
-  .step({ slug: 'summarize' }, (input) => {
-    // Simulate failure on first attempt
-    if (input.attemptNumber === 1) {
-      throw new Error('Simulated failure for retry demo');
-    }
-    return `Summary of: ${input.steps.fetch_article.title}`;
-  })
-  .step({ slug: 'extract_keywords' }, (input) => {
-    // Runs parallel with summarize
-    return ['keyword1', 'keyword2', 'keyword3'];
-  })
-  .step({ slug: 'publish' }, (input) => {
-    // Depends on both summarize and extract_keywords
-    return {
-      articleId: 'article_123',
-      summary: input.steps.summarize,
-      keywords: input.steps.extract_keywords
-    };
-  });
+  .step({ slug: 'fetch_article' }, async (input) => fetchArticle(input.run.url))
+  .step({ slug: 'summarize', dependsOn: ['fetch_article'] }, (input) =>
+    summarizeArticle(input.fetch_article.content, input.attemptNumber)
+  )
+  .step({ slug: 'extract_keywords', dependsOn: ['fetch_article'] }, (input) =>
+    extractKeywords(input.fetch_article.content)
+  )
+  .step(
+    { slug: 'publish', dependsOn: ['summarize', 'extract_keywords'] },
+    (input) =>
+      publishArticle({
+        summary: input.summarize,
+        keywords: input.extract_keywords,
+      })
+  );
 ```
 
 **Key patterns:**
+
 - Flow slug is `article_flow` (with underscore)
-- Parallel execution: summarize and extractKeywords run simultaneously (both depend only on fetch_article)
-- Retry simulation: Use `attemptNumber` param to fail first attempt
-- Flow config: `maxAttempts: 3, baseDelay: 1, timeout: 60`
+- Task functions defined separately for clarity
+- Parallel execution: `summarize` and `extract_keywords` run simultaneously (both depend only on `fetch_article`)
+- Retry simulation: `summarizeArticle` checks `attemptNumber` to fail first attempt
+- Diamond DAG shape: 1 → 2 parallel → 1
+- Minimal config: Only `slug` and `maxAttempts` (defaults used for baseDelay/timeout)
+- Explicit `dependsOn` arrays show flow structure clearly
 
 ---
 
-### 4. Create Edge Function Worker
+### 5. Create Edge Function Worker
 
 Create `apps/demo/supabase/functions/article_flow_worker/index.ts`:
 
@@ -104,7 +140,7 @@ EdgeWorker.start(ArticleFlow);
 
 ---
 
-### 5. Create Deno Import Map
+### 6. Create Deno Import Map
 
 Create `apps/demo/supabase/functions/article_flow_worker/deno.json`:
 
@@ -128,7 +164,7 @@ Create `apps/demo/supabase/functions/article_flow_worker/deno.json`:
 
 ---
 
-### 6. Configure Edge Function in config.toml
+### 7. Configure Edge Function in config.toml
 
 Edit `apps/demo/supabase/config.toml`, add at the end:
 
@@ -144,7 +180,7 @@ entrypoint = "./functions/article_flow_worker/index.ts"
 
 ---
 
-### 7. Set Environment Variables
+### 8. Set Environment Variables
 
 Create `apps/demo/supabase/.env.local` with `JINA_API_KEY` (optional for now):
 
@@ -154,7 +190,7 @@ JINA_API_KEY=your_jina_api_key_here
 
 ---
 
-### 8. Rebuild and Re-vendor
+### 9. Rebuild and Re-vendor
 
 ```bash
 pnpm nx build core dsl client
@@ -163,7 +199,7 @@ pnpm nx sync-edge-deps demo
 
 ---
 
-### 9. Test Edge Function Locally
+### 10. Test Edge Function Locally
 
 ```bash
 cd apps/demo
@@ -174,7 +210,7 @@ npx -y supabase@latest functions serve article_flow_worker
 
 ---
 
-### 10. Create pgflow State Store
+### 11. Create pgflow State Store
 
 Create `apps/demo/src/lib/stores/pgflow-state.svelte.ts`:
 
@@ -199,6 +235,7 @@ export const pgflowState = new PgflowState();
 **Purpose:** Central state management for flow execution, used by all UI components
 
 **Key patterns:**
+
 - Use Svelte 5 runes: `$state` and `$derived`
 - Export singleton instance
 - Will be updated in Phase 3 when building UI
@@ -207,14 +244,15 @@ export const pgflowState = new PgflowState();
 
 ## Validation Checklist
 
-- [ ] Article flow worker created (`article_flow_worker/`)
-- [ ] 4-step flow created with simulated retry
-- [ ] Worker configured in `config.toml` with `verify_jwt = false`
-- [ ] Deno import map created with all dependencies
-- [ ] pgflow State Store created and exported
-- [ ] All dependencies installed (`@xyflow/svelte`, `shiki`)
-- [ ] Edge Function serves successfully
-- [ ] Build succeeds
+- [x] All dependencies installed (`@xyflow/svelte`, `shiki`)
+- [x] Logo assets copied to `apps/demo/static/`
+- [x] Article flow worker created (`article_flow_worker/`)
+- [x] 4-step flow created with simulated retry
+- [x] Worker configured in `config.toml` with `verify_jwt = false`
+- [x] Deno import map created with all dependencies
+- [x] pgflow State Store created and exported
+- [x] Edge Function serves successfully
+- [x] Build succeeds
 
 ---
 

PHASE_3_DAG_DEBUG.md

@@ -19,7 +19,19 @@
 
 ## Tasks
 
-### 7. Create DAG Component
+### 1. Verify Dependencies
+
+Confirm `@xyflow/svelte` was installed in Phase 2:
+
+```bash
+ls apps/demo/node_modules/@xyflow/svelte
+```
+
+**If missing:** Run `cd apps/demo && pnpm add @xyflow/svelte && cd ../..`
+
+---
+
+### 2. Create DAG Component
 
 Create `apps/demo/src/lib/components/DAGVisualization.svelte`:
 
@@ -31,7 +43,7 @@ Use `@xyflow/svelte` to render 4 nodes:
 
 ---
 
-### 8. Create Debug Panel Component
+### 3. Create Debug Panel Component
 
 Create `apps/demo/src/lib/components/DebugPanel.svelte` with 3 sections:
 
@@ -43,7 +55,7 @@ Create `apps/demo/src/lib/components/DebugPanel.svelte` with 3 sections:
 
 ---
 
-### 9. Create Demo Page Layout
+### 4. Create Demo Page Layout
 
 Update `apps/demo/src/routes/+page.svelte`:
 
@@ -57,11 +69,11 @@ Update `apps/demo/src/routes/+page.svelte`:
 
 ---
 
-### 10. Add Brand Assets and Styles
+### 5. Add Brand Styles
 
-Copy logos from website to static folder:
+Verify logos were copied in Phase 2:
 ```bash
-cp pkgs/website/src/assets/pgflow-logo-*.svg apps/demo/static/
+ls apps/demo/static/pgflow-logo-*.svg
 ```
 
 Create `apps/demo/src/app.css` with pgflow brand colors:
@@ -74,7 +86,7 @@ Import styles in `apps/demo/src/routes/+layout.svelte`
 
 ---
 
-### 11. Test Complete Flow
+### 6. Test Complete Flow
 
 ```bash
 cd apps/demo

apps/demo/package.json

@@ -38,6 +38,9 @@
 	},
 	"dependencies": {
 		"@pgflow/client": "workspace:*",
-		"@supabase/supabase-js": "^2.78.0"
+		"@pgflow/dsl": "workspace:*",
+		"@supabase/supabase-js": "^2.78.0",
+		"@xyflow/svelte": "^1.4.1",
+		"shiki": "^3.14.0"
 	}
 }

apps/demo/project.json

@@ -6,11 +6,20 @@
   "targets": {
     "dev": {
       "executor": "nx:run-commands",
+      "dependsOn": ["^build"],
       "options": {
         "command": "vite dev",
         "cwd": "apps/demo"
       }
     },
+    "dev:remote": {
+      "executor": "nx:run-commands",
+      "dependsOn": ["^build"],
+      "options": {
+        "command": "vite dev --host 0.0.0.0",
+        "cwd": "apps/demo"
+      }
+    },
     "build": {
       "executor": "nx:run-commands",
       "options": {
@@ -34,6 +43,13 @@
         "command": "./scripts/sync-edge-deps.sh",
         "cwd": "apps/demo"
       }
+    },
+    "test": {
+      "executor": "nx:run-commands",
+      "options": {
+        "command": "deno test --allow-env --allow-net tests/",
+        "cwd": "apps/demo/supabase/functions/article_flow_worker"
+      }
     }
   }
 }