refactor: rename input parameters to flowInput and deps for clarity (#557)

# Rename input parameters in step handlers for better clarity

This PR renames the input parameters in step handlers to improve clarity and better reflect their purpose:

- Root steps (no dependencies) now use `flowInput` instead of `input.run` to directly access flow input
- Dependent steps now use `deps` instead of `input` to access outputs from dependency steps
- Context parameter is now consistently named `ctx` where used

These changes make the code more intuitive by:
1. Clearly distinguishing between flow input and dependency outputs
2. Using parameter names that better reflect their content
3. Maintaining consistent naming conventions throughout the codebase

The changes affect example code in the documentation as well as actual implementation code, ensuring consistency across the entire project.

Author: jumski

apps/demo/src/lib/data/flow-code.ts

@@ -26,50 +26,50 @@ export const FLOW_SECTIONS: Record<string, CodeSection> = {
 	fetchArticle: {
 		code: `  .step(
     { slug: 'fetchArticle' },
-    (input) => scrapeUrl(input.run.url)
+    (flowInput) => scrapeUrl(flowInput.url)
   )`,
 		mobileCode: `  .step(
     { slug: 'fetchArticle' },
-    (input) => scrapeUrl(
-      input.run.url
+    (flowInput) => scrapeUrl(
+      flowInput.url
     )
   )`
 	},
 	summarize: {
 		code: `  .step(
     { slug: 'summarize', dependsOn: ['fetchArticle'] },
-    (input) => summarize(schema, input.fetchArticle.content)
+    (deps) => summarize(schema, deps.fetchArticle.content)
   )`,
 		mobileCode: `  .step(
     {
       slug: 'summarize',
       dependsOn: ['fetchArticle']
     },
-    (input) => summarize(
+    (deps) => summarize(
       schema,
-      input.fetchArticle.content
+      deps.fetchArticle.content
     )
   )`
 	},
 	extractKeywords: {
 		code: `  .step(
     { slug: 'extractKeywords', dependsOn: ['fetchArticle'] },
-    (input) => extractKeywords(input.fetchArticle.content)
+    (deps) => extractKeywords(deps.fetchArticle.content)
   )`,
 		mobileCode: `  .step(
     {
       slug: 'extractKeywords',
       dependsOn: ['fetchArticle']
     },
-    (input) => extractKeywords(
-      input.fetchArticle.content
+    (deps) => extractKeywords(
+      deps.fetchArticle.content
     )
   )`
 	},
 	publish: {
 		code: `  .step(
     { slug: 'publish', dependsOn: ['summarize', 'extractKeywords'] },
-    (input) => publishArticle(input.summarize, input.extractKeywords)
+    (deps) => publishArticle(deps.summarize, deps.extractKeywords)
   );`,
 		mobileCode: `  .step(
     {
@@ -79,9 +79,9 @@ export const FLOW_SECTIONS: Record<string, CodeSection> = {
         'extractKeywords'
       ]
     },
-    (input) => publishArticle(
-      input.summarize,
-      input.extractKeywords
+    (deps) => publishArticle(
+      deps.summarize,
+      deps.extractKeywords
     )
   );`
 	}

apps/demo/supabase/functions/article_flow_worker/article_flow.ts

@@ -17,10 +17,10 @@ export default new Flow<{ url: string }>({
 	baseDelay: 1,
 	maxAttempts: 2
 })
-	.step({ slug: 'fetchArticle' }, async (input) => {
+	.step({ slug: 'fetchArticle' }, async (flowInput) => {
 		await sleep(SLEEP_MS);
 		const startTime = Date.now();
-		const result = await fetchArticle(input.run.url);
+		const result = await fetchArticle(flowInput.url);
 		const durationMs = Date.now() - startTime;
 		return {
 			...result,
@@ -30,16 +30,12 @@ export default new Flow<{ url: string }>({
 			}
 		};
 	})
-	.step({ slug: 'summarize', dependsOn: ['fetchArticle'], baseDelay: 1 }, async (input) =>
-		summarizeArticle(input.fetchArticle.content)
+	.step({ slug: 'summarize', dependsOn: ['fetchArticle'], baseDelay: 1 }, async (deps) =>
+		summarizeArticle(deps.fetchArticle.content)
 	)
-	.step({ slug: 'extractKeywords', dependsOn: ['fetchArticle'] }, async (input) =>
-		extractKeywords(input.fetchArticle.content)
+	.step({ slug: 'extractKeywords', dependsOn: ['fetchArticle'] }, async (deps) =>
+		extractKeywords(deps.fetchArticle.content)
 	)
-	.step({ slug: 'publish', dependsOn: ['summarize', 'extractKeywords'] }, async (input) =>
-		publishArticle(
-			input.summarize.summary,
-			input.summarize.sentiment,
-			input.extractKeywords.keywords
-		)
+	.step({ slug: 'publish', dependsOn: ['summarize', 'extractKeywords'] }, async (deps) =>
+		publishArticle(deps.summarize.summary, deps.summarize.sentiment, deps.extractKeywords.keywords)
 	);

pkgs/cli/__tests__/commands/install/create-flows-directory.test.ts

@@ -92,7 +92,7 @@ describe('createFlowsDirectory', () => {
     // Second step should depend on first
     expect(greetUserContent).toContain("dependsOn: ['fullName']");
     // Second step should access result from first step
-    expect(greetUserContent).toContain('input.fullName');
+    expect(greetUserContent).toContain('deps.fullName');
   });
 
   it('should not create files when they already exist', async () => {

pkgs/cli/examples/analyze_website.ts

@@ -14,11 +14,11 @@ const AnalyzeWebsite = new Flow<WebsiteAnalysisInput>({
   timeout: 10,
 })
   // First step: scrape the website
-  .step({ slug: 'website' }, async (input) => {
-    console.log(`Scraping website: ${input.run.url}`);
+  .step({ slug: 'website' }, async (flowInput) => {
+    console.log(`Scraping website: ${flowInput.url}`);
     // In a real implementation, this would call an actual web scraper
     return {
-      content: `Sample content from ${input.run.url}`,
+      content: `Sample content from ${flowInput.url}`,
       title: 'Sample Website Title',
       links: ['https://example.com/page1', 'https://example.com/page2'],
     };
@@ -31,8 +31,8 @@ const AnalyzeWebsite = new Flow<WebsiteAnalysisInput>({
       timeout: 30,
       maxAttempts: 5,
     },
-    async (input) => {
-      console.log(`Analyzing sentiment for: ${input.website.title}`);
+    async (deps) => {
+      console.log(`Analyzing sentiment for: ${deps.website.title}`);
       // In a real implementation, this would call a sentiment analysis service
       return {
         score: 0.75,
@@ -42,19 +42,19 @@ const AnalyzeWebsite = new Flow<WebsiteAnalysisInput>({
     }
   )
   // Third step: generate summary (depends on website step)
-  .step({ slug: 'summary', dependsOn: ['website'] }, async (input) => {
-    console.log(`Generating summary for: ${input.website.title}`);
+  .step({ slug: 'summary', dependsOn: ['website'] }, async (deps) => {
+    console.log(`Generating summary for: ${deps.website.title}`);
     // In a real implementation, this might use an AI service
     return {
-      aiSummary: `This is a summary of ${input.website.title}`,
-      wordCount: input.website.content.split(' ').length,
+      aiSummary: `This is a summary of ${deps.website.title}`,
+      wordCount: deps.website.content.split(' ').length,
     };
   })
   // Fourth step: save results to database (depends on sentiment and summary)
   .step(
     { slug: 'saveToDb', dependsOn: ['sentiment', 'summary'] },
-    async (input) => {
-      console.log(`Saving results to database for: ${input.run.url}`);
+    async (_deps, ctx) => {
+      console.log(`Saving results to database for: ${ctx.flowInput.url}`);
       // In a real implementation, this would save to a database
       return {
         status: 'success',

pkgs/cli/src/commands/install/create-flows-directory.ts

@@ -21,11 +21,11 @@ export const GreetUser = new Flow<Input>({
 })
   .step(
     { slug: 'fullName' },
-    (input) => \`\${input.run.firstName} \${input.run.lastName}\`
+    (flowInput) => \`\${flowInput.firstName} \${flowInput.lastName}\`
   )
   .step(
     { slug: 'greeting', dependsOn: ['fullName'] },
-    (input) => \`Hello, \${input.fullName}!\`
+    (deps) => \`Hello, \${deps.fullName}!\`
   );
 `;
 

pkgs/cli/supabase/flows/test_flow_e2e.ts

@@ -4,6 +4,6 @@ import { Flow } from '@pgflow/dsl';
 export const TestFlowE2E = new Flow<{ value: string }>({
   slug: 'test_flow_e2e',
   maxAttempts: 3,
-}).step({ slug: 'step1' }, async (input) => ({
-  result: `processed: ${input.run.value}`,
+}).step({ slug: 'step1' }, async (flowInput) => ({
+  result: `processed: ${flowInput.value}`,
 }));

pkgs/client/__tests__/e2e/full-stack-dsl.test.ts

@@ -19,20 +19,20 @@ describe('Full Stack DSL Integration', () => {
       const SimpleFlow = new Flow<{ url: string }>({
         slug: 'simple_dag_test',
       })
-        .step({ slug: 'fetch' }, async (_input) => ({
+        .step({ slug: 'fetch' }, async (_flowInput) => ({
           data: 'fetched content',
           status: 200,
           items: 10,
         }))
-        .step({ slug: 'process', dependsOn: ['fetch'] }, async (input) => ({
+        .step({ slug: 'process', dependsOn: ['fetch'] }, async (deps) => ({
           processed_data: 'cleaned and validated',
-          item_count: input.fetch.items * 2,
+          item_count: deps.fetch.items * 2,
           metadata: { stage: 'processed' },
         }))
-        .step({ slug: 'save', dependsOn: ['process'] }, async (input) => ({
+        .step({ slug: 'save', dependsOn: ['process'] }, async (deps) => ({
           saved: true,
           record_id: 'rec_12345',
-          final_count: input.process.item_count,
+          final_count: deps.process.item_count,
         }));
 
       // Clean up flow data to ensure clean state
@@ -95,7 +95,7 @@ describe('Full Stack DSL Integration', () => {
       let tasks = await readAndStart(sql, sqlClient, SimpleFlow.slug, 1, 5);
       expect(tasks).toHaveLength(1);
       expect(tasks[0].step_slug).toBe('fetch');
-      expect(tasks[0].input.run).toEqual(input);
+      expect(tasks[0].flow_input).toEqual(input);
 
       const fetchOutput = { data: 'fetched content', status: 200, items: 10 };
       await sqlClient.completeTask(tasks[0], fetchOutput);
@@ -111,7 +111,7 @@ describe('Full Stack DSL Integration', () => {
       tasks = await readAndStart(sql, sqlClient, SimpleFlow.slug, 1, 5);
       expect(tasks).toHaveLength(1);
       expect(tasks[0].step_slug).toBe('process');
-      expect(tasks[0].input.run).toEqual(input);
+      expect(tasks[0].flow_input).toEqual(input);
       expect(tasks[0].input.fetch).toEqual(fetchOutput); // Critical: dependency output included
 
       const processOutput = {

pkgs/client/__tests__/e2e/happy-path-e2e.test.ts

@@ -73,7 +73,7 @@ describe('Happy Path E2E Integration', () => {
       let tasks = await readAndStart(sql, sqlClient, testFlow.slug, 1, 5);
       expect(tasks).toHaveLength(1);
       expect(tasks[0].step_slug).toBe('fetch');
-      expect(tasks[0].input.run).toEqual(input);
+      expect(tasks[0].flow_input).toEqual(input);
 
       const fetchOutput = { data: 'fetched content', status: 200, items: 10 };
       await sqlClient.completeTask(tasks[0], fetchOutput);
@@ -88,7 +88,7 @@ describe('Happy Path E2E Integration', () => {
       tasks = await readAndStart(sql, sqlClient, testFlow.slug, 1, 5);
       expect(tasks).toHaveLength(1);
       expect(tasks[0].step_slug).toBe('process');
-      expect(tasks[0].input.run).toEqual(input);
+      expect(tasks[0].flow_input).toEqual(input);
       expect(tasks[0].input.fetch).toEqual(fetchOutput);
 
       const processOutput = { 

pkgs/client/__tests__/e2e/real-flow-execution.test.ts

@@ -41,7 +41,7 @@ describe('Real Flow Execution', () => {
       const tasks = await readAndStart(sql, sqlClient, testFlow.slug, 1, 5);
 
       expect(tasks).toHaveLength(1);
-      expect(tasks[0].input.run).toEqual(input);
+      expect(tasks[0].flow_input).toEqual(input);
 
       // Complete task with complex nested output to verify JSON parsing
       const taskOutput = {

pkgs/dsl/tests/context-inference.test.ts

@@ -77,10 +77,10 @@ interface ExplicitContext {
 }
 
 const ExplicitFlow = new Flow<{ userId: string }, ExplicitContext>({ slug: 'explicit-flow' })
-  .step({ slug: 'get-user' }, async (input, context) => {
+  .step({ slug: 'get-user' }, async (flowInput, context) => {
     // All ExplicitContext properties should be available
-    const user = await context.sql.query(`SELECT * FROM users WHERE id = ${input.run.userId}`);
-    await context.cache.set(`user:${input.run.userId}`, JSON.stringify(user));
+    const user = await context.sql.query(`SELECT * FROM users WHERE id = ${flowInput.userId}`);
+    await context.cache.set(`user:${flowInput.userId}`, JSON.stringify(user));
     context.pubsub.publish('user-fetched');
     return user;
   });
@@ -95,9 +95,9 @@ const _explicitTest: ExplicitFlowContext = {
 
 // Test 5: Mixed approach - explicit base with inferred additions
 const MixedFlow = new Flow<{ id: string }, { sql: TestSql }>({ slug: 'mixed-flow' })
-  .step({ slug: 'query' }, async (input, context) => {
+  .step({ slug: 'query' }, async (flowInput, context) => {
     // Has sql from explicit type
-    return await context.sql.query(`SELECT * FROM items WHERE id = ${input.run.id}`);
+    return await context.sql.query(`SELECT * FROM items WHERE id = ${flowInput.id}`);
   })
   .step({ slug: 'enhance' }, async (input, context: Context & { sql: TestSql, ai: { generate: () => string } }) => {
     // Adds ai requirement via inference