hanzoai_agent/scripts/seed_workflows.py

#!/usr/bin/env python3
"""
Seed the local Hanzo Agents SQLite store with realistic deep-research workflows.

The script generates configurable DAGs of workflow executions, associated runs,
steps, and timeline events so they show up inside the control-plane UI
(`http://localhost:8080/ui/workflows`).

Example:
    python scripts/seed_workflows.py --nodes-per-workflow 10000 --workflow-count 1
"""

from __future__ import annotations

import argparse
import json
import random
import sqlite3
import sys
import textwrap
import uuid
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Optional, Sequence, Tuple


# --- Data classes ---------------------------------------------------------


@dataclass(frozen=True)
class AgentNodeDefinition:
    node_id: str
    base_url: str
    version: str
    deployment_type: str
    invocation_url: str
    reasoners: Sequence[dict]  # Changed from Sequence[str] to Sequence[dict]
    skills: Sequence[dict]  # Changed from Sequence[str] to Sequence[dict]
    communication: dict
    health_status: str = "healthy"
    lifecycle_status: str = "active"
    features: Optional[dict] = None
    metadata: Optional[dict] = None


@dataclass
class WorkflowScenario:
    workflow_id: str
    workflow_name: str
    workflow_tags: List[str]
    session_id: str
    actor_id: str
    run_id: str
    root_execution_id: str
    root_request_id: str
    started_at: datetime
    completed_at: datetime


@dataclass
class NodeRecord:
    execution_id: str
    request_id: str
    parent_execution_id: Optional[str]
    depth: int
    agent_node_id: str
    reasoner_id: str
    status: str
    status_reason: Optional[str]
    error_message: Optional[str]
    started_at: datetime
    completed_at: datetime
    duration_ms: int
    input_payload: dict
    output_payload: dict
    notes: List[dict]


# --- Constants ------------------------------------------------------------

DEFAULT_DB_PATH = Path("~/.hanzo_agents/data/hanzo_agents.db").expanduser()

AGENT_NODE_POOL: Tuple[AgentNodeDefinition, ...] = (
    AgentNodeDefinition(
        node_id="atlas_scope_orchestrator",
        base_url="https://agents.internal/atlas-scope",
        version="1.14.3",
        deployment_type="long_running",
        invocation_url="https://agents.internal/atlas-scope/invoke",
        reasoners=(
            {
                "id": "Deep Scope Orchestrator",
                "input_schema": {},
                "output_schema": {},
                "memory_config": {
                    "auto_inject": [],
                    "memory_retention": "session",
                    "cache_results": False,
                },
            },
            {
                "id": "Context Window Planner",
                "input_schema": {},
                "output_schema": {},
                "memory_config": {
                    "auto_inject": [],
                    "memory_retention": "session",
                    "cache_results": False,
                },
            },
        ),
        skills=(
            {"id": "graph_navigation", "input_schema": {}, "tags": ["navigation"]},
            {"id": "source_enrichment", "input_schema": {}, "tags": ["enrichment"]},
        ),
        communication={"webhook": "https://callbacks.internal/atlas-scope"},
        features={"supports_multithread": True},
        metadata={"owner": "Atlas Research"},
    ),
    AgentNodeDefinition(
        node_id="source_link_parser",
        base_url="https://agents.internal/source-parser",
        version="2.3.1",
        deployment_type="ephemeral",
        invocation_url="https://agents.internal/source-parser/invoke",
        reasoners=(
            {
                "id": "Signal Enrichment Parser",
                "input_schema": {},
                "output_schema": {},
                "memory_config": {
                    "auto_inject": [],
                    "memory_retention": "session",
                    "cache_results": False,
                },
            },
            {
                "id": "Evidence Link Resolver",
                "input_schema": {},
                "output_schema": {},
                "memory_config": {
                    "auto_inject": [],
                    "memory_retention": "session",
                    "cache_results": False,
                },
            },
        ),
        skills=(
            {"id": "html_extraction", "input_schema": {}, "tags": ["extraction"]},
            {"id": "ocr", "input_schema": {}, "tags": ["extraction"]},
            {"id": "link_resolution", "input_schema": {}, "tags": ["resolution"]},
        ),
        communication={"queue": "source-parser-events"},
        features={"supports_batch": True},
        metadata={"coverage": "public+subscription"},
    ),
    AgentNodeDefinition(
        node_id="signal_prioritizer",
        base_url="https://agents.internal/signal-prioritizer",
        version="0.9.12",
        deployment_type="long_running",
        invocation_url="https://agents.internal/signal-prioritizer/invoke",
        reasoners=(
            {
                "id": "Signal Weighting Pipeline",
                "input_schema": {},
                "output_schema": {},
                "memory_config": {
                    "auto_inject": [],
                    "memory_retention": "session",
                    "cache_results": False,
                },
            },
            {
                "id": "Early Warning Ranker",
                "input_schema": {},
                "output_schema": {},
                "memory_config": {
                    "auto_inject": [],
                    "memory_retention": "session",
                    "cache_results": False,
                },
            },
        ),
        skills=(
            {"id": "ranking", "input_schema": {}, "tags": ["prioritization"]},
            {"id": "anomaly_detection", "input_schema": {}, "tags": ["detection"]},
        ),
        communication={"kafka_topic": "signal-priority"},
        features={"weighted_scores": True},
        metadata={"owner": "SignalOps"},
    ),
    AgentNodeDefinition(
        node_id="narrative_synthesizer",
        base_url="https://agents.internal/narrative-synth",
        version="3.0.2",
        deployment_type="long_running",
        invocation_url="https://agents.internal/narrative-synth/invoke",
        reasoners=(
            {
                "id": "Narrative Multi-Lens Synth",
                "input_schema": {},
                "output_schema": {},
                "memory_config": {
                    "auto_inject": [],
                    "memory_retention": "session",
                    "cache_results": False,
                },
            },
            {
                "id": "Cross Lens Summarizer",
                "input_schema": {},
                "output_schema": {},
                "memory_config": {
                    "auto_inject": [],
                    "memory_retention": "session",
                    "cache_results": False,
                },
            },
        ),
        skills=(
            {"id": "summarization", "input_schema": {}, "tags": ["synthesis"]},
            {"id": "storylining", "input_schema": {}, "tags": ["narrative"]},
        ),
        communication={"webhook": "https://callbacks.internal/narrative"},
        features={"supports_sections": True},
        metadata={"owner": "NarrativeOps"},
    ),
    AgentNodeDefinition(
        node_id="risk_modeler",
        base_url="https://agents.internal/risk-modeler",
        version="1.8.7",
        deployment_type="long_running",
        invocation_url="https://agents.internal/risk-modeler/invoke",
        reasoners=(
            {
                "id": "Risk Quant Scorer",
                "input_schema": {},
                "output_schema": {},
                "memory_config": {
                    "auto_inject": [],
                    "memory_retention": "session",
                    "cache_results": False,
                },
            },
            {
                "id": "Scenario Stress Analyzer",
                "input_schema": {},
                "output_schema": {},
                "memory_config": {
                    "auto_inject": [],
                    "memory_retention": "session",
                    "cache_results": False,
                },
            },
        ),
        skills=(
            {"id": "scenario_planning", "input_schema": {}, "tags": ["planning"]},
            {"id": "stress_testing", "input_schema": {}, "tags": ["testing"]},
        ),
        communication={"queue": "risk-modeler-events"},
        features={"supports_probabilistic": True},
        metadata={"coverage": "macroeconomic"},
    ),
    AgentNodeDefinition(
        node_id="market_lens_vectorizer",
        base_url="https://agents.internal/market-lens",
        version="0.6.5",
        deployment_type="ephemeral",
        invocation_url="https://agents.internal/market-lens/invoke",
        reasoners=(
            {
                "id": "Market Vector Indexer",
                "input_schema": {},
                "output_schema": {},
                "memory_config": {
                    "auto_inject": [],
                    "memory_retention": "session",
                    "cache_results": False,
                },
            },
            {
                "id": "Capital Flow Mapper",
                "input_schema": {},
                "output_schema": {},
                "memory_config": {
                    "auto_inject": [],
                    "memory_retention": "session",
                    "cache_results": False,
                },
            },
        ),
        skills=(
            {"id": "embedding", "input_schema": {}, "tags": ["vectorization"]},
            {"id": "semantic_alignment", "input_schema": {}, "tags": ["alignment"]},
        ),
        communication={"queue": "market-lens-jobs"},
        features={"supports_dimensionality": 1536},
        metadata={"owner": "MarketIntel"},
    ),
    AgentNodeDefinition(
        node_id="insight_validation_bureau",
        base_url="https://agents.internal/insight-validation",
        version="1.2.9",
        deployment_type="long_running",
        invocation_url="https://agents.internal/insight-validation/invoke",
        reasoners=(
            {
                "id": "Insight Validation Hub",
                "input_schema": {},
                "output_schema": {},
                "memory_config": {
                    "auto_inject": [],
                    "memory_retention": "session",
                    "cache_results": False,
                },
            },
            {
                "id": "Confidence Calibration Board",
                "input_schema": {},
                "output_schema": {},
                "memory_config": {
                    "auto_inject": [],
                    "memory_retention": "session",
                    "cache_results": False,
                },
            },
        ),
        skills=(
            {"id": "source_corrobortion", "input_schema": {}, "tags": ["validation"]},
            {"id": "confidence_estimation", "input_schema": {}, "tags": ["estimation"]},
        ),
        communication={"webhook": "https://callbacks.internal/insight-validation"},
        features={"supports_panel_reviews": True},
        metadata={"owner": "InsightOps"},
    ),
)

STATUS_WEIGHTS = (("succeeded", 1.0),)

REGIONS = (
    "EU robotics",
    "LATAM fintech",
    "APAC climate tech",
    "MENA mobility",
    "US defense tech",
    "Nordic energy storage",
    "India healthtech",
    "Sub-Saharan agri-tech",
)

OBJECTIVES = (
    "Map competitive pivots",
    "Surface latent demand",
    "Track supplier pressure",
    "Quantify regulatory heat",
    "Trace capital allocations",
    "Forecast leadership churn",
    "Score narrative momentum",
    "Detect stealth partnerships",
)

DELIVERABLES = (
    "executive_brief",
    "risk_dossier",
    "market_signal_digest",
    "threat_model",
    "investment_heatmap",
    "cx_opportunity_brief",
)


# --- Helpers --------------------------------------------------------------


def parse_args(argv: Sequence[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Seed the Hanzo Agents SQLite workflow tables with realistic data.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=textwrap.dedent(
            """
            Tips:
              • Use --nodes-per-workflow 10000 to mirror dense research DAGs.
              • Combine --workflow-count with a smaller node count to simulate breadth.
              • Re-run with the same --seed to get reproducible structures.
            """
        ),
    )
    parser.add_argument(
        "--db-path",
        type=Path,
        default=DEFAULT_DB_PATH,
        help=f"SQLite database path (default: {DEFAULT_DB_PATH})",
    )
    parser.add_argument(
        "--workflow-count",
        type=int,
        default=1,
        help="Number of distinct root workflows to generate.",
    )
    parser.add_argument(
        "--nodes-per-workflow",
        type=int,
        default=50,
        help="Number of execution nodes per workflow (excluding the root).",
    )
    parser.add_argument(
        "--workflow-prefix",
        type=str,
        default="wf_deep_context_probe",
        help="Prefix used when minting workflow IDs.",
    )
    parser.add_argument(
        "--session-prefix",
        type=str,
        default="session_deep_research",
        help="Prefix for session identifiers.",
    )
    parser.add_argument(
        "--actor-pool",
        type=str,
        nargs="+",
        default=["research_lead_marissa", "intel_director_akeem", "strategy_gm_li"],
        help="Actor IDs to sample from for ownership.",
    )
    parser.add_argument(
        "--team-id",
        type=str,
        default="intel_ops_enterprise",
        help="Team ID used for registering agent nodes.",
    )
    parser.add_argument(
        "--start-hours-ago",
        type=int,
        default=18,
        help="Anchor time (in hours before now) for the first workflow start.",
    )
    parser.add_argument(
        "--stagger-minutes",
        type=int,
        default=42,
        help="Minutes between successive workflow kickoffs.",
    )
    parser.add_argument(
        "--purge-prefix",
        action="store_true",
        help="Delete existing workflows whose IDs start with the workflow prefix before seeding.",
    )
    parser.add_argument(
        "--seed",
        type=int,
        default=None,
        help="Random seed for deterministic output.",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Build the workflow plan but skip database writes.",
    )
    return parser.parse_args(argv)


def ensure_agent_nodes(conn: sqlite3.Connection, team_id: str) -> None:
    """Upsert the canned agent nodes so UI views have metadata."""
    now_iso = isoformat(datetime.utcnow())
    for node in AGENT_NODE_POOL:
        payload = (
            node.node_id,
            team_id,
            node.base_url,
            node.version,
            node.deployment_type,
            node.invocation_url,
            json.dumps(node.reasoners).encode(),
            json.dumps(node.skills).encode(),
            json.dumps(node.communication).encode(),
            node.health_status,
            node.lifecycle_status,
            now_iso,
            now_iso,
            json.dumps(node.features or {}).encode(),
            json.dumps(node.metadata or {}).encode(),
        )
        conn.execute("DELETE FROM agent_nodes WHERE id = ?", (node.node_id,))
        conn.execute(
            """
            INSERT INTO agent_nodes (
                id, team_id, base_url, version, deployment_type, invocation_url,
                reasoners, skills, communication_config, health_status, lifecycle_status,
                last_heartbeat, registered_at, features, metadata
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            payload,
        )


def purge_workflows_with_prefix(conn: sqlite3.Connection, prefix: str) -> int:
    """Remove existing data for workflows whose IDs share the prefix."""
    pattern = f"{prefix}%"
    workflow_ids = [
        row[0]
        for row in conn.execute(
            "SELECT workflow_id FROM workflows WHERE workflow_id LIKE ?", (pattern,)
        )
    ]
    removed = 0
    for workflow_id in workflow_ids:
        run_ids = [
            row[0]
            for row in conn.execute(
                "SELECT run_id FROM workflow_runs WHERE root_workflow_id = ?",
                (workflow_id,),
            )
        ]
        for run_id in run_ids:
            conn.execute("DELETE FROM executions WHERE run_id = ?", (run_id,))
            conn.execute("DELETE FROM workflow_steps WHERE run_id = ?", (run_id,))
            conn.execute("DELETE FROM workflow_run_events WHERE run_id = ?", (run_id,))
            conn.execute(
                "DELETE FROM workflow_execution_events WHERE run_id = ?", (run_id,)
            )
        conn.execute(
            "DELETE FROM workflow_execution_events WHERE workflow_id = ?",
            (workflow_id,),
        )
        conn.execute(
            "DELETE FROM workflow_executions WHERE workflow_id = ?", (workflow_id,)
        )
        conn.execute(
            "DELETE FROM workflow_runs WHERE root_workflow_id = ?", (workflow_id,)
        )
        conn.execute("DELETE FROM workflows WHERE workflow_id = ?", (workflow_id,))
        removed += 1
    return removed


def choose_weighted(options: Sequence[Tuple[str, float]]) -> str:
    labels, weights = zip(*options)
    return random.choices(labels, weights=weights, k=1)[0]


def isoformat(dt: datetime) -> str:
    return dt.replace(microsecond=0).isoformat()


def generate_scenario(
    *,
    prefix: str,
    session_prefix: str,
    actor_pool: Sequence[str],
    idx: int,
    started_at: datetime,
    nodes_per_workflow: int,
) -> WorkflowScenario:
    run_id = f"run_{uuid.uuid4().hex[:12]}"
    root_exec = f"exec_root_{uuid.uuid4().hex[:12]}"
    root_req = f"af_req_{uuid.uuid4().hex[:12]}"
    suffix = started_at.strftime("%Y%m%d%H%M%S")
    workflow_id = f"{prefix}_{suffix}_{idx:03d}"

    region = random.choice(REGIONS)
    objective = random.choice(OBJECTIVES)
    workflow_name = f"{objective} – {region}"
    tags = [
        "deep_research",
        "market_intelligence",
        region.replace(" ", "_").lower(),
        objective.replace(" ", "_").lower(),
    ]
    session_id = f"{session_prefix}_{region.split()[0].lower()}_{idx:02d}"
    actor_id = random.choice(actor_pool)
    duration = timedelta(minutes=18 + int(nodes_per_workflow / 40))
    return WorkflowScenario(
        workflow_id=workflow_id,
        workflow_name=workflow_name,
        workflow_tags=tags,
        session_id=session_id,
        actor_id=actor_id,
        run_id=run_id,
        root_execution_id=root_exec,
        root_request_id=root_req,
        started_at=started_at,
        completed_at=started_at + duration,
    )


def synthesize_nodes(
    scenario: WorkflowScenario, *, nodes_per_workflow: int
) -> List[NodeRecord]:
    nodes: List[NodeRecord] = []

    root_input = {
        "objective": scenario.workflow_name,
        "initial_sources": random.sample(
            ["eu-parliament", "ifri", "techcrunch-eu", "g7-briefs", "cb-insights"], k=3
        ),
        "run_context": "executive-intelligence-brief",
    }
    root_output = {
        "summary": "Initial orchestration launched with validated scope.",
        "confidence": round(random.uniform(0.82, 0.95), 2),
    }
    root_notes = [
        {
            "author": scenario.actor_id,
            "note": "Scope locked with intelligence leadership.",
            "timestamp": isoformat(scenario.started_at),
        }
    ]

    root_end = scenario.started_at + timedelta(minutes=6)
    root_duration_ms = int((root_end - scenario.started_at).total_seconds() * 1000)
    nodes.append(
        NodeRecord(
            execution_id=scenario.root_execution_id,
            request_id=scenario.root_request_id,
            parent_execution_id=None,
            depth=0,
            agent_node_id="atlas_scope_orchestrator",
            reasoner_id=AGENT_NODE_POOL[0].reasoners[0]["id"],
            status="succeeded",
            status_reason=None,
            error_message=None,
            started_at=scenario.started_at,
            completed_at=root_end,
            duration_ms=root_duration_ms,
            input_payload=root_input,
            output_payload=root_output,
            notes=root_notes,
        )
    )

    for i in range(1, nodes_per_workflow + 1):
        parent_node = random.choice(nodes)
        status = choose_weighted(STATUS_WEIGHTS)
        status_reason = None
        error_message = None

        start_offset = random.randint(90, 480)
        node_start = parent_node.started_at + timedelta(seconds=start_offset)
        node_duration = random.randint(120, 1500)
        node_end = node_start + timedelta(seconds=node_duration)

        agent_choice = random.choice(AGENT_NODE_POOL)
        reasoner_choice = random.choice(agent_choice.reasoners)

        notes: List[dict] = []
        notes.append(
            {
                "author": random.choice(
                    ["analyst_jonah", "analyst_li", "analyst_manuela", "analyst_mira"]
                ),
                "note": "Reviewed output and advanced to synthesis track.",
                "timestamp": isoformat(node_end),
            }
        )

        nodes.append(
            NodeRecord(
                execution_id=f"exec_{i:05d}_{uuid.uuid4().hex[:10]}",
                request_id=f"af_req_{uuid.uuid4().hex[:12]}",
                parent_execution_id=parent_node.execution_id,
                depth=parent_node.depth + 1,
                agent_node_id=agent_choice.node_id,
                reasoner_id=reasoner_choice["id"],
                status=status,
                status_reason=status_reason,
                error_message=error_message,
                started_at=node_start,
                completed_at=node_end,
                duration_ms=node_duration * 1000,
                input_payload={
                    "focus": random.choice(
                        [
                            "supply_chain",
                            "policy_shift",
                            "funding_rounds",
                            "sentiment_trace",
                            "talent_flows",
                        ]
                    ),
                    "signals_processed": random.randint(12, 120),
                    "parent_execution": parent_node.execution_id,
                },
                output_payload={
                    "key_findings": random.randint(3, 8),
                    "priority_score": round(random.uniform(0.32, 0.98), 2),
                    "insight_hash": uuid.uuid4().hex[:16],
                },
                notes=notes,
            )
        )

    return nodes


def insert_workflow(
    conn: sqlite3.Connection, scenario: WorkflowScenario, nodes: Sequence[NodeRecord]
) -> None:
    total_executions = len(nodes)
    success_count = sum(1 for n in nodes if n.status == "succeeded")
    failed_count = sum(1 for n in nodes if n.status != "succeeded")
    max_depth = max(n.depth for n in nodes)
    workflow_status = "succeeded" if failed_count == 0 else "failed"
    total_duration_ms = int(
        (
            max(n.completed_at for n in nodes) - min(n.started_at for n in nodes)
        ).total_seconds()
        * 1000
    )
    tags_json = json.dumps(scenario.workflow_tags)

    conn.execute(
        """
        INSERT INTO workflows (
            workflow_id, workflow_name, workflow_tags, session_id, actor_id,
            parent_workflow_id, parent_execution_id, root_workflow_id,
            workflow_depth, total_executions, successful_executions, failed_executions,
            total_duration_ms, status, started_at, completed_at, created_at, updated_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            scenario.workflow_id,
            scenario.workflow_name,
            tags_json,
            scenario.session_id,
            scenario.actor_id,
            None,
            None,
            scenario.workflow_id,
            max_depth,
            total_executions,
            success_count,
            failed_count,
            total_duration_ms,
            workflow_status,
            isoformat(min(n.started_at for n in nodes)),
            isoformat(max(n.completed_at for n in nodes)),
            isoformat(scenario.started_at),
            isoformat(datetime.utcnow()),
        ),
    )

    run_metadata = {
        "objective": scenario.workflow_name,
        "deliverable": random.choice(DELIVERABLES),
        "regions": list({scenario.workflow_tags[2]}),
    }

    conn.execute(
        """
        INSERT INTO workflow_runs (
            run_id, root_workflow_id, root_execution_id, status, total_steps,
            completed_steps, failed_steps, state_version, last_event_sequence,
            metadata, created_at, updated_at, completed_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            scenario.run_id,
            scenario.workflow_id,
            scenario.root_execution_id,
            workflow_status,
            total_executions,
            success_count,
            failed_count,
            3,
            3,
            json.dumps(run_metadata).encode(),
            isoformat(scenario.started_at),
            isoformat(datetime.utcnow()),
            isoformat(max(n.completed_at for n in nodes)),
        ),
    )

    execution_rows = []
    event_rows = []
    step_rows = []
    step_lookup = {}
    execution_rows_simple = []

    for node in nodes:
        notes_payload = json.dumps(node.notes).encode()
        execution_rows.append(
            (
                scenario.workflow_id,
                node.execution_id,
                node.request_id,
                scenario.run_id,
                scenario.session_id,
                scenario.actor_id,
                node.agent_node_id,
                scenario.workflow_id,
                node.parent_execution_id,
                scenario.workflow_id,
                node.depth,
                node.reasoner_id,
                json.dumps(node.input_payload).encode(),
                json.dumps(node.output_payload).encode(),
                len(json.dumps(node.input_payload)),
                len(json.dumps(node.output_payload)),
                scenario.workflow_name,
                tags_json,
                node.status,
                isoformat(node.started_at),
                isoformat(node.completed_at),
                node.duration_ms,
                1,
                2,
                0,
                0,
                None,
                node.status_reason,
                None,
                None,
                node.error_message,
                0,
                notes_payload,
                isoformat(node.started_at),
                isoformat(node.completed_at),
            )
        )

        event_rows.extend(
            [
                (
                    node.execution_id,
                    scenario.workflow_id,
                    scenario.run_id,
                    node.parent_execution_id,
                    1,
                    0,
                    "status_changed",
                    "running",
                    None,
                    json.dumps({"detail": "Execution started"}),
                    isoformat(node.started_at),
                    isoformat(node.started_at),
                ),
                (
                    node.execution_id,
                    scenario.workflow_id,
                    scenario.run_id,
                    node.parent_execution_id,
                    2,
                    1,
                    "status_changed",
                    node.status,
                    node.status_reason,
                    json.dumps(
                        {
                            "detail": "Execution completed"
                            if node.status != "failed"
                            else "Execution failed"
                        }
                    ),
                    isoformat(node.completed_at),
                    isoformat(node.completed_at),
                ),
            ]
        )

        step_id = f"step_{uuid.uuid4().hex[:14]}"
        step_lookup[node.execution_id] = step_id
        parent_step_id = (
            step_lookup.get(node.parent_execution_id)
            if node.parent_execution_id
            else None
        )
        step_rows.append(
            (
                step_id,
                scenario.run_id,
                parent_step_id,
                node.execution_id,
                node.agent_node_id,
                f"agent://{node.agent_node_id}",
                "succeeded" if node.status.startswith("succeeded") else node.status,
                1,
                random.randint(0, 4),
                isoformat(node.started_at),
                None,
                None,
                node.error_message,
                b"{}",
                isoformat(node.started_at),
                isoformat(node.completed_at),
                None,
                None,
                isoformat(node.started_at),
                isoformat(node.completed_at),
            )
        )

        execution_rows_simple.append(
            (
                node.execution_id,
                scenario.run_id,
                node.parent_execution_id,
                node.agent_node_id,
                node.reasoner_id,
                node.agent_node_id,
                node.status,
                json.dumps(node.input_payload).encode(),
                json.dumps(node.output_payload).encode(),
                node.error_message,
                None,
                None,
                scenario.session_id,
                scenario.actor_id,
                isoformat(node.started_at),
                isoformat(node.completed_at),
                node.duration_ms,
                isoformat(node.started_at),
                isoformat(node.completed_at),
            )
        )

    placeholders = ",".join("?" for _ in execution_rows[0])
    conn.executemany(
        f"""
        INSERT INTO workflow_executions (
            workflow_id, execution_id, hanzo_agents_request_id, run_id, session_id,
            actor_id, agent_node_id, parent_workflow_id, parent_execution_id,
            root_workflow_id, workflow_depth, reasoner_id, input_data, output_data,
            input_size, output_size, workflow_name, workflow_tags, status,
            started_at, completed_at, duration_ms, state_version, last_event_sequence,
            active_children, pending_children, pending_terminal_status, status_reason,
            lease_owner, lease_expires_at, error_message, retry_count, notes, created_at,
            updated_at
        ) VALUES ({placeholders})
        """,
        execution_rows,
    )

    conn.executemany(
        """
        INSERT INTO workflow_execution_events (
            execution_id, workflow_id, run_id, parent_execution_id, sequence,
            previous_sequence, event_type, status, status_reason, payload,
            emitted_at, recorded_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        event_rows,
    )

    conn.executemany(
        """
        INSERT INTO workflow_steps (
            step_id, run_id, parent_step_id, execution_id, agent_node_id, target,
            status, attempt, priority, not_before, input_uri, result_uri, error_message,
            metadata, started_at, completed_at, leased_at, lease_timeout, created_at,
            updated_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        step_rows,
    )

    run_events = [
        (
            scenario.run_id,
            1,
            0,
            "run_started",
            "running",
            None,
            json.dumps({"kickoff": "approved"}),
            isoformat(scenario.started_at),
            isoformat(scenario.started_at),
        ),
        (
            scenario.run_id,
            2,
            1,
            "run_checkpoint",
            "running",
            None,
            json.dumps({"progress": 0.55}),
            isoformat(scenario.started_at + timedelta(minutes=9)),
            isoformat(scenario.started_at + timedelta(minutes=9)),
        ),
        (
            scenario.run_id,
            3,
            2,
            "run_completed",
            workflow_status,
            None,
            json.dumps({"deliverable": random.choice(DELIVERABLES)}),
            isoformat(max(n.completed_at for n in nodes)),
            isoformat(max(n.completed_at for n in nodes)),
        ),
    ]
    conn.executemany(
        """
        INSERT INTO workflow_run_events (
            run_id, sequence, previous_sequence, event_type, status,
            status_reason, payload, emitted_at, recorded_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        run_events,
    )

    conn.executemany(
        """
        INSERT INTO executions (
            execution_id, run_id, parent_execution_id, agent_node_id, reasoner_id,
            node_id, status, input_payload, result_payload, error_message,
            input_uri, result_uri, session_id, actor_id, started_at, completed_at,
            duration_ms, created_at, updated_at
        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        execution_rows_simple,
    )


def seed_database(args: argparse.Namespace) -> List[Tuple[str, str, int]]:
    if args.seed is not None:
        random.seed(args.seed)

    db_path = args.db_path.expanduser()
    if not db_path.exists():
        raise FileNotFoundError(
            f"SQLite database not found at {db_path}. Start the control plane once to create it."
        )

    with sqlite3.connect(db_path) as conn:
        conn.execute("PRAGMA foreign_keys = ON")
        conn.execute("BEGIN")

        ensure_agent_nodes(conn, args.team_id)
        if args.purge_prefix:
            purge_workflows_with_prefix(conn, args.workflow_prefix)

        inserted: List[Tuple[str, str, int]] = []
        base_start = datetime.utcnow() - timedelta(hours=args.start_hours_ago)

        for wf_index in range(args.workflow_count):
            scenario = generate_scenario(
                prefix=args.workflow_prefix,
                session_prefix=args.session_prefix,
                actor_pool=args.actor_pool,
                idx=wf_index,
                started_at=base_start
                + timedelta(minutes=wf_index * args.stagger_minutes),
                nodes_per_workflow=args.nodes_per_workflow,
            )
            nodes = synthesize_nodes(
                scenario, nodes_per_workflow=args.nodes_per_workflow
            )
            insert_workflow(conn, scenario, nodes)
            inserted.append((scenario.workflow_id, scenario.run_id, len(nodes)))

        conn.commit()

    return inserted


def main(argv: Sequence[str]) -> int:
    args = parse_args(argv)
    if args.nodes_per_workflow <= 0:
        print("nodes-per-workflow must be positive.", file=sys.stderr)
        return 1
    if args.workflow_count <= 0:
        print("workflow-count must be positive.", file=sys.stderr)
        return 1
    if args.dry_run:
        print("Dry-run mode currently not implemented. Remove --dry-run to write data.")
        return 0

    try:
        inserted = seed_database(args)
    except Exception as exc:
        print(f"Failed to seed database: {exc}", file=sys.stderr)
        return 2

    print(f"Seeded {len(inserted)} workflow(s) into {args.db_path.expanduser()}")
    for workflow_id, run_id, total_nodes in inserted[:10]:
        print(f"  • {workflow_id} / {run_id} ({total_nodes} executions incl. root)")
    if len(inserted) > 10:
        print("  • ...")

    if inserted:
        sample_workflow_id, sample_run_id, _ = inserted[0]
        print("\nValidate with:")
        print(
            "  curl 'http://localhost:8080/api/ui/v1/executions/enhanced?sort_by=started_at&sort_order=desc&page=1&limit=5'"
        )
        print(
            f"  curl 'http://localhost:8080/api/ui/v1/workflows/{sample_workflow_id}/dag'"
        )
        print(f"  curl 'http://localhost:8080/api/ui/v2/workflow-runs/{sample_run_id}'")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))