Compaction y continuidad sin perder trabajo

Conservaremos todos los originales en history, las versiones del resumen en summaries y un único puntero al resumen activo en runs. Cambiar ese puntero no borra mensajes. previous_id permite explicar de qué resumen partió cada versión.

import json
import sqlite3
import tempfile
from contextlib import contextmanager
from pathlib import Path


@contextmanager
def transaction(db):
    db.execute("BEGIN IMMEDIATE")
    try:
        yield
        db.execute("COMMIT")
    except BaseException:
        if db.in_transaction:
            db.execute("ROLLBACK")
        raise


def open_db(path):
    db = sqlite3.connect(path, autocommit=True)
    db.execute("PRAGMA foreign_keys=ON")
    return db


SCHEMA = """
CREATE TABLE runs (
    id TEXT PRIMARY KEY,
    active_summary_id INTEGER REFERENCES summaries(id)
);
CREATE TABLE history (
    run_id TEXT NOT NULL REFERENCES runs(id),
    seq INTEGER NOT NULL,
    text TEXT NOT NULL,
    PRIMARY KEY (run_id, seq)
);
CREATE TABLE summaries (
    id INTEGER PRIMARY KEY,
    run_id TEXT NOT NULL REFERENCES runs(id),
    coverage INTEGER NOT NULL,
    body TEXT NOT NULL,
    previous_id INTEGER REFERENCES summaries(id)
);
"""

Reutilizamos la transacción del capítulo de persistencia: yield cede la ejecución al cuerpo del with; después continúa en COMMIT si terminó bien o recibe la excepción para ejecutar ROLLBACK. BEGIN IMMEDIATE reserva el turno de escritura de SQLite. La preparación del resumen ocurre fuera de ese turno.

El candidato puede conservar una referencia al mensaje 1 del resumen anterior, aunque el material nuevo solo incluya 3 y 4. Por eso no validamos sus referencias únicamente contra el tramo nuevo. Consultamos originales del mismo run con secuencia no posterior al corte. Encontrar una fila demuestra que la referencia existe; no demuestra que la frase redactada esté respaldada por ella. Esa revisión semántica sigue siendo necesaria.

def install(db, run_id, base_id, cut, candidate):
    if not isinstance(candidate, list):
        raise ValueError("El resumen debe ser una lista de afirmaciones")
    with transaction(db):
        base = db.execute(
            "SELECT coverage FROM summaries WHERE id=? AND run_id=?",
            (base_id, run_id),
        ).fetchone()
        boundary = db.execute(
            "SELECT 1 FROM history WHERE run_id=? AND seq=?",
            (run_id, cut),
        ).fetchone()
        if base is None or cut <= base[0] or boundary is None:
            raise ValueError("La cobertura no extiende una base válida")
        for claim in candidate:
            if (not isinstance(claim, dict)
                    or set(claim) != {"text", "source_seq"}
                    or not isinstance(claim["text"], str)
                    or not claim["text"].strip()
                    or type(claim["source_seq"]) is not int):
                raise ValueError("Afirmación mal formada")
            source = db.execute(
                "SELECT 1 FROM history WHERE run_id=? AND seq=? AND seq<=?",
                (run_id, claim["source_seq"], cut),
            ).fetchone()
            if source is None:
                raise ValueError("Referencia ajena o fuera de cobertura")
        created = db.execute(
            "INSERT INTO summaries(run_id, coverage, body, previous_id) "
            "VALUES (?, ?, ?, ?)",
            (run_id, cut, json.dumps(candidate), base_id),
        )
        changed = db.execute(
            "UPDATE runs SET active_summary_id=? "
            "WHERE id=? AND active_summary_id=?",
            (created.lastrowid, run_id, base_id),
        )
        if changed.rowcount != 1:
            raise RuntimeError("La base cambió; prepara otro resumen")

fetchone() devuelve una fila como tupla o None si no hay coincidencia. base[0] lee su cobertura. lastrowid recupera la identidad de la fila recién insertada; rowcount dice cuántas filas cambió el UPDATE. Los ? son parámetros, no texto SQL construido con datos del modelo.

La comparación active_summary_id=base_id implementa compare-and-swap o CAS: «cambia el valor solo si sigue siendo el que leí». Si falla, la excepción revierte también la inserción del candidato; no dejamos una versión publicada a medias. El corte lo decide el runtime a partir de unidades completas, no un campo que el modelo pueda inventar.

En este ejemplo el resumen inicial contiene la restricción del mensaje 1. A y B se representan con listas redactadas a mano: así aislamos la corrección del almacenamiento de la calidad lingüística. Las llamadas en el orden indicado reproducen una carrera de manera determinista, sin depender de la velocidad de dos hilos.

with tempfile.TemporaryDirectory() as directory:
    path = Path(directory) / "compaction.db"
    db = open_db(path)
    db.executescript(SCHEMA)
    db.execute("INSERT INTO runs VALUES ('r-1', NULL)")
    originals = [
        (1, "La ejecución es de solo lectura."),
        (2, "Empieza la investigación de red."),
        (3, "La API resuelve el nombre db."),
        (4, "Falta comprobar el puerto configurado."),
    ]
    db.executemany("INSERT INTO history VALUES ('r-1', ?, ?)", originals)
    initial = [{"text": "Solo lectura", "source_seq": 1}]
    cursor = db.execute(
        "INSERT INTO summaries(run_id, coverage, body) VALUES ('r-1', 2, ?)",
        (json.dumps(initial),),
    )
    base_id = cursor.lastrowid
    db.execute("UPDATE runs SET active_summary_id=? WHERE id='r-1'", (base_id,))
    a = initial + [{"text": "El nombre db se resuelve", "source_seq": 3}]
    b = initial + [{"text": "Continuar diagnóstico DNS", "source_seq": 2}]

    # Llega mientras A y B preparan sus candidatos, después del corte de A.
    db.execute("INSERT INTO history VALUES ('r-1', 5, 'El usuario pide revisar TLS.')")
    install(db, "r-1", base_id, 4, a)
    try:
        install(db, "r-1", base_id, 3, b)
    except RuntimeError:
        print("B rechazado: su base ya no es la activa")
    else:
        raise AssertionError("Se aceptó un candidato obsoleto")
    db.close()

    # Reabrir comprueba qué conservamos, no solo nuestras variables en memoria.
    db = open_db(path)
    active_id, coverage, body = db.execute(
        "SELECT s.id, s.coverage, s.body FROM runs r "
        "JOIN summaries s ON s.id=r.active_summary_id WHERE r.id='r-1'"
    ).fetchone()
    tail = db.execute(
        "SELECT seq, text FROM history WHERE run_id=? AND seq>? ORDER BY seq",
        ("r-1", coverage),
    ).fetchall()
    assert coverage == 4 and json.loads(body)[0]["source_seq"] == 1
    assert [seq for seq, text in tail] == [5]
    assert db.execute("SELECT COUNT(*) FROM history").fetchone()[0] == 5
    assert db.execute("SELECT COUNT(*) FROM summaries").fetchone()[0] == 2
    print("Referencia heredada 1 válida; cobertura 4; cola conservada:", tail[0][0])
    db.close()

El resultado muestra el rechazo de B y la conservación del mensaje 5. Hay dos resúmenes —original y A— y cinco hechos originales. El contexto siguiente se forma con el resumen activo y esa cola; no con history cortado según la hora a la que terminó la inferencia.

El caso prueba instalación y recuperación de la selección. No prueba la calidad de un resumen generado ni una caída durante una escritura del motor. En un servicio real añade el estado del run a las condiciones de activación: conservar un resumen de una ejecución cancelada puede ser útil para consulta, pero esta operación no debe reactivarla.