"""Curated LangGraph tools for doc_redaction orchestration (no shell).""" from __future__ import annotations import json import os import re import subprocess import sys from pathlib import Path from typing import Any _SHARED_DIR = Path(__file__).resolve().parents[1] / "shared" if str(_SHARED_DIR) not in sys.path: sys.path.insert(0, str(_SHARED_DIR)) from remote_redaction import ( # noqa: E402 call_doc_redact, extract_server_paths, fetch_redaction_files, make_redaction_client, ) from session_workspace import session_workspace_dir # noqa: E402 _MAX_TEXT_BYTES = int(os.environ.get("LANGGRAPH_MAX_WORKSPACE_TEXT_BYTES", "1500000")) _MAX_SCRIPT_SECONDS = int(os.environ.get("LANGGRAPH_WORKSPACE_SCRIPT_TIMEOUT", "300")) def _session_root(session_hash: str | None) -> Path: if session_hash: return session_workspace_dir(session_hash) from session_workspace import workspace_base_dir return workspace_base_dir() _MAX_TEXT_BYTES = int(os.environ.get("LANGGRAPH_MAX_WORKSPACE_TEXT_BYTES", "1500000")) _MAX_SCRIPT_SECONDS = int(os.environ.get("LANGGRAPH_WORKSPACE_SCRIPT_TIMEOUT", "300")) _TOOL_ARG_KEY_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") _DOC_REDACT_PDF_KEYS = ( "pdf_relative_path", "pdf_path", "pdf", "document_file", ) _DOC_REDACT_DEST_KEYS = ( "dest_relative_dir", "dest_dir", "dest", "output_dir", ) _SCRIPT_PATH_KEYS = ( "relative_path", "path", "script", "script_path", "file", "filename", ) _PATH_ONLY_TOOL_KEYS = frozenset(_SCRIPT_PATH_KEYS) def _merge_tool_arg_dicts(*values: Any) -> dict[str, Any]: merged: dict[str, Any] = {} for value in values: if isinstance(value, dict): merged.update(value) return merged def _sanitize_tool_dict(payload: dict[str, Any]) -> dict[str, Any]: """Drop hallucinated tool-arg keys from weak local models (URLs, JSON fragments).""" clean: dict[str, Any] = {} for key, value in payload.items(): if isinstance(key, str) and _TOOL_ARG_KEY_RE.fullmatch(key): clean[key] = value return clean def _first_string(payload: dict[str, Any], keys: tuple[str, ...]) -> str: for key in keys: value = payload.get(key) if isinstance(value, str) and value.strip(): return value.strip() return "" def _looks_like_filesystem_path(text: str) -> bool: normalized = text.strip().replace("\\", "/") if not normalized: return False if normalized.startswith(("/", "~")): return True if re.match(r"^[A-Za-z]:/", normalized): return True return "/" in normalized or normalized.lower().endswith(".pdf") def _deep_flatten_tool_payload(*values: Any) -> dict[str, Any]: """ Recursively collect tool-arg fields from nested local-model structures. Weak models often nest the real args under an absolute path key, e.g. ``{"pdf_relative_path": {"/abs/path/doc.pdf": {"pdf_relative_path": "doc.pdf"}}}``. """ merged: dict[str, Any] = {} def absorb(key: str, val: Any) -> None: if isinstance(val, str) and val.strip(): merged[key] = val.strip() elif val is not None and not isinstance(val, (dict, list, tuple)): merged[key] = val def walk(node: Any) -> None: if isinstance(node, dict): for key, val in node.items(): if isinstance(key, str) and _TOOL_ARG_KEY_RE.fullmatch(key): if isinstance(val, dict): walk(val) else: absorb(key, val) elif isinstance(key, str) and _looks_like_filesystem_path(key): if isinstance(val, dict): walk(val) elif isinstance(val, str) and val.strip(): absorb("pdf_path", val) else: key_path = Path(key.replace("\\", "/")) if key_path.suffix.lower() in _OUTPUT_FILE_EXTENSIONS: name = key_path.name if name: absorb("pdf_path", name) else: walk(val) elif isinstance(node, str) and node.strip(): absorb("_literal", node) elif isinstance(node, (list, tuple)): for item in node: walk(item) for value in values: walk(value) return merged def _normalize_workspace_relative_path(path: str, session_hash: str | None) -> str: """Strip a session workspace prefix from absolute paths; keep basename as fallback.""" text = path.strip().replace("\\", "/") if not text: return text root = _session_root(session_hash).resolve() try: raw_path = Path(path.strip()) if raw_path.is_absolute(): resolved = raw_path.resolve() rel = os.path.relpath(str(resolved), str(root)) if not rel.startswith(".."): return rel.replace("\\", "/") except (OSError, ValueError): pass root_posix = root.as_posix().rstrip("/") lowered = text.lower() root_lower = root_posix.lower() idx = lowered.find(root_lower) if idx != -1: suffix = text[idx + len(root_posix) :].lstrip("/\\") if suffix: return suffix.replace("\\", "/") if "/" in text or ":" in text: name = Path(text).name if name: return name return text def _default_dest_for_pdf(pdf_relative_path: str) -> str: stem = Path(pdf_relative_path.replace("\\", "/")).stem return f"redact/{stem or 'document'}/output_redact" def _default_review_apply_dest_for_pdf(pdf_relative_path: str) -> str: stem = Path(pdf_relative_path.replace("\\", "/")).stem return f"redact/{stem or 'document'}/review/output_review_final" def _default_review_apply_dest_for_review_csv(review_csv_relative_path: str) -> str: normalized = review_csv_relative_path.replace("\\", "/") parts = Path(normalized).parts if "output_redact" in parts: idx = parts.index("output_redact") doc = Path(*parts[:idx]) return str(doc / "review" / "output_review_final").replace("\\", "/") return _default_review_apply_dest_for_pdf(review_csv_relative_path) _OUTPUT_FILE_EXTENSIONS = frozenset( { ".pdf", ".csv", ".json", ".txt", ".py", ".zip", ".png", ".jpg", ".jpeg", ".xlsx", } ) def _looks_like_file_relative_path(rel: str) -> bool: ext = Path(rel.replace("\\", "/")).suffix.lower() return bool(ext) and ext in _OUTPUT_FILE_EXTENSIONS def _ensure_workspace_output_dir( session_hash: str | None, dest_relative_dir: Any, *, pdf_relative_path: str | None = None, review_csv_relative_path: str | None = None, default_for: str = "doc_redact", ) -> Path: """ Resolve an output directory under the session workspace. Weak local models often pass the PDF path (or another file) as dest_relative_dir; on Windows ``Path.mkdir()`` then raises WinError 183 when that path is an existing file. """ rel = "" if dest_relative_dir is not None and dest_relative_dir != "": try: rel = _coerce_relative_path(dest_relative_dir, label="dest_relative_dir") except ValueError: rel = "" pdf_rel = "" if pdf_relative_path: try: pdf_rel = _coerce_relative_path( pdf_relative_path, label="pdf_relative_path" ) except ValueError: pdf_rel = str(pdf_relative_path).strip().replace("\\", "/") review_rel = "" if review_csv_relative_path: try: review_rel = _coerce_relative_path( review_csv_relative_path, label="review_csv_relative_path" ) except ValueError: review_rel = str(review_csv_relative_path).strip().replace("\\", "/") if not rel or _looks_like_file_relative_path(rel): if default_for == "review_apply": if review_rel: rel = _default_review_apply_dest_for_review_csv(review_rel) elif pdf_rel: rel = _default_review_apply_dest_for_pdf(pdf_rel) elif pdf_rel: rel = _default_dest_for_pdf(pdf_rel) if not rel: raise ValueError( "dest_relative_dir must be an output directory path, not a document file." ) candidate = _resolve_workspace_path(session_hash, rel) if candidate.is_file(): if default_for == "review_apply": rel = ( _default_review_apply_dest_for_review_csv(review_rel) if review_rel else _default_review_apply_dest_for_pdf(pdf_rel or candidate.name) ) else: rel = _default_dest_for_pdf(pdf_rel or candidate.name) candidate = _resolve_workspace_path(session_hash, rel) candidate.mkdir(parents=True, exist_ok=True) return candidate def _coerce_relative_path(value: Any, *, label: str = "path") -> str: """ Normalize tool path arguments. Local OpenAI-compatible models sometimes emit nested dicts or pass the full tool-args object as a single value; ``Path / dict`` then fails at runtime. """ if isinstance(value, Path): text = value.as_posix() elif isinstance(value, str): text = value.strip() elif isinstance(value, dict): payload = _sanitize_tool_dict(value) text = _first_string( payload, ( label, "relative_path", "path", *_DOC_REDACT_PDF_KEYS, *_DOC_REDACT_DEST_KEYS, "review_csv_relative_path", "redacted_pdf_relative_path", "ocr_words_csv_relative_path", "script", "script_path", "file", "filename", "value", ), ) if not text and len(payload) == 1: return _coerce_relative_path(next(iter(payload.values())), label=label) if not text: for key in ("relative_path", label, "path"): nested = payload.get(key) if isinstance(nested, dict): return _coerce_relative_path(nested, label=label) if not text: for nested in value.values(): try: return _coerce_relative_path(nested, label=label) except ValueError: continue if not text: raise ValueError(f"Tool {label} must be a string path, got dict: {value!r}") elif isinstance(value, (list, tuple)) and len(value) == 1: return _coerce_relative_path(value[0], label=label) else: text = str(value).strip() if not text: raise ValueError(f"Tool {label} is empty.") return text.replace("\\", "/") def _coerce_tool_text_content(value: Any, *, label: str = "content") -> str: """Normalize write_workspace_text body from messy local-model tool calls.""" if isinstance(value, str): return value if isinstance(value, (bytes, bytearray)): return bytes(value).decode("utf-8", errors="replace") if isinstance(value, dict): for key in (label, "content", "text", "body", "data", "source"): nested = value.get(key) if isinstance(nested, str): return nested if isinstance(nested, dict): return _coerce_tool_text_content(nested, label=label) str_values = [item for item in value.values() if isinstance(item, str)] if len(str_values) > 1: return max(str_values, key=len) if len(str_values) == 1: return str_values[0] payload = _sanitize_tool_dict(value) for key in (label, "content", "text", "body", "script", "data", "source"): nested = payload.get(key) if isinstance(nested, str): return nested if isinstance(nested, dict): return _coerce_tool_text_content(nested, label=label) str_values = [item for item in payload.values() if isinstance(item, str)] if len(str_values) == 1: return str_values[0] if len(payload) == 1: return _coerce_tool_text_content(next(iter(payload.values())), label=label) raise ValueError(f"Tool {label} must be text, got dict: {value!r}") if isinstance(value, (list, tuple)) and len(value) == 1: return _coerce_tool_text_content(value[0], label=label) raise ValueError( f"Tool {label} must be text, got {type(value).__name__}: {value!r}" ) def _should_resolve_script_path(payload: dict[str, Any], rel_raw: str) -> bool: """Only remap bare script names; leave explicit paths and non-.py files alone.""" if _first_string(payload, ("script", "script_path")): return True rel = rel_raw.replace("\\", "/") if "/" in rel: return False name = Path(rel).name if name.lower().endswith(".py"): return True return "." not in name def _parse_write_workspace_text_input( relative_path: Any, content: Any, ) -> tuple[str, str]: """Merge/normalize write_workspace_text args from messy local-model tool calls.""" merged = _merge_tool_arg_dicts(relative_path, content) payload = _sanitize_tool_dict(merged) rel_raw = _first_string(payload, _SCRIPT_PATH_KEYS) if not rel_raw: nested = payload.get("relative_path") if isinstance(nested, dict): rel_raw = _coerce_relative_path(nested, label="relative_path") if not rel_raw and isinstance(relative_path, str): rel_raw = relative_path.strip() if not rel_raw: raise ValueError( "write_workspace_text requires relative_path or script (e.g. fix_policy.py)." ) rel_raw = rel_raw.replace("\\", "/") content_raw: Any = merged.get("content") if isinstance(content_raw, dict): content_raw = _coerce_tool_text_content(content_raw) if content_raw is None and isinstance(content, str): content_raw = content if content_raw is None: for key, value in merged.items(): if key in _PATH_ONLY_TOOL_KEYS: continue if isinstance(value, dict): content_raw = _coerce_tool_text_content(value) break content_raw = value break if content_raw is None: raise ValueError("write_workspace_text requires content text.") return rel_raw, _coerce_tool_text_content(content_raw) def _resolve_script_relative_path(session_hash: str | None, script: str) -> str: """Map a script filename or relative path to a workspace-relative .py path.""" rel = script.replace("\\", "/").strip() if "/" in rel: return rel name = Path(rel).name if not name.lower().endswith(".py"): name = f"{name}.py" if name else "fix_policy.py" root = _session_root(session_hash).resolve() matches = sorted( (path for path in root.rglob(name) if path.is_file()), key=lambda path: len(path.relative_to(root).parts), ) if matches: return str(matches[0].relative_to(root)).replace("\\", "/") output_dirs = sorted( (path for path in root.rglob("output_redact") if path.is_dir()), key=lambda path: len(path.relative_to(root).parts), ) if output_dirs: target = output_dirs[0] return str((target / name).relative_to(root)).replace("\\", "/") return f"scripts/{name}" def _parse_doc_redact_tool_input( pdf_relative_path: Any, dest_relative_dir: Any | None, *, ocr_method: str | None, pii_method: str | None, session_hash: str | None = None, ) -> tuple[str, str, str | None, str | None]: """Merge/normalize doc_redact tool args from messy local-model tool calls.""" payload = _deep_flatten_tool_payload(pdf_relative_path, dest_relative_dir) pdf_raw = _first_string(payload, _DOC_REDACT_PDF_KEYS) if not pdf_raw: pdf_raw = str(payload.get("_literal") or "").strip() if not pdf_raw and isinstance(pdf_relative_path, str): pdf_raw = pdf_relative_path.strip() if not pdf_raw: raise ValueError( "doc_redact requires a PDF path (pdf_relative_path or pdf_path)." ) pdf_raw = _normalize_workspace_relative_path(pdf_raw, session_hash) pdf_rel = _coerce_relative_path(pdf_raw, label="pdf_relative_path") dest_raw = _first_string(payload, _DOC_REDACT_DEST_KEYS) if not dest_raw and isinstance(dest_relative_dir, str): dest_raw = dest_relative_dir.strip() dest_rel = ( _coerce_relative_path(dest_raw, label="dest_relative_dir") if dest_raw else _default_dest_for_pdf(pdf_rel) ) ocr = ocr_method or _first_string(payload, ("ocr_method",)) or None pii = pii_method or _first_string(payload, ("pii_method",)) or None return pdf_rel, dest_rel, ocr, pii def _parse_review_apply_tool_input( pdf_relative_path: Any, review_csv_relative_path: Any, dest_relative_dir: Any | None, *, session_hash: str | None = None, ) -> tuple[str, str, str]: """Merge/normalize review_apply tool args from messy local-model tool calls.""" payload = _deep_flatten_tool_payload( pdf_relative_path, review_csv_relative_path, dest_relative_dir ) pdf_raw = _first_string(payload, _DOC_REDACT_PDF_KEYS) if not pdf_raw: pdf_raw = str(payload.get("_literal") or "").strip() if not pdf_raw and isinstance(pdf_relative_path, str): pdf_raw = pdf_relative_path.strip() if not pdf_raw: raise ValueError( "review_apply requires a PDF path (pdf_relative_path or pdf_path)." ) pdf_raw = _normalize_workspace_relative_path(pdf_raw, session_hash) pdf_rel = _coerce_relative_path(pdf_raw, label="pdf_relative_path") review_raw = _first_string( payload, ( "review_csv_relative_path", "review_csv", "csv_path", "csv", "review_file", ), ) if not review_raw and isinstance(review_csv_relative_path, str): review_raw = review_csv_relative_path.strip() if not review_raw: raise ValueError( "review_apply requires a review CSV path (review_csv_relative_path)." ) review_rel = _coerce_relative_path(review_raw, label="review_csv_relative_path") dest_raw = _first_string(payload, _DOC_REDACT_DEST_KEYS) if not dest_raw and isinstance(dest_relative_dir, str): dest_raw = dest_relative_dir.strip() dest_rel = ( _coerce_relative_path(dest_raw, label="dest_relative_dir") if dest_raw else "" ) return pdf_rel, review_rel, dest_rel def _resolve_workspace_path(session_hash: str | None, relative_path: Any) -> Path: rel = _coerce_relative_path(relative_path) root = _session_root(session_hash).resolve() candidate = (root / rel).resolve() if not str(candidate).startswith(str(root)): raise ValueError(f"Path escapes session workspace: {rel}") return candidate def _resolve_workspace_pdf(session_hash: str | None, pdf_relative_path: str) -> Path: """Resolve a PDF under the session workspace; fall back to unique basename match.""" try: candidate = _resolve_workspace_path(session_hash, pdf_relative_path) if candidate.is_file(): return candidate except ValueError: candidate = None root = _session_root(session_hash).resolve() basename = Path(pdf_relative_path.replace("\\", "/")).name if not basename: raise FileNotFoundError(f"PDF not found in workspace: {pdf_relative_path}") matches = sorted( (path for path in root.rglob(basename) if path.is_file()), key=lambda path: len(path.relative_to(root).parts), ) if not matches: missing = candidate or (root / pdf_relative_path) raise FileNotFoundError(f"PDF not found in workspace: {missing}") if len(matches) > 1: rels = [str(path.relative_to(root)).replace("\\", "/") for path in matches[:5]] raise ValueError( "Multiple PDFs match " f"{basename!r} in the workspace; use a relative path. Matches: {rels}" ) return matches[0].resolve() def list_workspace_files(session_hash: str | None = None) -> str: """List files under the current session workspace.""" root = _session_root(session_hash) if not root.is_dir(): return json.dumps({"files": [], "root": str(root)}) files: list[str] = [] for path in sorted(root.rglob("*")): if path.is_file(): files.append(str(path.relative_to(root)).replace("\\", "/")) return json.dumps({"root": str(root), "files": files[:500]}) def run_doc_redact( pdf_relative_path: str, dest_relative_dir: str = "", *, session_hash: str | None = None, ocr_method: str | None = None, pii_method: str | None = None, deny_list: list[str] | None = None, allow_list: list[str] | None = None, ) -> str: """Run Pass 1 redaction via /doc_redact and download artifacts into the session workspace.""" try: pdf_rel, dest_rel, ocr_from_tool, pii_from_tool = _parse_doc_redact_tool_input( pdf_relative_path, dest_relative_dir, ocr_method=ocr_method, pii_method=pii_method, session_hash=session_hash, ) pdf = _resolve_workspace_pdf(session_hash, pdf_rel) dest = _ensure_workspace_output_dir( session_hash, dest_rel, pdf_relative_path=pdf_rel, default_for="doc_redact", ) result, saved = call_doc_redact( pdf, dest, ocr_method=ocr_from_tool or os.environ.get("AGENT_DEFAULT_OCR_METHOD"), pii_method=pii_from_tool or os.environ.get("AGENT_DEFAULT_PII_METHOD"), deny_list=deny_list, allow_list=allow_list, ) except (ValueError, FileNotFoundError) as exc: return json.dumps({"error": str(exc)}) message = result[1] if isinstance(result, (list, tuple)) and len(result) > 1 else "" payload = { "message": str(message or "doc_redact completed."), "saved_paths": [str(p) for p in saved], "server_paths": extract_server_paths(result), } return json.dumps(payload, indent=2) def _discover_ocr_words_csv(review_csv: Path) -> Path | None: """Find the word-level OCR CSV sibling of a *_review_file.csv.""" parent = review_csv.parent review_csv.name.lower() patterns = ( "*word*ocr*.csv", "*ocr*word*.csv", "*_words.csv", "*words*.csv", ) for pattern in patterns: for candidate in sorted(parent.glob(pattern)): if candidate.resolve() == review_csv.resolve(): continue if "_review_file" in candidate.name.lower(): continue return candidate for candidate in sorted(parent.glob("*.csv")): if candidate.resolve() == review_csv.resolve(): continue name = candidate.name.lower() if "_review_file" in name: continue if "word" in name or "ocr" in name: return candidate return None def read_workspace_text( relative_path: Any, *, session_hash: str | None = None, max_bytes: int | None = None, ) -> str: """Read a UTF-8 text file from the session workspace (CSV, JSON, Python script).""" try: rel = _coerce_relative_path(relative_path, label="relative_path") path = _resolve_workspace_path(session_hash, rel) except ValueError as exc: return json.dumps({"error": str(exc), "relative_path": str(relative_path)}) except FileNotFoundError as exc: return json.dumps({"error": str(exc), "relative_path": str(relative_path)}) if not path.is_file(): return json.dumps({"error": f"File not found: {rel}"}) limit = max_bytes if max_bytes is not None else _MAX_TEXT_BYTES size = path.stat().st_size if size > limit: return json.dumps( { "error": ( f"File too large to read ({size} bytes > {limit}). " "Use run_workspace_python_script on a .py file instead." ) } ) text = path.read_text(encoding="utf-8-sig") max_lines = int(os.environ.get("LANGGRAPH_READ_CSV_MAX_LINES", "60")) if path.suffix.lower() == ".csv" or path.name.lower().endswith(".csv"): lines = text.splitlines() if len(lines) > max_lines: preview = "\n".join(lines[:max_lines]) return ( f"CSV preview for {rel} (lines 1-{max_lines} of {len(lines)}). " "Edit the full file with write_workspace_text or run_workspace_python_script.\n\n" f"{preview}" ) return text def write_workspace_text( relative_path: Any, content: Any, *, session_hash: str | None = None, ) -> str: """Write UTF-8 text into the session workspace (preserve utf-8-sig for review CSVs).""" try: merged = _merge_tool_arg_dicts(relative_path, content) rel, body = _parse_write_workspace_text_input(relative_path, content) if _should_resolve_script_path(_sanitize_tool_dict(merged), rel): rel = _resolve_script_relative_path(session_hash, rel) path = _resolve_workspace_path(session_hash, rel) except ValueError as exc: return json.dumps({"error": str(exc)}) if len(body.encode("utf-8")) > _MAX_TEXT_BYTES: return json.dumps({"error": f"Content too large (>{_MAX_TEXT_BYTES} bytes)."}) path.parent.mkdir(parents=True, exist_ok=True) unchanged = False if path.is_file(): try: unchanged = path.read_text(encoding="utf-8-sig") == body except OSError: unchanged = False if not unchanged: path.write_text(body, encoding="utf-8-sig") root = _session_root(session_hash) rel_written = str(path.relative_to(root)).replace("\\", "/") payload: dict[str, Any] = { "written": rel_written, "bytes": path.stat().st_size, } if unchanged: payload["unchanged"] = True if path.suffix.lower() == ".py": payload["next_step"] = ( "Script already saved. Call run_workspace_python_script with " f"relative_path={rel_written!r} now — do not call write_workspace_text " "again unless the script body must change." ) return json.dumps(payload) def run_workspace_python_script( relative_path: Any, content: Any = None, *, session_hash: str | None = None, ) -> str: """Run a Python script already saved under the session workspace.""" merged = _merge_tool_arg_dicts(relative_path, content) written_path: str | None = None if isinstance(merged.get("content"), str): write_out = write_workspace_text( relative_path, content, session_hash=session_hash ) write_payload = json.loads(write_out) if write_payload.get("error"): return write_out written_path = write_payload.get("written") try: if written_path: rel = written_path else: payload = _sanitize_tool_dict(merged) rel = _first_string(payload, _SCRIPT_PATH_KEYS) if not rel: nested = payload.get("relative_path") if isinstance(nested, dict): rel = _coerce_relative_path(nested, label="relative_path") if not rel and not isinstance(relative_path, dict): rel = _coerce_relative_path(relative_path, label="relative_path") if not rel: raise ValueError( "run_workspace_python_script requires relative_path or script " "(e.g. fix_policy.py)." ) rel = rel.replace("\\", "/") if _should_resolve_script_path(payload, rel): rel = _resolve_script_relative_path(session_hash, rel) path = _resolve_workspace_path(session_hash, rel) except ValueError as exc: return json.dumps({"error": str(exc)}) if path.suffix.lower() != ".py": return json.dumps({"error": "Only .py scripts are allowed."}) completed = subprocess.run( [sys.executable, str(path)], cwd=str(path.parent), capture_output=True, text=True, timeout=_MAX_SCRIPT_SECONDS, check=False, ) return json.dumps( { "returncode": completed.returncode, "stdout": completed.stdout[-20000:], "stderr": completed.stderr[-20000:], }, indent=2, ) _REVIEW_APPROVED: dict[str, bool] = {} def approve_review_apply(session_hash: str | None = None) -> str: """Mark review_apply as approved for human-in-the-loop gating.""" key = session_hash or "" _REVIEW_APPROVED[key] = True return json.dumps({"approved": True, "session": key}) def run_review_apply( pdf_relative_path: str, review_csv_relative_path: str, dest_relative_dir: str, *, session_hash: str | None = None, ) -> str: """Apply an edited review CSV via /review_apply and download outputs.""" if os.environ.get("LANGGRAPH_REQUIRE_REVIEW_APPROVAL", "").strip().lower() in { "1", "true", "yes", }: key = session_hash or "" if not _REVIEW_APPROVED.pop(key, False): return json.dumps( { "error": ( "Human approval required before review_apply. " "Set LANGGRAPH_REQUIRE_REVIEW_APPROVAL=false to disable, or call " "approve_review_apply first." ) } ) from gradio_client import handle_file try: pdf_rel, review_rel, dest_rel = _parse_review_apply_tool_input( pdf_relative_path, review_csv_relative_path, dest_relative_dir, session_hash=session_hash, ) except ValueError as exc: return json.dumps({"error": str(exc)}) pdf = _resolve_workspace_pdf(session_hash, pdf_rel) review_csv = _resolve_workspace_path(session_hash, review_rel) dest = _ensure_workspace_output_dir( session_hash, dest_rel, pdf_relative_path=pdf_rel, review_csv_relative_path=review_rel, default_for="review_apply", ) client = make_redaction_client() result = client.predict( api_name="/review_apply", pdf_file=handle_file(str(pdf)), review_csv_file=handle_file(str(review_csv)), ) server_paths = extract_server_paths(result) saved = fetch_redaction_files(server_paths, dest) message = result[1] if isinstance(result, (list, tuple)) and len(result) > 1 else "" return json.dumps( { "message": str(message or "review_apply completed."), "saved_paths": [str(p) for p in saved], "server_paths": server_paths, }, indent=2, ) def _resolve_optional_redacted_pdf( session_hash: str | None, redacted_pdf_relative_path: Any, *, review_csv: Path, ) -> Path | None: """Resolve optional post-apply PDF; reject CSV / non-PDF mix-ups from the model.""" if redacted_pdf_relative_path is None: return None if ( isinstance(redacted_pdf_relative_path, str) and not redacted_pdf_relative_path.strip() ): return None rel = _coerce_relative_path( redacted_pdf_relative_path, label="redacted_pdf_relative_path" ) if not rel: return None lower = rel.lower().replace("\\", "/") name = Path(lower).name if ( lower.endswith((".csv", ".json", ".py", ".txt", ".md")) or "review_file" in name or name.endswith("_review.csv") ): raise ValueError( "redacted_pdf_relative_path must be a PDF (e.g. *_redacted.pdf). " f"Got {rel!r}. For pre-apply verify_coverage, omit " "redacted_pdf_relative_path entirely. For post-apply checks, pass the " "*_redacted.pdf produced by review_apply." ) if not lower.endswith(".pdf"): raise ValueError( "redacted_pdf_relative_path must end with .pdf " f"(got {rel!r}). Omit it for pre-apply checks." ) path = _resolve_workspace_path(session_hash, rel) if path.resolve() == review_csv.resolve(): raise ValueError( "redacted_pdf_relative_path must not be the review CSV. " "Omit it for pre-apply verify_coverage, or pass *_redacted.pdf." ) if not path.is_file(): raise FileNotFoundError(f"redacted PDF not found: {rel}") return path def run_verify_coverage( review_csv_relative_path: str, *, session_hash: str | None = None, redacted_pdf_relative_path: str | None = None, ocr_words_csv_relative_path: str | None = None, must_redact: list[str] | None = None, must_not_redact: list[str] | None = None, ) -> str: """Run Pass 1 coverage verification on workspace-local CSV/PDF paths.""" from redaction_langgraph.verify_coverage_lib import verify_redaction_coverage try: review_rel = _coerce_relative_path( review_csv_relative_path, label="review_csv_relative_path" ) review_csv = _resolve_workspace_path(session_hash, review_rel) if ocr_words_csv_relative_path: ocr_rel = _coerce_relative_path( ocr_words_csv_relative_path, label="ocr_words_csv_relative_path" ) ocr_words_csv = _resolve_workspace_path(session_hash, ocr_rel) else: discovered = _discover_ocr_words_csv(review_csv) if discovered is None: return json.dumps( { "error": ( "Could not find word-level OCR CSV beside the review CSV. " "Pass ocr_words_csv_relative_path explicitly." ), "review_csv": str(review_csv), } ) ocr_words_csv = discovered redacted_pdf = _resolve_optional_redacted_pdf( session_hash, redacted_pdf_relative_path, review_csv=review_csv, ) report = verify_redaction_coverage( review_csv, ocr_words_csv, must_redact=must_redact, must_not_redact=must_not_redact, redacted_pdf_path=redacted_pdf, ) except (ValueError, re.error, FileNotFoundError, OSError) as exc: return json.dumps( { "error": str(exc), "hint": ( "verify_coverage args: review_csv_relative_path (required), " "optional redacted_pdf_relative_path (*_redacted.pdf only; " "omit for pre-apply), optional ocr_words_csv_relative_path." ), }, indent=2, ) except Exception as exc: # noqa: BLE001 — keep LangGraph tool node alive return json.dumps( { "error": f"{type(exc).__name__}: {exc}", "hint": ( "verify_coverage failed unexpectedly. Check paths: review CSV vs " "optional *_redacted.pdf (never pass the review CSV as the PDF)." ), }, indent=2, ) payload = report.to_dict() payload["ocr_words_csv"] = str(ocr_words_csv) if redacted_pdf is not None: payload["redacted_pdf"] = str(redacted_pdf) return json.dumps(payload, indent=2, default=str) def build_langgraph_tools(session_hash: str | None): """Return LangChain tools bound to *session_hash* workspace.""" from langchain_core.tools import StructuredTool return [ StructuredTool.from_function( name="list_workspace_files", description="List files in the current session workspace.", func=lambda: list_workspace_files(session_hash), ), StructuredTool.from_function( name="doc_redact", description=( "Run initial document redaction (Pass 1) via /doc_redact. " "pdf_relative_path is workspace-relative (e.g. filename.pdf). " "dest_relative_dir is optional." ), func=lambda pdf_relative_path, dest_relative_dir="", ocr_method=None, pii_method=None: run_doc_redact( pdf_relative_path, dest_relative_dir, session_hash=session_hash, ocr_method=ocr_method, pii_method=pii_method, ), ), StructuredTool.from_function( name="approve_review_apply", description="Approve review_apply when LANGGRAPH_REQUIRE_REVIEW_APPROVAL is enabled.", func=lambda: approve_review_apply(session_hash), ), StructuredTool.from_function( name="review_apply", description=( "Apply an edited *_review_file.csv to the source PDF via /review_apply. " "Paths are relative to the session workspace." ), func=lambda pdf_relative_path, review_csv_relative_path, dest_relative_dir: run_review_apply( pdf_relative_path, review_csv_relative_path, dest_relative_dir, session_hash=session_hash, ), ), StructuredTool.from_function( name="verify_coverage", description=( "Verify Pass 1 redaction coverage on a *_review_file.csv (+ auto-discovered " "word OCR CSV). Returns pass_strict and pages needing fixes. " "For pre-apply checks, pass only review_csv_relative_path (omit " "redacted_pdf_relative_path). For post-apply checks, pass " "redacted_pdf_relative_path as the *_redacted.pdf from review_apply — " "never the review CSV. " "must_redact and must_not_redact: list of regex strings (one term per item), e.g. " '["Hyde", "Lauren\\\\s+Lilley", "Poss\\\\b"]. A single pipe-separated string is also accepted.' ), func=lambda review_csv_relative_path, redacted_pdf_relative_path=None, ocr_words_csv_relative_path=None, must_redact=None, must_not_redact=None: run_verify_coverage( review_csv_relative_path, session_hash=session_hash, redacted_pdf_relative_path=redacted_pdf_relative_path, ocr_words_csv_relative_path=ocr_words_csv_relative_path, must_redact=must_redact, must_not_redact=must_not_redact, ), ), StructuredTool.from_function( name="read_workspace_text", description="Read a text file (CSV, JSON, .py) from the session workspace.", func=lambda relative_path: read_workspace_text( relative_path, session_hash=session_hash ), ), StructuredTool.from_function( name="write_workspace_text", description=( "Write UTF-8 text into the session workspace (use utf-8-sig for review CSV edits). " "Keep content compact — prefer short .py scripts that read OCR/review CSVs and " "add rows programmatically; avoid huge hard-coded lists in the content argument " "(large/quote-heavy payloads often break tool-call JSON on local models)." ), func=lambda relative_path, content: write_workspace_text( relative_path, content, session_hash=session_hash ), ), StructuredTool.from_function( name="run_workspace_python_script", description=( "Execute a .py script saved in the session workspace (for pandas CSV policy edits). " "Prefer writing the script with write_workspace_text first, then call this with " "relative_path only (omit content) so tool args stay small." ), func=lambda relative_path, content=None: run_workspace_python_script( relative_path, content, session_hash=session_hash ), ), ]