claude_background_reconciliation.py

Rama main · réplica de lectura del repositorio original.

"""Bounded, data-only Claude task receipts; no I/O or provider transcripts."""
from __future__ import annotations

from collections import OrderedDict
from collections.abc import Mapping
import json


RECONCILED_EVENT = "claude_background_tasks_reconciled"
CONSUMED_EVENT = "claude_background_task_reconciliation_consumed"
MAX_TASKS = 64
MAX_TASK_BYTES = 6144
SESSION_FIELD = "claude_background_reconciliation"
STATUSES = frozenset({"running", "completed", "failed", "stopped", "killed", "tracking_lost"})


def _identifier(value, limit=256):
    return value if isinstance(value, str) and 0 < len(value) <= limit and all(
        ord(char) >= 32 and ord(char) != 127 for char in value
    ) else None


def normalize_task_receipts(receipts, *, run_id, provider_session_id, tracking_lost=False):
    """Allowlist exact task ownership and statuses, never commands or summaries."""
    if not _identifier(run_id) or not _identifier(provider_session_id):
        return []
    if not isinstance(receipts, (list, tuple)):
        return []
    result = OrderedDict()
    for receipt in receipts[:MAX_TASKS]:
        if not isinstance(receipt, Mapping):
            continue
        task_id = _identifier(receipt.get("task_id"))
        status = receipt.get("status")
        task_type = _identifier(receipt.get("task_type"), 64)
        if not task_id or not isinstance(status, str) or status not in STATUSES or not task_type:
            continue
        if receipt.get("owner_run_id") != run_id or receipt.get("provider_session_id") not in (None, "", provider_session_id):
            continue
        task = {
            "task_id": task_id, "task_type": task_type,
            "status": "tracking_lost" if tracking_lost and status == "running" else status,
            "owner_run_id": run_id, "provider_session_id": provider_session_id,
        }
        tool_id = _identifier(receipt.get("tool_use_id"))
        if tool_id:
            task["tool_use_id"] = tool_id
        result[task_id] = task
    return list(result.values())


def reconciliation_envelope(batches):
    """Latest exact-owner state wins; discarded states are counted, not invented."""
    tasks = OrderedDict()
    overflow = 0
    for batch in batches[-2:]:
        overflow += max(0, min(int(batch.get("overflow_count") or 0), 1_000_000))
        for task in batch.get("tasks", ()):
            key = (task["provider_session_id"], task["owner_run_id"], task["task_id"])
            tasks[key] = dict(task)
            tasks.move_to_end(key)
    selected = list(tasks.values())
    while len(selected) > MAX_TASKS or len(json.dumps(selected, ensure_ascii=True, separators=(",", ":")).encode("utf-8")) > MAX_TASK_BYTES:
        selected.pop(0)
        overflow += 1
    return {"tasks": selected, "overflow_count": min(overflow, 1_000_000)}


def pending_task_reconciliations(value, *, provider_session_id: str):
    """Read a single bounded persisted checkpoint without touching event logs."""
    if not isinstance(value, dict) or value.get("version") != 1 or value.get("provider_session_id") != provider_session_id:
        return []
    identifier = _identifier(value.get("reconciliation_id"))
    overflow = value.get("overflow_count", 0)
    source_tasks = value.get("tasks")
    if not identifier or type(overflow) is not int or overflow < 0 or not isinstance(source_tasks, list):
        return []
    tasks = []
    for receipt in source_tasks[:MAX_TASKS]:
        if isinstance(receipt, dict):
            tasks.extend(normalize_task_receipts([receipt], run_id=receipt.get("owner_run_id"), provider_session_id=provider_session_id))
    envelope = reconciliation_envelope([{"tasks": tasks, "overflow_count": overflow + max(0, len(source_tasks) - MAX_TASKS)}])
    return [{"reconciliation_id": identifier, **envelope}]


def pending_reconciliation_state(previous, batch, *, provider_session_id):
    pending = pending_task_reconciliations(previous, provider_session_id=provider_session_id)
    return {
        "version": 1, "provider_session_id": provider_session_id,
        "reconciliation_id": batch["reconciliation_id"],
        **reconciliation_envelope([*pending, batch]),
    }