"""llm_call_ledger — the platform's own AI usage, per call

Helix bills the company for AI spending and kept no record of its own.
Twenty-nine modules call the shared Bailian client; two recorded a
token. This table closes that gap.

Content-free by construction: there is no column for prompt or
completion text, so an ops table cannot quietly become a store of
employee support tickets and bounty reports.

Tokens, not money: this repo has no per-model price table and this
migration does not invent one. ``ai_spending`` (the vendor bill) stays
the money truth; this ledger answers the question the bill cannot —
which feature burned the tokens.

Revision ID: 619
Revises: 618
"""
from __future__ import annotations

import uuid

import sqlalchemy as sa

from alembic import op
from app.core.migration_helpers import id_type_for, table_kw_for

revision = "619"
# The real head on ``origin/main``, re-checked immediately before push.
# Numbers may have gaps and alembic does not care; ``down_revision`` is
# the only thing that decides the chain.
down_revision = "618"
branch_labels = None
depends_on = None

LEDGER = "llm_call_ledger"
PERSONS = "identity_persons"
JOBS = "scheduled_jobs"

#: Seeded enabled. The handler is a no-op when no LLM key is configured
#: (it records ``AI_NO_KEY`` on the row and does not stamp a generated-at,
#: so the rows are revisited once a key exists), and a disabled seed is
#: how a feature ships and is never switched on — the 2026-09-05 ledger
#: audit found 35 such rows.
_TRIAGE_JOB = (
    "automation_exception.triage",
    "自动化异常 · AI 分诊",
    "35 * * * *",
    "每小时给「规则说不出根因」的异常补一个根因与一句解释。**按签名调用,不按发生次数**"
    " —— 一个出现 4000 次的签名是一次模型调用,答案缓存在异常行上,签名文字变了才会重问。"
    "人已经判过 / 规则已经定了因 / 这个签名问过了 / 行已经解决,四种情况都不问模型,"
    "所以规则表越长,这条任务花的钱越少。模型只能从闭集 ROOT_CAUSES 里选根因,"
    "解释里的每个数字都必须在事实包里,否则整条回复作废。"
    "分诊不关闭任何东西、不通知任何人、不覆盖人工判定;"
    "每次调用记进 llm_call_ledger,可按 purpose='automation_exception.triage' 归集。",
)


def upgrade() -> None:
    bind = op.get_bind()
    inspector = sa.inspect(bind)
    if LEDGER in inspector.get_table_names():
        return

    # Borrowed from the parent, never hand-written: a mismatched
    # collation is error 3780 at deploy time on TiDB and is invisible
    # on local MySQL, where every parent shares one collation.
    pid = id_type_for(PERSONS)
    kw = table_kw_for(PERSONS)

    op.create_table(
        LEDGER,
        sa.Column("id", pid, primary_key=True),
        sa.Column("purpose", sa.String(64), nullable=False),
        sa.Column("model", sa.String(64), nullable=False),
        sa.Column("region", sa.String(32), nullable=True),
        sa.Column("operation", sa.String(16), nullable=False),
        sa.Column("ok", sa.Boolean(), nullable=False),
        # NULL means "the vendor did not report it", which is not zero
        # and must not be summed as if it were.
        sa.Column("input_tokens", sa.BigInteger(), nullable=True),
        sa.Column("output_tokens", sa.BigInteger(), nullable=True),
        sa.Column("duration_ms", sa.BigInteger(), nullable=True),
        # Our code or a normalised vendor one — never the vendor's
        # free-text message, which quotes rejected prompts back.
        sa.Column("error_code", sa.String(64), nullable=True),
        sa.Column("trace_id", sa.String(64), nullable=True),
        # Column type borrowed from the parent so the FK is accepted on
        # TiDB, where a mismatched collation is error 3780 at deploy
        # time. Never hand-written.
        sa.Column("actor_person_id", pid, nullable=True),
        sa.Column(
            "called_at", sa.DateTime(timezone=True),
            nullable=False, server_default=sa.func.now(),
        ),
        sa.ForeignKeyConstraint(
            ["actor_person_id"], [f"{PERSONS}.id"], ondelete="SET NULL",
        ),
        **kw,
    )
    # Every read of this table is time-bounded, so time leads.
    op.create_index(
        "ix_llm_ledger_called_purpose", LEDGER, ["called_at", "purpose"],
    )
    op.create_index(
        "ix_llm_ledger_called_model", LEDGER, ["called_at", "model"],
    )
    op.create_index("ix_llm_ledger_ok_called", LEDGER, ["ok", "called_at"])

    # Seed the triage cron. Guarded on the table existing and on the
    # handler not already being registered, so a re-run is a no-op.
    if JOBS in inspector.get_table_names():
        handler, display, cron, desc = _TRIAGE_JOB
        bind.execute(
            sa.text(
                f"INSERT INTO {JOBS} "
                "(id, name, display_name, description, handler_key, "
                " trigger_type, cron_expr, timezone, enabled, config, "
                " run_count, failure_count, created_at, updated_at) "
                "SELECT :id, :name, :display, :desc, :handler, 'cron', "
                ":cron, 'UTC', 1, :config, 0, 0, NOW(), NOW() "
                "FROM DUAL WHERE NOT EXISTS "
                f"(SELECT 1 FROM {JOBS} WHERE handler_key = :handler)",
            ),
            {
                "id": str(uuid.uuid4()),
                "name": handler,
                "display": display,
                "desc": desc,
                "handler": handler,
                "cron": cron,
                "config": "{}",
            },
        )


def downgrade() -> None:
    bind = op.get_bind()
    inspector = sa.inspect(bind)
    if JOBS in inspector.get_table_names():
        bind.execute(
            sa.text(f"DELETE FROM {JOBS} WHERE handler_key = :handler"),
            {"handler": _TRIAGE_JOB[0]},
        )
    if LEDGER not in inspector.get_table_names():
        return
    op.drop_index("ix_llm_ledger_ok_called", table_name=LEDGER)
    op.drop_index("ix_llm_ledger_called_model", table_name=LEDGER)
    op.drop_index("ix_llm_ledger_called_purpose", table_name=LEDGER)
    op.drop_table(LEDGER)