"""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()