Spaces:

defenestrate
/

asa

Sleeping

asa / scripts /lib /chat_eval /notes_summarizer.py

ASA Space: squashed history (instructions and settings live in the private pack)

526950e 24 days ago

10.6 kB

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

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