potato/codebook/changelog.py

"""
Codebook change-provenance overlay (Phase 2 C).

A **separate** audit/overlay layer for retroactive codebook edits
(merge / split / rename / recolor / move / delete) and the LLM
propose -> human-confirm flow. It is deliberately NOT part of the
`codes` records: `Codebook.labels()` / `as_tree()` feed the ICL prompt
verbatim, so authorship / change history must never join into them
(open-question #2 resolution).

Two tables (own migration, universal project.sqlite):
- ``codebook_change`` — append-only event log: every retroactive op,
  who/what/old->new/when, with ``actor_kind`` (human|model). Generalises
  the additive-only ``revision.codes_added_since`` so the review
  worklist / banner can say "X merged into Y", not just "N codes added".
- ``codebook_proposal`` — pending model-proposed edits awaiting human
  confirmation (status pending|confirmed|rejected).

Plus the temporal columns that make retroactive edits **append-only**:
- ``annotation_codes.invalidated_at`` / ``invalidated_by_change`` —
  a superseded link is marked, never DELETEd (NULL = live). NOTE:
  ``started_at``/``ended_at`` already mean *elapsed time on a span /
  agentic trace* — they are NOT validity and must not be reused.
- ``codes.archived_at`` — a merged-away source code is archived (leaves
  the palette + ICL prompt) but its row and history survive.
"""

from __future__ import annotations

import json
import time
import uuid
from typing import Any, Dict, List, Optional

from potato.persistence import Migration, get_db, register_migration

_CHANGE_MIGRATION = Migration(
    name="0003_codebook_change_provenance",
    sql="""
    ALTER TABLE annotation_codes ADD COLUMN invalidated_at REAL;
    ALTER TABLE annotation_codes ADD COLUMN invalidated_by_change TEXT;
    ALTER TABLE codes ADD COLUMN archived_at REAL;

    CREATE TABLE IF NOT EXISTS codebook_change (
        id              TEXT PRIMARY KEY,
        project         TEXT NOT NULL,
        code_id         TEXT,
        related_code_id TEXT,
        op              TEXT NOT NULL,
        old_value       TEXT,
        new_value       TEXT,
        actor           TEXT NOT NULL,
        actor_kind      TEXT NOT NULL DEFAULT 'human',
        created_at      REAL NOT NULL,
        revision        INTEGER NOT NULL DEFAULT 0
    );
    CREATE INDEX IF NOT EXISTS idx_cbchange_proj
        ON codebook_change (project, created_at);

    CREATE TABLE IF NOT EXISTS codebook_proposal (
        id          TEXT PRIMARY KEY,
        project     TEXT NOT NULL,
        op          TEXT NOT NULL,
        payload     TEXT NOT NULL,
        status      TEXT NOT NULL DEFAULT 'pending',
        actor       TEXT NOT NULL,
        actor_kind  TEXT NOT NULL DEFAULT 'model',
        created_at  REAL NOT NULL,
        decided_by  TEXT,
        decided_at  REAL,
        change_id   TEXT
    );
    CREATE INDEX IF NOT EXISTS idx_cbproposal_proj
        ON codebook_proposal (project, status, created_at);
    """,
)

# Defensive ordering: 0003 ALTERs `annotation_codes` and `codes`, so the
# 0001 CREATE and both 0002 ALTERs must register first regardless of
# import path (mirrors revision.py — import the Migration objects
# directly, not via package side effects, to avoid circular imports).
from potato.codebook.store import _CODEBOOK_MIGRATION as _CB_MIG
from potato.codebook.revision import (
    _REVISION_MIGRATION as _REV_MIG,
    _CODES_REV_MIGRATION as _CODES_REV_MIG,
)

register_migration(_CB_MIG)
register_migration(_REV_MIG)
register_migration(_CODES_REV_MIG)
register_migration(_CHANGE_MIGRATION)


def _db(task_dir: str):
    register_migration(_CHANGE_MIGRATION)
    return get_db(task_dir)


# ---- change log ----------------------------------------------------------

def log_change(
    task_dir: str, *, project: str, op: str, actor: str,
    actor_kind: str = "human", code_id: Optional[str] = None,
    related_code_id: Optional[str] = None,
    old_value: Optional[str] = None, new_value: Optional[str] = None,
    revision: int = 0,
) -> str:
    """Append one immutable change-log row; return its id."""
    cid = uuid.uuid4().hex
    conn = _db(task_dir)
    conn.execute(
        """INSERT INTO codebook_change
               (id, project, code_id, related_code_id, op, old_value,
                new_value, actor, actor_kind, created_at, revision)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        (cid, project, code_id, related_code_id, op, old_value,
         new_value, actor, actor_kind, time.time(), revision),
    )
    conn.commit()
    return cid


def changes_since(
    task_dir: str, project: str, revision: int
) -> List[Dict[str, Any]]:
    """Change-log rows recorded after `revision` — the non-additive
    counterpart to revision.codes_added_since, for the worklist/banner
    and the before->after delta view."""
    rows = _db(task_dir).execute(
        """SELECT id, code_id, related_code_id, op, old_value, new_value,
                  actor, actor_kind, created_at, revision
           FROM codebook_change
           WHERE project = ? AND revision > ?
           ORDER BY created_at ASC""",
        (project, revision),
    ).fetchall()
    return [dict(r) for r in rows]


def all_changes(
    task_dir: str, project: str
) -> List[Dict[str, Any]]:
    rows = _db(task_dir).execute(
        """SELECT id, code_id, related_code_id, op, old_value, new_value,
                  actor, actor_kind, created_at, revision
           FROM codebook_change
           WHERE project = ?
           ORDER BY created_at ASC""",
        (project,),
    ).fetchall()
    return [dict(r) for r in rows]


# ---- proposals (model -> human-confirm) ---------------------------------

def record_proposal(
    task_dir: str, *, project: str, op: str, payload: Dict[str, Any],
    actor: str, actor_kind: str = "model",
) -> Dict[str, Any]:
    pid = uuid.uuid4().hex
    now = time.time()
    conn = _db(task_dir)
    conn.execute(
        """INSERT INTO codebook_proposal
               (id, project, op, payload, status, actor, actor_kind,
                created_at)
           VALUES (?, ?, ?, ?, 'pending', ?, ?, ?)""",
        (pid, project, op, json.dumps(payload), actor, actor_kind, now),
    )
    conn.commit()
    return get_proposal(task_dir, pid)


def get_proposal(task_dir: str, proposal_id: str) -> Optional[Dict]:
    row = _db(task_dir).execute(
        "SELECT * FROM codebook_proposal WHERE id = ?", (proposal_id,)
    ).fetchone()
    if not row:
        return None
    d = dict(row)
    try:
        d["payload"] = json.loads(d["payload"])
    except Exception:
        pass
    return d


def list_proposals(
    task_dir: str, project: str, status: str = "pending"
) -> List[Dict[str, Any]]:
    rows = _db(task_dir).execute(
        """SELECT * FROM codebook_proposal
           WHERE project = ? AND status = ?
           ORDER BY created_at ASC""",
        (project, status),
    ).fetchall()
    out = []
    for r in rows:
        d = dict(r)
        try:
            d["payload"] = json.loads(d["payload"])
        except Exception:
            pass
        out.append(d)
    return out


def set_proposal_status(
    task_dir: str, proposal_id: str, *, status: str,
    decided_by: str, change_id: Optional[str] = None,
) -> bool:
    conn = _db(task_dir)
    cur = conn.execute(
        """UPDATE codebook_proposal
           SET status = ?, decided_by = ?, decided_at = ?, change_id = ?
           WHERE id = ? AND status = 'pending'""",
        (status, decided_by, time.time(), change_id, proposal_id),
    )
    conn.commit()
    return cur.rowcount > 0


def propose_change(
    task_dir: str, *, project: str, op: str, payload: Dict[str, Any],
    actor: str, actor_kind: str = "model",
) -> Dict[str, Any]:
    """Thin entry point for in-process (e.g. solo-mode LLM) producers
    that aren't going through the HTTP API. The contract is identical to
    POST /api/codebook/proposals."""
    return record_proposal(
        task_dir, project=project, op=op, payload=payload,
        actor=actor, actor_kind=actor_kind)