🚀 Feat: DataGenFlow extensibility system via dgf CLI

[DiTo97]

Description

The main issue in the DX of DataGenFlow is that engineers have to clone the repo and modify the source code to add blocks and templates, which makes it hard for them to build and maintain their own custom portfolio.

The roadmap for that would be:

• DataGenFlow Docker image (Docker Hub?)
• DataGenFlow CLI (dgf) for blocks and templates registration

This way engineering teams can maintain custom blocks and templates in their own repositories.

The most delicate part is blocks (templates are serializable), which may require third-party dependencies.

The implementation plan will live in a separate file due to character count constraints.

make sure to share feedback before I ask Claude Code to crunch it for us.

[DiTo97]

DataGenFlow Extensibility System: Technical Implementation Plan

Executive Summary

This document outlines the implementation plan for an extensibility system that allows users to register custom blocks and pipeline templates without modifying the DataGenFlow core codebase. The system enables a CLI-driven workflow where users launch the DataGenFlow container, register their extensions via file mounts, and immediately use them in the low-code UI.

Target DX:

# Launch with mounted extension directories
docker-compose up -d

# Add custom block (with dependency handling)
dgf blocks add ./my_sentiment_block.py --install-deps

# Add custom template
dgf templates add ./my_qa_pipeline.yaml

# Open UI → custom block/template available immediately

---

Table of Contents

1. Current State Analysis
2. System Architecture
3. Phase 1: Core Registry Enhancements
4. Phase 2: Hot Reload System
5. Phase 3: Dependency Management
6. Phase 4: CLI SDK
7. Phase 5: UI Integration
8. Phase 6: Docker & Deployment
9. Testing Strategy
10. Migration & Compatibility
11. Security Considerations
12. Timeline & Milestones
13. Open Questions & Decisions

---

1. Current State Analysis

1.1 Block Registration

File: lib/blocks/registry.py

Current behavior:

• BlockRegistry discovers blocks at instantiation via _discover_blocks()
• Scans hardcoded directories: lib/blocks/builtin/, lib/blocks/custom/, user_blocks/
• Singleton pattern: registry = BlockRegistry() at module load
• No runtime registration API
• No external path support
• No dependency awareness

# current implementation
class BlockRegistry:
    def __init__(self) -> None:
        self._blocks: dict[str, type[BaseBlock]] = {}
        self._discover_blocks()  # one-time scan at init

1.2 Template Registration

File: lib/templates/init.py

Current behavior:

• TemplateRegistry loads templates from single directory (lib/templates/)
• Singleton pattern: template_registry = TemplateRegistry() at module load
• No runtime registration API
• No external path support
• No user_templates/ equivalent

1.3 Identified Gaps

Gap	Impact
No runtime registration	Cannot add extensions without restart
No hot reload	Poor development experience
No external paths	Extensions must live in datagenflow repo
No dependency handling	Blocks with extra deps fail silently
No CLI tooling	Manual file copying required
No unavailable state	Broken blocks cause startup failures

---

2. System Architecture

2.1 High-Level Architecture

┌─────────────────────────────────────────────────────────────────────────┐
│  HOST MACHINE                                                            │
│                                                                          │
│  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────────────┐  │
│  │ user_blocks/    │  │ user_templates/ │  │ dgf CLI                 │  │
│  │  sentiment.py   │  │  my_pipeline.yaml│  │  $ dgf blocks add ...  │  │
│  │  translator.py  │  │  custom_qa.yaml │  │  $ dgf templates list  │  │
│  └────────┬────────┘  └────────┬────────┘  └───────────┬─────────────┘  │
│           │                    │                       │                 │
│           │    Docker Volume Mounts                    │ HTTP API        │
│           ▼                    ▼                       ▼                 │
│  ┌──────────────────────────────────────────────────────────────────┐   │
│  │  DOCKER CONTAINER                                                 │   │
│  │                                                                   │   │
│  │  ┌─────────────────────────────────────────────────────────────┐ │   │
│  │  │  /app/user_blocks/          /app/user_templates/            │ │   │
│  │  │  (mounted)                  (mounted)                       │ │   │
│  │  └──────────────────────┬──────────────────────────────────────┘ │   │
│  │                         │                                        │   │
│  │                         ▼                                        │   │
│  │  ┌─────────────────────────────────────────────────────────────┐ │   │
│  │  │  FileWatcher (watchdog)                                     │ │   │
│  │  │  - Monitors user_blocks/ and user_templates/                │ │   │
│  │  │  - Debounces rapid changes                                  │ │   │
│  │  │  - Triggers registry reload                                 │ │   │
│  │  └──────────────────────┬──────────────────────────────────────┘ │   │
│  │                         │                                        │   │
│  │                         ▼                                        │   │
│  │  ┌─────────────────────────────────────────────────────────────┐ │   │
│  │  │  DependencyManager                                          │ │   │
│  │  │  - Parses dependency declarations                           │ │   │
│  │  │  - Checks installed packages                                │ │   │
│  │  │  - Installs missing deps (if enabled)                       │ │   │
│  │  └──────────────────────┬──────────────────────────────────────┘ │   │
│  │                         │                                        │   │
│  │                         ▼                                        │   │
│  │  ┌──────────────────────────────┐  ┌────────────────────────────┐│   │
│  │  │  BlockRegistry               │  │  TemplateRegistry          ││   │
│  │  │  - _blocks: dict             │  │  - _templates: dict        ││   │
│  │  │  - _unavailable: dict        │  │  - reload()                ││   │
│  │  │  - register()                │  │  - register()              ││   │
│  │  │  - reload()                  │  │  - register_directory()    ││   │
│  │  │  - register_unavailable()    │  │                            ││   │
│  │  └──────────────────────────────┘  └────────────────────────────┘│   │
│  │                         │                                        │   │
│  │                         ▼                                        │   │
│  │  ┌─────────────────────────────────────────────────────────────┐ │   │
│  │  │  FastAPI Application                                        │ │   │
│  │  │  GET  /api/blocks         - list all blocks                 │ │   │
│  │  │  GET  /api/templates      - list all templates              │ │   │
│  │  │  POST /api/blocks/reload  - trigger manual reload           │ │   │
│  │  │  GET  /api/extensions/status - extension system health      │ │   │
│  │  └─────────────────────────────────────────────────────────────┘ │   │
│  │                                                                   │   │
│  └───────────────────────────────────────────────────────────────────┘   │
│                                                                          │
└──────────────────────────────────────────────────────────────────────────┘

2.2 Component Diagram

┌─────────────────────────────────────────────────────────────────────┐
│                        lib/blocks/                                   │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────────────┐   │
│  │  base.py     │    │  config.py   │    │  registry.py         │   │
│  │              │    │              │    │                      │   │
│  │  BaseBlock   │◄───│  BlockConfig │◄───│  BlockRegistry       │   │
│  │  BaseMulti-  │    │  Schema      │    │  - register()        │   │
│  │  plierBlock  │    │              │    │  - unregister()      │   │
│  │              │    │              │    │  - reload()          │   │
│  └──────────────┘    └──────────────┘    │  - list_blocks()     │   │
│                                          │  - get_block_class() │   │
│                                          └──────────┬───────────┘   │
│                                                     │               │
│  ┌──────────────────────────────────────────────────┼─────────────┐ │
│  │  NEW FILES                                       │             │ │
│  │                                                  ▼             │ │
│  │  ┌──────────────────┐    ┌──────────────────────────────────┐ │ │
│  │  │  dependencies.py │    │  hot_reload.py                   │ │ │
│  │  │                  │    │                                  │ │ │
│  │  │  - parse_deps()  │◄───│  BlockFileHandler                │ │ │
│  │  │  - check_missing │    │  - on_created()                  │ │ │
│  │  │  - install_deps  │    │  - on_modified()                 │ │ │
│  │  │                  │    │  - on_deleted()                  │ │ │
│  │  └──────────────────┘    │  - _debounced_reload()           │ │ │
│  │                          └──────────────────────────────────┘ │ │
│  └──────────────────────────────────────────────────────────────┘ │
│                                                                      │
└──────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────┐
│                        lib/templates/                                │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │  __init__.py                                                  │   │
│  │                                                               │   │
│  │  TemplateRegistry                                             │   │
│  │  - register()              # NEW                              │   │
│  │  - register_file()         # NEW                              │   │
│  │  - register_directory()    # NEW                              │   │
│  │  - unregister()            # NEW                              │   │
│  │  - reload()                # NEW                              │   │
│  │  - list_templates()                                           │   │
│  │  - get_template()                                             │   │
│  │                                                               │   │
│  └──────────────────────────────────────────────────────────────┘   │
│                                                                      │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │  hot_reload.py  # NEW                                         │   │
│  │                                                               │   │
│  │  TemplateFileHandler                                          │   │
│  │  - on_created()                                               │   │
│  │  - on_modified()                                              │   │
│  │  - on_deleted()                                               │   │
│  │                                                               │   │
│  └──────────────────────────────────────────────────────────────┘   │
│                                                                      │
└──────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────┐
│                        lib/  (new files)                             │
├─────────────────────────────────────────────────────────────────────┤
│                                                                      │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │  file_watcher.py                                              │   │
│  │                                                               │   │
│  │  ExtensionFileWatcher                                         │   │
│  │  - start()                                                    │   │
│  │  - stop()                                                     │   │
│  │  - _setup_observers()                                         │   │
│  │                                                               │   │
│  └──────────────────────────────────────────────────────────────┘   │
│                                                                      │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │  extension_manager.py                                         │   │
│  │                                                               │   │
│  │  ExtensionManager                                             │   │
│  │  - block_registry: BlockRegistry                              │   │
│  │  - template_registry: TemplateRegistry                        │   │
│  │  - file_watcher: ExtensionFileWatcher                         │   │
│  │  - dependency_manager: DependencyManager                      │   │
│  │  - initialize()                                               │   │
│  │  - get_status()                                               │   │
│  │                                                               │   │
│  └──────────────────────────────────────────────────────────────┘   │
│                                                                      │
└──────────────────────────────────────────────────────────────────────┘

2.3 Data Flow: Adding a Block

┌─────────┐     ┌─────────┐     ┌─────────────┐     ┌────────────┐
│  User   │     │  CLI    │     │  Volume     │     │  Container │
└────┬────┘     └────┬────┘     └──────┬──────┘     └──────┬─────┘
     │               │                 │                   │
     │  dgf blocks add ./block.py      │                   │
     │──────────────>│                 │                   │
     │               │                 │                   │
     │               │  Parse deps from file               │
     │               │─────────────────│                   │
     │               │                 │                   │
     │               │  Copy file to volume                │
     │               │────────────────>│                   │
     │               │                 │                   │
     │               │                 │  inotify event    │
     │               │                 │──────────────────>│
     │               │                 │                   │
     │               │                 │     FileWatcher   │
     │               │                 │     detects change│
     │               │                 │        │          │
     │               │                 │        ▼          │
     │               │                 │  DependencyManager│
     │               │                 │  checks/installs  │
     │               │                 │        │          │
     │               │                 │        ▼          │
     │               │                 │  BlockRegistry    │
     │               │                 │  imports & registers
     │               │                 │        │          │
     │               │  GET /api/blocks│        │          │
     │               │────────────────────────────────────>│
     │               │                 │                   │
     │               │  Block list (includes new block)    │
     │               │<────────────────────────────────────│
     │               │                 │                   │
     │  Success: "SentimentBlock" registered               │
     │<──────────────│                 │                   │
     │               │                 │                   │

---

3. Phase 1: Core Registry Enhancements

Goal: Add programmatic registration APIs to both registries without breaking existing behavior.

3.1 BlockRegistry Enhancements

File: registry.py

import importlib
import inspect
import logging
import sys
import threading
from pathlib import Path
from typing import Any

from lib.blocks.base import BaseBlock, BaseMultiplierBlock

logger = logging.getLogger(__name__)


class BlockRegistrationError(Exception):
    pass


class BlockRegistry:
    def __init__(self) -> None:
        self._blocks: dict[str, type[BaseBlock]] = {}
        self._unavailable: dict[str, str] = {}  # block_name -> error_reason
        self._sources: dict[str, str] = {}  # block_name -> source_path
        self._lock = threading.RLock()
        self._discover_blocks()

    def register(
        self, 
        block_class: type[BaseBlock], 
        *, 
        source: str = "programmatic",
        force: bool = False,
    ) -> None:
        self._validate_block_class(block_class)
        name = block_class.__name__
        
        with self._lock:
            if name in self._blocks and not force:
                existing_source = self._sources.get(name, "unknown")
                raise BlockRegistrationError(
                    f"block '{name}' already registered from {existing_source}"
                )
            
            self._blocks[name] = block_class
            self._sources[name] = source
            self._unavailable.pop(name, None)  # clear any unavailable state
            
        logger.info(f"registered block: {name} (source: {source})")

    def register_unavailable(self, name: str, reason: str, source: str = "unknown") -> None:
        with self._lock:
            self._unavailable[name] = reason
            self._sources[name] = source
            self._blocks.pop(name, None)  # remove from available if present
            
        logger.warning(f"block '{name}' marked unavailable: {reason}")

    def unregister(self, block_name: str) -> bool:
        with self._lock:
            removed = self._blocks.pop(block_name, None) is not None
            self._unavailable.pop(block_name, None)
            self._sources.pop(block_name, None)
            
        if removed:
            logger.info(f"unregistered block: {block_name}")
        return removed

    def register_from_module(self, module_path: str) -> list[str]:
        registered = []
        try:
            module = importlib.import_module(module_path)
            for name, obj in inspect.getmembers(module, inspect.isclass):
                if (
                    issubclass(obj, BaseBlock) 
                    and obj not in (BaseBlock, BaseMultiplierBlock)
                    and obj.__module__ == module.__name__
                ):
                    try:
                        self.register(obj, source=module_path)
                        registered.append(name)
                    except BlockRegistrationError as e:
                        logger.warning(str(e))
        except Exception as e:
            logger.error(f"failed to import module {module_path}: {e}")
            raise BlockRegistrationError(f"failed to import {module_path}: {e}") from e
            
        return registered

    def register_from_file(self, file_path: Path) -> list[str]:
        if file_path.name.startswith("_"):
            return []
            
        # determine module name based on location
        module_name = self._file_to_module_name(file_path)
        
        # force reimport if already loaded
        if module_name in sys.modules:
            del sys.modules[module_name]
        
        return self.register_from_module(module_name)

    def register_from_directory(self, dir_path: Path | str) -> list[str]:
        dir_path = Path(dir_path)
        if not dir_path.exists():
            logger.warning(f"directory does not exist: {dir_path}")
            return []
            
        registered = []
        for py_file in dir_path.glob("*.py"):
            try:
                registered.extend(self.register_from_file(py_file))
            except Exception as e:
                logger.warning(f"failed to load {py_file}: {e}")
                
        return registered

    def reload_user_blocks(self) -> dict[str, list[str]]:
        result = {"registered": [], "failed": [], "removed": []}
        user_blocks_path = Path("user_blocks")
        
        if not user_blocks_path.exists():
            return result
        
        # track which blocks we see in this reload
        seen_blocks: set[str] = set()
        
        for py_file in user_blocks_path.glob("*.py"):
            if py_file.name.startswith("_"):
                continue
            try:
                names = self.register_from_file(py_file)
                seen_blocks.update(names)
                result["registered"].extend(names)
            except Exception as e:
                result["failed"].append(f"{py_file.name}: {e}")
        
        # remove blocks that no longer exist
        with self._lock:
            user_block_names = [
                name for name, source in self._sources.items()
                if source.startswith("user_blocks")
            ]
            for name in user_block_names:
                if name not in seen_blocks:
                    self.unregister(name)
                    result["removed"].append(name)
        
        return result

    def get_block_class(self, block_type: str) -> type[BaseBlock] | None:
        return self._blocks.get(block_type)

    def list_blocks(self) -> list[dict[str, Any]]:
        blocks = []
        
        with self._lock:
            # available blocks
            for block_class in self._blocks.values():
                schema = block_class.get_schema()
                schema["available"] = True
                schema["source"] = self._sources.get(block_class.__name__, "unknown")
                blocks.append(schema)
            
            # unavailable blocks
            for name, reason in self._unavailable.items():
                blocks.append({
                    "type": name,
                    "name": name,
                    "available": False,
                    "error": reason,
                    "source": self._sources.get(name, "unknown"),
                })
        
        return blocks

    def _validate_block_class(self, cls: type) -> None:
        if not inspect.isclass(cls):
            raise BlockRegistrationError(f"{cls} is not a class")
        if not issubclass(cls, BaseBlock):
            raise BlockRegistrationError(f"{cls.__name__} must inherit from BaseBlock")
        
        required_attrs = ["name", "description", "inputs", "outputs"]
        for attr in required_attrs:
            if not hasattr(cls, attr):
                raise BlockRegistrationError(
                    f"{cls.__name__} missing required attribute: {attr}"
                )

    def _file_to_module_name(self, file_path: Path) -> str:
        # convert file path to module path
        # user_blocks/sentiment.py -> user_blocks.sentiment
        parts = file_path.with_suffix("").parts
        return ".".join(parts)

    def _discover_blocks(self) -> None:
        scan_dirs = [
            "lib/blocks/builtin",
            "lib/blocks/custom",
            "user_blocks",
        ]
        for blocks_dir in scan_dirs:
            self.register_from_directory(blocks_dir)


# singleton instance
registry = BlockRegistry()

3.2 TemplateRegistry Enhancements

File: init.py

import json
import logging
import threading
from pathlib import Path
from typing import Any

import yaml

logger = logging.getLogger(__name__)


class TemplateRegistrationError(Exception):
    pass


class TemplateRegistry:
    def __init__(self, templates_dir: Path | None = None):
        if templates_dir is None:
            templates_dir = Path(__file__).parent
        self.templates_dir = templates_dir
        self.seeds_dir = templates_dir / "seeds"
        self.user_templates_dir = Path("user_templates")
        
        self._templates: dict[str, dict[str, Any]] = {}
        self._sources: dict[str, str] = {}  # template_id -> source_path
        self._lock = threading.RLock()
        
        self._load_templates()
        self._load_user_templates()

    def register(
        self,
        template_id: str,
        template_data: dict[str, Any],
        *,
        source: str = "programmatic",
        force: bool = False,
    ) -> None:
        self._validate_template(template_data)
        
        with self._lock:
            if template_id in self._templates and not force:
                existing_source = self._sources.get(template_id, "unknown")
                raise TemplateRegistrationError(
                    f"template '{template_id}' already registered from {existing_source}"
                )
            
            self._templates[template_id] = template_data
            self._sources[template_id] = source
            
        logger.info(f"registered template: {template_id} (source: {source})")

    def register_file(self, yaml_path: Path | str) -> str:
        yaml_path = Path(yaml_path)
        if not yaml_path.exists():
            raise TemplateRegistrationError(f"file not found: {yaml_path}")
        
        try:
            with open(yaml_path, "r") as f:
                template_data = yaml.safe_load(f)
        except yaml.YAMLError as e:
            raise TemplateRegistrationError(f"invalid YAML in {yaml_path}: {e}") from e
        
        template_id = yaml_path.stem
        self.register(template_id, template_data, source=str(yaml_path), force=True)
        return template_id

    def register_directory(self, dir_path: Path | str) -> list[str]:
        dir_path = Path(dir_path)
        if not dir_path.exists():
            logger.warning(f"directory does not exist: {dir_path}")
            return []
        
        registered = []
        for yaml_file in dir_path.glob("*.yaml"):
            try:
                template_id = self.register_file(yaml_file)
                registered.append(template_id)
            except TemplateRegistrationError as e:
                logger.warning(str(e))
        
        return registered

    def unregister(self, template_id: str) -> bool:
        with self._lock:
            removed = self._templates.pop(template_id, None) is not None
            self._sources.pop(template_id, None)
        
        if removed:
            logger.info(f"unregistered template: {template_id}")
        return removed

    def reload_user_templates(self) -> dict[str, list[str]]:
        result = {"registered": [], "removed": []}
        
        if not self.user_templates_dir.exists():
            return result
        
        seen_templates: set[str] = set()
        
        for yaml_file in self.user_templates_dir.glob("*.yaml"):
            try:
                template_id = self.register_file(yaml_file)
                seen_templates.add(template_id)
                result["registered"].append(template_id)
            except TemplateRegistrationError as e:
                logger.warning(str(e))
        
        # remove templates that no longer exist
        with self._lock:
            user_template_ids = [
                tid for tid, source in self._sources.items()
                if "user_templates" in source
            ]
            for tid in user_template_ids:
                if tid not in seen_templates:
                    self.unregister(tid)
                    result["removed"].append(tid)
        
        return result

    def list_templates(self) -> list[dict[str, Any]]:
        with self._lock:
            return [
                {
                    "id": template_id,
                    "name": template["name"],
                    "description": template.get("description", ""),
                    "example_seed": template.get("example_seed"),
                    "source": self._sources.get(template_id, "unknown"),
                }
                for template_id, template in self._templates.items()
            ]

    def get_template(self, template_id: str) -> dict[str, Any] | None:
        return self._templates.get(template_id)

    def _validate_template(self, data: dict[str, Any]) -> None:
        required_fields = ["name", "blocks"]
        for field in required_fields:
            if field not in data:
                raise TemplateRegistrationError(f"template missing required field: {field}")
        
        if not isinstance(data["blocks"], list):
            raise TemplateRegistrationError("template 'blocks' must be a list")

    def _load_templates(self) -> None:
        for template_file in self.templates_dir.glob("*.yaml"):
            try:
                self.register_file(template_file)
            except TemplateRegistrationError:
                pass

    def _load_user_templates(self) -> None:
        self.register_directory(self.user_templates_dir)


# singleton instance
template_registry = TemplateRegistry()

3.3 Configuration Updates

File: config.py

# ...existing code...

class Settings:
    # ...existing settings...
    
    # extension system settings
    USER_BLOCKS_PATH: str = os.getenv("DATAGENFLOW_USER_BLOCKS_PATH", "user_blocks")
    USER_TEMPLATES_PATH: str = os.getenv("DATAGENFLOW_USER_TEMPLATES_PATH", "user_templates")
    
    # dependency management
    AUTO_INSTALL_DEPS: bool = os.getenv("DATAGENFLOW_AUTO_INSTALL_DEPS", "false").lower() in ("true", "1", "yes")
    PIP_INDEX_URL: str | None = os.getenv("DATAGENFLOW_PIP_INDEX_URL")
    PIP_EXTRA_ARGS: str = os.getenv("DATAGENFLOW_PIP_EXTRA_ARGS", "")
    
    # hot reload
    ENABLE_HOT_RELOAD: bool = os.getenv("DATAGENFLOW_HOT_RELOAD", "true").lower() in ("true", "1", "yes")
    HOT_RELOAD_DEBOUNCE_MS: int = int(os.getenv("DATAGENFLOW_HOT_RELOAD_DEBOUNCE_MS", "500"))

    @classmethod
    def ensure_extension_dirs(cls) -> None:
        Path(cls.USER_BLOCKS_PATH).mkdir(parents=True, exist_ok=True)
        Path(cls.USER_TEMPLATES_PATH).mkdir(parents=True, exist_ok=True)

3.4 Deliverables

File	Changes
registry.py	Full rewrite with registration API
init.py	Export new classes
init.py	Full rewrite with registration API
config.py	New settings
test_registry.py	Update tests for new API
test_templates.py	Update tests for new API

---

4. Phase 2: Hot Reload System

Goal: Implement file watching to automatically reload blocks and templates when files change.

4.1 File Watcher Module

File: lib/file_watcher.py

import asyncio
import logging
import threading
from pathlib import Path
from typing import Callable

from watchdog.events import FileSystemEvent, FileSystemEventHandler
from watchdog.observers import Observer

from config import settings

logger = logging.getLogger(__name__)


class DebouncedHandler(FileSystemEventHandler):
    def __init__(
        self,
        callback: Callable[[Path, str], None],
        debounce_ms: int = 500,
    ):
        self.callback = callback
        self.debounce_ms = debounce_ms
        self._pending: dict[str, threading.Timer] = {}
        self._lock = threading.Lock()

    def _schedule_callback(self, path: Path, event_type: str) -> None:
        key = str(path)
        
        with self._lock:
            # cancel any pending callback for this path
            if key in self._pending:
                self._pending[key].cancel()
            
            # schedule new callback
            timer = threading.Timer(
                self.debounce_ms / 1000,
                self._execute_callback,
                args=(path, event_type),
            )
            self._pending[key] = timer
            timer.start()

    def _execute_callback(self, path: Path, event_type: str) -> None:
        with self._lock:
            self._pending.pop(str(path), None)
        
        try:
            self.callback(path, event_type)
        except Exception as e:
            logger.exception(f"error in file watcher callback: {e}")

    def on_created(self, event: FileSystemEvent) -> None:
        if not event.is_directory:
            self._schedule_callback(Path(event.src_path), "created")

    def on_modified(self, event: FileSystemEvent) -> None:
        if not event.is_directory:
            self._schedule_callback(Path(event.src_path), "modified")

    def on_deleted(self, event: FileSystemEvent) -> None:
        if not event.is_directory:
            self._schedule_callback(Path(event.src_path), "deleted")


class BlockFileHandler(DebouncedHandler):
    def __init__(self, registry: "BlockRegistry"):
        from lib.blocks.dependencies import DependencyManager
        
        self.registry = registry
        self.dependency_manager = DependencyManager()
        super().__init__(self._handle_block_change, settings.HOT_RELOAD_DEBOUNCE_MS)

    def _handle_block_change(self, path: Path, event_type: str) -> None:
        if not path.suffix == ".py" or path.name.startswith("_"):
            return
        
        logger.info(f"block file {event_type}: {path}")
        
        if event_type == "deleted":
            # find and unregister blocks from this file
            source = str(path)
            to_remove = [
                name for name, src in self.registry._sources.items()
                if src == source
            ]
            for name in to_remove:
                self.registry.unregister(name)
            return
        
        # created or modified - check deps and reload
        try:
            missing_deps = self.dependency_manager.check_file_dependencies(path)
            
            if missing_deps:
                if settings.AUTO_INSTALL_DEPS:
                    logger.info(f"installing dependencies for {path.name}: {missing_deps}")
                    self.dependency_manager.install(missing_deps)
                else:
                    # mark block as unavailable
                    block_name = path.stem  # use filename as placeholder
                    self.registry.register_unavailable(
                        block_name,
                        f"missing dependencies: {', '.join(missing_deps)}",
                        source=str(path),
                    )
                    return
            
            self.registry.register_from_file(path)
            
        except Exception as e:
            logger.error(f"failed to reload block {path}: {e}")
            self.registry.register_unavailable(
                path.stem,
                str(e),
                source=str(path),
            )


class TemplateFileHandler(DebouncedHandler):
    def __init__(self, registry: "TemplateRegistry"):
        self.registry = registry
        super().__init__(self._handle_template_change, settings.HOT_RELOAD_DEBOUNCE_MS)

    def _handle_template_change(self, path: Path, event_type: str) -> None:
        if not path.suffix == ".yaml":
            return
        
        logger.info(f"template file {event_type}: {path}")
        
        if event_type == "deleted":
            template_id = path.stem
            self.registry.unregister(template_id)
            return
        
        try:
            self.registry.register_file(path)
        except Exception as e:
            logger.error(f"failed to reload template {path}: {e}")


class ExtensionFileWatcher:
    def __init__(
        self,
        block_registry: "BlockRegistry",
        template_registry: "TemplateRegistry",
    ):
        self.block_registry = block_registry
        self.template_registry = template_registry
        self._observer: Observer | None = None

    def start(self) -> None:
        if not settings.ENABLE_HOT_RELOAD:
            logger.info("hot reload disabled")
            return
        
        self._observer = Observer()
        
        # watch user_blocks
        user_blocks_path = Path(settings.USER_BLOCKS_PATH)
        if user_blocks_path.exists():
            self._observer.schedule(
                BlockFileHandler(self.block_registry),
                str(user_blocks_path),
                recursive=False,
            )
            logger.info(f"watching {user_blocks_path} for block changes")
        
        # watch user_templates
        user_templates_path = Path(settings.USER_TEMPLATES_PATH)
        if user_templates_path.exists():
            self._observer.schedule(
                TemplateFileHandler(self.template_registry),
                str(user_templates_path),
                recursive=False,
            )
            logger.info(f"watching {user_templates_path} for template changes")
        
        self._observer.start()
        logger.info("extension file watcher started")

    def stop(self) -> None:
        if self._observer:
            self._observer.stop()
            self._observer.join()
            self._observer = None
            logger.info("extension file watcher stopped")

4.2 Application Integration

File: app.py (modifications)

# add imports
from lib.file_watcher import ExtensionFileWatcher

# modify lifespan
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    import os
    import litellm

    await storage.init_db()
    settings.ensure_extension_dirs()

    # start file watcher for hot reload
    file_watcher = ExtensionFileWatcher(registry, template_registry)
    file_watcher.start()
    app.state.file_watcher = file_watcher

    # configure langfuse integration if credentials are set
    if os.getenv("LANGFUSE_PUBLIC_KEY") and os.getenv("LANGFUSE_SECRET_KEY"):
        litellm.success_callback = ["langfuse"]
        logger.info("Langfuse observability enabled")

    yield
    
    # cleanup
    file_watcher.stop()
    await storage.close()


# add reload endpoint
@api_router.post("/extensions/reload")
async def reload_extensions() -> dict[str, Any]:
    block_result = registry.reload_user_blocks()
    template_result = template_registry.reload_user_templates()
    
    return {
        "blocks": block_result,
        "templates": template_result,
    }


@api_router.get("/extensions/status")
async def extensions_status() -> dict[str, Any]:
    blocks = registry.list_blocks()
    templates = template_registry.list_templates()
    
    return {
        "blocks": {
            "total": len(blocks),
            "available": sum(1 for b in blocks if b.get("available", True)),
            "unavailable": sum(1 for b in blocks if not b.get("available", True)),
        },
        "templates": {
            "total": len(templates),
        },
        "hot_reload_enabled": settings.ENABLE_HOT_RELOAD,
        "auto_install_deps": settings.AUTO_INSTALL_DEPS,
    }

4.3 Dependencies

Add to pyproject.toml:

dependencies = [
    # ...existing...
    "watchdog>=3.0.0",
]

4.4 Deliverables

File	Changes
lib/file_watcher.py	New file
app.py	Integrate file watcher, add endpoints
pyproject.toml	Add watchdog dependency
tests/test_file_watcher.py	New tests

---

5. Phase 3: Dependency Management

Goal: Parse, validate, and optionally install dependencies declared in block files.

5.1 Dependency Manager Module

File: lib/blocks/dependencies.py

import importlib.metadata
import logging
import re
import subprocess
import sys
from pathlib import Path

from packaging.requirements import Requirement
from packaging.version import Version

from config import settings

logger = logging.getLogger(__name__)


class DependencyError(Exception):
    pass


class DependencyManager:
    # pattern to match dependency declarations in docstrings
    DOCSTRING_PATTERN = re.compile(
        r"(?:dependencies|requires):\s*\n((?:\s*-\s*[\w\-\[\]<>=!,.]+\s*\n?)+)",
        re.IGNORECASE | re.MULTILINE,
    )
    DEP_LINE_PATTERN = re.compile(r"^\s*-\s*([\w\-\[\]<>=!,.]+)\s*$", re.MULTILINE)

    def parse_file_dependencies(self, file_path: Path) -> list[str]:
        content = file_path.read_text()
        
        deps: list[str] = []
        
        # check for docstring dependencies
        match = self.DOCSTRING_PATTERN.search(content)
        if match:
            deps_block = match.group(1)
            deps.extend(self.DEP_LINE_PATTERN.findall(deps_block))
        
        # check for requirements.txt in same directory
        req_file = file_path.parent / "requirements.txt"
        if req_file.exists():
            for line in req_file.read_text().splitlines():
                line = line.strip()
                if line and not line.startswith("#"):
                    deps.append(line)
        
        # check for per-block requirements file
        block_req_file = file_path.with_suffix(".requirements.txt")
        if block_req_file.exists():
            for line in block_req_file.read_text().splitlines():
                line = line.strip()
                if line and not line.startswith("#"):
                    deps.append(line)
        
        return list(set(deps))  # deduplicate

    def check_file_dependencies(self, file_path: Path) -> list[str]:
        deps = self.parse_file_dependencies(file_path)
        return self.check_missing(deps)

    def check_missing(self, requirements: list[str]) -> list[str]:
        missing = []
        
        for req_str in requirements:
            try:
                req = Requirement(req_str)
                try:
                    installed_version = importlib.metadata.version(req.name)
                    # check if installed version satisfies requirement
                    if req.specifier and not req.specifier.contains(installed_version):
                        missing.append(req_str)
                except importlib.metadata.PackageNotFoundError:
                    missing.append(req_str)
            except Exception as e:
                logger.warning(f"invalid requirement '{req_str}': {e}")
                missing.append(req_str)
        
        return missing

    def install(self, requirements: list[str]) -> None:
        if not requirements:
            return
        
        cmd = [sys.executable, "-m", "pip", "install", "--quiet"]
        
        if settings.PIP_INDEX_URL:
            cmd.extend(["--index-url", settings.PIP_INDEX_URL])
        
        if settings.PIP_EXTRA_ARGS:
            cmd.extend(settings.PIP_EXTRA_ARGS.split())
        
        cmd.extend(requirements)
        
        logger.info(f"installing: {' '.join(requirements)}")
        
        try:
            subprocess.check_call(cmd)
            logger.info("installation complete")
        except subprocess.CalledProcessError as e:
            raise DependencyError(f"pip install failed: {e}") from e

    def get_dependency_info(self, requirements: list[str]) -> list[dict]:
        info = []
        
        for req_str in requirements:
            try:
                req = Requirement(req_str)
                try:
                    installed_version = importlib.metadata.version(req.name)
                    satisfies = (
                        not req.specifier or req.specifier.contains(installed_version)
                    )
                    info.append({
                        "requirement": req_str,
                        "name": req.name,
                        "installed_version": installed_version,
                        "satisfies": satisfies,
                        "status": "ok" if satisfies else "version_mismatch",
                    })
                except importlib.metadata.PackageNotFoundError:
                    info.append({
                        "requirement": req_str,
                        "name": req.name,
                        "installed_version": None,
                        "satisfies": False,
                        "status": "not_installed",
                    })
            except Exception as e:
                info.append({
                    "requirement": req_str,
                    "name": None,
                    "error": str(e),
                    "status": "invalid",
                })
        
        return info

5.2 Dependencies

Add to pyproject.toml:

dependencies = [
    # ...existing...
    "packaging>=23.0",
]

5.3 API Endpoint

File: app.py (additions)

from lib.blocks.dependencies import DependencyManager

dependency_manager = DependencyManager()


@api_router.post("/blocks/check-dependencies")
async def check_block_dependencies(
    file_content: str = Body(..., description="Python file content")
) -> dict[str, Any]:
    import tempfile
    
    # write to temp file for parsing
    with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
        f.write(file_content)
        temp_path = Path(f.name)
    
    try:
        deps = dependency_manager.parse_file_dependencies(temp_path)
        info = dependency_manager.get_dependency_info(deps)
        missing = [d for d in info if d["status"] != "ok"]
        
        return {
            "dependencies": info,
            "missing": [d["requirement"] for d in missing],
            "all_satisfied": len(missing) == 0,
        }
    finally:
        temp_path.unlink()

5.4 Deliverables

File	Changes
lib/blocks/dependencies.py	New file
app.py	Add dependency check endpoint
pyproject.toml	Add packaging dependency
tests/blocks/test_dependencies.py	New tests

---

6. Phase 4: CLI SDK

Goal: Create a standalone CLI tool for managing DataGenFlow extensions.

6.1 Project Structure

datagenflow-cli/
├── pyproject.toml
├── README.md
├── src/
│   └── dgf/
│       ├── __init__.py
│       ├── __main__.py
│       ├── cli.py
│       ├── config.py
│       ├── client.py
│       ├── commands/
│       │   ├── __init__.py
│       │   ├── blocks.py
│       │   ├── templates.py
│       │   ├── image.py
│       │   └── status.py
│       └── utils/
│           ├── __init__.py
│           ├── files.py
│           └── validation.py
└── tests/
    └── ...

6.2 CLI Entry Point

File: src/dgf/cli.py

import typer
from rich.console import Console

from dgf.commands import blocks, templates, image, status
from dgf.config import config

app = typer.Typer(
    name="dgf",
    help="DataGenFlow CLI - Manage blocks and templates",
    no_args_is_help=True,
)
console = Console()

app.add_typer(blocks.app, name="blocks", help="Manage custom blocks")
app.add_typer(templates.app, name="templates", help="Manage pipeline templates")
app.add_typer(image.app, name="image", help="Build custom Docker images")


@app.command()
def status():
    """Show DataGenFlow server status and extension info"""
    from dgf.client import DataGenFlowClient
    
    client = DataGenFlowClient(config.endpoint)
    
    try:
        health = client.health()
        ext_status = client.extensions_status()
        
        console.print(f"[green]✓[/green] Server: {config.endpoint}")
        console.print(f"  Blocks: {ext_status['blocks']['available']} available, "
                      f"{ext_status['blocks']['unavailable']} unavailable")
        console.print(f"  Templates: {ext_status['templates']['total']}")
        console.print(f"  Hot reload: {'enabled' if ext_status['hot_reload_enabled'] else 'disabled'}")
        console.print(f"  Auto-install deps: {'enabled' if ext_status['auto_install_deps'] else 'disabled'}")
    except Exception as e:
        console.print(f"[red]✗[/red] Cannot connect to {config.endpoint}: {e}")
        raise typer.Exit(1)


@app.command()
def configure(
    endpoint: str = typer.Option(None, help="DataGenFlow API endpoint"),
    blocks_path: str = typer.Option(None, help="Path to user_blocks volume mount"),
    templates_path: str = typer.Option(None, help="Path to user_templates volume mount"),
):
    """Configure CLI settings"""
    if endpoint:
        config.endpoint = endpoint
    if blocks_path:
        config.blocks_path = blocks_path
    if templates_path:
        config.templates_path = templates_path
    
    config.save()
    console.print("[green]✓[/green] Configuration saved")


def main():
    app()


if __name__ == "__main__":
    main()

6.3 Blocks Commands

File: src/dgf/commands/blocks.py

from pathlib import Path

import typer
from rich.console import Console
from rich.table import Table

from dgf.client import DataGenFlowClient
from dgf.config import config
from dgf.utils.files import copy_to_volume
from dgf.utils.validation import validate_block_file, parse_block_dependencies

app = typer.Typer()
console = Console()


@app.command("list")
def list_blocks():
    """List all registered blocks"""
    client = DataGenFlowClient(config.endpoint)
    blocks = client.list_blocks()
    
    table = Table(title="Registered Blocks")
    table.add_column("Name", style="cyan")
    table.add_column("Type", style="magenta")
    table.add_column("Category", style="green")
    table.add_column("Status", style="yellow")
    table.add_column("Source", style="dim")
    
    for block in blocks:
        status = "[green]✓[/green]" if block.get("available", True) else f"[red]✗ {block.get('error', '')}[/red]"
        table.add_row(
            block.get("name", block["type"]),
            block["type"],
            block.get("category", "general"),
            status,
            block.get("source", ""),
        )
    
    console.print(table)


@app.command("add")
def add_block(
    path: Path = typer.Argument(..., help="Path to block Python file or directory"),
    install_deps: bool = typer.Option(False, "--install-deps", "-i", help="Install missing dependencies"),
    force: bool = typer.Option(False, "--force", "-f", help="Overwrite existing files"),
):
    """Add a custom block to DataGenFlow"""
    if not path.exists():
        console.print(f"[red]Error:[/red] Path not found: {path}")
        raise typer.Exit(1)
    
    if not config.blocks_path:
        console.print("[red]Error:[/red] blocks_path not configured. Run: dgf configure --blocks-path <path>")
        raise typer.Exit(1)
    
    files_to_add = []
    if path.is_file():
        files_to_add = [path]
    else:
        files_to_add = list(path.glob("*.py"))
    
    for file_path in files_to_add:
        if file_path.name.startswith("_"):
            continue
        
        # validate block file
        console.print(f"Validating {file_path.name}...")
        validation = validate_block_file(file_path)
        
        if not validation["valid"]:
            console.print(f"[red]✗[/red] {file_path.name}: {validation['error']}")
            continue
        
        # check dependencies
        deps = parse_block_dependencies(file_path)
        if deps:
            client = DataGenFlowClient(config.endpoint)
            dep_info = client.check_dependencies(file_path.read_text())
            
            if dep_info["missing"]:
                console.print(f"  Missing dependencies: {', '.join(dep_info['missing'])}")
                if not install_deps:
                    console.print("  [yellow]Warning:[/yellow] Block may not load. Use --install-deps to install.")
        
        # copy to volume
        dest = Path(config.blocks_path) / file_path.name
        if dest.exists() and not force:
            console.print(f"[yellow]Skipping[/yellow] {file_path.name} (exists, use --force to overwrite)")
            continue
        
        copy_to_volume(file_path, dest)
        console.print(f"[green]✓[/green] Added {file_path.name}")
        
        # also copy requirements file if exists
        req_file = file_path.with_suffix(".requirements.txt")
        if req_file.exists():
            copy_to_volume(req_file, Path(config.blocks_path) / req_file.name)


@app.command("remove")
def remove_block(
    name: str = typer.Argument(..., help="Block filename (without .py) or block class name"),
):
    """Remove a custom block"""
    if not config.blocks_path:
        console.print("[red]Error:[/red] blocks_path not configured")
        raise typer.Exit(1)
    
    blocks_dir = Path(config.blocks_path)
    
    # try filename first
    block_file = blocks_dir / f"{name}.py"
    if not block_file.exists():
        # search for file containing the class name
        for py_file in blocks_dir.glob("*.py"):
            content = py_file.read_text()
            if f"class {name}" in content:
                block_file = py_file
                break
    
    if not block_file.exists():
        console.print(f"[red]Error:[/red] Block not found: {name}")
        raise typer.Exit(1)
    
    block_file.unlink()
    console.print(f"[green]✓[/green] Removed {block_file.name}")


@app.command("deps")
def check_deps(
    path: Path = typer.Argument(..., help="Path to block Python file"),
):
    """Check dependencies for a block file"""
    if not path.exists():
        console.print(f"[red]Error:[/red] File not found: {path}")
        raise typer.Exit(1)
    
    deps = parse_block_dependencies(path)
    
    if not deps:
        console.print("No dependencies declared")
        return
    
    client = DataGenFlowClient(config.endpoint)
    dep_info = client.check_dependencies(path.read_text())
    
    table = Table(title=f"Dependencies for {path.name}")
    table.add_column("Requirement", style="cyan")
    table.add_column("Installed", style="magenta")
    table.add_column("Status", style="green")
    
    for dep in dep_info["dependencies"]:
        status_icon = {
            "ok": "[green]✓[/green]",
            "not_installed": "[red]✗ not installed[/red]",
            "version_mismatch": "[yellow]! version mismatch[/yellow]",
            "invalid": "[red]✗ invalid[/red]",
        }.get(dep["status"], dep["status"])
        
        table.add_row(
            dep["requirement"],
            dep.get("installed_version") or "-",
            status_icon,
        )
    
    console.print(table)


@app.command("validate")
def validate_block(
    path: Path = typer.Argument(..., help="Path to block Python file"),
):
    """Validate a block file without adding it"""
    if not path.exists():
        console.print(f"[red]Error:[/red] File not found: {path}")
        raise typer.Exit(1)
    
    validation = validate_block_file(path)
    
    if validation["valid"]:
        console.print(f"[green]✓[/green] {path.name} is valid")
        console.print(f"  Blocks found: {', '.join(validation['blocks'])}")
    else:
        console.print(f"[red]✗[/red] {path.name} is invalid: {validation['error']}")
        raise typer.Exit(1)


@app.command("scaffold")
def scaffold_block(
    name: str = typer.Argument(..., help="Block class name (e.g., SentimentAnalyzer)"),
    output: Path = typer.Option(Path("."), "-o", "--output", help="Output directory"),
    category: str = typer.Option("general", "-c", "--category", help="Block category"),
):
    """Generate a block template file"""
    filename = _camel_to_snake(name) + ".py"
    output_path = output / filename
    
    template = f'''"""
Block: {name}
Dependencies:
  # - some-package>=1.0.0
"""
from typing import Any

from lib.blocks.base import BaseBlock
from lib.entities.block_execution_context import BlockExecutionContext


class {name}(BaseBlock):
    name = "{_camel_to_spaces(name)}"
    description = "TODO: Add description"
    category = "{category}"
    inputs = ["text"]  # TODO: Define inputs
    outputs = ["result"]  # TODO: Define outputs

    def __init__(self, param: str = "default"):
        self.param = param

    async def execute(self, context: BlockExecutionContext) -> dict[str, Any]:
        text = context.get_state("text", "")
        
        # TODO: Implement block logic
        result = text
        
        return {{"result": result}}
'''
    
    output_path.write_text(template)
    console.print(f"[green]✓[/green] Created {output_path}")


def _camel_to_snake(name: str) -> str:
    import re
    return re.sub(r"(?<!^)(?=[A-Z])", "_", name).lower()


def _camel_to_spaces(name: str) -> str:
    import re
    return re.sub(r"(?<!^)(?=[A-Z])", " ", name)

6.4 Templates Commands

File: src/dgf/commands/templates.py

from pathlib import Path

import typer
from rich.console import Console
from rich.table import Table

from dgf.client import DataGenFlowClient
from dgf.config import config
from dgf.utils.files import copy_to_volume

app = typer.Typer()
console = Console()


@app.command("list")
def list_templates():
    """List all available templates"""
    client = DataGenFlowClient(config.endpoint)
    templates = client.list_templates()
    
    table = Table(title="Available Templates")
    table.add_column("ID", style="cyan")
    table.add_column("Name", style="magenta")
    table.add_column("Description", style="green")
    table.add_column("Source", style="dim")
    
    for template in templates:
        table.add_row(
            template["id"],
            template["name"],
            template.get("description", "")[:50],
            template.get("source", ""),
        )
    
    console.print(table)


@app.command("add")
def add_template(
    path: Path = typer.Argument(..., help="Path to template YAML file or directory"),
    force: bool = typer.Option(False, "--force", "-f", help="Overwrite existing files"),
):
    """Add a custom template to DataGenFlow"""
    if not path.exists():
        console.print(f"[red]Error:[/red] Path not found: {path}")
        raise typer.Exit(1)
    
    if not config.templates_path:
        console.print("[red]Error:[/red] templates_path not configured. Run: dgf configure --templates-path <path>")
        raise typer.Exit(1)
    
    files_to_add = []
    if path.is_file():
        files_to_add = [path]
    else:
        files_to_add = list(path.glob("*.yaml")) + list(path.glob("*.yml"))
    
    for file_path in files_to_add:
        # validate template
        console.print(f"Validating {file_path.name}...")
        validation = validate_template_file(file_path)
        
        if not validation["valid"]:
            console.print(f"[red]✗[/red] {file_path.name}: {validation['error']}")
            continue
        
        # copy to volume
        dest = Path(config.templates_path) / file_path.name
        if dest.exists() and not force:
            console.print(f"[yellow]Skipping[/yellow] {file_path.name} (exists, use --force to overwrite)")
            continue
        
        copy_to_volume(file_path, dest)
        console.print(f"[green]✓[/green] Added {file_path.name}")


@app.command("remove")
def remove_template(
    name: str = typer.Argument(..., help="Template ID (filename without .yaml)"),
):
    """Remove a custom template"""
    if not config.templates_path:
        console.print("[red]Error:[/red] templates_path not configured")
        raise typer.Exit(1)
    
    templates_dir = Path(config.templates_path)
    template_file = templates_dir / f"{name}.yaml"
    
    if not template_file.exists():
        template_file = templates_dir / f"{name}.yml"
    
    if not template_file.exists():
        console.print(f"[red]Error:[/red] Template not found: {name}")
        raise typer.Exit(1)
    
    template_file.unlink()
    console.print(f"[green]✓[/green] Removed {template_file.name}")


@app.command("validate")
def validate_template(
    path: Path = typer.Argument(..., help="Path to template YAML file"),
):
    """Validate a template file without adding it"""
    if not path.exists():
        console.print(f"[red]Error:[/red] File not found: {path}")
        raise typer.Exit(1)
    
    validation = validate_template_file(path)
    
    if validation["valid"]:
        console.print(f"[green]✓[/green] {path.name} is valid")
        console.print(f"  Name: {validation['name']}")
        console.print(f"  Blocks: {validation['block_count']}")
    else:
        console.print(f"[red]✗[/red] {path.name} is invalid: {validation['error']}")
        raise typer.Exit(1)


@app.command("scaffold")
def scaffold_template(
    name: str = typer.Argument(..., help="Template name"),
    output: Path = typer.Option(Path("."), "-o", "--output", help="Output directory"),
):
    """Generate a template YAML file"""
    filename = name.lower().replace(" ", "_") + ".yaml"
    output_path = output / filename
    
    template = f'''name: "{name}"
description: "TODO: Add description"

# example seed data structure
example_seed:
  text: "Sample input text"

# pipeline blocks
blocks:
  - type: TextGenerator
    name: "Generate Content"
    config:
      model: "gpt-4o-mini"
      prompt_template: |
        Process the following text:
        {{{{ text }}}}
    inputs:
      text: "{{{{ seed.text }}}}"
    outputs:
      generated: "result"
'''
    
    output_path.write_text(template)
    console.print(f"[green]✓[/green] Created {output_path}")


def validate_template_file(path: Path) -> dict:
    """Validate a template YAML file"""
    import yaml
    
    try:
        with open(path) as f:
            data = yaml.safe_load(f)
        
        if not isinstance(data, dict):
            return {"valid": False, "error": "Template must be a YAML mapping"}
        
        if "name" not in data:
            return {"valid": False, "error": "Missing required field: name"}
        
        if "blocks" not in data:
            return {"valid": False, "error": "Missing required field: blocks"}
        
        if not isinstance(data["blocks"], list):
            return {"valid": False, "error": "blocks must be a list"}
        
        if len(data["blocks"]) == 0:
            return {"valid": False, "error": "blocks list cannot be empty"}
        
        # validate each block has a type
        for i, block in enumerate(data["blocks"]):
            if "type" not in block:
                return {"valid": False, "error": f"Block {i} missing 'type' field"}
        
        return {
            "valid": True,
            "name": data["name"],
            "block_count": len(data["blocks"]),
        }
    
    except yaml.YAMLError as e:
        return {"valid": False, "error": f"Invalid YAML: {e}"}
    except Exception as e:
        return {"valid": False, "error": str(e)}

6.5 Image Commands

File: src/dgf/commands/image.py

from pathlib import Path

import typer
from rich.console import Console

from dgf.config import config
from dgf.utils.validation import parse_block_dependencies

app = typer.Typer()
console = Console()


@app.command("scaffold")
def scaffold_dockerfile(
    blocks_dir: Path = typer.Option(None, "--blocks-dir", "-b", help="Directory containing blocks"),
    output: Path = typer.Option(Path("Dockerfile.custom"), "-o", "--output", help="Output Dockerfile path"),
    base_image: str = typer.Option("datagenflow:latest", "--base", help="Base image"),
):
    """Generate a Dockerfile for custom image with dependencies"""
    deps: set[str] = set()
    
    if blocks_dir and blocks_dir.exists():
        for py_file in blocks_dir.glob("*.py"):
            if py_file.name.startswith("_"):
                continue
            file_deps = parse_block_dependencies(py_file)
            deps.update(file_deps)
        
        # also check for requirements.txt
        req_file = blocks_dir / "requirements.txt"
        if req_file.exists():
            for line in req_file.read_text().splitlines():
                line = line.strip()
                if line and not line.startswith("#"):
                    deps.add(line)
    
    dockerfile_content = f'''# Auto-generated Dockerfile for DataGenFlow with custom dependencies
# Generated by: dgf image scaffold

FROM {base_image}

# Install custom block dependencies
'''
    
    if deps:
        deps_str = " \\\n    ".join(sorted(deps))
        dockerfile_content += f'''RUN pip install --no-cache-dir \\
    {deps_str}
'''
    else:
        dockerfile_content += "# No additional dependencies detected\n"
    
    dockerfile_content += '''
# User blocks and templates are mounted at runtime:
#   -v ./my-blocks:/app/user_blocks
#   -v ./my-templates:/app/user_templates
'''
    
    output.write_text(dockerfile_content)
    console.print(f"[green]✓[/green] Created {output}")
    
    if deps:
        console.print(f"  Dependencies included: {len(deps)}")
        for dep in sorted(deps):
            console.print(f"    - {dep}")
    
    console.print(f"\nBuild with: docker build -f {output} -t my-datagenflow:latest .")


@app.command("build")
def build_image(
    dockerfile: Path = typer.Option(Path("Dockerfile.custom"), "-f", "--dockerfile", help="Dockerfile path"),
    tag: str = typer.Option("my-datagenflow:latest", "-t", "--tag", help="Image tag"),
    no_cache: bool = typer.Option(False, "--no-cache", help="Build without cache"),
):
    """Build a custom Docker image"""
    import subprocess
    
    if not dockerfile.exists():
        console.print(f"[red]Error:[/red] Dockerfile not found: {dockerfile}")
        console.print("Run 'dgf image scaffold' first to generate a Dockerfile")
        raise typer.Exit(1)
    
    cmd = ["docker", "build", "-f", str(dockerfile), "-t", tag]
    if no_cache:
        cmd.append("--no-cache")
    cmd.append(".")
    
    console.print(f"Building image: {tag}")
    console.print(f"Command: {' '.join(cmd)}\n")
    
    try:
        subprocess.run(cmd, check=True)
        console.print(f"\n[green]✓[/green] Successfully built {tag}")
    except subprocess.CalledProcessError as e:
        console.print(f"\n[red]✗[/red] Build failed with exit code {e.returncode}")
        raise typer.Exit(1)

6.6 CLI Client

File: src/dgf/client.py

from typing import Any
import httpx


class DataGenFlowClient:
    def __init__(self, endpoint: str, timeout: float = 30.0):
        self.endpoint = endpoint.rstrip("/")
        self.timeout = timeout
    
    def _request(self, method: str, path: str, **kwargs) -> Any:
        url = f"{self.endpoint}{path}"
        with httpx.Client(timeout=self.timeout) as client:
            response = client.request(method, url, **kwargs)
            response.raise_for_status()
            return response.json()
    
    def health(self) -> dict[str, Any]:
        return self._request("GET", "/health")
    
    def list_blocks(self) -> list[dict[str, Any]]:
        return self._request("GET", "/api/blocks")
    
    def list_templates(self) -> list[dict[str, Any]]:
        return self._request("GET", "/api/templates")
    
    def extensions_status(self) -> dict[str, Any]:
        return self._request("GET", "/api/extensions/status")
    
    def reload_extensions(self) -> dict[str, Any]:
        return self._request("POST", "/api/extensions/reload")
    
    def check_dependencies(self, file_content: str) -> dict[str, Any]:
        return self._request(
            "POST",
            "/api/blocks/check-dependencies",
            json={"file_content": file_content},
        )

6.7 CLI Configuration

File: src/dgf/config.py

from pathlib import Path
from typing import Any
import json


class CLIConfig:
    CONFIG_FILE = Path.home() / ".dgf" / "config.json"
    
    def __init__(self):
        self._config: dict[str, Any] = self._load()
    
    def _load(self) -> dict[str, Any]:
        if self.CONFIG_FILE.exists():
            try:
                return json.loads(self.CONFIG_FILE.read_text())
            except Exception:
                return {}
        return {}
    
    def save(self) -> None:
        self.CONFIG_FILE.parent.mkdir(parents=True, exist_ok=True)
        self.CONFIG_FILE.write_text(json.dumps(self._config, indent=2))
    
    @property
    def endpoint(self) -> str:
        return self._config.get("endpoint", "http://localhost:8000")
    
    @endpoint.setter
    def endpoint(self, value: str) -> None:
        self._config["endpoint"] = value
    
    @property
    def blocks_path(self) -> str | None:
        return self._config.get("blocks_path")
    
    @blocks_path.setter
    def blocks_path(self, value: str) -> None:
        self._config["blocks_path"] = value
    
    @property
    def templates_path(self) -> str | None:
        return self._config.get("templates_path")
    
    @templates_path.setter
    def templates_path(self, value: str) -> None:
        self._config["templates_path"] = value


config = CLIConfig()

6.8 CLI Utilities

File: src/dgf/utils/files.py

from pathlib import Path
import shutil


def copy_to_volume(src: Path, dest: Path) -> None:
    dest.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(src, dest)

File: src/dgf/utils/validation.py

from pathlib import Path
import ast
import re


def validate_block_file(path: Path) -> dict:
    """Validate a Python block file"""
    try:
        content = path.read_text()
        tree = ast.parse(content)
        
        blocks_found = []
        for node in ast.walk(tree):
            if isinstance(node, ast.ClassDef):
                # check if class has BaseBlock in bases
                for base in node.bases:
                    base_name = ""
                    if isinstance(base, ast.Name):
                        base_name = base.id
                    elif isinstance(base, ast.Attribute):
                        base_name = base.attr
                    
                    if "BaseBlock" in base_name or "Block" in base_name:
                        blocks_found.append(node.name)
                        break
        
        if not blocks_found:
            return {"valid": False, "error": "No block classes found (must inherit from BaseBlock)"}
        
        return {"valid": True, "blocks": blocks_found}
    
    except SyntaxError as e:
        return {"valid": False, "error": f"Syntax error: {e}"}
    except Exception as e:
        return {"valid": False, "error": str(e)}


def parse_block_dependencies(path: Path) -> list[str]:
    """Parse dependencies from a block file"""
    content = path.read_text()
    deps = []
    
    # pattern for docstring dependencies
    pattern = re.compile(
        r"(?:dependencies|requires):\s*\n((?:\s*-\s*[\w\-\[\]<>=!,.]+\s*\n?)+)",
        re.IGNORECASE | re.MULTILINE,
    )
    dep_line = re.compile(r"^\s*-\s*([\w\-\[\]<>=!,.]+)\s*$", re.MULTILINE)
    
    match = pattern.search(content)
    if match:
        deps_block = match.group(1)
        deps.extend(dep_line.findall(deps_block))
    
    # check for per-file requirements
    req_file = path.with_suffix(".requirements.txt")
    if req_file.exists():
        for line in req_file.read_text().splitlines():
            line = line.strip()
            if line and not line.startswith("#"):
                deps.append(line)
    
    return list(set(deps))

6.9 CLI Package Configuration

File: datagenflow-cli/pyproject.toml

[project]
name = "datagenflow-cli"
version = "0.1.0"
description = "CLI tool for managing DataGenFlow extensions"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
    "typer>=0.9.0",
    "rich>=13.0.0",
    "httpx>=0.25.0",
    "pyyaml>=6.0",
]

[project.scripts]
dgf = "dgf.cli:main"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/dgf"]

6.10 Deliverables

File	Description
src/dgf/cli.py	Main CLI entry point
src/dgf/config.py	CLI configuration management
src/dgf/client.py	HTTP client for DataGenFlow API
src/dgf/commands/blocks.py	Block management commands
src/dgf/commands/templates.py	Template management commands
src/dgf/commands/image.py	Docker image commands
src/dgf/utils/files.py	File utilities
src/dgf/utils/validation.py	Validation utilities
pyproject.toml	Package configuration
tests/	CLI test suite

---

7. Phase 5: UI Integration

Goal: Update the frontend to display unavailable blocks and extension status.

7.1 Block Palette Updates

File: frontend/src/components/BlockPalette.tsx

interface Block {
  type: string;
  name: string;
  description: string;
  category: string;
  available: boolean;  // NEW
  error?: string;      // NEW
  source?: string;     // NEW
}

function BlockItem({ block }: { block: Block }) {
  const isUnavailable = !block.available;
  
  return (
    <div 
      className={cn(
        "block-item",
        isUnavailable && "opacity-50 cursor-not-allowed"
      )}
      title={isUnavailable ? block.error : block.description}
      draggable={!isUnavailable}
    >
      <div className="flex items-center gap-2">
        <span>{block.name}</span>
        {isUnavailable && (
          <Badge variant="destructive" size="sm">
            Unavailable
          </Badge>
        )}
        {block.source?.includes("user_blocks") && (
          <Badge variant="outline" size="sm">
            Custom
          </Badge>
        )}
      </div>
      {isUnavailable && block.error && (
        <p className="text-xs text-muted-foreground mt-1">
          {block.error}
        </p>
      )}
    </div>
  );
}

7.2 Extension Status Component

File: frontend/src/components/ExtensionStatus.tsx

import { useQuery } from "@tanstack/react-query";
import { fetchExtensionStatus } from "@/services/api";

export function ExtensionStatus() {
  const { data: status } = useQuery({
    queryKey: ["extensionStatus"],
    queryFn: fetchExtensionStatus,
    refetchInterval: 10000,  // poll every 10s
  });
  
  if (!status) return null;
  
  return (
    <div className="extension-status p-4 border rounded-lg">
      <h3 className="font-semibold mb-2">Extensions</h3>
      
      <div className="grid grid-cols-2 gap-4 text-sm">
        <div>
          <span className="text-muted-foreground">Blocks:</span>
          <span className="ml-2">
            {status.blocks.available} available
            {status.blocks.unavailable > 0 && (
              <span className="text-destructive ml-1">
                ({status.blocks.unavailable} unavailable)
              </span>
            )}
          </span>
        </div>
        
        <div>
          <span className="text-muted-foreground">Templates:</span>
          <span className="ml-2">{status.templates.total}</span>
        </div>
        
        <div>
          <span className="text-muted-foreground">Hot Reload:</span>
          <span className={cn(
            "ml-2",
            status.hot_reload_enabled ? "text-green-500" : "text-muted-foreground"
          )}>
            {status.hot_reload_enabled ? "Enabled" : "Disabled"}
          </span>
        </div>
        
        <div>
          <span className="text-muted-foreground">Auto-install Deps:</span>
          <span className={cn(
            "ml-2",
            status.auto_install_deps ? "text-green-500" : "text-muted-foreground"
          )}>
            {status.auto_install_deps ? "Enabled" : "Disabled"}
          </span>
        </div>
      </div>
    </div>
  );
}

7.3 API Service Updates

File: frontend/src/services/api.ts

export async function fetchExtensionStatus(): Promise<ExtensionStatus> {
  const response = await fetch(`${API_BASE}/api/extensions/status`);
  return response.json();
}

export async function reloadExtensions(): Promise<ReloadResult> {
  const response = await fetch(`${API_BASE}/api/extensions/reload`, {
    method: "POST",
  });
  return response.json();
}

interface ExtensionStatus {
  blocks: {
    total: number;
    available: number;
    unavailable: number;
  };
  templates: {
    total: number;
  };
  hot_reload_enabled: boolean;
  auto_install_deps: boolean;
}

7.4 Deliverables

File	Changes
BlockPalette.tsx	Show unavailable blocks with badges
ExtensionStatus.tsx	New component
api.ts	New API functions
types/index.ts	New type definitions

---

8. Phase 6: Docker & Deployment

Goal: Update Docker configuration to support extension volumes.

8.1 Updated Docker Compose

File: docker/docker-compose.yml

version: "3.8"

services:
  datagenflow:
    build:
      context: ..
      dockerfile: docker/Dockerfile
    image: datagenflow:latest
    ports:
      - "8000:8000"
    volumes:
      # extension mounts
      - ./user_blocks:/app/user_blocks
      - ./user_templates:/app/user_templates
      # pip cache for faster dependency installs
      - pip-cache:/root/.cache/pip
      # data persistence
      - ./data:/app/data
    environment:
      # extension settings
      - DATAGENFLOW_USER_BLOCKS_PATH=/app/user_blocks
      - DATAGENFLOW_USER_TEMPLATES_PATH=/app/user_templates
      - DATAGENFLOW_HOT_RELOAD=true
      - DATAGENFLOW_AUTO_INSTALL_DEPS=false
      # llm settings
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

volumes:
  pip-cache:

8.2 Updated Dockerfile

File: docker/Dockerfile

FROM python:3.12-slim

WORKDIR /app

# install system dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
    curl \
    && rm -rf /var/lib/apt/lists/*

# install python dependencies
COPY pyproject.toml ./
RUN pip install --no-cache-dir -e .

# copy application code
COPY . .

# create extension directories
RUN mkdir -p user_blocks user_templates data

# expose port
EXPOSE 8000

# health check
HEALTHCHECK --interval=30s --timeout=10s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

# run application
CMD ["python", "-m", "uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

8.3 Quick Start Script

File: docker/start.sh

#!/bin/bash
set -e

# create directories if they don't exist
mkdir -p user_blocks user_templates data

# check for .env file
if [ ! -f .env ]; then
    echo "Warning: .env file not found. Creating from example..."
    cp .env.example .env
    echo "Please edit .env with your API keys before running."
    exit 1
fi

# start containers
docker-compose up -d

echo ""
echo "DataGenFlow is starting..."
echo ""
echo "  UI:       http://localhost:8000"
echo "  API:      http://localhost:8000/api"
echo ""
echo "Extension directories:"
echo "  Blocks:    ./user_blocks/"
echo "  Templates: ./user_templates/"
echo ""
echo "Use 'dgf' CLI to manage extensions:"
echo "  dgf blocks add ./my_block.py"
echo "  dgf templates add ./my_template.yaml"
echo ""

8.4 Deliverables

File	Changes
docker-compose.yml	Add volume mounts, env vars
Dockerfile	Create extension directories
start.sh	Quick start script
README.md	Update with extension docs

---

9. Testing Strategy

9.1 Unit Tests

Registry Tests:

# tests/blocks/test_registry.py

def test_register_block():
    registry = BlockRegistry()
    registry.register(MockBlock)
    assert "MockBlock" in registry._blocks

def test_register_duplicate_raises():
    registry = BlockRegistry()
    registry.register(MockBlock)
    with pytest.raises(BlockRegistrationError):
        registry.register(MockBlock)

def test_register_with_force():
    registry = BlockRegistry()
    registry.register(MockBlock)
    registry.register(MockBlock, force=True)  # should not raise

def test_register_unavailable():
    registry = BlockRegistry()
    registry.register_unavailable("BrokenBlock", "missing deps")
    assert "BrokenBlock" in registry._unavailable

def test_list_blocks_includes_unavailable():
    registry = BlockRegistry()
    registry.register(MockBlock)
    registry.register_unavailable("BrokenBlock", "error")
    blocks = registry.list_blocks()
    assert len(blocks) == 2
    assert any(b["available"] == False for b in blocks)

Dependency Tests:

# tests/blocks/test_dependencies.py

def test_parse_docstring_dependencies():
    content = '''
    """
    Dependencies:
      - requests>=2.28.0
      - pandas>=1.5.0
    """
    '''
    deps = DependencyManager().parse_file_dependencies(mock_file(content))
    assert "requests>=2.28.0" in deps
    assert "pandas>=1.5.0" in deps

def test_check_missing_returns_uninstalled():
    manager = DependencyManager()
    missing = manager.check_missing(["nonexistent-package-xyz"])
    assert "nonexistent-package-xyz" in missing

def test_check_missing_returns_empty_for_installed():
    manager = DependencyManager()
    missing = manager.check_missing(["pytest"])  # should be installed
    assert missing == []

9.2 Integration Tests

Hot Reload Tests:

# tests/integration/test_hot_reload.py

@pytest.fixture
def temp_blocks_dir(tmp_path):
    blocks_dir = tmp_path / "user_blocks"
    blocks_dir.mkdir()
    return blocks_dir

async def test_file_watcher_detects_new_block(temp_blocks_dir):
    registry = BlockRegistry()
    watcher = ExtensionFileWatcher(registry, mock_template_registry)
    watcher.start()
    
    # write a new block file
    block_file = temp_blocks_dir / "test_block.py"
    block_file.write_text(VALID_BLOCK_CODE)
    
    # wait for debounce
    await asyncio.sleep(1)
    
    assert "TestBlock" in registry._blocks
    watcher.stop()

async def test_file_watcher_removes_deleted_block(temp_blocks_dir):
    # setup: create and register block
    block_file = temp_blocks_dir / "test_block.py"
    block_file.write_text(VALID_BLOCK_CODE)
    
    registry = BlockRegistry()
    registry.register_from_file(block_file)
    
    watcher = ExtensionFileWatcher(registry, mock_template_registry)
    watcher.start()
    
    # delete the file
    block_file.unlink()
    
    await asyncio.sleep(1)
    
    assert "TestBlock" not in registry._blocks
    watcher.stop()

9.3 CLI Tests

# datagenflow-cli/tests/test_blocks_commands.py

from typer.testing import CliRunner
from dgf.cli import app

runner = CliRunner()

def test_blocks_list():
    result = runner.invoke(app, ["blocks", "list"])
    assert result.exit_code == 0

def test_blocks_validate_valid():
    result = runner.invoke(app, ["blocks", "validate", "fixtures/valid_block.py"])
    assert result.exit_code == 0
    assert "is valid" in result.stdout

def test_blocks_validate_invalid():
    result = runner.invoke(app, ["blocks", "validate", "fixtures/invalid_block.py"])
    assert result.exit_code == 1
    assert "is invalid" in result.stdout

def test_blocks_scaffold():
    with runner.isolated_filesystem():
        result = runner.invoke(app, ["blocks", "scaffold", "MyTestBlock"])
        assert result.exit_code == 0
        assert Path("my_test_block.py").exists()

9.4 Test Coverage Goals

Component	Target Coverage
BlockRegistry	95%
TemplateRegistry	95%
DependencyManager	90%
ExtensionFileWatcher	85%
CLI commands	90%
API endpoints	90%

---

10. Migration & Compatibility

10.1 Backward Compatibility

All existing functionality must continue to work:

Feature	Compatibility
user_blocks/ directory	✅ Preserved, now with hot reload
Built-in blocks	✅ Unchanged
Existing templates	✅ Unchanged
API endpoints	✅ Unchanged (new endpoints added)
Pipeline execution	✅ Unchanged

10.2 Migration Steps

For users upgrading from previous versions:

1. No code changes required — existing user_blocks/ continue to work
2. Optional: Add user_templates/ directory for custom templates
3. Optional: Install dgf CLI for improved DX
4. Optional: Update docker-compose.yml with new volume mounts

10.3 Deprecations

None planned for v1.0.

---

11. Security Considerations

11.1 Threat Model

Threat	Risk	Mitigation
Malicious block code	High	User blocks run in same process; document trust requirements
Dependency injection attacks	Medium	AUTO_INSTALL_DEPS=false by default
Path traversal in file operations	Medium	Validate paths stay within expected directories
Arbitrary code execution via API	High	No code execution endpoints exposed
Volume mount attacks	Low	Standard Docker volume security

11.2 Security Defaults

# secure defaults in config.py
AUTO_INSTALL_DEPS = False  # require explicit opt-in
HOT_RELOAD = True          # convenience feature, low risk

11.3 Recommendations

1. Document security model — Users must understand that blocks run with full application privileges
2. Validate block sources — Encourage users to review block code before adding
3. Use custom images for production — Pre-build images with vetted dependencies
4. Network isolation — Run DataGenFlow in isolated network when processing sensitive data

---

12. Timeline & Milestones

12.1 Development Schedule

Phase	Duration	Start	End
Phase 1: Core Registry	1 week	Week 1	Week 1
Phase 2: Hot Reload	1 week	Week 2	Week 2
Phase 3: Dependencies	1 week	Week 3	Week 3
Phase 4: CLI	2 weeks	Week 4	Week 5
Phase 5: UI	1 week	Week 6	Week 6
Phase 6: Docker	1 week	Week 6	Week 6
Testing & Polish	1 week	Week 7	Week 7
Documentation	1 week	Week 7	Week 7

Total: 7 weeks

12.2 Milestones

Milestone	Target Date	Criteria
M1: Registry MVP	End of Week 2	Blocks can be registered/unregistered programmatically
M2: Hot Reload Working	End of Week 3	File changes auto-reload blocks
M3: CLI Alpha	End of Week 5	dgf blocks add and dgf templates add working
M4: Feature Complete	End of Week 6	All features implemented
M5: Release Candidate	End of Week 7	Tests passing, docs complete

12.3 Release Plan

• v0.1.0-alpha — After M3, internal testing
• v0.1.0-beta — After M4, community testing
• v0.1.0 — After M5, stable release

---

13. Open Questions & Decisions

13.1 Open Questions

ID	Question	Options	Recommendation	Status
Q1	Should we support nested directories in user_blocks/?	Yes / No	No (keep simple for v1)	Open
Q2	Should CLI be bundled with main package or separate?	Bundled / Separate	Separate (smaller install)	Decided
Q3	Should we implement entry point discovery?	Yes / No / Later	Later (v1.1)	Decided
Q4	Max file size for hot reload?	None / 1MB / 10MB	10MB with warning	Open
Q5	Should we support YAML blocks (no-code)?	Yes / No / Later	Later (v2.0)	Decided

13.2 Technical Decisions Made

Decision	Rationale
Use watchdog for file watching	Cross-platform, well-maintained, async-compatible
Use typer for CLI	Modern, type-safe, auto-generates help
Use httpx for CLI HTTP client	Async-capable, modern API, good error handling
Debounce file events	Prevent rapid reloads during file saves
Separate CLI package	Keeps main package lean, optional install

13.3 Risks

Risk	Probability	Impact	Mitigation
Hot reload causes memory leaks	Medium	Medium	Test thoroughly, document restart recommendation
Python module reloading edge cases	Medium	Low	Fall back to restart, document limitations
CLI version mismatch with server	Low	Medium	Version check on CLI startup
File watcher misses events	Low	Low	Manual reload endpoint as fallback

---

Appendix A: CLI Command Reference

# Configuration
dgf configure --endpoint http://localhost:8000
dgf configure --blocks-path ./my-blocks
dgf configure --templates-path ./my-templates

# Status
dgf status

# Blocks
dgf blocks list
dgf blocks add ./my_block.py [--install-deps] [--force]
dgf blocks add ./blocks_dir/ [--install-deps] [--force]
dgf blocks remove <name>
dgf blocks validate ./my_block.py
dgf blocks deps ./my_block.py
dgf blocks scaffold MyBlockName [-o ./output] [-c category]

# Templates
dgf templates list
dgf templates add ./my_template.yaml [--force]
dgf templates remove <id>
dgf templates validate ./my_template.yaml
dgf templates scaffold "My Template" [-o ./output]

# Docker Images
dgf image scaffold [--blocks-dir ./blocks] [-o Dockerfile.custom]
dgf image build [-f Dockerfile.custom] [-t my-datagenflow:latest]

---

Appendix B: Environment Variables

Variable	Default	Description
DATAGENFLOW_USER_BLOCKS_PATH	user_blocks	Path to user blocks directory
DATAGENFLOW_USER_TEMPLATES_PATH	user_templates	Path to user templates directory
DATAGENFLOW_HOT_RELOAD	true	Enable file watching for hot reload
DATAGENFLOW_HOT_RELOAD_DEBOUNCE_MS	500	Debounce delay for file events
DATAGENFLOW_AUTO_INSTALL_DEPS	false	Auto-install block dependencies
DATAGENFLOW_PIP_INDEX_URL	(none)	Custom PyPI index URL
DATAGENFLOW_PIP_EXTRA_ARGS	(none)	Extra pip install arguments

---

Appendix C: Block File Format

"""
Block: My Custom Block
Description: Does something useful

Dependencies:
  - requests>=2.28.0
  - pandas>=1.5.0
"""
from typing import Any

from lib.blocks.base import BaseBlock
from lib.entities.block_execution_context import BlockExecutionContext


class MyCustomBlock(BaseBlock):
    name = "My Custom Block"
    description = "Does something useful"
    category = "general"  # generators, validators, metrics, seeders, general, integrations
    inputs = ["text"]
    outputs = ["result"]

    def __init__(self, param: str = "default"):
        self.param = param

    async def execute(self, context: BlockExecutionContext) -> dict[str, Any]:
        text = context.get_state("text", "")
        
        # block logic here
        result = process(text)
        
        return {"result": result}

---

Appendix D: Template File Format

name: "My Pipeline Template"
description: "Generates Q&A pairs from text"

# example seed data for testing
example_seed:
  text: "Sample input text"
  metadata:
    source: "example"

# pipeline definition
blocks:
  - type: TextGenerator
    name: "Generate Questions"
    config:
      model: "gpt-4o-mini"
      temperature: 0.7
      prompt_template: |
        Generate 3 questions about this text:
        {{ text }}
    inputs:
      text: "{{ seed.text }}"
    outputs:
      questions: "generated_questions"

  - type: TextGenerator
    name: "Generate Answers"
    config:
      model: "gpt-4o-mini"
      prompt_template: |
        Answer the following questions based on the text:
        
        Text: {{ text }}
        Questions: {{ questions }}
    inputs:
      text: "{{ seed.text }}"
      questions: "{{ state.generated_questions }}"
    outputs:
      answers: "generated_answers"