
===============================================================================
STABILIZE WORKFLOW ENGINE - COMPLETE REFERENCE FOR CODE GENERATION
===============================================================================

Stabilize is a Python DAG-based workflow orchestration engine. Workflows consist
of Stages (nodes in the DAG) containing Tasks (atomic work units). Stages can
run sequentially or in parallel based on their dependencies.

CRITICAL RULES:

1. NEVER USE IN-MEMORY SQLITE (sqlite:///:memory:)
   Always use disk-based SQLite with a file path:
       WRONG:  SqliteWorkflowStore("sqlite:///:memory:", ...)
       RIGHT:  SqliteWorkflowStore("sqlite:///./stabilize.db", ...)

   In-memory databases lose ALL data on crash/restart, making workflows
   unrecoverable. For production, use PostgreSQL. For development, use
   disk-based SQLite with a proper file path.

2. FOR SHELL WORKFLOWS: Import the BUILT-IN ShellTask:
       from stabilize import ShellTask
       registry.register("shell", ShellTask)

   DO NOT define your own ShellTask class - use the built-in one! It automatically
   substitutes {key} placeholders with upstream outputs (e.g., {stdout} becomes actual output).

===============================================================================
1. COMPLETE WORKING EXAMPLE - COPY THIS AS YOUR STARTING TEMPLATE
===============================================================================

#!/usr/bin/env python3
"""Minimal working Stabilize workflow example."""

from stabilize import (
    # Core models
    Workflow, StageExecution, TaskExecution, WorkflowStatus,
    # Infrastructure (use SqliteWorkflowStore/SqliteQueue or Postgres variants)
    Orchestrator, QueueProcessor,
    Queue, SqliteQueue,                    # Queue abstract + SQLite impl
    WorkflowStore, SqliteWorkflowStore,    # Store abstract + SQLite impl
    # Tasks - use built-in tasks, do NOT define your own
    Task, TaskResult, TaskRegistry,
    ShellTask,      # For shell/command execution
    PythonTask,     # For Python code execution (uses script/INPUT/RESULT)
    DockerTask,     # For Docker container execution
    HTTPTask,       # For HTTP requests
)
from stabilize.events import (
    configure_event_sourcing,  # One-line setup for event sourcing
    SqliteEventStore,          # SQLite event store
    get_event_bus,             # Access the event bus for subscriptions
    EventType,                 # Event type enum for filtering
)


# Step 1: USE BUILT-IN TASKS - Do NOT define your own Task classes!
# Available built-in tasks:
#   - ShellTask: For shell/command execution
#   - PythonTask: For Python code execution (uses script/INPUT/RESULT)
#   - DockerTask: For Docker container execution
#   - HTTPTask: For HTTP API requests
#
# Only define custom Task classes if the built-in tasks don't meet your needs.
# Example custom task (rarely needed):
#
# class MyCustomTask(Task):
#     def execute(self, stage: StageExecution) -> TaskResult:
#         value = stage.context.get("key")
#         return TaskResult.success(outputs={"result": value})


# Step 2: Setup infrastructure with event sourcing
def setup_pipeline_runner(store: WorkflowStore, queue: Queue, db_path: str) -> tuple[QueueProcessor, Orchestrator]:
    """Create processor and orchestrator with task registered."""
    task_registry = TaskRegistry()
    # Register built-in tasks
    task_registry.register("shell", ShellTask)
    task_registry.register("python", PythonTask)
    task_registry.register("docker", DockerTask)
    task_registry.register("http", HTTPTask)

    # Enable event sourcing — all handler events are recorded automatically
    event_store = SqliteEventStore(f"sqlite:///{db_path}", create_tables=True)
    configure_event_sourcing(event_store)

    processor = QueueProcessor(queue, store=store, task_registry=task_registry)

    orchestrator = Orchestrator(queue)
    return processor, orchestrator


# Step 3: Create and run workflow
def main():
    # Initialize storage (disk-based SQLite - ALWAYS use disk, never in-memory)
    db_path = "./stabilize.db"  # Use appropriate path for your project
    store = SqliteWorkflowStore(f"sqlite:///{db_path}", create_tables=True)
    queue = SqliteQueue(f"sqlite:///{db_path}", table_name="queue_messages")
    queue._create_table()
    processor, orchestrator = setup_pipeline_runner(store, queue, db_path)

    # Create workflow with stages using built-in tasks
    workflow = Workflow.create(
        application="my-app",
        name="My Pipeline",
        stages=[
            StageExecution(
                ref_id="1",
                type="shell",  # Use built-in ShellTask
                name="First Stage",
                context={"command": "echo 'Hello World'"},
                tasks=[
                    TaskExecution.create(
                        name="Run Shell",
                        implementing_class="shell",    # Must match registered name
                        stage_start=True,              # REQUIRED for first task
                        stage_end=True,                # REQUIRED for last task
                    ),
                ],
            ),
        ],
    )

    # Execute workflow
    store.store(workflow)
    orchestrator.start(workflow)
    processor.process_all(timeout=30.0)

    # Check result
    result = store.retrieve(workflow.id)
    print(f"Status: {result.status}")
    print(f"Output: {result.stages[0].outputs}")


if __name__ == "__main__":
    main()

===============================================================================
1.1 SHELL PIPELINE TEMPLATE - USE FOR ANY SHELL/COMMAND WORKFLOWS
===============================================================================
For shell commands, IMPORT the built-in ShellTask (do NOT define your own):

from stabilize import ShellTask, TaskRegistry

registry = TaskRegistry()
registry.register("shell", ShellTask)

ShellTask Context Parameters:
  command (str)         - The shell command to execute (required)
  timeout (int)         - Timeout in seconds (default: 60)
  cwd (str)             - Working directory
  env (dict)            - Additional environment variables
  shell (bool|str)      - True for default shell, or path like "/bin/bash"
  stdin (str)           - Input to send to command stdin
  max_output_size (int) - Max bytes for output (default: 10MB)
  expected_codes (list) - Exit codes treated as success (default: [0])
  secrets (list)        - Context keys to mask in logs
  binary (bool)         - Capture output as bytes (default: False)
  continue_on_failure   - Return failed_continue instead of terminal

ShellTask Outputs:
  stdout, stderr, returncode, truncated (bool), stdout_b64 (if binary)

# Example: Pipeline with upstream output substitution
stages=[
    StageExecution(
        ref_id="1", type="shell", name="Get Data",
        context={"command": "git status"},
        tasks=[TaskExecution.create("Run", "shell", stage_start=True, stage_end=True)],
    ),
    StageExecution(
        ref_id="2", type="shell", name="Save Data",
        requisite_stage_ref_ids={"1"},
        context={"command": "echo '{stdout}' > /tmp/output.txt"},  # {stdout} auto-replaced
        tasks=[TaskExecution.create("Save", "shell", stage_start=True, stage_end=True)],
    ),
]

# Example: With environment and working directory
context={"command": "npm install", "cwd": "/app", "env": {"NODE_ENV": "production"}}

# Example: With secrets masking
context={"command": "curl -H 'Auth: {token}' api.com", "token": "secret", "secrets": ["token"]}

# Example: Allow grep's exit code 1 (no match found)
context={"command": "grep pattern file.txt", "expected_codes": [0, 1]}

===============================================================================
1.2 HTTP PIPELINE TEMPLATE - USE FOR ANY HTTP/API WORKFLOWS
===============================================================================
For HTTP requests, IMPORT the built-in HTTPTask (do NOT define your own):

from stabilize import HTTPTask, TaskRegistry

registry = TaskRegistry()
registry.register("http", HTTPTask)

HTTPTask Context Parameters:
  url (str)             - Request URL (required)
  method (str)          - GET, POST, PUT, DELETE, PATCH, HEAD, OPTIONS (default: GET)

  Request Body (mutually exclusive):
    body (str|bytes)    - Raw request body
    json (dict)         - JSON body (auto-serialized, sets Content-Type)
    form (dict)         - Form-encoded body (application/x-www-form-urlencoded)

  Headers & Auth:
    headers (dict)      - Custom request headers
    auth (list)         - Basic auth as [username, password]
    bearer_token (str)  - Bearer token for Authorization header

  File Upload:
    upload_file (str)   - Path to file to upload (multipart/form-data)
    upload_field (str)  - Form field name (default: "file")
    upload_form (dict)  - Additional form fields with upload

  File Download:
    download_to (str)   - Path to save response body

  Timeouts & Retries:
    timeout (int)       - Request timeout in seconds (default: 30)
    retries (int)       - Number of retries (default: 0)
    retry_delay (float) - Delay between retries (default: 1.0)
    retry_on_status (list) - Status codes to retry (default: [502, 503, 504])

  Response Handling:
    expected_status (int|list) - Expected status code(s)
    parse_json (bool)   - Auto-parse JSON response (default: False)
    max_response_size (int) - Max bytes (default: 10MB)

  Other:
    verify_ssl (bool)   - Verify SSL certs (default: True)
    continue_on_failure - Return failed_continue instead of terminal

HTTPTask Outputs:
  status_code, headers, body, body_json (if parse_json), elapsed_ms, url, content_type, content_length

# Example: Simple GET with JSON parsing
context={"url": "https://api.example.com/users", "parse_json": True}

# Example: POST with JSON body
context={"url": "https://api.example.com/users", "method": "POST", "json": {"name": "John"}}

# Example: With Bearer token authentication
context={"url": "https://api.example.com/private", "bearer_token": "my-token"}

# Example: File upload
context={"url": "https://api.example.com/upload", "method": "POST", "upload_file": "/path/to/file.pdf"}

# Example: Download file
context={"url": "https://example.com/report.pdf", "download_to": "/tmp/report.pdf"}

# Example: With retries for unreliable endpoints
context={"url": "https://api.example.com/data", "retries": 3, "retry_delay": 2.0}

===============================================================================
1.3 PYTHON PIPELINE TEMPLATE - USE FOR PYTHON CODE EXECUTION
===============================================================================
For Python code execution, IMPORT the built-in PythonTask (do NOT define your own):

from stabilize import PythonTask, TaskRegistry

registry = TaskRegistry()
registry.register("python", PythonTask)

PythonTask Context Parameters:
  # Execution Mode (choose one):
  script (str)          - Inline Python code to execute
  script_file (str)     - Path to Python script file
  module (str)          - Module path (e.g., "myapp.tasks")
  function (str)        - Function name (required with module)

  # Inputs:
  inputs (dict)         - Input data, available as INPUT in script
  args (list)           - Command line arguments

  # Execution:
  python_path (str)     - Python interpreter (default: current)
  timeout (int)         - Timeout in seconds (default: 60)
  cwd (str)             - Working directory
  env (dict)            - Additional environment variables
  continue_on_failure   - Return failed_continue instead of terminal

PythonTask Outputs:
  stdout (str)            - Script stdout
  stderr (str)            - Script stderr
  exit_code (int)         - Script exit code
  result (Any)            - Value of RESULT variable if set (JSON-serializable)

Script Convention:
  - Access inputs via INPUT dict (includes upstream outputs + explicit inputs)
  - Set return value via RESULT variable (must be JSON-serializable)
  - IMPORTANT: The RESULT variable becomes the 'result' key in outputs!
    So downstream stages access it as: stage.context['result'] or INPUT['result']

# Example: Inline script with INPUT and RESULT
stages=[
    StageExecution(
        ref_id="1", type="python", name="Calculate",
        context={
            "script": """
numbers = INPUT["values"]
RESULT = {"sum": sum(numbers), "avg": sum(numbers) / len(numbers)}
""",
            "inputs": {"values": [1, 2, 3, 4, 5]}
        },
        tasks=[TaskExecution.create("Run", "python", stage_start=True, stage_end=True)],
    ),
    # Stage 1 outputs: {"result": {"sum": 15, "avg": 3.0}, "stdout": "", ...}
    # Stage 2 can access the RESULT via INPUT["result"]
    StageExecution(
        ref_id="2", type="python", name="UseResult",
        requisite_stage_ref_ids={"1"},
        context={
            "script": """
# Access upstream RESULT via INPUT["result"]
prev_result = INPUT["result"]
RESULT = {"doubled_sum": prev_result["sum"] * 2}
"""
        },
        tasks=[TaskExecution.create("Double", "python", stage_start=True, stage_end=True)],
    ),
]

# Example: Module + function mode (calls myapp.validators.validate(INPUT))
context={"module": "myapp.validators", "function": "validate", "inputs": {"data": {...}}}

# Example: Script file
context={"script_file": "/path/to/script.py", "inputs": {"config": {...}}}

===============================================================================
1.4 DOCKER PIPELINE TEMPLATE - USE FOR CONTAINER EXECUTION
===============================================================================
For Docker container execution, IMPORT the built-in DockerTask (do NOT define your own):

from stabilize import DockerTask, TaskRegistry

registry = TaskRegistry()
registry.register("docker", DockerTask)

DockerTask Actions:
  run     - Run a container (default)
  exec    - Execute command in running container
  build   - Build image from Dockerfile
  pull    - Pull image from registry
  ps      - List containers
  images  - List images
  logs    - Get container logs
  stop    - Stop container
  rm      - Remove container

DockerTask Context Parameters (run action):
  image (str)           - Docker image (required)
  command (str|list)    - Command to run in container
  entrypoint (str|list) - Override container entrypoint
  name (str)            - Container name
  user (str)            - Run as user (e.g., "1000:1000")
  hostname (str)        - Container hostname

  # Mounts & Network:
  volumes (list)        - Volume mounts as "host:container"
  ports (list)          - Port mappings as "host:container"
  network (str)         - Docker network name
  dns (list)            - Custom DNS servers
  extra_hosts (list)    - Add host mappings as "host:ip"

  # Environment:
  environment (dict)    - Environment variables
  workdir (str)         - Working directory

  # Resources:
  memory (str)          - Memory limit (e.g., "512m", "2g")
  memory_swap (str)     - Memory + swap limit
  cpus (str)            - CPU limit (e.g., "0.5", "2")
  gpus (str)            - GPU access (e.g., "all", "device=0")
  shm_size (str)        - Shared memory size

  # Security:
  privileged (bool)     - Privileged mode
  cap_add (list)        - Add Linux capabilities
  cap_drop (list)       - Drop Linux capabilities
  security_opt (list)   - Security options
  read_only (bool)      - Read-only root filesystem

  # Other:
  remove (bool)         - Remove after exit (default: True)
  detach (bool)         - Run in background
  init (bool)           - Run init inside container
  platform (str)        - Target platform (e.g., "linux/amd64")
  pull (str)            - Pull policy: "always", "never", "missing"
  labels (dict)         - Container labels
  timeout (int)         - Command timeout (default: 300)
  continue_on_failure   - Return failed_continue instead of terminal

DockerTask Outputs:
  stdout, stderr, exit_code, container_id (if detach), image_id (if build)

# Example: Simple container run
context={"action": "run", "image": "python:3.11", "command": "python -c 'print(1+1)'"}

# Example: With volumes and environment
context={
    "action": "run",
    "image": "node:18",
    "volumes": ["/app:/app"],
    "environment": {"NODE_ENV": "production"},
    "workdir": "/app",
    "command": "npm test"
}

# Example: GPU container with resource limits
context={
    "action": "run",
    "image": "pytorch/pytorch:latest",
    "gpus": "all",
    "memory": "8g",
    "shm_size": "2g",
    "volumes": ["/data:/data"],
    "command": "python train.py"
}

# Example: Build and tag image
context={"action": "build", "tag": "myapp:latest", "context": "./docker"}

===============================================================================
1.5 SSH PIPELINE TEMPLATE - USE FOR REMOTE COMMAND EXECUTION
===============================================================================
For SSH remote commands, IMPORT the built-in SSHTask (do NOT define your own):

from stabilize import SSHTask, TaskRegistry

registry = TaskRegistry()
registry.register("ssh", SSHTask)

SSHTask Context Parameters:
  host (str)              - Remote hostname or IP address (required)
  command (str)           - Command to execute on remote host (required)
  user (str)              - SSH username (default: current user)
  port (int)              - SSH port (default: 22)
  key_file (str)          - Path to private key file (optional)
  timeout (int)           - Command timeout in seconds (default: 60)
  connect_timeout (int)   - SSH connection timeout in seconds (default: 10)
  strict_host_key (bool)  - Strict host key checking (default: False)
  continue_on_failure     - Return failed_continue instead of terminal on error

SSHTask Outputs:
  stdout, stderr, exit_code, host, user

# Example: Simple remote command
context={"host": "server.example.com", "command": "uptime"}

# Example: With key file authentication
context={
    "host": "server.example.com",
    "user": "deploy",
    "key_file": "/home/user/.ssh/deploy_key",
    "command": "systemctl status nginx",
}

# Example: Health check with continue_on_failure
context={
    "host": "server.example.com",
    "command": "test -f /app/healthcheck",
    "continue_on_failure": True,
}

# Example: Non-standard port with longer timeout
context={
    "host": "secure.example.com",
    "port": 2222,
    "command": "df -h",
    "timeout": 120,
}

===============================================================================
1.6 LONG-RUNNING TASKS - SERVERS AND BACKGROUND SERVICES
===============================================================================
For long-running tasks (HTTP servers, background workers, monitors), use the
`restart_on_failure` option to automatically restart when they crash.

CRITICAL CONCEPTS:

1. restart_on_failure vs continue_on_failure:
   - continue_on_failure=True: Task fails → marked FAILED_CONTINUE → NEVER retried
     → workflow continues with dead service (BAD for servers!)
   - restart_on_failure=True: Task fails → raises TransientError → automatically
     retried with exponential backoff → service restarts (CORRECT for servers!)

2. Parallel Processing:
   - process_all() is SYNCHRONOUS - processes ONE message at a time
   - processor.start() enables PARALLEL thread pool processing
   - For multiple long-running tasks, you MUST use processor.start()

ShellTask Long-Running Options:
  restart_on_failure (bool) - If True, raises TransientError on failure to trigger
                              automatic retry with backoff (up to 10 retries)
  timeout (int)             - Long timeout for servers (e.g., 3600 for 1 hour)

# Example: Parallel HTTP Servers with Auto-Restart
#!/usr/bin/env python3
"""Parallel long-running servers with automatic restart on failure."""

from stabilize import (
    Workflow, StageExecution, TaskExecution,
    Orchestrator, QueueProcessor, SqliteQueue, SqliteWorkflowStore,
    TaskRegistry, ShellTask,
)
from stabilize.queue.processor import QueueProcessorConfig
from stabilize.recovery import recover_on_startup
import time


def main():
    db_path = "/tmp/servers_workflow.db"
    store = SqliteWorkflowStore(f"sqlite:///{db_path}", create_tables=True)
    queue = SqliteQueue(f"sqlite:///{db_path}", table_name="queue_messages")
    queue._create_table()

    registry = TaskRegistry()
    registry.register("shell", ShellTask)

    # IMPORTANT: Use QueueProcessorConfig with max_workers for parallel execution
    processor = QueueProcessor(
        queue,
        config=QueueProcessorConfig(max_workers=4, poll_frequency_ms=100),
        store=store,
        task_registry=registry,
    )

    WORKFLOW_ID = "my-servers"

    # Check for existing workflow (recovery scenario)
    if store.exists(WORKFLOW_ID):
        existing = store.retrieve(WORKFLOW_ID)
        if existing.status.is_complete:
            print(f"Previous workflow completed: {existing.status.name}")
            return

        # Recover in-progress workflow
        print("Recovering workflow...")
        recover_on_startup(store, queue, application="demo")
    else:
        # Create new workflow with parallel servers
        workflow = Workflow(
            id=WORKFLOW_ID,
            application="demo",
            name="Parallel Servers",
            stages=[
                StageExecution(
                    ref_id="server1",
                    type="shell",
                    name="Server 1 (port 19001)",
                    context={
                        "command": "cd /tmp/server1 && python3 -m http.server 19001",
                        "timeout": 3600,           # 1 hour timeout
                        "restart_on_failure": True,  # Auto-restart on crash!
                    },
                    requisite_stage_ref_ids=set(),  # No dependencies - runs in parallel
                    tasks=[TaskExecution.create("S1", "shell", stage_start=True, stage_end=True)],
                ),
                StageExecution(
                    ref_id="server2",
                    type="shell",
                    name="Server 2 (port 19002)",
                    context={
                        "command": "cd /tmp/server2 && python3 -m http.server 19002",
                        "timeout": 3600,
                        "restart_on_failure": True,
                    },
                    requisite_stage_ref_ids=set(),  # No dependencies - runs in parallel
                    tasks=[TaskExecution.create("S2", "shell", stage_start=True, stage_end=True)],
                ),
                StageExecution(
                    ref_id="monitor",
                    type="shell",
                    name="Health Monitor",
                    context={
                        "command": "while true; do curl -s localhost:19001 && curl -s localhost:19002; sleep 30; done",
                        "timeout": 3600,
                        "restart_on_failure": True,
                    },
                    requisite_stage_ref_ids=set(),  # No dependencies - runs in parallel
                    tasks=[TaskExecution.create("Mon", "shell", stage_start=True, stage_end=True)],
                ),
            ],
        )
        store.store(workflow)
        Orchestrator(queue).start(workflow)

    # IMPORTANT: Use processor.start() for PARALLEL execution, not process_all()!
    try:
        processor.start()  # Starts thread pool for parallel processing
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("Stopping...")
        processor.stop()
        print("Run again to recover the workflow.")


if __name__ == "__main__":
    main()

Key Points for Long-Running Tasks:
  1. Use restart_on_failure=True for services that must stay running
  2. Use processor.start() not process_all() for parallel execution
  3. Set long timeout values (3600+ seconds)
  4. Use recover_on_startup() to resume after crashes
  5. All parallel stages should have requisite_stage_ref_ids=set()
  6. Kill a server by PID to test restart: kill <PID>, then verify new PID appears

Testing Auto-Restart:
  1. Start the workflow
  2. Find server PID: ps aux | grep "http.server 19001"
  3. Kill it: kill <PID>
  4. Watch logs - server should restart automatically within seconds
  5. Verify new PID: ps aux | grep "http.server 19001"

===============================================================================
2. CORE CLASSES API
===============================================================================

2.1 Workflow
-------------
Factory: Workflow.create(application, name, stages, trigger=None, pipeline_config_id=None)

Fields:
  id: str                    - Unique ULID identifier (auto-generated)
  status: WorkflowStatus     - Current execution status
  stages: list[StageExecution] - All stages in the workflow
  application: str           - Application name
  name: str                  - Pipeline name

Methods:
  stage_by_id(stage_id) -> StageExecution    - Get stage by internal ID
  stage_by_ref_id(ref_id) -> StageExecution  - Get stage by reference ID
  get_context() -> dict                      - Get merged outputs from all stages


2.2 StageExecution
-------------------
Constructor: StageExecution(ref_id, type, name, context, tasks, requisite_stage_ref_ids=set())

Fields:
  ref_id: str                         - UNIQUE reference ID for DAG (e.g., "1", "deploy", "build")
  type: str                           - Stage type (usually matches task name)
  name: str                           - Human-readable name
  context: dict[str, Any]             - INPUT parameters for this stage
  outputs: dict[str, Any]             - OUTPUT values for downstream stages (populated by tasks)
  tasks: list[TaskExecution]          - Tasks to execute (sequentially)
  requisite_stage_ref_ids: set[str]   - Dependencies (ref_ids of upstream stages)
  status: WorkflowStatus              - Current status

DAG Dependencies:
  - Empty set: Stage runs immediately (initial stage)
  - {"A"}: Stage runs after stage with ref_id="A" completes
  - {"A", "B"}: Stage waits for BOTH A and B to complete (join point)


2.3 TaskExecution
------------------
Factory: TaskExecution.create(name, implementing_class, stage_start=False, stage_end=False)

Fields:
  name: str                  - Human-readable task name
  implementing_class: str    - MUST match the name used in TaskRegistry.register()
  stage_start: bool          - MUST be True for first task in stage
  stage_end: bool            - MUST be True for last task in stage
  status: WorkflowStatus     - Current status

CRITICAL: If a stage has only one task, set BOTH stage_start=True AND stage_end=True


2.4 WorkflowStatus
-------------------
All status values:
  NOT_STARTED     - Not yet started
  RUNNING         - Currently executing
  PAUSED          - Paused, can be resumed
  SUSPENDED       - Waiting for external trigger
  SUCCEEDED       - Completed successfully
  FAILED_CONTINUE - Failed but pipeline continues
  TERMINAL        - Failed, pipeline halts
  CANCELED        - Execution was canceled
  STOPPED         - Execution was stopped
  SKIPPED         - Stage/task was skipped
  REDIRECT        - Decision branch redirect
  BUFFERED        - Buffered, waiting

Properties:
  .is_complete: bool    - Has finished executing
  .is_halt: bool        - Blocks downstream stages
  .is_successful: bool  - SUCCEEDED or SKIPPED
  .is_failure: bool     - TERMINAL, STOPPED, or FAILED_CONTINUE

===============================================================================
3. TASK IMPLEMENTATION
===============================================================================

3.1 Task Interface (Abstract Base Class)
-----------------------------------------
from stabilize import Task

class MyTask(Task):
    def execute(self, stage: StageExecution) -> TaskResult:
        # Read from stage.context (includes upstream outputs)
        value = stage.context.get("key")

        # Return TaskResult
        return TaskResult.success(outputs={"output_key": "value"})

    # Optional: Handle timeout (for RetryableTask)
    def on_timeout(self, stage: StageExecution) -> TaskResult | None:
        return TaskResult.terminal(error="Task timed out")

    # Optional: Handle cancellation
    def on_cancel(self, stage: StageExecution) -> TaskResult | None:
        return TaskResult.canceled()


3.2 TaskResult Factory Methods - CRITICAL REFERENCE
----------------------------------------------------
from stabilize import TaskResult

SUCCESS - Task completed successfully, pipeline continues:
    TaskResult.success(outputs=None, context=None)
    Parameters:
      outputs: dict  - Values available to downstream stages
      context: dict  - Values stored in stage.context (stage-scoped)

RUNNING - Task needs to poll again (for RetryableTask):
    TaskResult.running(context=None)
    Parameters:
      context: dict  - Updated state for next poll iteration

TERMINAL - Task failed, pipeline HALTS:
    TaskResult.terminal(error, context=None)
    Parameters:
      error: str     - Error message (REQUIRED)
      context: dict  - Additional context data
    WARNING: Does NOT accept 'outputs' parameter!

FAILED_CONTINUE - Task failed but pipeline continues:
    TaskResult.failed_continue(error, outputs=None, context=None)
    Parameters:
      error: str     - Error message (REQUIRED)
      outputs: dict  - Values still available downstream
      context: dict  - Additional context data

SKIPPED - Task was skipped:
    TaskResult.skipped()

CANCELED - Task was canceled:
    TaskResult.canceled(outputs=None)

STOPPED - Task was stopped:
    TaskResult.stopped(outputs=None)

REDIRECT - Indicates decision branch redirect:
    TaskResult.redirect(context=None)
    Parameters:
      context: dict  - Context for the redirect

JUMP_TO - Dynamic routing to another stage:
    TaskResult.jump_to(target_stage_ref_id, context=None, outputs=None)
    Parameters:
      target_stage_ref_id: str  - ref_id of stage to jump to (REQUIRED)
      context: dict            - Context to merge into target stage
      outputs: dict            - Outputs to preserve

    Use for dynamic flow control, retry loops, and conditional branching.
    The target stage is reset and re-executed with merged context.

    Example - Router Task:
        class RouterTask(Task):
            def execute(self, stage: StageExecution) -> TaskResult:
                if stage.context.get("tests_passed"):
                    return TaskResult.success()
                else:
                    return TaskResult.jump_to(
                        "implement_stage",
                        context={"retry_reason": "tests failed"}
                    )

    Jump count is tracked in execution.context["_jump_count"] to prevent
    infinite loops (default max: 10, configurable via _max_jumps context key).

Builder Pattern (for complex results):
    TaskResult.builder(status).context({...}).outputs({...}).build()

    Methods:
      .context(dict)           - Set the full context
      .outputs(dict)           - Set the full outputs
      .add_context(key, value) - Add a single context value
      .add_output(key, value)  - Add a single output value
      .build()                 - Build and return the TaskResult


3.3 RetryableTask - For Polling Operations
-------------------------------------------
from datetime import timedelta
from stabilize import RetryableTask

class PollTask(RetryableTask):
    def get_timeout(self) -> timedelta:
        """Maximum time before task times out."""
        return timedelta(minutes=30)

    def get_backoff_period(self, stage: StageExecution, duration: timedelta) -> timedelta:
        """Time to wait between poll attempts."""
        return timedelta(seconds=10)

    def execute(self, stage: StageExecution) -> TaskResult:
        status = check_external_system()

        if status == "complete":
            return TaskResult.success(outputs={"status": "done"})
        elif status == "failed":
            return TaskResult.terminal(error="External system failed")
        else:
            # Keep polling - will be called again after backoff
            return TaskResult.running(context={"last_check": time.time()})


3.4 SkippableTask - Conditional Execution
------------------------------------------
from stabilize.tasks.interface import SkippableTask  # Advanced, not in main exports

class ConditionalTask(SkippableTask):
    def is_enabled(self, stage: StageExecution) -> bool:
        """Return False to skip this task."""
        return stage.context.get("should_run", True)

    def do_execute(self, stage: StageExecution) -> TaskResult:
        """Actual task logic (only called if is_enabled returns True)."""
        return TaskResult.success()


3.5 Additional Built-in Tasks
------------------------------
from stabilize.tasks.interface import CallableTask, NoOpTask, WaitTask

OverridableTimeoutRetryableTask:
    A RetryableTask that allows the stage to override timeout via 'stageTimeoutMs'
    context value. Useful when timeout should be configurable per-stage.

CallableTask:
    Wraps a callable function as a task without creating a class.

    def my_task(stage: StageExecution) -> TaskResult:
        return TaskResult.success(outputs={"result": "done"})

    task = CallableTask(my_task)
    registry.register("my_task", task)

NoOpTask:
    A task that does nothing and returns success immediately.
    Useful for testing, placeholder stages, or synchronization points.

    registry.register("noop", NoOpTask)

WaitTask:
    Built-in RetryableTask that waits for a specified duration.
    Reads 'waitTime' (seconds) from stage.context.

    StageExecution(
        ref_id="wait",
        type="wait",
        name="Wait 30 seconds",
        context={"waitTime": 30},
        tasks=[TaskExecution.create("Wait", "wait", stage_start=True, stage_end=True)],
    )

===============================================================================
4. TASK REGISTRY
===============================================================================

from stabilize import TaskRegistry

registry = TaskRegistry()

# Register a task class
registry.register("my_task", MyTask)

# Register with aliases
registry.register("http", HTTPTask, aliases=["http_request", "web_request"])

# The implementing_class in TaskExecution MUST match the registered name:
TaskExecution.create(
    name="Do something",
    implementing_class="my_task",  # Must match registry.register() name
    stage_start=True,
    stage_end=True,
)

===============================================================================
5. DAG PATTERNS
===============================================================================

5.1 Sequential Stages (A -> B -> C)
------------------------------------
stages=[
    StageExecution(ref_id="A", ..., requisite_stage_ref_ids=set()),      # Initial
    StageExecution(ref_id="B", ..., requisite_stage_ref_ids={"A"}),      # After A
    StageExecution(ref_id="C", ..., requisite_stage_ref_ids={"B"}),      # After B
]


5.2 Parallel Stages
--------------------
       A
      / \
     B   C    <- B and C run in parallel after A
      \ /
       D

stages=[
    StageExecution(ref_id="A", ..., requisite_stage_ref_ids=set()),
    StageExecution(ref_id="B", ..., requisite_stage_ref_ids={"A"}),    # Parallel
    StageExecution(ref_id="C", ..., requisite_stage_ref_ids={"A"}),    # Parallel
    StageExecution(ref_id="D", ..., requisite_stage_ref_ids={"B", "C"}), # Join
]


5.3 Complex DAG
----------------
     A
    /|\
   B C D     <- All parallel after A
   |/ \|
   E   F     <- E waits for B,C; F waits for C,D
    \ /
     G       <- G waits for E and F

stages=[
    StageExecution(ref_id="A", ..., requisite_stage_ref_ids=set()),
    StageExecution(ref_id="B", ..., requisite_stage_ref_ids={"A"}),
    StageExecution(ref_id="C", ..., requisite_stage_ref_ids={"A"}),
    StageExecution(ref_id="D", ..., requisite_stage_ref_ids={"A"}),
    StageExecution(ref_id="E", ..., requisite_stage_ref_ids={"B", "C"}),
    StageExecution(ref_id="F", ..., requisite_stage_ref_ids={"C", "D"}),
    StageExecution(ref_id="G", ..., requisite_stage_ref_ids={"E", "F"}),
]

===============================================================================
6. CONTEXT AND OUTPUTS DATA FLOW
===============================================================================

stage.context  - INPUT: Parameters passed when creating the stage
                 Also includes outputs from upstream stages (automatic lookup)

stage.outputs  - OUTPUT: Values produced by tasks for downstream stages
                 Set via TaskResult.success(outputs={...})

Example flow:
  Stage A context: {"input": "hello"}
  Stage A task returns: TaskResult.success(outputs={"result": "processed"})
  Stage B context: {"input": "hello", "result": "processed"}  <- Includes A's output

Accessing in tasks:
  def execute(self, stage):
      # Read from context (includes upstream outputs)
      upstream_result = stage.context.get("result")  # From upstream stage

      # Write to outputs (available downstream)
      return TaskResult.success(outputs={"my_output": "value"})

IMPORTANT - Shell Tasks with Upstream Outputs:
  Use the BUILT-IN ShellTask which automatically substitutes {key} placeholders:

  from stabilize import ShellTask
  registry.register("shell", ShellTask)

  The built-in ShellTask handles: cwd, env, stdin, timeout, expected_codes, secrets, binary mode.
  See section 1.1 for full parameter documentation.

===============================================================================
7. COMMON MISTAKES AND HOW TO FIX THEM
===============================================================================

MISTAKE 1: Using 'outputs' parameter with TaskResult.terminal()
---------------------------------------------------------------
WRONG:
    return TaskResult.terminal(error="Failed", outputs={"data": value})

RIGHT:
    return TaskResult.terminal(error="Failed", context={"data": value})

terminal() only accepts: error (required), context (optional)


MISTAKE 2: Forgetting stage_start and stage_end on tasks
---------------------------------------------------------
WRONG:
    TaskExecution.create(name="X", implementing_class="y")

RIGHT:
    TaskExecution.create(name="X", implementing_class="y", stage_start=True, stage_end=True)


MISTAKE 3: implementing_class doesn't match registered name
------------------------------------------------------------
WRONG:
    registry.register("http_task", HTTPTask)
    TaskExecution.create(..., implementing_class="HTTPTask")  # Class name, not registered name

RIGHT:
    registry.register("http_task", HTTPTask)
    TaskExecution.create(..., implementing_class="http_task")  # Matches registered name


MISTAKE 4: Duplicate ref_id values
-----------------------------------
WRONG:
    StageExecution(ref_id="1", name="Stage A", ...)
    StageExecution(ref_id="1", name="Stage B", ...)  # Same ref_id!

RIGHT:
    StageExecution(ref_id="1", name="Stage A", ...)
    StageExecution(ref_id="2", name="Stage B", ...)  # Unique ref_ids


MISTAKE 5: Missing store or task_registry in QueueProcessor
-------------------------------------------------------------
Both store and task_registry are REQUIRED for auto-registration of handlers:
    processor = QueueProcessor(queue, store=store, task_registry=registry)
Without both, no handlers are registered and the engine won't process messages.


MISTAKE 6: Forgetting requisite_stage_ref_ids for sequential stages
--------------------------------------------------------------------
WRONG - Stages may run in parallel, stage 2 won't have stage 1 outputs:
    StageExecution(ref_id="1", context={"command": "git status"}, ...),
    StageExecution(ref_id="2", context={"command": "echo {stdout}"}, ...),  # No dependency!

RIGHT - Stage 2 waits for stage 1 and receives its outputs:
    StageExecution(ref_id="1", context={"command": "git status"}, ...),
    StageExecution(ref_id="2", requisite_stage_ref_ids={"1"}, context={"command": "echo {stdout}"}, ...),

Without requisite_stage_ref_ids, stages run in parallel and upstream outputs are NOT available.


MISTAKE 7: Using $variable instead of {variable} for upstream outputs
----------------------------------------------------------------------
WRONG - Shell variable syntax doesn't work:
    context={"command": "echo $stdout > file.txt"}  # $stdout is shell variable, not context

RIGHT - Use {key} placeholders that ShellTask substitutes:
    context={"command": "echo '{stdout}' > file.txt"}  # {stdout} replaced by task


MISTAKE 8: Defining your own ShellTask instead of using built-in
-----------------------------------------------------------------
WRONG - Defining custom ShellTask that may lack features:
    class ShellTask(Task):
        def execute(self, stage):
            command = stage.context.get("command")
            result = subprocess.run(command, shell=True, ...)

RIGHT - Use the built-in ShellTask which handles everything:
    from stabilize import ShellTask
    registry.register("shell", ShellTask)


MISTAKE 9: Using in-memory SQLite database
------------------------------------------
WRONG - Data is lost on crash/restart, workflows become unrecoverable:
    store = SqliteWorkflowStore("sqlite:///:memory:", create_tables=True)
    queue = SqliteQueue("sqlite:///:memory:", table_name="queue_messages")

RIGHT - Always use disk-based SQLite with a file path:
    db_path = "./stabilize.db"  # Or absolute path like "/var/lib/stabilize/data.db"
    store = SqliteWorkflowStore(f"sqlite:///{db_path}", create_tables=True)
    queue = SqliteQueue(f"sqlite:///{db_path}", table_name="queue_messages")

For production deployments, use PostgreSQL for better concurrency and reliability.


MISTAKE 10: Accessing PythonTask RESULT incorrectly in downstream stages
------------------------------------------------------------------------
WRONG - Accessing RESULT dict keys directly:
    # Stage 1 sets: RESULT = {"sum": 15}
    # Stage 2 tries: INPUT["sum"]  <- KeyError! "sum" is not a top-level key

RIGHT - Access via the "result" key:
    # Stage 1 sets: RESULT = {"sum": 15}
    # Stage 1 outputs: {"result": {"sum": 15}, "stdout": "", "stderr": "", "exit_code": 0}
    # Stage 2 accesses: INPUT["result"]["sum"]  <- Correct!

The RESULT variable in PythonTask becomes the "result" key in stage outputs.


MISTAKE 11: Using continue_on_failure for long-running servers
--------------------------------------------------------------
WRONG - Server dies and never restarts:
    context={
        "command": "python3 -m http.server 8000",
        "timeout": 3600,
        "continue_on_failure": True,  # BAD: Server crashes → marked FAILED_CONTINUE → dead forever
    }

RIGHT - Server auto-restarts on crash:
    context={
        "command": "python3 -m http.server 8000",
        "timeout": 3600,
        "restart_on_failure": True,  # GOOD: Server crashes → TransientError → retry with backoff
    }

continue_on_failure: Task fails → FAILED_CONTINUE (terminal) → workflow continues, service dead
restart_on_failure: Task fails → TransientError → exponential backoff retry → service restarts


MISTAKE 12: Using process_all() for parallel long-running tasks
---------------------------------------------------------------
WRONG - Server 1 blocks, Server 2 never starts:
    # Server 1 runs for 3600 seconds, process_all waits, Server 2 never gets a turn
    processor.process_all(timeout=3600)

RIGHT - All servers run in parallel:
    processor.start()  # Starts thread pool, processes messages in parallel
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        processor.stop()

process_all() is SYNCHRONOUS - processes ONE message at a time.
processor.start() uses thread pool for PARALLEL processing.
See section 1.6 for complete long-running tasks example.

===============================================================================
8. COMPLETE EXAMPLE: SEQUENTIAL PIPELINE WITH ERROR HANDLING
===============================================================================

#!/usr/bin/env python3
from stabilize import (
    Workflow, StageExecution, TaskExecution, WorkflowStatus,
    Orchestrator, QueueProcessor, SqliteQueue, SqliteWorkflowStore,
    Task, TaskResult, TaskRegistry,
)


class ValidateTask(Task):
    def execute(self, stage: StageExecution) -> TaskResult:
        data = stage.context.get("data")
        if not data:
            return TaskResult.terminal(error="No data provided")
        return TaskResult.success(outputs={"validated": True, "data": data})


class ProcessTask(Task):
    def execute(self, stage: StageExecution) -> TaskResult:
        data = stage.context.get("data")
        validated = stage.context.get("validated")
        if not validated:
            return TaskResult.terminal(error="Data not validated")
        result = data.upper()
        return TaskResult.success(outputs={"processed_data": result})


class NotifyTask(Task):
    def execute(self, stage: StageExecution) -> TaskResult:
        processed = stage.context.get("processed_data")
        # Even if notification fails, we don't want to fail the pipeline
        try:
            send_notification(processed)
            return TaskResult.success(outputs={"notified": True})
        except Exception as e:
            # Use failed_continue to not halt the pipeline
            return TaskResult.failed_continue(
                error=f"Notification failed: {e}",
                outputs={"notified": False}
            )


def setup_pipeline_runner(store, queue, db_path):
    registry = TaskRegistry()
    registry.register("validate", ValidateTask)
    registry.register("process", ProcessTask)
    registry.register("notify", NotifyTask)

    # Enable event sourcing
    event_store = SqliteEventStore(f"sqlite:///{db_path}", create_tables=True)
    configure_event_sourcing(event_store)

    processor = QueueProcessor(queue, store=store, task_registry=registry)

    return processor, Orchestrator(queue)


def main():
    # ALWAYS use disk-based SQLite, never in-memory (data loss on crash)
    db_path = "./stabilize.db"
    store = SqliteWorkflowStore(f"sqlite:///{db_path}", create_tables=True)
    queue = SqliteQueue(f"sqlite:///{db_path}", table_name="queue_messages")
    queue._create_table()
    processor, orchestrator = setup_pipeline_runner(store, queue, db_path)

    workflow = Workflow.create(
        application="data-pipeline",
        name="Process Data",
        stages=[
            StageExecution(
                ref_id="validate",
                type="validate",
                name="Validate Input",
                context={"data": "hello world"},
                tasks=[TaskExecution.create("Validate", "validate", stage_start=True, stage_end=True)],
            ),
            StageExecution(
                ref_id="process",
                type="process",
                name="Process Data",
                requisite_stage_ref_ids={"validate"},
                context={},  # Will receive 'data' from upstream
                tasks=[TaskExecution.create("Process", "process", stage_start=True, stage_end=True)],
            ),
            StageExecution(
                ref_id="notify",
                type="notify",
                name="Send Notification",
                requisite_stage_ref_ids={"process"},
                context={},
                tasks=[TaskExecution.create("Notify", "notify", stage_start=True, stage_end=True)],
            ),
        ],
    )

    store.store(workflow)
    orchestrator.start(workflow)
    processor.process_all(timeout=30.0)

    result = store.retrieve(workflow.id)
    print(f"Final status: {result.status}")
    for stage in result.stages:
        print(f"  {stage.name}: {stage.status} - {stage.outputs}")


if __name__ == "__main__":
    main()

===============================================================================
9. COMPLETE EXAMPLE: PARALLEL STAGES WITH JOIN
===============================================================================

#!/usr/bin/env python3
from stabilize import (
    Workflow, StageExecution, TaskExecution,
    Orchestrator, QueueProcessor, SqliteQueue, SqliteWorkflowStore,
    Task, TaskResult, TaskRegistry,
)


class FetchDataTask(Task):
    def execute(self, stage: StageExecution) -> TaskResult:
        source = stage.context.get("source")
        # Simulate fetching data from different sources
        data = f"data_from_{source}"
        return TaskResult.success(outputs={f"{source}_data": data})


class AggregateTask(Task):
    def execute(self, stage: StageExecution) -> TaskResult:
        # Collect data from all upstream parallel stages
        api_data = stage.context.get("api_data")
        db_data = stage.context.get("db_data")
        cache_data = stage.context.get("cache_data")
        combined = f"{api_data} + {db_data} + {cache_data}"
        return TaskResult.success(outputs={"combined_data": combined})


def setup_pipeline_runner(store, queue, db_path):
    registry = TaskRegistry()
    registry.register("fetch", FetchDataTask)
    registry.register("aggregate", AggregateTask)

    # Enable event sourcing
    event_store = SqliteEventStore(f"sqlite:///{db_path}", create_tables=True)
    configure_event_sourcing(event_store)

    processor = QueueProcessor(queue, store=store, task_registry=registry)

    return processor, Orchestrator(queue)


def main():
    # ALWAYS use disk-based SQLite, never in-memory (data loss on crash)
    db_path = "./stabilize.db"
    store = SqliteWorkflowStore(f"sqlite:///{db_path}", create_tables=True)
    queue = SqliteQueue(f"sqlite:///{db_path}", table_name="queue_messages")
    queue._create_table()
    processor, orchestrator = setup_pipeline_runner(store, queue, db_path)

    #        Start
    #       /  |  \
    #     API  DB  Cache    <- Run in parallel
    #       \  |  /
    #      Aggregate        <- Join point

    workflow = Workflow.create(
        application="parallel-fetch",
        name="Parallel Data Fetch",
        stages=[
            StageExecution(
                ref_id="api",
                type="fetch",
                name="Fetch from API",
                context={"source": "api"},
                tasks=[TaskExecution.create("Fetch API", "fetch", stage_start=True, stage_end=True)],
            ),
            StageExecution(
                ref_id="db",
                type="fetch",
                name="Fetch from Database",
                context={"source": "db"},
                tasks=[TaskExecution.create("Fetch DB", "fetch", stage_start=True, stage_end=True)],
            ),
            StageExecution(
                ref_id="cache",
                type="fetch",
                name="Fetch from Cache",
                context={"source": "cache"},
                tasks=[TaskExecution.create("Fetch Cache", "fetch", stage_start=True, stage_end=True)],
            ),
            StageExecution(
                ref_id="aggregate",
                type="aggregate",
                name="Aggregate Results",
                requisite_stage_ref_ids={"api", "db", "cache"},  # Wait for ALL three
                context={},
                tasks=[TaskExecution.create("Aggregate", "aggregate", stage_start=True, stage_end=True)],
            ),
        ],
    )

    store.store(workflow)
    orchestrator.start(workflow)
    processor.process_all(timeout=30.0)

    result = store.retrieve(workflow.id)
    print(f"Final status: {result.status}")
    print(f"Combined data: {result.stages[-1].outputs.get('combined_data')}")


if __name__ == "__main__":
    main()

===============================================================================
10. COMPLETE IMPORTS REFERENCE
===============================================================================

# RECOMMENDED: Single consolidated import (most common classes)
from stabilize import (
    # Core models
    Workflow, StageExecution, TaskExecution, WorkflowStatus,
    # Infrastructure
    Orchestrator, QueueProcessor, SqliteQueue, SqliteWorkflowStore,
    # Tasks
    Task, RetryableTask, TaskResult, TaskRegistry,
    ShellTask, HTTPTask, DockerTask, SSHTask, PythonTask,
)
# Note: Handlers are auto-registered by QueueProcessor(queue, store=store, task_registry=registry)

# Advanced imports (for specialized use cases)
from stabilize.persistence.store import WorkflowStore      # Abstract base for custom stores
from stabilize.queue import Queue                    # Abstract base for custom queues
from stabilize.tasks.interface import (                    # Advanced task types
    SkippableTask, OverridableTimeoutRetryableTask,
    CallableTask, NoOpTask, WaitTask,
)
from stabilize.tasks.result import TaskResultBuilder       # For complex result building

# Verification System (NEW)
from stabilize.verification import (
    VerifyResult, VerifyStatus, Verifier, OutputVerifier, CallableVerifier,
)

# Structured Conditions (NEW)
from stabilize.conditions import (
    Condition, ConditionSet, ConditionType, ConditionReason,
)

# Assertion Helpers (NEW)
from stabilize.assertions import (
    assert_context, assert_context_type, assert_context_in,
    assert_output, assert_output_type,
    assert_config, assert_verified, assert_true,
    assert_stage_ready, assert_not_none, assert_non_empty,
    ContextError, OutputError, ConfigError, VerificationError,
    PreconditionError, StageNotReadyError,
)

# Configuration Validation (NEW)
from stabilize.validation import (
    validate_context, validate_outputs, is_valid,
    SchemaValidator, ValidationError,
    SHELL_TASK_SCHEMA, WAIT_TASK_SCHEMA,
)

# Error Handling & Reliability (NEW)
from stabilize.errors import (
    TransientError, PermanentError, TaskTimeoutError,
    is_transient, is_permanent,
)
from stabilize.models.status import (
    can_transition, validate_transition, InvalidStateTransitionError,
)
from stabilize.recovery import WorkflowRecovery, recover_on_startup

# Agentic building blocks (NEW) - see Section 20
from stabilize import (
    ApprovalTask, approve, reject, send_signal, get_signal,   # human-in-the-loop
    WorkflowStream, StreamItem, emit_progress,                 # live streaming
    register_reducer,                                          # fan-in reducers (output_reducers)
)
from stabilize.llm import (                                    # LLM agent toolkit (stdlib-only)
    LLMClient, LLMTask, AgentLoopTask, tool, ToolRegistry,
)

===============================================================================
11. VERIFICATION SYSTEM (NEW)
===============================================================================

The verification system validates stage outputs after task completion,
before downstream stages start. This ensures data integrity in pipelines.

11.1 VerifyResult - Verification Result Type
---------------------------------------------
from stabilize.verification import VerifyResult, VerifyStatus

# Create results using factory methods:
VerifyResult.ok(message="All checks passed")           # Verification passed
VerifyResult.retry(message="Still waiting", details={}) # Will retry
VerifyResult.failed(message="Check failed", details={}) # Terminal failure
VerifyResult.skipped(message="Not applicable")          # Skipped

# Check result status:
result.is_ok        # True if verification passed
result.is_retry     # True if should retry
result.is_failed    # True if terminal failure
result.is_terminal  # True if OK, FAILED, or SKIPPED (won't retry)

11.2 OutputVerifier - Check Required Outputs
--------------------------------------------
from stabilize.verification import OutputVerifier

# Verify that specific outputs exist with correct types
verifier = OutputVerifier(
    required_keys=["url", "status_code"],
    type_checks={"status_code": int},
)

class MyTask(Task):
    def execute(self, stage: StageExecution) -> TaskResult:
        # ... task logic ...
        result = verifier.verify(stage)
        if not result.is_ok:
            return TaskResult.terminal(result.message)
        return TaskResult.success(outputs={"url": url, "status_code": 200})

11.3 Custom Verifier
---------------------
from stabilize.verification import Verifier, VerifyResult

class URLVerifier(Verifier):
    def verify(self, stage: StageExecution) -> VerifyResult:
        url = stage.outputs.get("url")
        if not url:
            return VerifyResult.failed("No URL in outputs")

        # Check if URL is reachable
        try:
            response = requests.head(url, timeout=5)
            if response.ok:
                return VerifyResult.ok(f"URL {url} is reachable")
            return VerifyResult.retry(f"URL returned {response.status_code}")
        except Exception as e:
            return VerifyResult.retry(f"URL check failed: {e}")

    @property
    def max_retries(self) -> int:
        return 5  # Override default of 3

    @property
    def retry_delay_seconds(self) -> float:
        return 2.0  # Override default of 1.0

===============================================================================
12. STRUCTURED CONDITIONS (NEW)
===============================================================================

Conditions provide detailed status information with reasons and timestamps,
inspired by Kubernetes conditions.

12.1 Condition - Status with Context
------------------------------------
from stabilize.conditions import Condition, ConditionType, ConditionReason

# Create conditions using factory methods:
Condition.ready(status=True, reason=ConditionReason.TASKS_SUCCEEDED, message="Done")
Condition.progressing(status=True, reason=ConditionReason.IN_PROGRESS)
Condition.verified(status=True, reason=ConditionReason.VERIFICATION_PASSED)
Condition.failed(reason=ConditionReason.TASK_FAILED, message="Task timed out")
Condition.config_valid(status=True)

# Update a condition (immutable - returns new instance)
updated = condition.update(status=False, reason=ConditionReason.IN_PROGRESS)

# Serialize for storage
data = condition.to_dict()  # {"type": "Ready", "status": true, ...}
condition = Condition.from_dict(data)

12.2 ConditionSet - Manage Multiple Conditions
----------------------------------------------
from stabilize.conditions import ConditionSet

conditions = ConditionSet()

# Set/update conditions
conditions.set(Condition.ready(True, ConditionReason.TASKS_SUCCEEDED))
conditions.set(Condition.progressing(False, ConditionReason.STAGE_COMPLETED))

# Quick status checks
conditions.is_ready       # True if Ready condition is True
conditions.is_progressing # True if Progressing condition is True
conditions.is_verified    # True if Verified condition is True
conditions.has_failed     # True if Failed condition exists
conditions.is_config_valid # True if ConfigValid is True (default: True)

# Get specific condition
ready = conditions.get(ConditionType.READY)
if ready:
    print(f"Ready: {ready.status}, Reason: {ready.reason}")

# Serialize
data_list = conditions.to_list()
conditions = ConditionSet.from_list(data_list)

===============================================================================
13. ASSERTION HELPERS (NEW)
===============================================================================

Assertion helpers provide clean error handling with descriptive exceptions.

13.1 Context Assertions
-----------------------
from stabilize.assertions import (
    assert_context, assert_context_type, assert_context_in,
    ContextError,
)

class MyTask(Task):
    def execute(self, stage: StageExecution) -> TaskResult:
        # Assert key exists and get value (raises ContextError if missing)
        api_key = assert_context(stage, "api_key", "API key is required")

        # Assert key exists with specific type
        timeout = assert_context_type(stage, "timeout", int, "Timeout must be int")

        # Assert value is in allowed list
        env = assert_context_in(stage, "env", ["dev", "staging", "prod"])

        # ... rest of task logic
        return TaskResult.success()

13.2 Output Assertions
----------------------
from stabilize.assertions import assert_output, assert_output_type, OutputError

# Assert output exists
result = assert_output(stage, "deployment_id")

# Assert output with type
count = assert_output_type(stage, "item_count", int)

13.3 Configuration & Verification Assertions
--------------------------------------------
from stabilize.assertions import assert_config, assert_verified, ConfigError

# Assert configuration is valid
assert_config(timeout > 0, "Timeout must be positive", field="timeout")

# Assert verification condition
assert_verified(response.ok, "API check failed", details={"status": response.status_code})

13.4 Stage Readiness Assertions
-------------------------------
from stabilize.assertions import assert_stage_ready, assert_no_upstream_failures

# Assert all upstream stages complete
assert_stage_ready(stage, "Cannot start: upstream incomplete")

# Assert no upstream failures
assert_no_upstream_failures(stage)

13.5 General Assertions
-----------------------
from stabilize.assertions import assert_true, assert_not_none, assert_non_empty

assert_true(condition, "Condition not met")
user = assert_not_none(get_user(id), f"User {id} not found")
items = assert_non_empty(stage.context.get("items", []), "Items required")

13.6 Exception Hierarchy
------------------------
StabilizeError (base)
├── StabilizeFatalError (unrecoverable - halts pipeline)
│   ├── ContextError (missing/invalid context)
│   └── ConfigError (invalid configuration)
└── StabilizeExpectedError (may allow retry)
    ├── PreconditionError (general precondition)
    ├── OutputError (missing/invalid output)
    ├── VerificationError (verification failed)
    └── StageNotReadyError (upstream incomplete)

===============================================================================
14. CONFIGURATION VALIDATION (NEW)
===============================================================================

JSON Schema-based validation for stage contexts and configurations.

14.1 Validate Context
---------------------
from stabilize.validation import validate_context, ValidationError

DEPLOY_SCHEMA = {
    "type": "object",
    "required": ["cluster", "image"],
    "properties": {
        "cluster": {"type": "string", "minLength": 1},
        "image": {"type": "string", "pattern": r"^[a-z0-9./-]+:[a-z0-9.-]+$"},
        "replicas": {"type": "integer", "minimum": 1, "default": 1},
        "timeout": {"type": "integer", "minimum": 0},
    },
}

class DeployTask(Task):
    def execute(self, stage: StageExecution) -> TaskResult:
        errors = validate_context(stage.context, DEPLOY_SCHEMA)
        if errors:
            return TaskResult.terminal(f"Invalid config: {errors[0]}")

        # Config is valid, proceed
        cluster = stage.context["cluster"]
        image = stage.context["image"]
        # ...

14.2 Built-in Schemas
---------------------
from stabilize.validation import SHELL_TASK_SCHEMA, WAIT_TASK_SCHEMA

# SHELL_TASK_SCHEMA validates: command (required), timeout, cwd, env, etc.
# WAIT_TASK_SCHEMA validates: waitTime (required, >= 0)

14.3 Quick Validation Check
---------------------------
from stabilize.validation import is_valid

if not is_valid(stage.context, DEPLOY_SCHEMA):
    return TaskResult.terminal("Invalid configuration")

14.4 Supported Validations
--------------------------
Type:         "type": "string" | "integer" | "number" | "boolean" | "array" | "object" | "null"
Union:        "type": ["string", "integer"]
Required:     "required": ["field1", "field2"]
Enum:         "enum": ["value1", "value2"]
Const:        "const": "fixed_value"

String:       "minLength", "maxLength", "pattern"
Number:       "minimum", "maximum", "exclusiveMinimum", "exclusiveMaximum", "multipleOf"
Array:        "minItems", "maxItems", "uniqueItems", "items" (schema for array elements)
Object:       "properties", "additionalProperties", "minProperties", "maxProperties"

===============================================================================
15. ERROR HANDLING & RELIABILITY (NEW)
===============================================================================

Stabilize has enterprise-grade reliability features for production deployments.

15.1 Transient vs Permanent Errors
----------------------------------
from stabilize.errors import TransientError, PermanentError, is_transient

# Transient errors are automatically retried with exponential backoff
raise TransientError("Connection timeout")  # Will retry

# Permanent errors immediately fail the task
raise PermanentError("Invalid input")  # No retry, marks task as terminal

# Classification helper - checks exception class name for keywords
is_transient(ConnectionError("timeout"))  # True - has "connection"
is_transient(TimeoutError())              # True - has "timeout"
is_transient(ValueError("bad input"))     # False - standard exception

Keywords that make an error transient:
  - "timeout", "temporary", "transient", "connection", "network"
  - "unavailable", "retry", "throttl", "rate", "limit", "5xx"

15.2 Automatic Retry with Exponential Backoff
---------------------------------------------
Transient errors are retried with exponential backoff:
  - Attempt 1: ~1 second delay
  - Attempt 2: ~2 seconds delay
  - Attempt 3: ~4 seconds delay
  - ...continues doubling up to 60 seconds max
  - ±25% jitter added to prevent thundering herd

Maximum 10 retry attempts before marking as terminal.

15.2.1 Stateful Retries with context_update (NEW)
-------------------------------------------------
TransientError now supports preserving state across retries:

from stabilize.errors import TransientError

class ProgressTask(Task):
    def execute(self, stage: StageExecution) -> TaskResult:
        processed = stage.context.get("processed_items", 0)
        try:
            new_processed = process_batch(processed)
            return TaskResult.success(outputs={"total": new_processed})
        except RateLimitError:
            # Preserve progress for next retry
            raise TransientError(
                "Rate limited",
                retry_after=30,
                context_update={"processed_items": processed + 10}
            )

Parameters for TransientError:
  message: str             - Error message (required)
  code: int               - Optional error code
  cause: Exception        - Optional original exception
  retry_after: float      - Seconds to wait before retry
  context_update: dict    - Dict to merge into stage.context on retry (NEW)

The context_update is merged into stage.context before rescheduling,
allowing tasks to track progress and resume from where they left off.

15.3 Message Deduplication (Idempotency)
----------------------------------------
Messages are deduplicated to prevent duplicate processing:

# Automatic - no code changes needed
# Each message has a unique ID tracked in processed_messages table
# Re-processing the same message is skipped

This ensures:
- Crash recovery doesn't cause duplicate side effects
- Network retries don't duplicate work
- At-least-once delivery becomes effectively-once processing

15.4 State Transition Validation
--------------------------------
from stabilize.models.status import can_transition, validate_transition

# Check if transition is valid
can_transition(WorkflowStatus.NOT_STARTED, WorkflowStatus.RUNNING)  # True
can_transition(WorkflowStatus.SUCCEEDED, WorkflowStatus.RUNNING)    # False

# Validate with exception
validate_transition(
    WorkflowStatus.SUCCEEDED,
    WorkflowStatus.RUNNING,
    entity_type="workflow",
    entity_id="wf-123",
)  # Raises InvalidStateTransitionError

Valid transitions:
  NOT_STARTED → RUNNING, CANCELED, SKIPPED
  RUNNING     → SUCCEEDED, FAILED_CONTINUE, TERMINAL, CANCELED, PAUSED, STOPPED
  PAUSED      → RUNNING, CANCELED
  Terminal states (SUCCEEDED, TERMINAL, CANCELED, STOPPED, SKIPPED) → no transitions

15.5 Timeout Enforcement
------------------------
Tasks are executed with timeout enforcement using thread interruption:

# Default timeout: 5 minutes for regular tasks
# RetryableTask has two limits:
#   get_timeout() / get_dynamic_timeout(stage): TOTAL time across all executions,
#     measured from the task's first start
#   get_execution_timeout(stage): ONE execute() call; defaults to the total limit

class MyRetryableTask(RetryableTask):
    def get_timeout(self) -> timedelta:
        return timedelta(minutes=30)

    def get_dynamic_timeout(self, stage: StageExecution) -> timedelta:
        # Can use stage context to determine timeout
        return timedelta(milliseconds=stage.context.get("stageTimeoutMs", 300000))

    def get_execution_timeout(self, stage: StageExecution) -> timedelta:
        return timedelta(seconds=30)

# When timeout occurs, task.on_timeout(stage) is called if defined
def on_timeout(self, stage: StageExecution) -> TaskResult | None:
    # Cleanup and return partial result, or None for default behavior
    return TaskResult.failed_continue(error="Timed out", outputs={"partial": data})

15.6 Crash Recovery
-------------------
from stabilize.recovery import WorkflowRecovery, recover_on_startup

# At application startup, recover in-progress workflows
recovery = WorkflowRecovery(store, queue)
results = recovery.recover_pending_workflows()

# Convenience function
recover_on_startup(store, queue)  # Returns list of RecoveryResult

Recovery automatically:
- Finds workflows in RUNNING/NOT_STARTED state
- Re-queues their current stages for continuation
- Uses idempotency to prevent duplicate work

===============================================================================
16. SYNTHETIC STAGES
===============================================================================

Synthetic stages are dynamically injected stages that run before, after, or on
failure of a parent stage. Use them for setup, cleanup, validation, or rollback.

16.1 SyntheticStageOwner Enum
-----------------------------
from stabilize.models.stage import SyntheticStageOwner

SyntheticStageOwner.STAGE_BEFORE  # Runs before parent's tasks
SyntheticStageOwner.STAGE_AFTER   # Runs after parent completes successfully

16.2 StageDefinitionBuilder
---------------------------
Create custom builders to define synthetic stages for your stage types:

from stabilize import StageExecution, TaskExecution
from stabilize.stages.builder import StageDefinitionBuilder
from stabilize.dag.graph import StageGraphBuilder
from stabilize.models.stage import SyntheticStageOwner

class DeployStageBuilder(StageDefinitionBuilder):
    @property
    def type(self) -> str:
        return "deploy"

    def build_tasks(self, stage: StageExecution) -> list[TaskExecution]:
        return [
            TaskExecution.create(
                name="Deploy Application",
                implementing_class="shell",
                stage_start=True,
                stage_end=True,
            ),
        ]

    def before_stages(self, stage: StageExecution, graph: StageGraphBuilder) -> None:
        # Validation runs BEFORE the deploy tasks
        validation = StageExecution.create_synthetic(
            type="shell",
            name="Validate Configuration",
            parent=stage,
            owner=SyntheticStageOwner.STAGE_BEFORE,
            context={"command": "validate-config.sh"},
            tasks=[TaskExecution.create("Validate", "shell", stage_start=True, stage_end=True)],
        )
        graph.add(validation)

    def after_stages(self, stage: StageExecution, graph: StageGraphBuilder) -> None:
        # Notification runs AFTER deploy completes successfully
        notify = StageExecution.create_synthetic(
            type="http",
            name="Send Notification",
            parent=stage,
            owner=SyntheticStageOwner.STAGE_AFTER,
            context={"url": "https://hooks.slack.com/...", "method": "POST"},
            tasks=[TaskExecution.create("Notify", "http", stage_start=True, stage_end=True)],
        )
        graph.add(notify)

    def on_failure_stages(self, stage: StageExecution, graph: StageGraphBuilder) -> None:
        # Rollback runs ONLY if deploy fails
        rollback = StageExecution.create_synthetic(
            type="shell",
            name="Rollback Deployment",
            parent=stage,
            owner=SyntheticStageOwner.STAGE_AFTER,  # After stages handle both success/failure
            context={"command": "rollback.sh"},
            tasks=[TaskExecution.create("Rollback", "shell", stage_start=True, stage_end=True)],
        )
        graph.add(rollback)

16.3 Registering Custom Builders
--------------------------------
from stabilize.stages.builder import register_builder, StageDefinitionBuilderFactory

# Option 1: Use global factory
register_builder(DeployStageBuilder())

# Option 2: Create custom factory
factory = StageDefinitionBuilderFactory()
factory.register(DeployStageBuilder())

16.4 Execution Order
--------------------
1. before_stages() synthetic stages execute first (in dependency order)
2. Parent stage's tasks execute
3. after_stages() synthetic stages execute (success path)
4. on_failure_stages() synthetic stages execute (failure path only)

The ContinueParentStageHandler manages transitions between synthetic stages
and notifies the parent when all children complete.

===============================================================================
17. CONCURRENCY & RACE CONDITIONS
===============================================================================

Stabilize handles concurrent execution safely through optimistic locking and
idempotent message processing.

17.1 Optimistic Locking
-----------------------
Every stage and task has a `version` column. Updates check the version:

UPDATE stage_executions SET
    status = :status,
    version = version + 1
WHERE id = :id AND version = :expected_version

If another process modified the row, the WHERE clause fails and raises:

from stabilize.errors import ConcurrencyError

try:
    repository.store_stage(stage)
except ConcurrencyError:
    # Another process modified this stage - handle accordingly
    pass

17.2 StartStage Race Condition
------------------------------
When multiple upstream stages complete simultaneously, each may try to start
the same downstream stage:

    A         B
     \       /
      \     /
       \   /
         C    <- Both A and B complete, both try to start C

The StartStageHandler safely handles this race:

from stabilize.errors import ConcurrencyError

try:
    with self.repository.transaction(self.queue) as txn:
        txn.store_stage(stage)  # First caller wins, second gets ConcurrencyError
        txn.push_message(StartTask(...))
except ConcurrencyError:
    # Another handler already started this stage - safe to ignore
    logger.debug("Stage %s already started by another handler", stage.name)
    return

This is safe because:
- The stage is already being processed by another handler
- No work is lost - the first handler will complete the stage
- Message deduplication prevents duplicate task execution

17.3 Best Practices
-------------------
1. Always use transactions for state + message atomicity
2. Catch ConcurrencyError only when you can safely retry or ignore
3. Use message deduplication for idempotent processing
4. Let the engine handle retries for transient failures

===============================================================================
18. EVENT SOURCING
===============================================================================

Stabilize includes a built-in event sourcing system. When enabled, every
workflow, stage, and task state transition is recorded as an immutable event.

18.1 Setup (Required)
---------------------
Always enable event sourcing in your setup_pipeline_runner:

from stabilize.events import (
    configure_event_sourcing,
    SqliteEventStore,
    get_event_bus,
    EventType,
)

# In setup_pipeline_runner, before QueueProcessor creation:
event_store = SqliteEventStore(f"sqlite:///{db_path}", create_tables=True)
configure_event_sourcing(event_store)

# That's it — handlers automatically record events.

18.2 Subscribing to Events
--------------------------
Subscribe to the event bus for real-time monitoring:

bus = get_event_bus()

# Log all events
bus.subscribe("logger", lambda e: print(f"{e.event_type.value}: {e.entity_id}"))

# Filter by event type
bus.subscribe(
    "failure-alert",
    lambda e: handle_failure(e),
    event_types={EventType.WORKFLOW_FAILED, EventType.TASK_FAILED},
)

18.3 Event Replay and State Reconstruction
-------------------------------------------
Reconstruct workflow state from events at any point in time:

from stabilize.events import EventReplayer

replayer = EventReplayer(event_store)
state = replayer.rebuild_workflow_state(workflow.id)

# Time-travel: state at a specific sequence number
partial = replayer.rebuild_workflow_state(workflow.id, as_of_sequence=50)

18.4 Projections
----------------
Build analytics views from events:

from stabilize.events import StageMetricsProjection, WorkflowTimelineProjection

# Metrics (subscribe to bus for live updates)
metrics = StageMetricsProjection()
bus.subscribe("metrics", metrics.apply)
# After workflows run:
for stage_type, m in metrics.get_state().items():
    print(f"{stage_type}: {m.success_rate:.0f}% success")

# Timeline (apply events manually from store)
timeline_proj = WorkflowTimelineProjection(workflow.id)
for event in event_store.get_events_for_workflow(workflow.id):
    timeline_proj.apply(event)
timeline = timeline_proj.get_state()

18.5 Event Types
----------------
Workflow: workflow.created, workflow.started, workflow.completed, workflow.failed,
         workflow.canceled, workflow.paused, workflow.resumed
Stage:    stage.started, stage.completed, stage.failed, stage.skipped, stage.canceled
Task:     task.started, task.completed, task.failed, task.retried
State:    status.changed, context.updated, outputs.updated, jump.executed

18.6 Event Stores
-----------------
- SqliteEventStore("sqlite:///path.db", create_tables=True)  # Development
- SqliteEventStore("sqlite:///:memory:", create_tables=True)  # Testing only
- PostgresEventStore("postgresql://...")                       # Production

===============================================================================
19. ADVANCED WORKFLOW CONTROL-FLOW PATTERNS (20 of 43 WCPs)
===============================================================================

Stabilize implements 20 of the 43 Workflow Control-Flow Patterns from van der Aalst
et al. These are configured via fields on StageExecution.

19.1 Join Types
---------------
from stabilize.models.stage import JoinType

JoinType.AND           # (Default) Wait for ALL upstreams (WCP-3)
JoinType.OR            # Wait only for activated branches from OR-split (WCP-7)
JoinType.MULTI_MERGE   # Fire once per upstream completion, no sync (WCP-8).
                       # Serialised; outputs hold the last firing; the
                       # merge's downstream still runs once.
JoinType.DISCRIMINATOR # Fire on first upstream completion, ignore rest (WCP-9)
JoinType.N_OF_M        # Fire when N of M upstreams complete (WCP-30)

# Example: Discriminator join
StageExecution(
    ref_id="triage",
    join_type=JoinType.DISCRIMINATOR,
    requisite_stage_ref_ids={"check_breathing", "check_pulse"},
    ...
)

# Example: N-of-M join (3 of 5 reviewers)
StageExecution(
    ref_id="proceed",
    join_type=JoinType.N_OF_M,
    join_threshold=3,
    requisite_stage_ref_ids={"r1", "r2", "r3", "r4", "r5"},
    ...
)

19.2 Split Types (OR-Split / Multi-Choice, WCP-6)
---------------------------------------------------
from stabilize.models.stage import SplitType

SplitType.AND  # (Default) Activate ALL downstream stages
SplitType.OR   # Evaluate conditions per downstream, activate matching ones

# Example: OR-split with conditions
StageExecution(
    ref_id="triage",
    split_type=SplitType.OR,
    split_conditions={
        "police": "emergency_type == 'crime' or emergency_type == 'accident'",
        "ambulance": "injury_severity > 0",
        "fire": "fire_detected == True",
    },
    ...
)

Conditions are evaluated using a safe expression evaluator. Supported:
  - Comparisons: ==, !=, <, <=, >, >=, in, not in
  - Boolean: and, or, not
  - Literals: strings, numbers, True, False, None
  - Context lookups: key_name, nested.key, dict["key"]
Does NOT support: function calls, imports, assignments, arbitrary code.

BRANCH PRUNING. A branch the split does not choose does not run, and neither
does anything downstream of it. A stage is pruned when EVERY incoming edge is
dead -- an edge is dead if the split de-selected that stage, or if the upstream
was itself pruned. Pruned stages are marked SKIPPED.

The decision is made per stage, where all of its upstreams are visible: a stage
with one de-selected parent and one live parent still runs, exactly once. A join
whose every branch was de-selected is pruned rather than firing on a path no
token reached.

stageEnabled is deliberately NOT a prune: a stage disabled that way still lets
its successor run. Use SplitType.OR with split_conditions to gate a whole branch.

19.3 Deferred Choice (WCP-16)
------------------------------
# Race between branches — first to start wins, others are cancelled
StageExecution(ref_id="agent_contact", deferred_choice_group="complaint_response", ...)
StageExecution(ref_id="escalate_to_manager", deferred_choice_group="complaint_response", ...)

19.4 Milestone Gating (WCP-18)
-------------------------------
# Stage only enabled while milestone stage is in required status
StageExecution(
    ref_id="route_change",
    milestone_ref_id="issue_ticket",
    milestone_status="RUNNING",
    ...
)

19.5 Mutual Exclusion / Critical Section (WCP-17, 39, 40)
-----------------------------------------------------------
# Stages with same mutex_key cannot run simultaneously
StageExecution(ref_id="update_inventory", mutex_key="shared_db", ...)
StageExecution(ref_id="update_ledger", mutex_key="shared_db", ...)

19.6 Cancel Region (WCP-25)
-----------------------------
from stabilize.queue.messages import CancelRegion

# Tag stages with a region name
StageExecution(ref_id="access_evidence_1", cancel_region="evidence_access", ...)
StageExecution(ref_id="access_evidence_2", cancel_region="evidence_access", ...)

# Cancel all stages in the region
queue.push(CancelRegion(
    execution_type="workflow",
    execution_id=workflow.id,
    region="evidence_access",
))

19.7 Signals - Suspend/Resume (WCP-23, WCP-24)
------------------------------------------------
# Task suspends itself waiting for external signal
class ApprovalTask(Task):
    def execute(self, stage: StageExecution) -> TaskResult:
        signal_name = stage.context.get("_signal_name")
        if signal_name == "approved":
            return TaskResult.success(
                outputs={"approved_by": stage.context.get("_signal_data", {}).get("user")}
            )
        return TaskResult.suspend()  # Wait for signal

# Send signal (transient - discarded if stage not suspended)
from stabilize.queue.messages import SignalStage

queue.push(SignalStage(
    execution_type="workflow",
    execution_id=workflow.id,
    stage_id=stage.id,
    signal_name="approved",
    signal_data={"user": "alice"},
    persistent=False,  # True = buffer until stage is ready (WCP-24)
))

Signal data available in resumed task via:
  stage.context["_signal_name"]  # Signal name
  stage.context["_signal_data"]  # Signal payload dict

19.8 Multi-Instance Patterns (WCP-12-15)
-----------------------------------------
from stabilize.stages.multi_instance_builder import MultiInstanceBuilder

parent = StageExecution(ref_id="review", type="review", name="Review", ...)

# WCP-13: Fixed count, wait for all
instance_stages = MultiInstanceBuilder.create_fixed(
    parent_stage=parent, count=6, instance_name_prefix="Reviewer",
)

# WCP-12: Fire and forget (no synchronization)
instance_stages = MultiInstanceBuilder.create_fixed(
    parent_stage=parent, count=3, sync_on_complete=False,
)

# WCP-14: Count from context
parent.context["num_reviewers"] = 4
instance_stages = MultiInstanceBuilder.create_from_context(
    parent_stage=parent, count_key="num_reviewers",
)

# Collection-based: one instance per item
parent.context["items"] = ["a", "b", "c"]
instance_stages = MultiInstanceBuilder.create_from_collection(
    parent_stage=parent, collection_key="items", item_context_key="current_item",
)

# WCP-15: Dynamic (add instances during execution). initial_count may be 0 or
# more; a seeded parent still accepts later AddMultiInstance messages.
instance_stages = MultiInstanceBuilder.create_dynamic(parent_stage=parent, initial_count=2)

# N-of-M with MI: proceed once 3 of 5 complete. The remaining instances still
# run to completion -- the join threshold releases the DOWNSTREAM, it does not
# stop upstream work.
# cancel_remaining=True RAISES NotImplementedError rather than being accepted
# and ignored: every instance is dispatched when the parent completes, so by the
# time a threshold is reached the rest are already RUNNING and the engine has no
# cancellation channel into a running task.
instance_stages = MultiInstanceBuilder.create_fixed(
    parent_stage=parent, count=5, join_threshold=3,
)

19.9 Structured Loops (WCP-21)
-------------------------------
from stabilize.stages.loop_builder import LoopBuilder

# While loop: check condition first, then execute body
stages = LoopBuilder.while_loop(
    condition="iteration_count < max_iterations",
    body_stages=[stage_a, stage_b],
    loop_ref_prefix="retry_loop",
    max_iterations=100,
)

# Repeat-until: execute body first, then check condition
stages = LoopBuilder.repeat_until(
    condition="tests_passed == True",
    body_stages=[stage_a, stage_b],
    loop_ref_prefix="test_loop",
)

# Loop state travels as OUTPUTS. A body stage reads the current values from
# stage.context and returns updated ones; mutating stage.context in place does
# NOT persist (the stage is reloaded before the result is processed).
class Attempt(Task):
    def execute(self, stage):
        n = stage.context.get("attempts", 0)
        return TaskResult.success(outputs={"attempts": n + 1,
                                           "tests_passed": run_tests()})

# The condition's identifiers must be published by something, or the stage
# fails naming the identifier rather than looping silently to the bound.
# Reaching max_iterations exits the loop and CONTINUES past it, carrying
# loop_exhausted=True downstream and marking the loop's last stage
# FAILED_CONTINUE. Nested loops work: an inner loop gets a full budget on
# every pass of the outer one.

19.10 Sub-Workflows / Recursion (WCP-22)
-----------------------------------------
from stabilize.tasks.sub_workflow import SubWorkflowTask

registry.register("sub_workflow", SubWorkflowTask)

StageExecution(
    ref_id="resolve_sub_defects",
    type="sub_workflow",
    context={
        "_sub_workflow_config": {
            "name": "Resolve Sub-Defect",
            "application": "defect-tracker",
            "stages": [...],
            "context": {"defect_id": "DEF-456"},
        },
    },
    ...
)
# Recursion depth tracked via _recursion_depth (default max: 10)

19.11 StageExecution Control-Flow Fields Summary
--------------------------------------------------
Field                    Type               Default   Patterns
join_type                JoinType           AND       WCP-7,8,9,28-33
join_threshold           int                0         WCP-30 (N-of-M)
split_type               SplitType          AND       WCP-6
split_conditions         dict[str,str]      {}        WCP-6
mi_config                MultiInstanceConfig None     WCP-12-15
deferred_choice_group    str|None           None      WCP-16
milestone_ref_id         str|None           None      WCP-18
milestone_status         str|None           None      WCP-18
mutex_key                str|None           None      WCP-17,39,40
cancel_region            str|None           None      WCP-25

===============================================================================
20. AGENTIC WORKFLOWS (LLM AGENTS, TOOLS, HUMAN-IN-THE-LOOP, STREAMING)
===============================================================================

Stabilize ships an optional toolkit for building LLM agent systems on top of the
engine. It is standard-library only and is NOT imported by the core engine, so
it adds nothing to non-agent workflows. Because agents run on the engine, they
inherit its durability: a killed process resumes exactly where it stopped, every
step is event-sourced, and the full control-flow pattern set (Section 19) is
available.

KEY RULES FOR AGENTIC WORKFLOWS:
  - Build a model client ONCE and register task INSTANCES that hold it
    (registry.register("agent", AgentLoopTask(client=client, tools=tools))).
    Registering the bare class will fail at runtime: it has no client.
  - API keys come from the environment (OLLAMA_API_KEY / OPENAI_API_KEY).
    NEVER hard-code a key in generated code.
  - Prefer the high-level tasks (LLMTask, AgentLoopTask) and helpers
    (ApprovalTask/approve, emit_progress, output_reducers) over hand-rolling
    HTTP calls or raw SignalStage.

20.1 MODEL CLIENT
-----------------
from stabilize.llm import LLMClient

# Ollama endpoint (local or ollama.com cloud):
client = LLMClient(model="glm-5.2", base_url="https://ollama.com", api="ollama")
# OpenAI-compatible endpoint:
# client = LLMClient(model="gpt-4o", base_url="https://api.openai.com/v1", api="openai")
# api_key defaults to OLLAMA_API_KEY / OPENAI_API_KEY from the environment.

20.2 ONE-SHOT LLM TASK
----------------------
from stabilize.llm import LLMTask
registry.register("llm", LLMTask(client=client))   # register an INSTANCE holding the client

# Stage CONTEXT keys:  {"prompt": "...", optional "system": "...", optional "temperature": 0.2}
# Stage OUTPUTS:       {"completion": "<text>", "llm_raw": {...}}

20.3 TOOL-CALLING REACT AGENT
-----------------------------
from stabilize.llm import AgentLoopTask, ToolRegistry, tool

@tool
def knowledge_base(name: str) -> str:
    """Look up a spec by name."""          # the docstring becomes the tool description
    return json.dumps(SPECS.get(name, {})) # tools return strings

tools = ToolRegistry().add(knowledge_base)  # .add(fn) per tool; chainable
registry.register("agent", AgentLoopTask(client=client, tools=tools))

# Stage CONTEXT keys:  {"prompt": "...", optional "max_iterations": 8 (loop cap)}
# Stage OUTPUTS:       {"answer": "<text>", "tool_invocations": [{"tool","result"}...], "iterations": N}
# AgentLoopTask runs the WHOLE model->tool-calls->results->model loop inside ONE
# durable task, bounded by max_iterations. @tool derives the JSON schema from the
# function signature and docstring; ToolRegistry dispatches the model's calls.

20.4 HUMAN-IN-THE-LOOP APPROVAL (durable, survives restarts)
------------------------------------------------------------
from stabilize import ApprovalTask, approve, reject, WorkflowStatus
registry.register("approval", ApprovalTask)

# An approval stage SUSPENDS durably until a decision arrives. After process_all,
# find the suspended gate and send the decision:
gate = next(s for s in store.retrieve(workflow.id).stages if s.ref_id == "approve")
if gate.status == WorkflowStatus.SUSPENDED:
    approve(queue, workflow.id, gate.id, {"note": "ok"}, user="alice")  # or reject(...)
    # user= is recorded as the actor on the stage.resumed event.
    processor.process_all(timeout=30)                          # resumes and finishes
# The suspension is persisted (waits minutes/days, survives a restart). A signal
# sent BEFORE the stage suspends is buffered, so approvals are never lost.
# Approved -> stage outputs {"approved": True, "approval": {...payload}}.
# Rejected -> terminal by default; set context {"approval_reject_continues": True}
# to continue the pipeline (FAILED_CONTINUE) instead.

20.5 LIVE STREAMING (progress + tokens)
---------------------------------------
from stabilize import WorkflowStream, emit_progress

# Inside a task, push progress (any keyword data is carried on the event):
emit_progress(stage, "researching", agent="researcher:0", percent=40)

# In the caller, consume events live (callback form, non-blocking):
stream = WorkflowStream(workflow.id)
stream.on_event(lambda item: print(item.event_type, item.data.get("message", "")))
# ... run the workflow ...
stream.close()
# Progress events have event_type == "custom.progress". Lifecycle events
# (stage.completed, workflow.completed, ...) flow through the same stream.
# WorkflowStream(workflow.id, event_store=es).follow(include_history=True) iterates
# history then live until the workflow ends.

20.6 FAN-IN REDUCERS (combine parallel agent outputs, don't overwrite)
----------------------------------------------------------------------
# By default, parallel branches writing the SAME output key overwrite (last wins).
# On the JOIN stage, set output_reducers to combine each branch's value:
StageExecution(
    ref_id="gather",
    type="synthesizer",
    join_type=JoinType.AND,                        # or N_OF_M / DISCRIMINATOR
    requisite_stage_ref_ids={"agent_0", "agent_1", "agent_2"},
    output_reducers={"finding": "collect"},        # combine "finding" from all branches
    tasks=[...],
)
# Built-in reducers: collect|append (gather into a list), extend (flatten lists),
# sum, max, min, merge (dict), first, last. Register custom ones with
# register_reducer("name", fn). In the join task, stage.context["finding"] is the
# combined value (e.g. a list). Keys without a reducer keep last-writer-wins.

20.7 AGENTIC CONTROL-FLOW ON THE ENGINE (reuse Section 19)
----------------------------------------------------------
# Bounded agent loop (retry / refine): a task jumps back to an earlier stage.
return TaskResult.jump_to("plan", context={"feedback": "tighten the numbers"})
#   -> resets plan + downstream and re-runs; bounded by execution context "_max_jumps".
# Proceed on a quorum of parallel agents:  JoinType.N_OF_M, join_threshold=K.
# Race strategies, first result wins:       JoinType.DISCRIMINATOR.
# A sub-agent as its own child workflow:    SubWorkflowTask (Section 19).

20.8 COMPLETE AGENTIC TEMPLATE (copy and adapt)
-----------------------------------------------
#!/usr/bin/env python3
"""Fan-out of tool-using agents -> reducer join -> synthesis -> human approval."""
import json, os
from stabilize import (
    Workflow, StageExecution, TaskExecution, TaskRegistry, WorkflowStatus,
    SqliteWorkflowStore, SqliteQueue, QueueProcessor, Orchestrator, JoinType,
    ApprovalTask, approve, Task, TaskResult,
)
from stabilize.llm import LLMClient, AgentLoopTask, ToolRegistry, tool

SPECS = {"atlas-70b": {"weights_gb": 140}, "h100": {"vram_gb": 80}}

@tool
def knowledge_base(name: str) -> str:
    """Look up a spec by name (atlas-70b, h100)."""
    return json.dumps(SPECS.get(name.lower(), {"error": "unknown"}))

@tool
def calculate(expression: str) -> str:
    """Evaluate a simple arithmetic expression like '140 / 80'."""
    import ast, operator
    ops = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv}
    def ev(n):
        if isinstance(n, ast.Expression): return ev(n.body)
        if isinstance(n, ast.Constant): return n.value
        if isinstance(n, ast.BinOp): return ops[type(n.op)](ev(n.left), ev(n.right))
        raise ValueError("unsupported")
    return str(ev(ast.parse(expression, mode="eval")))

class Synthesizer(Task):
    def execute(self, stage):
        findings = stage.context.get("finding", [])          # reducer-gathered list
        if isinstance(findings, str): findings = [findings]
        return TaskResult.success(outputs={"summary": " | ".join(str(f)[:120] for f in findings)})

client = LLMClient(model="glm-5.2", base_url="https://ollama.com", api="ollama")  # OLLAMA_API_KEY from env
tools = ToolRegistry().add(knowledge_base).add(calculate)

registry = TaskRegistry()
registry.register("agent", AgentLoopTask(client=client, tools=tools))
registry.register("synth", Synthesizer)
registry.register("approval", ApprovalTask)

def _t(name, impl):
    return TaskExecution.create(name=name, implementing_class=impl, stage_start=True, stage_end=True)

SUBQ = ["How much VRAM do atlas-70b weights need?", "How many h100 GPUs fit the weights?"]
agents = [
    StageExecution(ref_id=f"agent_{i}", type="agent", name=f"Agent {i}",
        context={"prompt": q + " Use knowledge_base and calculate. State the number.", "output_key": "finding"},
        tasks=[_t(f"a{i}", "agent")])
    for i, q in enumerate(SUBQ)
]
gather = StageExecution(ref_id="gather", type="synth", name="Synthesize",
    requisite_stage_ref_ids={a.ref_id for a in agents},
    join_type=JoinType.AND, output_reducers={"finding": "collect"},
    tasks=[_t("g", "synth")])
approve_stage = StageExecution(ref_id="approve", type="approval", name="Approve",
    requisite_stage_ref_ids={"gather"}, tasks=[_t("ap", "approval")])

workflow = Workflow.create(application="agentic-demo", name="Fan-out agents",
    stages=[*agents, gather, approve_stage])

store = SqliteWorkflowStore("sqlite:///./agentic.db", create_tables=True)   # DISK, never :memory:
queue = SqliteQueue("sqlite:///./agentic.db"); queue._create_table()
processor = QueueProcessor(queue, store=store, task_registry=registry)
orchestrator = Orchestrator(queue, store=store)

store.store(workflow)
orchestrator.start(workflow)
processor.process_all(timeout=180)                     # runs until the approval gate suspends

gate = next(s for s in store.retrieve(workflow.id).stages if s.ref_id == "approve")
if gate.status == WorkflowStatus.SUSPENDED:
    approve(queue, workflow.id, gate.id, {"note": "ok"}, user="alice")
    processor.process_all(timeout=30)

result = store.retrieve(workflow.id)
print("status:", result.status.name)                   # SUCCEEDED
print("summary:", next(s for s in result.stages if s.ref_id == "gather").outputs.get("summary"))

===============================================================================
END OF REFERENCE
===============================================================================
