teleo-infrastructure/teleo-pipeline.py

#!/usr/bin/env python3
"""Teleo Pipeline v2 — single async daemon replacing 7 cron scripts.

Four stages: Ingest → Validate → Evaluate → Merge
SQLite WAL state store. systemd-managed. Graceful shutdown.
"""

import asyncio
import logging
import signal
import sys
from datetime import datetime, timezone

# Add parent dir to path so lib/ is importable
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))

from lib import config, db, log as logmod
from lib.health import start_health_server, stop_health_server
from lib.breaker import CircuitBreaker
from lib.merge import merge_cycle
from lib.validate import validate_cycle
from lib.evaluate import evaluate_cycle, kill_active_subprocesses

logger = logging.getLogger("pipeline")

# Global shutdown event — stages check this between iterations
shutdown_event = asyncio.Event()


async def stage_loop(name: str, interval: int, func, conn, breaker: CircuitBreaker):
    """Generic stage loop with interval, shutdown check, and circuit breaker."""
    logger.info("Stage %s started (interval=%ds)", name, interval)
    while not shutdown_event.is_set():
        try:
            if not breaker.allow_request():
                logger.debug("Stage %s: breaker OPEN, skipping cycle", name)
            else:
                workers = breaker.max_workers()
                succeeded, failed = await func(conn, max_workers=workers)
                if failed > 0 and succeeded == 0:
                    breaker.record_failure()
                elif succeeded > 0:
                    breaker.record_success()
        except Exception:
            logger.exception("Stage %s: unhandled error in cycle", name)
            breaker.record_failure()

        # Wait for interval or shutdown, whichever comes first
        try:
            await asyncio.wait_for(shutdown_event.wait(), timeout=interval)
            break  # shutdown_event was set
        except asyncio.TimeoutError:
            pass  # interval elapsed, continue loop

    logger.info("Stage %s stopped", name)


# --- Stage stubs (Phase 1 — replaced in later phases) ---

async def ingest_cycle(conn, max_workers=None):
    """Stage 1: Scan inbox, extract claims. (stub)"""
    return 0, 0


# validate_cycle imported from lib.validate


# evaluate_cycle imported from lib.evaluate


# merge_cycle imported from lib.merge


# --- Shutdown ---

def handle_signal(sig):
    """Signal handler — sets shutdown event."""
    logger.info("Received %s, initiating graceful shutdown...", sig.name)
    shutdown_event.set()


async def kill_subprocesses():
    """Kill any lingering Claude CLI subprocesses (delegates to evaluate module)."""
    await kill_active_subprocesses()


async def cleanup_orphan_worktrees():
    """Remove any orphan worktrees from previous crashes."""
    import glob
    import shutil
    # Use specific prefix to avoid colliding with other /tmp users (Ganymede)
    orphans = glob.glob("/tmp/teleo-extract-*") + glob.glob("/tmp/teleo-merge-*")
    for path in orphans:
        logger.warning("Cleaning orphan worktree: %s", path)
        try:
            proc = await asyncio.create_subprocess_exec(
                "git", "worktree", "remove", "--force", path,
                cwd=str(config.REPO_DIR),
                stdout=asyncio.subprocess.DEVNULL,
                stderr=asyncio.subprocess.DEVNULL,
            )
            await asyncio.wait_for(proc.wait(), timeout=10)
        except Exception:
            shutil.rmtree(path, ignore_errors=True)
    # Prune stale worktree metadata entries from bare repo (Ganymede)
    try:
        proc = await asyncio.create_subprocess_exec(
            "git", "worktree", "prune",
            cwd=str(config.REPO_DIR),
            stdout=asyncio.subprocess.DEVNULL,
            stderr=asyncio.subprocess.DEVNULL,
        )
        await asyncio.wait_for(proc.wait(), timeout=10)
    except Exception:
        logger.warning("git worktree prune failed, continuing")


# --- Main ---

async def main():
    logmod.setup_logging()
    logger.info("Teleo Pipeline v2 starting")

    # Clean orphan worktrees from prior crashes (Ganymede's requirement)
    await cleanup_orphan_worktrees()

    # Initialize database
    conn = db.get_connection()
    db.migrate(conn)
    logger.info("Database ready at %s", config.DB_PATH)

    # Initialize circuit breakers
    breakers = {
        "ingest": CircuitBreaker("ingest", conn),
        "validate": CircuitBreaker("validate", conn),
        "evaluate": CircuitBreaker("evaluate", conn),
        "merge": CircuitBreaker("merge", conn),
    }

    # Recover interrupted state from crashes
    # Atomic recovery: all three resets in one transaction (Ganymede)
    # Increment transient_retries on recovered sources to prevent infinite cycling (Vida)
    with db.transaction(conn):
        # Sources stuck in 'extracting' — increment retry counter, move to error if exhausted
        c1 = conn.execute(
            """UPDATE sources SET
                transient_retries = transient_retries + 1,
                status = CASE
                    WHEN transient_retries + 1 >= ? THEN 'error'
                    ELSE 'unprocessed'
                END,
                last_error = CASE
                    WHEN transient_retries + 1 >= ? THEN 'crash recovery: retry budget exhausted'
                    ELSE last_error
                END,
                updated_at = datetime('now')
            WHERE status = 'extracting'""",
            (config.TRANSIENT_RETRY_MAX, config.TRANSIENT_RETRY_MAX),
        )
        # PRs stuck in 'merging' → approved (Ganymede's Q4 answer)
        c2 = conn.execute(
            "UPDATE prs SET status = 'approved' WHERE status = 'merging'"
        )
        # PRs stuck in 'reviewing' → open
        c3 = conn.execute(
            "UPDATE prs SET status = 'open' WHERE status = 'reviewing'"
        )
    recovered = c1.rowcount + c2.rowcount + c3.rowcount
    if recovered:
        logger.info("Recovered %d interrupted rows from prior crash", recovered)

    # Register signal handlers
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, handle_signal, sig)

    # Start health API
    health_runners = []
    await start_health_server(health_runners)

    # Start stage loops
    stages = [
        asyncio.create_task(
            stage_loop("ingest", config.INGEST_INTERVAL, ingest_cycle, conn, breakers["ingest"]),
            name="ingest",
        ),
        asyncio.create_task(
            stage_loop("validate", config.VALIDATE_INTERVAL, validate_cycle, conn, breakers["validate"]),
            name="validate",
        ),
        asyncio.create_task(
            stage_loop("evaluate", config.EVAL_INTERVAL, evaluate_cycle, conn, breakers["evaluate"]),
            name="evaluate",
        ),
        asyncio.create_task(
            stage_loop("merge", config.MERGE_INTERVAL, merge_cycle, conn, breakers["merge"]),
            name="merge",
        ),
    ]

    logger.info("All stages running")

    # Wait for shutdown signal
    await shutdown_event.wait()
    logger.info("Shutdown event received, waiting for stages to finish...")

    # Give stages time to finish current work
    try:
        await asyncio.wait_for(asyncio.gather(*stages, return_exceptions=True), timeout=60)
    except asyncio.TimeoutError:
        logger.warning("Stages did not finish within 60s, force-cancelling")
        for task in stages:
            task.cancel()
        await asyncio.gather(*stages, return_exceptions=True)

    # Kill lingering subprocesses
    await kill_subprocesses()

    # Stop health API
    await stop_health_server(health_runners)

    # Close DB
    conn.close()
    logger.info("Teleo Pipeline v2 shut down cleanly")


if __name__ == "__main__":
    asyncio.run(main())