"""SQLite schema for graphlemon's own structured storage.

U-001 scope: the 15 platform entities from docs/sdd/01-M-模型设计.md §1.0-1.14
(PH0-PH4). `deployment`/`deploy_health_signal` (§1.15/§1.16, added here in U-034,
2026-08-02) are PH5-only. `work_graph_loop_limit_config` (§1.11.1, 2026-08-08,
Work Graph self-planning + loop-back) added post-PH5 -- 18 tables total.

Interpretive conventions used below, where 01-M leaves the physical type implicit:
- Booleans are stored as INTEGER CHECK(x IN (0, 1)) -- sqlite3 has no native bool type.
- A `*_at` column that records "when this row was created" (or, for skill_iteration,
  when the row's one-shot event happened) is NOT NULL DEFAULT CURRENT_TIMESTAMP.
  A `*_at` column that records a later lifecycle event which may not have happened
  yet (started_at, completed_at, purged_at, ...) is nullable with no default -- the
  application sets it only when that event actually occurs.
- JSON-typed columns (graph_manifest, nodes, edges, payload, ...) are stored as TEXT;
  (de)serialization is an application-layer concern, not enforced by this schema.
- task.assigned_org_node is documented as "not null, fk" but there is no standalone
  org-node lookup table to reference (planner/coder/validator only exist as entries
  inside org_graph_definition.nodes, a JSON column) -- implemented as a CHECK enum
  instead, matching skill.owner_org_node / agent_profile.org_node / work_node_execution.subagent_type
  (work_graph_run.org_node existed the same way until it was removed, Org Graph unification
  Phase 2c §2.1, 2026-08-12 -- see that column's own removal comment above work_graph_run).
- bug_record.root_cause_category gained 'core_engine' in U-034 (2026-08-02, PH5) --
  05 §5.8: self-target scenarios (`.claude/agents/*.md` identity file changes, Graph
  Engine scheduler/executor core logic changes, this schema's own changes), routed by
  `c_org_loop_route` to `p_deploy_supervisor_blue_green` instead of fast_lane/full_replan.
- bug_record and skill have no created_at in 01-M (skill is Git-versioned, so git log
  covers "when"; bug_record has no such backing and this looks like a possible spec
  gap, but adding an undocumented column is out of scope for U-001 -- flagged, not fixed).
- deployment.previous_commit_hash/previous_instance_dir are "not null" per 01-M 1.15,
  but the very first deployment ever run has no prior `finalized` deployment to point
  at -- that bootstrap case is c_deploy_supervisor_stage's problem (U-037), not this
  schema's: it must supply the currently-running commit/checkout dir as the "previous"
  values on that first call, not leave the schema-level NOT NULL as aspirational.
"""

import sqlite3

TABLE_NAMES = (
    "project",
    "org_graph_definition",
    "skill",
    "task",
    "work_graph_run",
    "work_node_execution",
    "artifact",
    "bug_record",
    "skill_iteration",
    "agent_profile",
    "budget_guardrail_config",
    "org_concurrency_config",
    "work_graph_loop_limit_config",
    "execution_event",
    "execution_log_file",
    "log_retention_config",
    "deployment",
    "deploy_health_signal",
)

SCHEMA_SQL = """
-- 1.0 project
CREATE TABLE project (
    project_id TEXT PRIMARY KEY,
    name TEXT NOT NULL,
    target_folder_path TEXT NOT NULL UNIQUE,
    has_git_repo INTEGER NOT NULL CHECK (has_git_repo IN (0, 1)),
    graphlemon_subdir_initialized INTEGER NOT NULL DEFAULT 0
        CHECK (graphlemon_subdir_initialized IN (0, 1)),
    status TEXT NOT NULL CHECK (status IN ('active', 'archived')),
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

-- 1.1 org_graph_definition
CREATE TABLE org_graph_definition (
    graph_id TEXT PRIMARY KEY,
    version TEXT NOT NULL,
    nodes TEXT NOT NULL,
    edges TEXT NOT NULL,
    file_path TEXT NOT NULL,
    status TEXT NOT NULL CHECK (status IN ('draft', 'active', 'deprecated'))
);

-- 1.7 skill (created before task/work_node_execution: work_node_execution.skill_ref
-- references it)
CREATE TABLE skill (
    skill_id TEXT PRIMARY KEY,
    -- H (2026-08-06, Role Package abstraction): was CHECK(owner_org_node IN
    -- ('planner','coder','validator')) -- relaxed to a plain non-empty-string
    -- check so a new role, added purely via roles/<role_id>/role.json (no
    -- schema.py/agent_profile.py code change), isn't rejected by a stale
    -- hardcoded enum. See docs/done/role-package-abstraction-design.md §4.
    owner_org_node TEXT NOT NULL CHECK (owner_org_node != ''),
    file_path TEXT NOT NULL,
    current_git_commit_sha TEXT NOT NULL,
    source TEXT NOT NULL CHECK (source IN ('migrated_from_hermes', 'new', 'auto_iterated')),
    migrated_from_profile TEXT CHECK (migrated_from_profile IN ('code', 'planning', 'qa')),
    status TEXT NOT NULL CHECK (status IN ('active', 'archived'))
);

-- 1.2 task
CREATE TABLE task (
    task_id TEXT PRIMARY KEY,
    project_id TEXT NOT NULL REFERENCES project(project_id),
    goal_text TEXT NOT NULL,
    task_type TEXT NOT NULL CHECK (task_type IN ('goal', 'bug_fix')),
    origin_ref TEXT,
    created_via TEXT NOT NULL DEFAULT 'api' CHECK (created_via IN ('api', 'human_console')),
    status TEXT NOT NULL CHECK (status IN (
        'queued', 'planning_brd_prd_sdd', 'reviewing_planner_output', 'planning_failed',
        'coding', 'coding_failed', 'validating',
        'routed_fast_lane', 'routed_full_replan',
        'blocked_pending_human', 'done'
    )),
    route TEXT CHECK (route IN ('fast_lane', 'full_replan')),
    loop_round INTEGER NOT NULL DEFAULT 0,
    blocked_reason TEXT,
    -- H (2026-08-06): see skill.owner_org_node's comment above -- same relaxation.
    assigned_org_node TEXT NOT NULL CHECK (assigned_org_node != ''),
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_task_status ON task(status);
CREATE INDEX idx_task_assigned_org_node ON task(assigned_org_node);
CREATE INDEX idx_task_project_id ON task(project_id);

-- 1.3 work_graph_run
CREATE TABLE work_graph_run (
    run_id TEXT PRIMARY KEY,
    task_id TEXT NOT NULL REFERENCES task(task_id),
    -- org_node column removed (Org Graph unification Phase 2c §2.1,
    -- 2026-08-12, user-confirmed 2026-08-08): a merged graph's one run
    -- genuinely spans several org_nodes (planner/coder/validator all
    -- append their own nodes into the SAME run), so a single run-level
    -- value was never accurate once Phase 2b/2c landed -- the real
    -- per-node ownership already lives in work_node_execution.subagent_type
    -- (see §1.2 of the design doc). See migrate_drop_work_graph_run_org_node.
    graph_manifest TEXT NOT NULL,
    orchestration_script_ref TEXT,
    status TEXT NOT NULL CHECK (status IN (
        'planning', 'running', 'merging', 'verifying',
        'reflecting', 'blocked', 'done', 'failed'
    )),
    started_at TEXT,
    completed_at TEXT,
    result_artifact_refs TEXT,
    -- D (2026-08-06, time-travel debugging): both NULL for a normal run.
    -- Set together when this run was forked from a historical node instead
    -- of planned fresh -- see src/work_graph_fork.py. v1 only forks from a
    -- parent run that has already reached a terminal status, so a forked
    -- run's task_id never collides with ux_work_graph_run_active below
    -- (that index only covers non-terminal statuses) -- no change needed
    -- to the index itself.
    parent_run_id TEXT REFERENCES work_graph_run(run_id),
    fork_of_node_exec_id TEXT REFERENCES work_node_execution(node_exec_id)
);

-- Org Graph unification Phase 2c §2.2 (2026-08-12): narrowed from
-- (task_id, org_node) to plain task_id -- a merged graph means a task has
-- ONE run spanning its whole lifecycle now, not one per org_node. 'blocked'
-- stays counted as active on purpose (not just 'planning'/'running'/
-- 'merging'/'verifying'/'reflecting'): resuming a blocked run appends into
-- the SAME row (plan_work_graph(..., append_to_run_id=...)), it never
-- INSERTs a fresh one, so a blocked run must still hold the (task_id) slot.
-- Only 'done'/'failed' are genuinely terminal.
CREATE UNIQUE INDEX ux_work_graph_run_active ON work_graph_run(task_id)
    WHERE status NOT IN ('done', 'failed');

-- 1.4 work_node_execution
CREATE TABLE work_node_execution (
    node_exec_id TEXT PRIMARY KEY,
    run_id TEXT NOT NULL REFERENCES work_graph_run(run_id),
    node_id TEXT NOT NULL,
    subagent_type TEXT NOT NULL,
    skill_ref TEXT NOT NULL REFERENCES skill(skill_id),
    input_schema_ref TEXT NOT NULL,
    output_schema_ref TEXT NOT NULL,
    input_payload TEXT NOT NULL,
    output_payload TEXT,
    model_tier TEXT NOT NULL DEFAULT 'standard' CHECK (model_tier IN ('cheap', 'standard', 'premium')),
    status TEXT NOT NULL CHECK (status IN (
        'pending', 'running', 'reflecting', 'retrying',
        'waiting_for_quota', 'blocked', 'done', 'failed'
    )),
    retry_count INTEGER NOT NULL DEFAULT 0,
    reflection_notes TEXT,
    git_worktree_ref TEXT,
    quota_wait_started_at TEXT,
    quota_limit_type TEXT CHECK (quota_limit_type IN ('session', 'weekly', 'opus'))
);

-- 1.5 artifact
CREATE TABLE artifact (
    artifact_id TEXT PRIMARY KEY,
    artifact_type TEXT NOT NULL CHECK (artifact_type IN (
        'brd', 'prd', 'sdd', 'benchmark', 'bug_record', 'code_diff', 'test_report',
        'test_dataset', 'test_script',
        'validator_benchmark', 'validator_test_dataset', 'validator_test_script'
    )),
    task_id TEXT NOT NULL REFERENCES task(task_id),
    produced_by_run_id TEXT NOT NULL REFERENCES work_graph_run(run_id),
    file_path TEXT NOT NULL,
    git_commit_sha TEXT NOT NULL,
    depends_on_artifact_ids TEXT,
    validation_status TEXT NOT NULL DEFAULT 'pending' CHECK (validation_status IN ('pending', 'passed', 'failed')),
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE UNIQUE INDEX ux_artifact_task_type_commit ON artifact(task_id, artifact_type, git_commit_sha);

-- 1.6 bug_record (skill_iteration_suggestion_id forward-references skill_iteration,
-- created further below -- SQLite does not validate FK target existence at CREATE
-- TABLE time, only at DML time, so table creation order here is fine)
CREATE TABLE bug_record (
    bug_id TEXT PRIMARY KEY,
    task_id TEXT REFERENCES task(task_id),
    source TEXT NOT NULL CHECK (source IN ('validator', 'human')),
    reported_by TEXT,
    description TEXT NOT NULL,
    repro_steps TEXT,
    root_cause_category TEXT CHECK (root_cause_category IN ('planning', 'coder', 'skill_defect', 'core_engine')),
    root_cause_detail TEXT,
    related_artifact_ids TEXT,
    suggested_route TEXT CHECK (suggested_route IN ('fast_lane', 'full_replan')),
    skill_iteration_suggestion_id TEXT REFERENCES skill_iteration(iteration_id),
    status TEXT NOT NULL CHECK (status IN ('open', 'triaging', 'routed', 'resolved'))
);

-- 1.8 skill_iteration
CREATE TABLE skill_iteration (
    iteration_id TEXT PRIMARY KEY,
    skill_id TEXT NOT NULL REFERENCES skill(skill_id),
    triggered_by_bug_id TEXT NOT NULL REFERENCES bug_record(bug_id),
    before_git_commit_sha TEXT NOT NULL,
    after_git_commit_sha TEXT NOT NULL,
    change_summary TEXT NOT NULL,
    applied_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    rollback_of_iteration_id TEXT REFERENCES skill_iteration(iteration_id)
);

-- 1.9 agent_profile
CREATE TABLE agent_profile (
    -- H (2026-08-06): see skill.owner_org_node's comment above -- same relaxation.
    org_node TEXT PRIMARY KEY CHECK (org_node != ''),
    soul_file_path TEXT NOT NULL,
    agents_file_path TEXT,
    -- Org Graph unification Phase 1 (2026-08-08): NOT NULL dropped -- a role with no
    -- legacy Hermes profile to migrate from (e.g. planner_reviewer, role.json's
    -- migrated_from_profile: null) has nothing truthy to put here. Matches
    -- skill.migrated_from_profile, which was already nullable for the same reason.
    migrated_from_profile TEXT CHECK (migrated_from_profile IN ('planning', 'code', 'qa')),
    current_git_commit_sha TEXT NOT NULL
);

-- 1.10 budget_guardrail_config
CREATE TABLE budget_guardrail_config (
    config_id TEXT PRIMARY KEY DEFAULT 'default',
    scope TEXT NOT NULL CHECK (scope IN ('global', 'per_task', 'per_day')),
    limit_type TEXT NOT NULL CHECK (limit_type IN ('token_count', 'call_count')),
    limit_value INTEGER NOT NULL,
    current_usage INTEGER NOT NULL DEFAULT 0,
    on_exceed_action TEXT NOT NULL DEFAULT 'pause_and_notify',
    updated_by TEXT NOT NULL,
    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

-- 1.11 org_concurrency_config
CREATE TABLE org_concurrency_config (
    config_id TEXT PRIMARY KEY DEFAULT 'default',
    max_concurrent_goals INTEGER NOT NULL DEFAULT 1,
    updated_by TEXT NOT NULL,
    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

-- 1.12 execution_event
CREATE TABLE execution_event (
    event_id TEXT PRIMARY KEY,
    project_id TEXT NOT NULL REFERENCES project(project_id),
    task_id TEXT REFERENCES task(task_id),
    run_id TEXT REFERENCES work_graph_run(run_id),
    node_exec_id TEXT REFERENCES work_node_execution(node_exec_id),
    event_type TEXT NOT NULL CHECK (event_type IN (
        'task_status_changed', 'node_status_changed', 'artifact_created',
        'bug_recorded', 'loop_routed', 'budget_updated', 'skill_iterated',
        'log_appended', 'quota_wait_started', 'loop_back_triggered', 'planner_review_verdict',
        'concurrency_wait'
    )),
    payload TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_execution_event_project_created ON execution_event(project_id, created_at);
CREATE INDEX idx_execution_event_task_created ON execution_event(task_id, created_at);

-- 1.11.1 work_graph_loop_limit_config (Work Graph self-planning + loop-back,
-- docs/done/work-graph-self-planning-and-loop-back-design.md §6)
CREATE TABLE work_graph_loop_limit_config (
    config_id TEXT PRIMARY KEY,
    max_loop_attempts INTEGER NOT NULL,
    updated_by TEXT,
    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

-- 1.13 execution_log_file
CREATE TABLE execution_log_file (
    log_file_id TEXT PRIMARY KEY,
    node_exec_id TEXT NOT NULL UNIQUE REFERENCES work_node_execution(node_exec_id),
    project_id TEXT NOT NULL REFERENCES project(project_id),
    file_path TEXT NOT NULL,
    format TEXT NOT NULL CHECK (format IN ('jsonl', 'raw_text')),
    size_bytes INTEGER NOT NULL DEFAULT 0,
    status TEXT NOT NULL CHECK (status IN ('active', 'closed', 'purged')),
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    purged_at TEXT
);

CREATE INDEX idx_execution_log_file_project ON execution_log_file(project_id);

-- 1.14 log_retention_config
CREATE TABLE log_retention_config (
    config_id TEXT PRIMARY KEY DEFAULT 'default',
    retention_days INTEGER NOT NULL DEFAULT 14,
    max_total_size_mb INTEGER,
    updated_by TEXT NOT NULL,
    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

-- 1.15 deployment (PH5, U-034, 2026-08-02 -- see docs/done/supervisor-auto-rollback-design.md)
CREATE TABLE deployment (
    deployment_id TEXT PRIMARY KEY,
    git_commit_hash TEXT NOT NULL,
    previous_commit_hash TEXT NOT NULL,
    status TEXT NOT NULL CHECK (status IN (
        'staged', 'candidate_starting', 'health_check_running', 'health_check_passed',
        'health_check_failed', 'cutover_in_progress', 'live', 'monitoring',
        'anomaly_detected', 'rollback_in_progress', 'rolled_back', 'rollback_failed',
        'blocked_pending_human', 'monitoring_window_clean', 'finalized'
    )),
    candidate_port INTEGER,
    candidate_db_path TEXT,
    -- Not in supervisor-auto-rollback-design.md §3's field table (found
    -- missing while building U-037's c_deploy_supervisor_stage): the
    -- checkout directory a candidate is actually running from has to be
    -- persisted somewhere, or a Deploy Supervisor restart between staging
    -- and cutover would lose track of it entirely -- the same class of gap
    -- REQ-06 exists to prevent everywhere else. Once a deployment reaches
    -- 'live'/'finalized', this same directory IS what the next deployment's
    -- `previous_instance_dir` bootstraps from (see deploy_stage.py).
    candidate_instance_dir TEXT,
    started_at TEXT,
    cutover_at TEXT,
    monitoring_closed_at TEXT,
    rollback_at TEXT,
    rollback_reason TEXT,
    previous_instance_dir TEXT NOT NULL
);

-- only one non-terminal deployment at a time (01-M 1.15's own constraint) --
-- expression index on a constant, same "global singleton while active" idiom
-- as a partial unique index scoped by a real column, just with no column to
-- scope by here (there's exactly one graphlemon self-target, not one per key)
CREATE UNIQUE INDEX ux_deployment_active ON deployment((1))
    WHERE status NOT IN ('finalized', 'rolled_back', 'blocked_pending_human');

-- 1.16 deploy_health_signal (PH5, U-034, 2026-08-02)
CREATE TABLE deploy_health_signal (
    signal_id TEXT PRIMARY KEY,
    deployment_id TEXT NOT NULL REFERENCES deployment(deployment_id),
    sampled_at TEXT NOT NULL,
    signal_type TEXT NOT NULL CHECK (signal_type IN (
        'process_alive', 'http_health', 'canary_task', 'validator_approval_rate'
    )),
    status TEXT NOT NULL CHECK (status IN ('ok', 'degraded', 'fail')),
    detail TEXT
);

CREATE INDEX idx_deploy_health_signal_deployment_sampled ON deploy_health_signal(deployment_id, sampled_at);
"""


def connect(db_path=":memory:", check_same_thread=True):
    """Open a sqlite3 connection with foreign-key enforcement turned on.

    SQLite disables FK enforcement by default per-connection; every caller needs
    this pragma, so it lives here rather than being repeated at every call site.

    check_same_thread=False is for callers like backend_server.py that share
    one connection across a ThreadingHTTPServer's per-request threads -- the
    caller is then responsible for serializing access itself (a lock), since
    turning this off only disables sqlite3's own thread-identity check, it
    doesn't make concurrent use of one connection object safe by itself.
    """
    conn = sqlite3.connect(db_path, check_same_thread=check_same_thread)
    conn.execute("PRAGMA foreign_keys = ON")
    return conn


def create_schema(conn):
    """Create all 17 tables (15 from U-001 + deployment/deploy_health_signal
    from U-034, PH5) + indexes on an open connection."""
    conn.executescript(SCHEMA_SQL)
    conn.commit()


def migrate_expand_artifact_types(conn):
    """2026-08-06: adds test_dataset/test_script/validator_benchmark/
    validator_test_dataset/validator_test_script to artifact.artifact_type's
    CHECK enum on an ALREADY-POPULATED database -- SQLite has no
    `ALTER TABLE ... MODIFY CONSTRAINT`, so this is the standard "create a
    new table with the wanted constraint, copy every row, drop the old
    table, rename" dance. No other table has a foreign key into `artifact`
    (checked), so this doesn't need to touch anything beyond the table
    itself and its one index.

    Idempotent and safe to call unconditionally (including against a
    freshly create_schema()'d database, which already has the expanded
    constraint) -- checks the live constraint text first and no-ops if it's
    already expanded, so callers never need to special-case "is this a
    fresh or an upgraded database".
    """
    row = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='artifact'").fetchone()
    if row is None or "'test_dataset'" in row[0]:
        return
    conn.executescript(
        """
        CREATE TABLE artifact_new (
            artifact_id TEXT PRIMARY KEY,
            artifact_type TEXT NOT NULL CHECK (artifact_type IN (
                'brd', 'prd', 'sdd', 'benchmark', 'bug_record', 'code_diff', 'test_report',
                'test_dataset', 'test_script',
                'validator_benchmark', 'validator_test_dataset', 'validator_test_script'
            )),
            task_id TEXT NOT NULL REFERENCES task(task_id),
            produced_by_run_id TEXT NOT NULL REFERENCES work_graph_run(run_id),
            file_path TEXT NOT NULL,
            git_commit_sha TEXT NOT NULL,
            depends_on_artifact_ids TEXT,
            validation_status TEXT NOT NULL DEFAULT 'pending' CHECK (validation_status IN ('pending', 'passed', 'failed')),
            created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        INSERT INTO artifact_new SELECT * FROM artifact;
        DROP TABLE artifact;
        ALTER TABLE artifact_new RENAME TO artifact;
        CREATE UNIQUE INDEX ux_artifact_task_type_commit ON artifact(task_id, artifact_type, git_commit_sha);
        """
    )
    conn.commit()


def migrate_add_work_graph_fork_columns(conn):
    """D (2026-08-06, time-travel debugging): adds parent_run_id/
    fork_of_node_exec_id to work_graph_run on an ALREADY-POPULATED
    database. Unlike migrate_expand_artifact_types, this is a plain `ALTER
    TABLE ... ADD COLUMN` (no CHECK constraint involved, so no need for the
    create-copy-drop-rename dance). Idempotent and safe to call
    unconditionally, including against a freshly create_schema()'d database
    (which already has both columns)."""
    cols = {row[1] for row in conn.execute("PRAGMA table_info(work_graph_run)").fetchall()}
    if "parent_run_id" not in cols:
        conn.execute("ALTER TABLE work_graph_run ADD COLUMN parent_run_id TEXT REFERENCES work_graph_run(run_id)")
    if "fork_of_node_exec_id" not in cols:
        conn.execute(
            "ALTER TABLE work_graph_run ADD COLUMN fork_of_node_exec_id TEXT "
            "REFERENCES work_node_execution(node_exec_id)"
        )
    conn.commit()


def _table_sql(conn, table_name):
    row = conn.execute(
        "SELECT sql FROM sqlite_master WHERE type='table' AND name=?", (table_name,)
    ).fetchone()
    return row[0] if row else None


def migrate_relax_org_node_constraints(conn):
    """H (2026-08-06, Role Package abstraction): relaxes the 4 tables whose
    org_node-ish column was CHECK-constrained to exactly ('planner',
    'coder', 'validator') down to CHECK(x != '') -- a new role, added
    purely via roles/<role_id>/role.json (see src/agent_profile.py), no
    longer gets rejected by a schema-level enum baked in at U-001.

    **Why all 4 need the create-copy-drop-rename dance, not a plain ALTER
    TABLE**: SQLite has no `ALTER TABLE ... DROP CONSTRAINT` for CHECK.
    `skill`/`task`/`work_graph_run` are referenced by several other tables'
    foreign keys (work_node_execution, artifact, bug_record,
    execution_event, work_graph_run's own self-reference) -- but since only
    the PARENT table's own CHECK text changes (the FK column definitions
    in those child tables, e.g. `REFERENCES task(task_id)`, don't change at
    all), only these 4 parent tables need rebuilding, none of their
    children. Followed the SQLite-documented procedure for schema changes
    under active foreign keys ("Making Other Kinds Of Table Schema
    Changes"): PRAGMA foreign_keys=OFF for the duration (FK enforcement
    can't be toggled mid-transaction), rebuild, PRAGMA foreign_key_check to
    confirm nothing went dangling, PRAGMA foreign_keys=ON again.

    Depends on migrate_add_work_graph_fork_columns() having already run
    (called here unconditionally, it's idempotent) -- work_graph_run's
    rebuild needs to know about D's parent_run_id/fork_of_node_exec_id
    columns whether or not this specific database has seen that migration
    yet.

    Idempotent: each table's rebuild is skipped if its stored CREATE TABLE
    text no longer contains the old enum CHECK, so a fresh
    create_schema()'d database (already on the new constraint) is a no-op,
    same as migrate_expand_artifact_types's own convention.
    """
    migrate_add_work_graph_fork_columns(conn)
    conn.commit()

    conn.execute("PRAGMA foreign_keys = OFF")
    try:
        if "owner_org_node IN (" in (_table_sql(conn, "skill") or ""):
            conn.executescript(
                """
                CREATE TABLE skill_new (
                    skill_id TEXT PRIMARY KEY,
                    owner_org_node TEXT NOT NULL CHECK (owner_org_node != ''),
                    file_path TEXT NOT NULL,
                    current_git_commit_sha TEXT NOT NULL,
                    source TEXT NOT NULL CHECK (source IN ('migrated_from_hermes', 'new', 'auto_iterated')),
                    migrated_from_profile TEXT CHECK (migrated_from_profile IN ('code', 'planning', 'qa')),
                    status TEXT NOT NULL CHECK (status IN ('active', 'archived'))
                );
                INSERT INTO skill_new SELECT * FROM skill;
                DROP TABLE skill;
                ALTER TABLE skill_new RENAME TO skill;
                """
            )

        if "assigned_org_node IN (" in (_table_sql(conn, "task") or ""):
            conn.executescript(
                """
                CREATE TABLE task_new (
                    task_id TEXT PRIMARY KEY,
                    project_id TEXT NOT NULL REFERENCES project(project_id),
                    goal_text TEXT NOT NULL,
                    task_type TEXT NOT NULL CHECK (task_type IN ('goal', 'bug_fix')),
                    origin_ref TEXT,
                    created_via TEXT NOT NULL DEFAULT 'api' CHECK (created_via IN ('api', 'human_console')),
                    status TEXT NOT NULL CHECK (status IN (
                        'queued', 'planning_brd_prd_sdd', 'planning_failed',
                        'coding', 'coding_failed', 'validating',
                        'routed_fast_lane', 'routed_full_replan',
                        'blocked_pending_human', 'done'
                    )),
                    route TEXT CHECK (route IN ('fast_lane', 'full_replan')),
                    loop_round INTEGER NOT NULL DEFAULT 0,
                    blocked_reason TEXT,
                    assigned_org_node TEXT NOT NULL CHECK (assigned_org_node != ''),
                    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
                    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
                );
                INSERT INTO task_new SELECT * FROM task;
                DROP TABLE task;
                ALTER TABLE task_new RENAME TO task;
                CREATE INDEX idx_task_status ON task(status);
                CREATE INDEX idx_task_assigned_org_node ON task(assigned_org_node);
                CREATE INDEX idx_task_project_id ON task(project_id);
                """
            )

        if "org_node IN (" in (_table_sql(conn, "work_graph_run") or ""):
            conn.executescript(
                """
                CREATE TABLE work_graph_run_new (
                    run_id TEXT PRIMARY KEY,
                    task_id TEXT NOT NULL REFERENCES task(task_id),
                    org_node TEXT NOT NULL CHECK (org_node != ''),
                    graph_manifest TEXT NOT NULL,
                    orchestration_script_ref TEXT,
                    status TEXT NOT NULL CHECK (status IN (
                        'planning', 'running', 'merging', 'verifying',
                        'reflecting', 'blocked', 'done', 'failed'
                    )),
                    started_at TEXT,
                    completed_at TEXT,
                    result_artifact_refs TEXT,
                    parent_run_id TEXT REFERENCES work_graph_run(run_id),
                    fork_of_node_exec_id TEXT REFERENCES work_node_execution(node_exec_id)
                );
                INSERT INTO work_graph_run_new SELECT * FROM work_graph_run;
                DROP TABLE work_graph_run;
                ALTER TABLE work_graph_run_new RENAME TO work_graph_run;
                CREATE UNIQUE INDEX ux_work_graph_run_active ON work_graph_run(task_id, org_node)
                    WHERE status IN ('planning', 'running', 'merging', 'verifying');
                """
            )

        if "org_node TEXT PRIMARY KEY CHECK (org_node IN (" in (_table_sql(conn, "agent_profile") or ""):
            conn.executescript(
                """
                CREATE TABLE agent_profile_new (
                    org_node TEXT PRIMARY KEY CHECK (org_node != ''),
                    soul_file_path TEXT NOT NULL,
                    agents_file_path TEXT,
                    migrated_from_profile TEXT NOT NULL CHECK (migrated_from_profile IN ('planning', 'code', 'qa')),
                    current_git_commit_sha TEXT NOT NULL
                );
                INSERT INTO agent_profile_new SELECT * FROM agent_profile;
                DROP TABLE agent_profile;
                ALTER TABLE agent_profile_new RENAME TO agent_profile;
                """
            )

        conn.commit()
        violations = conn.execute("PRAGMA foreign_key_check").fetchall()
        if violations:
            raise RuntimeError(
                "migrate_relax_org_node_constraints left dangling foreign keys: {}".format(violations)
            )
    finally:
        conn.execute("PRAGMA foreign_keys = ON")


def migrate_relax_agent_profile_migrated_from_profile(conn):
    """Org Graph unification Phase 1 (2026-08-08): drops the NOT NULL on
    agent_profile.migrated_from_profile on an ALREADY-POPULATED database --
    surfaced by adding the planner_reviewer role, which (like any future
    role with no legacy Hermes profile) has role.json migrated_from_profile:
    null, and record_agent_profiles() passes that straight through into this
    column. skill.migrated_from_profile was already nullable for the exact
    same reason (see its own comment); this table's NOT NULL was a leftover
    nobody hit until a role with no migration history actually existed.
    Rebuilds to the final relaxed org_node form regardless of whether
    migrate_relax_org_node_constraints has already run against this
    database -- harmless either way, this table needs rebuilding regardless.
    Idempotent and safe to call unconditionally, including against a freshly
    create_schema()'d database (which already has the nullable column)."""
    row = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='agent_profile'").fetchone()
    if row is None or "migrated_from_profile TEXT NOT NULL CHECK" not in row[0]:
        return
    conn.execute("PRAGMA foreign_keys = OFF")
    try:
        conn.executescript(
            """
            CREATE TABLE agent_profile_new (
                org_node TEXT PRIMARY KEY CHECK (org_node != ''),
                soul_file_path TEXT NOT NULL,
                agents_file_path TEXT,
                migrated_from_profile TEXT CHECK (migrated_from_profile IN ('planning', 'code', 'qa')),
                current_git_commit_sha TEXT NOT NULL
            );
            INSERT INTO agent_profile_new SELECT * FROM agent_profile;
            DROP TABLE agent_profile;
            ALTER TABLE agent_profile_new RENAME TO agent_profile;
            """
        )
        conn.commit()
        violations = conn.execute("PRAGMA foreign_key_check").fetchall()
        if violations:
            raise RuntimeError(
                "migrate_relax_agent_profile_migrated_from_profile left dangling foreign keys: {}".format(violations)
            )
    finally:
        conn.execute("PRAGMA foreign_keys = ON")


def migrate_add_work_graph_loop_limit_config(conn):
    """2026-08-08 (Work Graph self-planning + loop-back,
    docs/done/work-graph-self-planning-and-loop-back-design.md §6): adds
    work_graph_loop_limit_config on an ALREADY-POPULATED database. Plain
    `CREATE TABLE IF NOT EXISTS` -- a brand-new table, no CHECK constraint
    to rebuild around, unlike migrate_expand_artifact_types. Idempotent and
    safe to call unconditionally, including against a freshly
    create_schema()'d database (which already has it)."""
    conn.execute(
        "CREATE TABLE IF NOT EXISTS work_graph_loop_limit_config ("
        "config_id TEXT PRIMARY KEY, max_loop_attempts INTEGER NOT NULL, "
        "updated_by TEXT, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP)"
    )
    conn.commit()


def migrate_add_loop_back_event_type(conn):
    """2026-08-08 (Work Graph self-planning + loop-back,
    docs/done/work-graph-self-planning-and-loop-back-design.md §7): adds
    'loop_back_triggered' to execution_event.event_type's CHECK enum on an
    ALREADY-POPULATED database -- same create-copy-drop-rename dance as
    migrate_expand_artifact_types (SQLite has no ALTER-based constraint
    migration). Idempotent and safe to call unconditionally, including
    against a freshly create_schema()'d database (which already has the
    expanded constraint)."""
    row = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='execution_event'").fetchone()
    if row is None or "'loop_back_triggered'" in row[0]:
        return
    conn.executescript(
        """
        CREATE TABLE execution_event_new (
            event_id TEXT PRIMARY KEY,
            project_id TEXT NOT NULL REFERENCES project(project_id),
            task_id TEXT REFERENCES task(task_id),
            run_id TEXT REFERENCES work_graph_run(run_id),
            node_exec_id TEXT REFERENCES work_node_execution(node_exec_id),
            event_type TEXT NOT NULL CHECK (event_type IN (
                'task_status_changed', 'node_status_changed', 'artifact_created',
                'bug_recorded', 'loop_routed', 'budget_updated', 'skill_iterated',
                'log_appended', 'quota_wait_started', 'loop_back_triggered'
            )),
            payload TEXT NOT NULL,
            created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        INSERT INTO execution_event_new SELECT * FROM execution_event;
        DROP TABLE execution_event;
        ALTER TABLE execution_event_new RENAME TO execution_event;
        CREATE INDEX idx_execution_event_project_created ON execution_event(project_id, created_at);
        CREATE INDEX idx_execution_event_task_created ON execution_event(task_id, created_at);
        """
    )
    conn.commit()


def migrate_add_reviewing_planner_output_status(conn):
    """2026-08-08 (Org Graph unification Phase 1,
    docs/done/org-graph-unification-phase1-review-gate-design.md §3.1): adds
    'reviewing_planner_output' to task.status's CHECK enum on an
    ALREADY-POPULATED database -- same create-copy-drop-rename dance as
    migrate_expand_artifact_types/migrate_add_loop_back_event_type. Idempotent
    and safe to call unconditionally, including against a freshly
    create_schema()'d database (which already has the expanded constraint).

    **2026-08-12 real-restart incident fix**: `task` is a PARENT table --
    work_graph_run/artifact/execution_event/bug_record all REFERENCES
    task(task_id) -- unlike execution_event (a leaf table nothing points
    at), which is why migrate_add_planner_review_verdict_event_type doesn't
    need this. Rebuilding a referenced parent table with
    `PRAGMA foreign_keys=ON` (schema.connect()'s default -- see its own
    docstring) trips FK enforcement on the DROP TABLE step the instant any
    child row exists, even though the surviving task_id column's type/PK
    are unchanged. Caught for real restarting the live production
    backend_server against ~/.graphlemon/graphlemon.db (months of real
    work_graph_run/artifact/execution_event history) -- every unit test for
    this migration had passed because none of them seeded a single
    FK-referencing child row, so DROP TABLE never actually exercised FK
    enforcement in-test. Now follows the exact
    migrate_relax_org_node_constraints() precedent (that function's own
    docstring explains the same PRAGMA foreign_keys=OFF/rebuild/
    foreign_key_check/ON dance in more depth) instead of the leaf-table
    executescript() shortcut this originally (wrongly) copied."""
    row = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='task'").fetchone()
    if row is None or "'reviewing_planner_output'" in row[0]:
        return
    conn.execute("PRAGMA foreign_keys = OFF")
    try:
        conn.executescript(
            """
            CREATE TABLE task_new (
                task_id TEXT PRIMARY KEY,
                project_id TEXT NOT NULL REFERENCES project(project_id),
                goal_text TEXT NOT NULL,
                task_type TEXT NOT NULL CHECK (task_type IN ('goal', 'bug_fix')),
                origin_ref TEXT,
                created_via TEXT NOT NULL DEFAULT 'api' CHECK (created_via IN ('api', 'human_console')),
                status TEXT NOT NULL CHECK (status IN (
                    'queued', 'planning_brd_prd_sdd', 'reviewing_planner_output', 'planning_failed',
                    'coding', 'coding_failed', 'validating',
                    'routed_fast_lane', 'routed_full_replan',
                    'blocked_pending_human', 'done'
                )),
                route TEXT CHECK (route IN ('fast_lane', 'full_replan')),
                loop_round INTEGER NOT NULL DEFAULT 0,
                blocked_reason TEXT,
                assigned_org_node TEXT NOT NULL CHECK (assigned_org_node != ''),
                created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
                updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
            );
            INSERT INTO task_new SELECT * FROM task;
            DROP TABLE task;
            ALTER TABLE task_new RENAME TO task;
            CREATE INDEX idx_task_status ON task(status);
            CREATE INDEX idx_task_assigned_org_node ON task(assigned_org_node);
            CREATE INDEX idx_task_project_id ON task(project_id);
            """
        )
        conn.commit()
        violations = conn.execute("PRAGMA foreign_key_check").fetchall()
        if violations:
            raise RuntimeError(
                "migrate_add_reviewing_planner_output_status left dangling foreign keys: {}".format(violations)
            )
    finally:
        conn.execute("PRAGMA foreign_keys = ON")


def migrate_add_planner_review_verdict_event_type(conn):
    """2026-08-08 (Org Graph unification Phase 1,
    docs/done/org-graph-unification-phase1-review-gate-design.md §3.2): adds
    'planner_review_verdict' to execution_event.event_type's CHECK enum on an
    ALREADY-POPULATED database -- same create-copy-drop-rename dance as
    migrate_add_loop_back_event_type. Idempotent and safe to call
    unconditionally, including against a freshly create_schema()'d database
    (which already has the expanded constraint)."""
    row = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='execution_event'").fetchone()
    if row is None or "'planner_review_verdict'" in row[0]:
        return
    conn.executescript(
        """
        CREATE TABLE execution_event_new (
            event_id TEXT PRIMARY KEY,
            project_id TEXT NOT NULL REFERENCES project(project_id),
            task_id TEXT REFERENCES task(task_id),
            run_id TEXT REFERENCES work_graph_run(run_id),
            node_exec_id TEXT REFERENCES work_node_execution(node_exec_id),
            event_type TEXT NOT NULL CHECK (event_type IN (
                'task_status_changed', 'node_status_changed', 'artifact_created',
                'bug_recorded', 'loop_routed', 'budget_updated', 'skill_iterated',
                'log_appended', 'quota_wait_started', 'loop_back_triggered', 'planner_review_verdict'
            )),
            payload TEXT NOT NULL,
            created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        INSERT INTO execution_event_new SELECT * FROM execution_event;
        DROP TABLE execution_event;
        ALTER TABLE execution_event_new RENAME TO execution_event;
        CREATE INDEX idx_execution_event_project_created ON execution_event(project_id, created_at);
        CREATE INDEX idx_execution_event_task_created ON execution_event(task_id, created_at);
        """
    )
    conn.commit()


def migrate_add_concurrency_wait_event_type(conn):
    """2026-08-13 (real incident, task 71868db6): adds 'concurrency_wait' to
    execution_event.event_type's CHECK enum on an ALREADY-POPULATED database
    -- same create-copy-drop-rename dance as migrate_add_planner_review_
    verdict_event_type, one accumulated step later. task_execution.py::
    _mark_blocked() now logs this event type when
    org_node_lifecycle.ConcurrencyLimitReachedError fires, instead of
    routing max_concurrent_goals backpressure through blocked_pending_human/
    execution_error like a real failure. Idempotent and safe to call
    unconditionally, including against a freshly create_schema()'d database
    (which already has the expanded constraint)."""
    row = conn.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='execution_event'").fetchone()
    if row is None or "'concurrency_wait'" in row[0]:
        return
    conn.executescript(
        """
        CREATE TABLE execution_event_new (
            event_id TEXT PRIMARY KEY,
            project_id TEXT NOT NULL REFERENCES project(project_id),
            task_id TEXT REFERENCES task(task_id),
            run_id TEXT REFERENCES work_graph_run(run_id),
            node_exec_id TEXT REFERENCES work_node_execution(node_exec_id),
            event_type TEXT NOT NULL CHECK (event_type IN (
                'task_status_changed', 'node_status_changed', 'artifact_created',
                'bug_recorded', 'loop_routed', 'budget_updated', 'skill_iterated',
                'log_appended', 'quota_wait_started', 'loop_back_triggered', 'planner_review_verdict',
                'concurrency_wait'
            )),
            payload TEXT NOT NULL,
            created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        INSERT INTO execution_event_new SELECT * FROM execution_event;
        DROP TABLE execution_event;
        ALTER TABLE execution_event_new RENAME TO execution_event;
        CREATE INDEX idx_execution_event_project_created ON execution_event(project_id, created_at);
        CREATE INDEX idx_execution_event_task_created ON execution_event(task_id, created_at);
        """
    )
    conn.commit()


def migrate_drop_work_graph_run_org_node(conn):
    """Org Graph unification Phase 2c §2.1/§2.2 (2026-08-12, user-confirmed
    2026-08-08): drops work_graph_run.org_node on an ALREADY-POPULATED
    database and narrows ux_work_graph_run_active from (task_id, org_node)
    to plain task_id in the SAME rebuild -- see both columns'/the index's
    own comments on the CREATE TABLE above for why (a merged graph's one
    run genuinely spans several org_nodes now, so neither the column nor
    the old per-org_node uniqueness grain is accurate anymore). Both
    changes have to happen in one rebuild since the old index is defined
    on the column being dropped.

    work_graph_run is a PARENT table (work_node_execution/artifact/
    execution_event/bug_record/work_graph_run's own self-reference all
    REFERENCES it) -- same PRAGMA foreign_keys=OFF/rebuild/
    foreign_key_check/ON dance as migrate_add_reviewing_planner_output_status
    for the exact same reason (see that function's own docstring for the
    real-restart incident this precedent comes from); a plain executescript()
    without toggling foreign_keys would trip FK enforcement on DROP TABLE
    the instant any child row exists.

    Depends on migrate_relax_org_node_constraints() having already run
    against this database (called here unconditionally, it's idempotent) --
    that migration is what first relaxed org_node's CHECK from the old
    3-value enum; this one goes further and removes the column outright,
    but only after that intermediate state, matching the accumulated-history
    convention every other CHECK-enum migration in this file already follows
    (each embeds the full column/constraint set valid as of ITS place in the
    sequence, not retroactively edited by later migrations).

    Idempotent, and re-entrant against a database left PARTWAY through a
    previous, interrupted attempt (real 2026-08-13 self-hosting incident:
    executescript() does not wrap the whole script in one transaction, so a
    prior run of this function got as far as dropping the org_node column
    and renaming the table in, then died on the final CREATE UNIQUE INDEX
    step -- see the _reconcile_duplicate_active_work_graph_runs() docstring
    for why that step can fail against real data). Column-drop and index-
    creation are therefore checked and completed independently rather than
    as one all-or-nothing gate, so re-running this function against that
    exact half-migrated state finishes the job instead of silently
    no-op'ing (the OLD idempotency check here was just `if "org_node" not
    in cols: return`, which would have skipped forever on that database,
    permanently leaving ux_work_graph_run_active missing)."""
    migrate_relax_org_node_constraints(conn)
    conn.commit()

    _reconcile_duplicate_active_work_graph_runs(conn)
    conn.commit()

    cols = {row[1] for row in conn.execute("PRAGMA table_info(work_graph_run)").fetchall()}
    has_index = conn.execute(
        "SELECT 1 FROM sqlite_master WHERE type = 'index' AND name = 'ux_work_graph_run_active'"
    ).fetchone() is not None

    if "org_node" not in cols:
        if not has_index:
            conn.execute(
                "CREATE UNIQUE INDEX ux_work_graph_run_active ON work_graph_run(task_id) "
                "WHERE status NOT IN ('done', 'failed')"
            )
            conn.commit()
        return

    conn.execute("PRAGMA foreign_keys = OFF")
    try:
        conn.executescript(
            """
            CREATE TABLE work_graph_run_new (
                run_id TEXT PRIMARY KEY,
                task_id TEXT NOT NULL REFERENCES task(task_id),
                graph_manifest TEXT NOT NULL,
                orchestration_script_ref TEXT,
                status TEXT NOT NULL CHECK (status IN (
                    'planning', 'running', 'merging', 'verifying',
                    'reflecting', 'blocked', 'done', 'failed'
                )),
                started_at TEXT,
                completed_at TEXT,
                result_artifact_refs TEXT,
                parent_run_id TEXT REFERENCES work_graph_run(run_id),
                fork_of_node_exec_id TEXT REFERENCES work_node_execution(node_exec_id)
            );
            INSERT INTO work_graph_run_new (
                run_id, task_id, graph_manifest, orchestration_script_ref, status,
                started_at, completed_at, result_artifact_refs, parent_run_id, fork_of_node_exec_id
            )
            SELECT
                run_id, task_id, graph_manifest, orchestration_script_ref, status,
                started_at, completed_at, result_artifact_refs, parent_run_id, fork_of_node_exec_id
            FROM work_graph_run;
            DROP TABLE work_graph_run;
            ALTER TABLE work_graph_run_new RENAME TO work_graph_run;
            CREATE UNIQUE INDEX ux_work_graph_run_active ON work_graph_run(task_id)
                WHERE status NOT IN ('done', 'failed');
            """
        )
        conn.commit()
        violations = conn.execute("PRAGMA foreign_key_check").fetchall()
        if violations:
            raise RuntimeError(
                "migrate_drop_work_graph_run_org_node left dangling foreign keys: {}".format(violations)
            )
    finally:
        conn.execute("PRAGMA foreign_keys = ON")


def _reconcile_duplicate_active_work_graph_runs(conn):
    """Real production data (2026-08-13 incident) can have MULTIPLE
    non-terminal work_graph_run rows for the same task_id -- a legitimate
    leftover of the pre-Phase-2c architecture where each org_node got its
    own row instead of sharing one merged run. ux_work_graph_run_active's
    UNIQUE(task_id) WHERE status NOT IN ('done', 'failed') can't be created
    while such duplicates exist (10 affected task_ids found against the
    live self-hosting database, e.g. one task had 8 concurrently-'open'
    rows from the old per-org_node model).

    Keeps the most-recently-started non-terminal row per task_id as the
    active one and marks every other non-terminal row for that task_id as
    'failed' -- 'blocked' was considered but rejected: ux_work_graph_run_active's
    WHERE clause is `status NOT IN ('done', 'failed')`, so a 'blocked' row
    still counts as active and would keep tripping the same UNIQUE
    violation this function exists to prevent (confirmed by re-running this
    against real duplicate data -- the index creation still failed with
    'blocked'). 'failed' is the only terminal-compatible status that
    doesn't misrepresent these rows as successfully completed ('done'
    would); it's an honest description too, since these are abandoned runs
    from before the merged-graph architecture -- nothing will ever resume
    against them directly again (Coder/Validator now only ever append into
    the newest run for a task). Plain Python grouping rather than a SQL
    window function, matching this file's existing style (no window
    functions used elsewhere in this module) and avoiding any SQLite-
    version assumption.

    Safe to call unconditionally and repeatedly: a database with no
    duplicates (the common case) does zero UPDATEs.

    Ranks candidates by (started_at, rowid), NOT (started_at, run_id):
    real data confirmed started_at is NULL on every one of these legacy
    rows (never backfilled -- they predate this column being populated),
    which would make a run_id tiebreak pick among them by UUID string
    value, unrelated to actual chronology. rowid reflects true insertion
    order (SQLite assigns it monotonically increasing unless the table is
    declared WITHOUT ROWID, which this one isn't), so it's the only
    honest proxy available for "most recently created" when started_at
    can't answer that."""
    rows = conn.execute(
        "SELECT rowid, run_id, task_id, started_at FROM work_graph_run "
        "WHERE status NOT IN ('done', 'failed')"
    ).fetchall()
    by_task = {}
    for rowid, run_id, task_id, started_at in rows:
        by_task.setdefault(task_id, []).append((started_at or "", rowid, run_id))

    to_block = []
    for task_id, entries in by_task.items():
        if len(entries) <= 1:
            continue
        entries.sort()
        to_block.extend(run_id for _, _, run_id in entries[:-1])

    for run_id in to_block:
        conn.execute("UPDATE work_graph_run SET status = 'failed' WHERE run_id = ?", (run_id,))