"""Reclassify deploy-killed runs from ``failed`` to ``interrupted``.

Revision ID: 614
Revises: 613
Create Date: 2026-09-05

A worker killed mid-run left its ``scheduled_job_runs`` row at
``running``; six hours later the stale-row reaper stamped it ``failed``
with ``auto-fail: stale running row swept after Nh (worker likely died
mid-run)``. Production, 30 days: **386 of 452 scheduler failures were
that message**, against ~15 deploys a day and an eight-second drain
window, and every sampled one had already re-run successfully within
fifteen minutes of the next boot. They sit unacknowledged in the red
queue, dominate ``helix_scheduled_job_unacked_failures``, and make the
failure rate a fiction.

**Why this is a correction and not a rewrite of history.** The repo's
rule is that an acknowledged failure keeps ``status='failed'`` — we never
launder a real failure into a success, because the audit trail and
``failure_count`` depend on the truth. Nothing here contradicts that:

* The rows are matched by the reaper's own machine-written message, so
  no human-diagnosed failure can be caught by the predicate.
* ``interrupted`` is not a success. It says the process went away, which
  is what the message already said in prose; only the column disagreed.
* ``run_count`` / ``failure_count`` are untouched, because the bulk
  UPDATE that created these rows never incremented them either — the
  counters have always excluded these runs. Nothing double-moves.
* The original message is preserved verbatim with a marker appended, so
  the reclassification is legible from the row itself.

Downgrade restores ``failed`` and strips the marker.
"""
from __future__ import annotations

import sqlalchemy as sa

from alembic import op

revision = "614"
down_revision = "613"
branch_labels = None
depends_on = None

#: The reaper's own wording, matched as a prefix. Narrow on purpose: a
#: failure an operator diagnosed and wrote up must never be swept into
#: this bucket, so the predicate keys on a string only the reaper emits.
LEGACY_PREFIX = "auto-fail: stale running row swept after"
MARKER = " [reclassified as interrupted by migration 614]"


def _table_exists(bind, table: str) -> bool:
    return bool(
        bind.execute(
            sa.text(
                "SELECT 1 FROM information_schema.TABLES "
                "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = :t LIMIT 1",
            ),
            {"t": table},
        ).scalar(),
    )


def upgrade() -> None:
    bind = op.get_bind()
    if not _table_exists(bind, "scheduled_job_runs"):
        return
    bind.execute(
        sa.text(
            "UPDATE scheduled_job_runs "
            "SET status = 'interrupted', "
            "    error_message = CONCAT(error_message, :marker) "
            "WHERE status = 'failed' "
            "  AND error_message LIKE :pattern "
            "  AND error_message NOT LIKE :already",
        ),
        {
            "marker": MARKER,
            "pattern": f"{LEGACY_PREFIX}%",
            "already": f"%{MARKER}%",
        },
    )


def downgrade() -> None:
    bind = op.get_bind()
    if not _table_exists(bind, "scheduled_job_runs"):
        return
    bind.execute(
        sa.text(
            "UPDATE scheduled_job_runs "
            "SET status = 'failed', "
            "    error_message = REPLACE(error_message, :marker, '') "
            "WHERE status = 'interrupted' "
            "  AND error_message LIKE :already",
        ),
        {"marker": MARKER, "already": f"%{MARKER}%"},
    )