ThothII/harness/tht/taskdoc.py

"""Per-step task document generator (spec D16, §4.9).

Emette un singolo documento compatto per fase/step, derivato dagli artefatti precedenti
e dalla vista effective delle decisioni, con byte budget enforced (target <20k token
per un modello 35B/<200k). MAI incorpora physical.yaml (~190k token, fatale).

D15+D16 complementari: il task doc e' generato dalla vista effective_decisions, quindi
post-rollback riflette automaticamente lo stato corretto (le decisioni stale di fasi
> current_phase sono escluse).
"""
from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path

from tht.phase import effective_decisions
from tht.workflow import load_workflow

MAX_BODY_BYTES = 80_000  # ~20k token (target per task document di una fase)
_TRUNCATION_MARKER = "\n\n[...troncato per il budget di contesto D16...]"


@dataclass
class TaskDoc:
    phase: int
    body: str
    byte_budget_ok: bool


def _slice_schema_linking(raw: str, promoted_tables: list[str] | None) -> str:
    """Riduce schema_linking.json alle sole tabelle promosse (D16 §4.9: 'solo le
    tabelle/colonne promosse, non tutto lo schema'). Senza promoted_tables passa il
    contenuto invariato (e' gia' la superficie decisionale, non lo schema fisico).
    Su JSON malformato ritorna il raw (il bound piu' sotto lo tronca se enorme)."""
    if not promoted_tables:
        return raw
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        return raw
    allow = set(promoted_tables)

    def table_of(cand: dict) -> str:
        return str(cand.get("name", "")).split(".")[0]

    if isinstance(data, dict) and isinstance(data.get("candidates"), list):
        data["candidates"] = [c for c in data["candidates"] if table_of(c) in allow]
    return json.dumps(data, ensure_ascii=False, indent=2)


def generate_task_doc(
    session_dir: Path | str,
    phase: int,
    promoted_tables: list[str] | None = None,
) -> TaskDoc:
    """Genera il documento di task per la fase `phase`.

    Contenuto (compatti, mai artefatti integrali fatali):
    - Domanda (question.md) se presente.
    - Schema linking (schema_linking.json) solo da fase >= 4.
    - Brief delle decisioni effective (esclude stale post-rollback, esclude ritirate).
    - Header del task con il numero/nome della fase.
    """
    session_dir = Path(session_dir)
    parts: list[str] = []

    q = session_dir / "question.md"
    if q.exists():
        parts.append("## Domanda\n" + q.read_text())

    sl = session_dir / "schema_linking.json"
    if sl.exists() and phase >= 4:
        sliced = _slice_schema_linking(sl.read_text(), promoted_tables)
        parts.append("## Schema linking (deciso)\n```json\n" + sliced + "\n```")

    # Brief decisioni effective (D15-aware)
    eff = effective_decisions(session_dir)
    if eff:
        lines = [f"- {d.type} | {d.subject} | {d.detail}" for d in eff]
        parts.append("## Decisioni effettive (effective)\n" + "\n".join(lines))

    # Header fase
    try:
        wf = load_workflow()
        name = wf.phase_name(phase)
        header = f"## Task: fase {phase} ({name})"
    except Exception:
        header = f"## Task: fase {phase}"
    parts.append(header)

    body = "\n\n".join(parts)
    # Enforcement del bound (D16): non solo segnalare -- troncare. Un task doc oltre
    # budget collasserebbe un 35B; meglio un documento troncato e marcato.
    encoded = body.encode()
    budget = MAX_BODY_BYTES - len(_TRUNCATION_MARKER.encode())
    if len(encoded) > MAX_BODY_BYTES:
        body = encoded[:budget].decode("utf-8", "ignore") + _TRUNCATION_MARKER
        return TaskDoc(phase=phase, body=body, byte_budget_ok=False)
    return TaskDoc(phase=phase, body=body, byte_budget_ok=True)