Declarative Pipeline DAG Engine (depends_on), Parallel Fetch/Scrape & Dynamic TUI Task Queueing

[Aryan10]

[Feature Proposal] Declarative Pipeline DAG Engine (depends_on), Parallel Fetch/Scrape Stage Execution & Dynamic TUI Task Queueing

Issue Title: Declarative Pipeline DAG Engine (depends_on), Parallel Fetch/Scrape Stage Execution & Dynamic TUI Task Queueing
Labels: feature, pipeline, tui, architecture
Status: Open (Proposal)

---

1. Overview & Objectives

Currently, the news ingestion pipeline executes stages sequentially (fetch → scrape → generate → embed) with hardcoded execution logic and pre-rendered, static stage progress bars in the Textual TUI.

This issue proposes introducing a declarative, dependency-driven Pipeline DAG Engine using a depends_on model (inspired by Docker Compose / Airflow). This will enable:

1. Parallel Execution: Independent ingestion stages (such as fetch and scrape) running concurrently to dramatically cut down pipeline duration.
2. Flexible Serial/Parallel Configuration: Easily configurable stage dependencies and parallelism rules.
3. Dynamic & Persistent TUI Task Blocks: Replacing static TUI progress bars with an on-demand task queue renderer that shows an initial "No tasks running yet!" state, instantiates task blocks when actions start, and persists completed task blocks in the status section history.
4. Hierarchical Locks & Waiting States: Graceful task queueing with explicit status messaging (e.g., "Waiting for dependency 'fetch'...").

---

2. Key Requirements

A. Declarative Pipeline DAG Configuration (pipeline/config/pipeline_dag.yaml)

Define the pipeline topology declaratively so stages can be configured for serial or parallel execution:

version: "1.0"
stages:
  fetch:
    description: "Fetch raw API news payloads from configured sources"
    depends_on: []
    allow_parallel: true

  scrape:
    description: "Web scrape HTML contents from target URLs"
    depends_on: []
    allow_parallel: true

  generate:
    description: "LLM extraction, classification, and markdown synthesis"
    depends_on:
      - fetch
      - scrape
    allow_parallel: false

  embed:
    description: "Vector embedding generation for RAG retrieval"
    depends_on:
      - generate
    allow_parallel: false

B. Standalone vs. Full Pipeline Queueing

• Direct Action (Standalone): When a user triggers an individual action directly (e.g., clicking Scrape Articles or running python pipeline/cli.py scrape), it executes immediately without enforcing depends_on prerequisites.
• Full Pipeline Run: When Run Pipeline (All) is triggered, the DAG scheduler enqueues all tasks, resolving dependencies dynamically. Tasks with no unfulfilled dependencies (e.g., fetch and scrape) launch concurrently in parallel threads.

C. Hierarchical Locking & Waiting States

• Tasks waiting on unfulfilled dependencies enter a WAITING state.
• In the TUI and logs, waiting tasks explicitly report their blocking dependencies:
  • [WAITING] Stage 'generate' waiting for dependencies: ['fetch', 'scrape']...
• A lock manager ensures atomic state transitions and prevents race conditions between parallel worker threads.

D. Dynamic & Persistent TUI Status Section (pipeline/tui/)

• Initial State: The Pipeline Status section displays a clean placeholder banner:
  No tasks running yet!
• On-Demand Block Creation: When a stage starts (standalone or pipeline), a progress block for that specific task is dynamically mounted to the status panel.
• Persistence: Completed task blocks remain visible in the status panel (displaying ✓ Completed (duration: 12.4s)) rather than disappearing or resetting to empty bars, giving users a persistent history of their pipeline runs.

---

3. Proposed Architecture & Event Lifecycle

                      [ User Action / CLI ]
                                │
                        (Enqueue DAG Stage)
                                │
                                ▼
                       ┌─────────────────┐
                       │   DAG Engine    │
                       └────────┬────────┘
                                │
               ┌────────────────┴────────────────┐
               ▼                                 ▼
       ┌───────────────┐                 ┌───────────────┐
       │  Stage: fetch │ (Parallel)      │ Stage: scrape │
       └───────┬───────┘                 └───────┬───────┘
               │                                 │
               └────────────────┬────────────────┘
                                │ (depends_on)
                                ▼
                       ┌─────────────────┐
                       │ Stage: generate │
                       └────────┬────────┘
                                │ (depends_on)
                                ▼
                       ┌─────────────────┐
                       │  Stage: embed   │
                       └─────────────────┘

New Event Types (pipeline/runner.py):

• TaskQueued(task_id, stage, depends_on)
• TaskWaiting(task_id, stage, blocking_dependencies)
• TaskStarted(task_id, stage, total_items)
• TaskProgress(task_id, stage, current, total, detail)
• TaskCompleted(task_id, stage, duration_seconds)
• TaskFailed(task_id, stage, error_message)

---

4. Implementation Steps

1. DAG Config & Engine (pipeline/dag_runner.py):
   • Implement DAGScheduler to parse stage dependencies, manage hierarchical locks, and execute independent nodes using Python ThreadPoolExecutor.
2. Parallel Stage Execution:
   • Update run_fetch and run_scrape to execute concurrently when enqueued without mutual dependencies.
3. TUI Dynamic Task Renderer (pipeline/tui/app.py):
   • Remove static #pb-fetch, #pb-scrape, #pb-generate, #pb-embed elements from initial compose().
   • Render Static("No tasks running yet!") by default.
   • Dynamically instantiate TaskProgressWidget components on TaskQueued or TaskStarted events.
4. CLI & Unit Testing:
   • Add DAG runner tests in tests/test_dag_runner.py verifying serial, parallel, and dependency locking execution paths.

---

This GitHub issue proposal has been drafted for review prior to code implementation.