Blame for scripts/lib/chat_eval/notes_summarizer.py

"""Per-faculty notes summarizer.

Pure-function entry point `run_for_user` reads turns + current notes +
prior audit rows, calls the LLM, and returns (new_notes_body, audit_row).
No HF I/O — callers (the cli command) wire that in via notes_io.

Failure semantics:
- SummarizerTransientError: caller records the failure row but does NOT
  push the notes file and does NOT advance the cursor. Next day retries.
- SummarizerTerminalError: caller exits 1 — operator must investigate.
- Other unexpected exceptions: bubble up (treat as terminal).
"""
from __future__ import annotations
import json
import logging
from pathlib import Path
from typing import Literal, Protocol

from scripts.lib.chat_eval.turn_log import iter_turns


logger = logging.getLogger(__name__)


_NOTES_HARD_CAP_BYTES = 100_000  # 100KB safety guard against runaway output


Outcome = Literal[
    "updated",
    "no_new_turns",
    "no_change",
    "transient_failure",
    "output_too_large",
    "terminal_failure",
]


class SummarizerError(Exception):
    """Base for summarizer-side failures."""


class SummarizerTransientError(SummarizerError):
    """Retryable on the next cron tick (provider overload, 5xx, 429, etc.)."""


class SummarizerTerminalError(SummarizerError):
    """Non-retryable — operator must investigate (auth, validation, etc.)."""


class _LLM(Protocol):
    def complete(self, system: str, prompt: str) -> str: ...


_SUMMARIZER_SYSTEM_PROMPT = """You are maintaining a personal investigation notebook for a Duke Fuqua faculty member who has been using a chat tool to review how their course (Managerial Economics, MGRECON780) was delivered last year. You will be given (1) the user's current notes file in markdown, and (2) a JSONL block of new conversation turns the user had with the bot since the last update.

Produce the updated notes file. Rules:
- The notes file is a working notebook a colleague could read in two minutes. Use short bullet points organised by investigation theme.
- Preserve every distinct finding or open question from the existing notes that is still relevant.
- Consolidate duplicates and near-duplicates. If the user revisited a topic, merge — don't duplicate.
- If the user explicitly said to forget something, drop it.
- If the user contradicted a prior finding, replace the older claim with the corrected one.
- Drop or compress to one line any finding older than 90 days, unless the new turns revisit it.
- Never include student names. Faculty names are fine.
- Do not include retrieval-doc IDs, session hashes, or any bot-internal identifiers.
- Output only the new notes file body — no preamble, no commentary, no markdown code fences around it.
"""


def _latest_cursor_for(login: str, prior_log_rows: list[dict]) -> str | None:
    """Return the most recent last_processed_ts for `login` across all
    successful or no-op rows. None if no prior row exists (bootstrap)."""
    last_ts: str | None = None
    for row in prior_log_rows:
        if row.get("login") != login:
            continue
        if row.get("outcome") not in ("updated", "no_new_turns", "no_change"):
            continue
        cursor = row.get("last_processed_ts")
        if cursor is None:
            continue
        if last_ts is None or cursor > last_ts:
            last_ts = cursor
    return last_ts


def _select_new_turns(turns_path: Path, login: str, since_ts: str | None) -> list[dict]:
    """Read turns_path; return turns where user_login==login,
    mode=='review', and ts > since_ts (or all faculty turns when
    since_ts is None).

    iter_turns(since=...) drops rows with ts < since; the strict-gt
    edge case (ts == since_ts) is handled by the Python-side filter below.
    """
    selected: list[dict] = []
    for t in iter_turns(turns_path, since=since_ts):
        if t.get("user_login") != login:
            continue
        if t.get("mode") != "review":
            continue
        ts = t.get("ts", "")
        if since_ts is not None and ts <= since_ts:
            continue
        selected.append(t)
    return selected


def _build_user_prompt(current_notes_body: str, new_turns: list[dict]) -> str:
    """Compose the user-message payload for the summarizer LLM."""
    turns_block = "\n".join(
        json.dumps({
            "ts": t.get("ts"),
            "user_message": t.get("user_message", ""),
            "response": t.get("response", ""),
        })
        for t in new_turns
    )
    return (
        f"## Current notes\n\n{current_notes_body or '(empty — first update)'}\n\n"
        f"## New conversation turns ({len(new_turns)})\n\n{turns_block}\n"
    )


def _audit_row(
    *,
    login: str,
    now_ts: str,
    outcome: Outcome,
    since_ts: str | None,
    last_processed_ts: str | None,
    turns_processed: int,
    notes_chars_before: int,
    notes_chars_after: int | None,
    archived_to: str | None,
    haiku_model: str | None,
    elapsed_s: float,
    failure_reason: str | None,
) -> dict:
    return {
        "ts": now_ts,
        "login": login,
        "outcome": outcome,
        "since_ts": since_ts,
        "last_processed_ts": last_processed_ts,
        "turns_processed": turns_processed,
        "notes_chars_before": notes_chars_before,
        "notes_chars_after": notes_chars_after,
        "archived_to": archived_to,
        "haiku_model": haiku_model,
        "elapsed_s": round(elapsed_s, 3),
        "failure_reason": failure_reason,
    }


def run_for_user(
    login: str,
    *,
    turns_path: Path,
    current_notes_body: str,
    prior_log_rows: list[dict],
    llm: _LLM,
    now_ts: str,
    haiku_model: str = "claude-haiku-4-5-20251001",
) -> tuple[str | None, dict]:
    """Run the summarizer once for `login`.

    Returns (new_notes_body, audit_row).
      new_notes_body == None means "do not overwrite the notes file"
        (no-op, no-change, transient failure, output too large).
      new_notes_body == str means the caller should archive the prior
        content and write the new body.

    On terminal failure, raises SummarizerTerminalError. The caller is
    expected to record a failure row separately and exit non-zero.
    """
    import time
    t0 = time.monotonic()
    since_ts = _latest_cursor_for(login, prior_log_rows)

    new_turns = _select_new_turns(turns_path, login, since_ts)
    if not new_turns:
        return None, _audit_row(
            login=login, now_ts=now_ts, outcome="no_new_turns",
            since_ts=since_ts, last_processed_ts=since_ts,
            turns_processed=0,
            notes_chars_before=len(current_notes_body),
            notes_chars_after=None, archived_to=None,
            haiku_model=None, elapsed_s=time.monotonic() - t0,
            failure_reason=None,
        )

    latest_ts = max(t.get("ts", "") for t in new_turns)
    user_prompt = _build_user_prompt(current_notes_body, new_turns)

    try:
        output = llm.complete(_SUMMARIZER_SYSTEM_PROMPT, user_prompt)
    except SummarizerTransientError as e:
        return None, _audit_row(
            login=login, now_ts=now_ts, outcome="transient_failure",
            since_ts=since_ts, last_processed_ts=since_ts,  # CURSOR STAYS
            turns_processed=0,
            notes_chars_before=len(current_notes_body),
            notes_chars_after=None, archived_to=None,
            haiku_model=haiku_model, elapsed_s=time.monotonic() - t0,
            failure_reason=str(e),
        )

    output = (output or "").strip()
    if len(output.encode("utf-8")) > _NOTES_HARD_CAP_BYTES:
        return None, _audit_row(
            login=login, now_ts=now_ts, outcome="output_too_large",
            since_ts=since_ts, last_processed_ts=since_ts,
            turns_processed=len(new_turns),
            notes_chars_before=len(current_notes_body),
            notes_chars_after=None, archived_to=None,
            haiku_model=haiku_model, elapsed_s=time.monotonic() - t0,
            failure_reason=f"output exceeded {_NOTES_HARD_CAP_BYTES} bytes",
        )

    if output == current_notes_body.strip():
        return None, _audit_row(
            login=login, now_ts=now_ts, outcome="no_change",
            since_ts=since_ts, last_processed_ts=latest_ts,
            turns_processed=len(new_turns),
            notes_chars_before=len(current_notes_body),
            notes_chars_after=None, archived_to=None,
            haiku_model=haiku_model, elapsed_s=time.monotonic() - t0,
            failure_reason=None,
        )

    return output, _audit_row(
        login=login, now_ts=now_ts, outcome="updated",
        since_ts=since_ts, last_processed_ts=latest_ts,
        turns_processed=len(new_turns),
        notes_chars_before=len(current_notes_body),
        notes_chars_after=len(output),
        archived_to=None,  # CLI fills this in when it knows the path
        haiku_model=haiku_model, elapsed_s=time.monotonic() - t0,
        failure_reason=None,
    )


def make_anthropic_client(*, model: str = "claude-haiku-4-5-20251001"):
    """Construct a callable LLM matching the run_for_user `llm` protocol,
    backed by the Anthropic SDK. SDK transient errors raise
    SummarizerTransientError; everything else raises SummarizerTerminalError."""
    import anthropic

    _TRANSIENT_STATUS_CODES = frozenset({408, 409, 429})
    client = anthropic.Anthropic(max_retries=5)

    class _Client:
        model_name = model
        def complete(self, system: str, prompt: str) -> str:
            try:
                msg = client.messages.create(
                    model=model,
                    max_tokens=4000,
                    system=system,
                    messages=[{"role": "user", "content": prompt}],
                )
            except anthropic.APIConnectionError as e:
                raise SummarizerTransientError(f"connection: {e}") from e
            except anthropic.APIStatusError as e:
                if (
                    getattr(e, "status_code", 0) in _TRANSIENT_STATUS_CODES
                    or getattr(e, "status_code", 0) >= 500
                ):
                    raise SummarizerTransientError(
                        f"status {e.status_code}: {e}"
                    ) from e
                raise SummarizerTerminalError(f"status {e.status_code}: {e}") from e
            except anthropic.APIError as e:
                raise SummarizerTerminalError(f"api: {e}") from e
            parts = []
            for block in msg.content:
                if getattr(block, "type", "") == "text":
                    parts.append(getattr(block, "text", ""))
            return "".join(parts)

    return _Client()