feat: add integration helpers and comprehensive tests for conditional flow logic

- Introduced new helper functions for creating flows and retrieving step states with skip and
error info
- Added detailed integration tests covering conditional step execution, skipping, cascade
behavior, and optional dependencies
- Included tests for 'if', 'ifNot', 'whenUnmet', 'whenFailed' conditions, and cascade vs
non-cascade skips
- Created a new test file for retries exhausted scenarios (placeholder for future tests)

Author: jumski

pkgs/edge-worker/tests/integration/flow/_conditionalHelpers.ts

@@ -0,0 +1,118 @@
+import type { postgres } from '../../sql.ts';
+import type { Json } from '@pgflow/core';
+
+// ============= Test Helpers for Conditional Flow Integration Tests =============
+
+/**
+ * Step definition for conditional flows
+ *
+ * Note: baseDelay, timeout, and startDelay are in seconds (integers) per
+ * pgflow.steps schema. The SQL function handles conversion to INTERVAL type.
+ */
+export interface ConditionalStepDef {
+  slug: string;
+  deps: string[];
+  maxAttempts?: number;
+  baseDelay?: number;
+  timeout?: number;
+  startDelay?: number;
+  stepType?: 'single' | 'map';
+  requiredInputPattern?: Json;
+  forbiddenInputPattern?: Json;
+  whenUnmet?: 'skip' | 'skip-cascade' | 'fail';
+  whenFailed?: 'skip' | 'skip-cascade' | 'fail';
+}
+
+/**
+ * Create a flow with conditional steps using full add_step parameters
+ */
+export const createConditionalFlow = async (
+  sql: postgres.Sql,
+  flowSlug: string,
+  steps: ConditionalStepDef[]
+) => {
+  await sql`select pgflow.create_flow(${flowSlug});`;
+  for (const step of steps) {
+    const deps = step.deps.length > 0 ? step.deps : null;
+    const maxAttempts = step.maxAttempts ?? null;
+    const baseDelay = step.baseDelay ?? null;
+    const timeout = step.timeout ?? null;
+    const startDelay = step.startDelay ?? null;
+    const stepType = step.stepType ?? 'single';
+    const requiredPattern = step.requiredInputPattern
+      ? JSON.stringify(step.requiredInputPattern)
+      : null;
+    const forbiddenPattern = step.forbiddenInputPattern
+      ? JSON.stringify(step.forbiddenInputPattern)
+      : null;
+    const whenUnmet = step.whenUnmet ?? null;
+    const whenFailed = step.whenFailed ?? null;
+
+    await sql`
+      select pgflow.add_step(
+        ${flowSlug},
+        ${step.slug},
+        deps_slugs => ${deps}::text[],
+        max_attempts => ${maxAttempts}::smallint,
+        base_delay => ${baseDelay}::interval,
+        timeout => ${timeout}::interval,
+        start_delay => ${startDelay}::interval,
+        step_type => ${stepType}::pgflow.step_type,
+        required_input_pattern => ${requiredPattern}::jsonb,
+        forbidden_input_pattern => ${forbiddenPattern}::jsonb,
+        when_unmet => ${whenUnmet}::pgflow.skip_option,
+        when_failed => ${whenFailed}::pgflow.skip_option
+      );
+    `;
+  }
+};
+
+/**
+ * Extended step state info including skip details
+ */
+export interface StepStateWithSkip {
+  step_slug: string;
+  status: string;
+  skip_reason: string | null;
+  skipped_at: string | null;
+}
+
+/**
+ * Get step states with skip information
+ */
+export const getStepStatesWithSkip = async (
+  sql: postgres.Sql,
+  runId: string
+): Promise<StepStateWithSkip[]> => {
+  return await sql<StepStateWithSkip[]>`
+    SELECT step_slug, status, skip_reason, skipped_at
+    FROM pgflow.step_states
+    WHERE run_id = ${runId}
+    ORDER BY step_slug;
+  `;
+};
+
+/**
+ * Extended task info including error details
+ */
+export interface TaskWithError {
+  step_slug: string;
+  status: string;
+  error_message: string | null;
+  attempts_count: number;
+}
+
+/**
+ * Get step tasks with error information
+ */
+export const getStepTasksWithError = async (
+  sql: postgres.Sql,
+  runId: string
+): Promise<TaskWithError[]> => {
+  return await sql<TaskWithError[]>`
+    SELECT step_slug, status, error_message, attempts_count
+    FROM pgflow.step_tasks
+    WHERE run_id = ${runId}
+    ORDER BY step_slug, task_index;
+  `;
+};