| """Persist active Gradio redaction task state for multi-turn post-run eval.""" |
|
|
| from __future__ import annotations |
|
|
| import sys |
| import threading |
| from dataclasses import dataclass, field |
| from pathlib import Path |
| from typing import Any |
|
|
| from session_workspace import ( |
| read_active_redaction_task_file, |
| session_workspace_dir, |
| write_active_redaction_task_file, |
| ) |
|
|
| _LOCK = threading.Lock() |
| _REPO_ROOT = Path(__file__).resolve().parents[2] |
| _EVAL_DIR = Path(__file__).resolve().parents[1] / "eval" |
|
|
|
|
| def _ensure_eval_imports() -> None: |
| text = str(_EVAL_DIR) |
| if text not in sys.path: |
| sys.path.insert(0, text) |
|
|
|
|
| @dataclass |
| class ActiveRedactionTask: |
| document_name: str |
| task_instructions: str |
| follow_up_steer_count: int = 0 |
| cumulative_tool_calls: list[dict[str, Any]] = field(default_factory=list) |
| eval_gold_path: str | None = None |
| eval_gold_match_level: str = "none" |
| eval_gold_match_reason: str = "" |
| encourage_vlm_faces: bool = False |
| encourage_vlm_signatures: bool = False |
|
|
|
|
| def _task_from_raw(raw: dict[str, Any]) -> ActiveRedactionTask | None: |
| doc = str(raw.get("document_name") or "").strip() |
| if not doc: |
| return None |
| instructions = str(raw.get("task_instructions") or "").strip() |
| count = raw.get("follow_up_steer_count", 0) |
| try: |
| follow_up_steer_count = max(0, int(count)) |
| except (TypeError, ValueError): |
| follow_up_steer_count = 0 |
| calls = raw.get("cumulative_tool_calls") |
| if not isinstance(calls, list): |
| calls = [] |
| cumulative_tool_calls = [ |
| c for c in calls if isinstance(c, dict) and str(c.get("tool") or "").strip() |
| ] |
| eval_gold_path = str(raw.get("eval_gold_path") or "").strip() or None |
| return ActiveRedactionTask( |
| document_name=doc, |
| task_instructions=instructions, |
| follow_up_steer_count=follow_up_steer_count, |
| cumulative_tool_calls=cumulative_tool_calls, |
| eval_gold_path=eval_gold_path, |
| eval_gold_match_level=str(raw.get("eval_gold_match_level") or "none"), |
| eval_gold_match_reason=str(raw.get("eval_gold_match_reason") or ""), |
| encourage_vlm_faces=bool(raw.get("encourage_vlm_faces")), |
| encourage_vlm_signatures=bool(raw.get("encourage_vlm_signatures")), |
| ) |
|
|
|
|
| def _read_task(session_hash: str) -> ActiveRedactionTask | None: |
| raw = read_active_redaction_task_file(session_hash) |
| if raw is None: |
| return None |
| return _task_from_raw(raw) |
|
|
|
|
| def _write_task(session_hash: str, task: ActiveRedactionTask) -> bool: |
| payload = { |
| "document_name": task.document_name, |
| "task_instructions": task.task_instructions, |
| "follow_up_steer_count": task.follow_up_steer_count, |
| "cumulative_tool_calls": task.cumulative_tool_calls, |
| "eval_gold_path": task.eval_gold_path, |
| "eval_gold_match_level": task.eval_gold_match_level, |
| "eval_gold_match_reason": task.eval_gold_match_reason, |
| "encourage_vlm_faces": bool(task.encourage_vlm_faces), |
| "encourage_vlm_signatures": bool(task.encourage_vlm_signatures), |
| } |
| return write_active_redaction_task_file(session_hash, payload) |
|
|
|
|
| def _resolve_pdf_path( |
| session_hash: str, document_name: str, pdf_path: Path | None |
| ) -> Path | None: |
| if pdf_path is not None and pdf_path.is_file(): |
| return pdf_path |
| candidate = session_workspace_dir(session_hash) / document_name |
| return candidate if candidate.is_file() else None |
|
|
|
|
| def begin_redaction_task( |
| session_hash: str, |
| document_name: str, |
| task_instructions: str, |
| *, |
| pdf_path: Path | None = None, |
| encourage_vlm_faces: bool = False, |
| encourage_vlm_signatures: bool = False, |
| ) -> str: |
| """ |
| Start a new redaction task in *session_hash* (resets steer count and tool trace). |
| |
| Returns a short status line for the activity log (gold match result when auto-gold on). |
| """ |
| doc = (document_name or "").strip() |
| if not doc or not (session_hash or "").strip(): |
| return "" |
| task = ActiveRedactionTask( |
| document_name=doc, |
| task_instructions=(task_instructions or "").strip(), |
| follow_up_steer_count=0, |
| cumulative_tool_calls=[], |
| encourage_vlm_faces=bool(encourage_vlm_faces), |
| encourage_vlm_signatures=bool(encourage_vlm_signatures), |
| ) |
| log_line = "" |
| try: |
| _ensure_eval_imports() |
| from agent_eval_env import agent_eval_auto_gold_enabled, resolve_gold_root |
| from gold_match import resolve_gold_case_for_task |
|
|
| if agent_eval_auto_gold_enabled(): |
| upload = _resolve_pdf_path(session_hash, doc, pdf_path) |
| if upload is not None: |
| result = resolve_gold_case_for_task( |
| pdf_path=upload, |
| document_name=doc, |
| task_instructions=task.task_instructions, |
| gold_root=resolve_gold_root(_REPO_ROOT), |
| ) |
| task.eval_gold_match_level = result.match_level |
| task.eval_gold_match_reason = result.reason |
| if result.gold_dir is not None: |
| task.eval_gold_path = str(result.gold_dir.resolve()) |
| log_line = ( |
| f"**Eval gold case:** `{result.case_id or result.gold_dir.name}` " |
| f"({result.match_level})" |
| ) |
| else: |
| log_line = ( |
| f"**Eval skipped:** {result.reason or 'no matching gold case'}" |
| ) |
| else: |
| task.eval_gold_match_reason = ( |
| "upload PDF not found in session workspace" |
| ) |
| log_line = f"**Eval skipped:** {task.eval_gold_match_reason}" |
| except Exception as exc: |
| task.eval_gold_match_reason = f"gold match error: {exc}" |
| log_line = f"**Eval skipped:** {task.eval_gold_match_reason}" |
|
|
| with _LOCK: |
| _write_task(session_hash, task) |
| return log_line |
|
|
|
|
| def note_follow_up_or_steer_message(session_hash: str) -> bool: |
| """ |
| Record one user follow-up or in-run steer message for the active task. |
| |
| Returns True when an active task was updated. |
| """ |
| if not (session_hash or "").strip(): |
| return False |
| with _LOCK: |
| task = _read_task(session_hash) |
| if task is None: |
| return False |
| task.follow_up_steer_count += 1 |
| return _write_task(session_hash, task) |
|
|
|
|
| def _tool_calls_from_events(events: list[Any]) -> list[dict[str, Any]]: |
| calls: list[dict[str, Any]] = [] |
| for event in events: |
| kind = str(getattr(event, "kind", "") or "") |
| if kind != "tool_start": |
| continue |
| name = str(getattr(event, "tool_name", None) or "tool") |
| args = getattr(event, "tool_args", None) |
| if not isinstance(args, dict): |
| args = {} |
| calls.append({"tool": name, "args": args}) |
| return calls |
|
|
|
|
| def append_turn_tool_calls(session_hash: str, events: list[Any]) -> None: |
| """Merge tool-start events from the latest agent turn into the task trace.""" |
| if not (session_hash or "").strip() or not events: |
| return |
| new_calls = _tool_calls_from_events(events) |
| if not new_calls: |
| return |
| with _LOCK: |
| task = _read_task(session_hash) |
| if task is None: |
| return |
| task.cumulative_tool_calls.extend(new_calls) |
| _write_task(session_hash, task) |
|
|
|
|
| def get_active_task_for_eval(session_hash: str) -> ActiveRedactionTask | None: |
| """Return the current task snapshot for post-run eval (or None).""" |
| if not (session_hash or "").strip(): |
| return None |
| return _read_task(session_hash) |
|
|
