agent_routes.py

"""
FastAPI routes for programmatic / agent callers.

HTTP paths align with Gradio ``api_name`` values in app.py. See GET /agent/operations
for the full map. Uses cli_redact.main(direct_mode_args=...) where a CLI task exists.
"""

from __future__ import annotations

import io
import os
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional

from fastapi import APIRouter, Depends, Header, HTTPException
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, field_validator

from tools.config import (
    AWS_LLM_PII_OPTION,
    AWS_PII_OPTION,
    INFERENCE_SERVER_PII_OPTION,
    INPUT_FOLDER,
    LOCAL_OCR_MODEL_OPTIONS,
    LOCAL_PII_OPTION,
    LOCAL_TRANSFORMERS_LLM_PII_OPTION,
    OUTPUT_FOLDER,
)
from tools.secure_path_utils import validate_path_safety

router = APIRouter(tags=["Agent"])

REPO_ROOT = Path(__file__).resolve().parent
_MAX_INSTRUCTION_LEN = 16_000

# NOTE: Paths from request bodies are untrusted. Avoid Path.resolve() on untrusted
# input (CodeQL py/path-injection); instead normalize via os.path and enforce
# containment under trusted roots.

# Mirrors app.py api_name values (Gradio).
GRADIO_API_NAMES: tuple[str, ...] = (
    "redact_document",
    "load_and_prepare_documents_or_data",
    "apply_review_redactions",
    "review_apply",
    "pdf_summarise",
    "tabular_redact",
    "word_level_ocr_text_search",
    "redact_data",
    "find_duplicate_pages",
    "find_duplicate_tabular",
    "summarise_document",
    "combine_review_csvs",
    "combine_review_pdfs",
    "export_review_redaction_overlay",
    "export_review_page_ocr_visualisation",
    "verify_redaction_coverage",
)


def _allowed_path_roots() -> list[Path]:
    # Return roots without resolving. These are trusted config values, but avoiding
    # Path.resolve() keeps CodeQL happy and matches our "no resolve on untrusted"
    # approach elsewhere.
    roots = [REPO_ROOT]
    for folder in (INPUT_FOLDER, OUTPUT_FOLDER):
        if folder:
            roots.append(Path(str(folder)))
    return roots


def _sanitize_untrusted_path_input(path_str: str) -> str:
    """Basic raw-input validation before any path normalization."""
    if not isinstance(path_str, str):
        raise HTTPException(status_code=400, detail="Path must be a string.")
    cleaned = path_str.strip()
    if not cleaned:
        raise HTTPException(status_code=400, detail="Path must not be empty.")
    if "\x00" in cleaned:
        raise HTTPException(status_code=400, detail="Path contains invalid null byte.")
    return cleaned


def _normalize_untrusted_path_to_abs(path_str: str) -> str:
    """
    Expand ~, then normalize to an absolute path.

    Relative paths are interpreted relative to REPO_ROOT (matching prior behaviour).
    """
    safe_input = _sanitize_untrusted_path_input(path_str)
    expanded = os.path.expanduser(safe_input)
    if os.path.isabs(expanded):
        return os.path.normpath(os.path.abspath(expanded))
    return os.path.normpath(os.path.abspath(os.path.join(str(REPO_ROOT), expanded)))


def _must_be_under_allowed_roots(candidate_abs: str, original: str) -> None:
    """Enforce candidate is contained under repo, INPUT_FOLDER, or OUTPUT_FOLDER."""
    candidate_real = os.path.realpath(str(candidate_abs))
    allowed_roots = [
        os.path.realpath(os.path.abspath(str(p))) for p in _allowed_path_roots()
    ]
    for root in allowed_roots:
        try:
            common = os.path.commonpath([candidate_real, root])
        except ValueError:
            # Different drive on Windows or invalid path mix
            continue
        if common == root:
            return
    raise HTTPException(
        status_code=403,
        detail="Path must be under the app repo, INPUT_FOLDER, or OUTPUT_FOLDER",
    )


def _path_must_be_allowed_file(path_str: str) -> str:
    """Resolve path, ensure it is under an allowed root and exists as a file."""
    candidate_abs = _normalize_untrusted_path_to_abs(path_str)
    candidate_real = os.path.realpath(candidate_abs)

    # Validate both "safe path" patterns and containment under trusted roots.
    _must_be_under_allowed_roots(candidate_real, path_str)
    ok = any(
        validate_path_safety(candidate_real, base_path=str(root))
        for root in _allowed_path_roots()
    )
    if not ok:
        raise HTTPException(status_code=400, detail=f"Unsafe path rejected: {path_str}")
    try:
        candidate_path = Path(candidate_real)
        if not candidate_path.is_file():
            raise HTTPException(
                status_code=400, detail=f"Not a file or missing: {candidate_real}"
            )
    except OSError:
        raise HTTPException(
            status_code=400, detail=f"Not a file or missing: {candidate_real}"
        )
    return candidate_real


def _path_must_be_allowed_directory(path_str: str, *, must_exist: bool = True) -> str:
    """
    Normalize and validate a directory path under allowed roots.

    By default the directory must already exist; callers can opt out (e.g. output_dir
    that will be created later by the CLI).
    """
    candidate_abs = _normalize_untrusted_path_to_abs(path_str)
    candidate_real = os.path.realpath(candidate_abs)

    _must_be_under_allowed_roots(candidate_real, path_str)
    ok = any(
        validate_path_safety(candidate_real, base_path=str(root))
        for root in _allowed_path_roots()
    )
    if not ok:
        raise HTTPException(status_code=400, detail=f"Unsafe path rejected: {path_str}")
    if must_exist:
        try:
            if not Path(candidate_real).is_dir():
                raise HTTPException(
                    status_code=400, detail=f"Not a directory: {candidate_real}"
                )
        except OSError:
            raise HTTPException(
                status_code=400, detail=f"Not a directory: {candidate_real}"
            )
    return candidate_real


def _optional_agent_api_key(x_agent_api_key: Optional[str] = Header(None)) -> None:
    expected = os.environ.get("AGENT_API_KEY", "").strip()
    if not expected:
        return
    if not x_agent_api_key or x_agent_api_key.strip() != expected:
        raise HTTPException(
            status_code=401,
            detail="Set header X-Agent-API-Key to match AGENT_API_KEY environment variable",
        )


class AgentRedactDocumentRequest(BaseModel):
    """Parity with Gradio api_name ``redact_document``."""

    input_files: list[str] = Field(
        ...,
        min_length=1,
        description="Paths to input files (PDF, images, or tabular/Word for anonymisation)",
    )
    instruction: Optional[str] = Field(
        None,
        description="Optional instructions for LLM-based PII detection (custom_llm_instructions)",
    )
    output_dir: Optional[str] = None
    input_dir: Optional[str] = None
    ocr_method: Optional[str] = Field(
        None,
        description=(
            "High-level OCR/text mode. Accepted values: 'Local OCR', "
            "'AWS Textract', 'Local text'. To choose a specific local OCR engine "
            "(e.g. paddle/tesseract/vlm), set "
            "overrides.chosen_local_ocr_model."
        ),
    )
    pii_detector: Optional[str] = Field(
        None,
        description=(
            "PII detection method. Recommended configured labels: "
            f"'{LOCAL_PII_OPTION}', '{AWS_PII_OPTION}', '{AWS_LLM_PII_OPTION}', "
            f"'{INFERENCE_SERVER_PII_OPTION}', '{LOCAL_TRANSFORMERS_LLM_PII_OPTION}', "
            "'None'."
        ),
    )
    overrides: Optional[dict[str, Any]] = Field(
        None,
        description=(
            "Optional CLI flag overrides; keys must match argparse destination names. "
            "For local OCR model selection, set 'chosen_local_ocr_model' "
            f"(allowed models depend on deployment; configured options: {LOCAL_OCR_MODEL_OPTIONS})."
        ),
    )

    model_config = {
        "json_schema_extra": {
            "examples": [
                {
                    "input_files": [
                        "example_data/example_of_emails_sent_to_a_professor_before_applying.pdf"
                    ],
                    "instruction": "Do not redact the university name.",
                    "ocr_method": "Local OCR",
                    "pii_detector": LOCAL_PII_OPTION,
                    "overrides": {"chosen_local_ocr_model": "paddle"},
                }
            ]
        }
    }

    @field_validator("instruction")
    @classmethod
    def _cap_instruction(cls, v: Optional[str]) -> Optional[str]:
        if v is None:
            return v
        if len(v) > _MAX_INSTRUCTION_LEN:
            raise ValueError(f"instruction exceeds {_MAX_INSTRUCTION_LEN} characters")
        return v


class AgentRedactDataRequest(AgentRedactDocumentRequest):
    """Parity with Gradio api_name ``redact_data``; same CLI task as redact_document."""


class AgentTaskResponse(BaseModel):
    status: str
    gradio_api_name: str
    task: str
    output_dir: str
    input_dir: str
    message: str
    log_excerpt: Optional[str] = None
    output_paths: Optional[list[str]] = None


class AgentVerifyRedactionRequest(BaseModel):
    review_csv_path: str = Field(..., description="Path to *_review_file.csv")
    ocr_words_csv_path: str = Field(
        ..., description="Path to *_ocr_results_with_words_*.csv from the same run"
    )
    must_redact: Optional[List[str]] = Field(
        None,
        description="Regex patterns for terms that must be covered by review boxes.",
    )
    must_not_redact: Optional[List[str]] = Field(
        None,
        description="Regex patterns for terms that must not appear in review rows.",
    )
    redacted_pdf_path: Optional[str] = Field(
        None, description="Optional applied *_redacted.pdf for text-layer leak checks."
    )
    total_pages: Optional[int] = Field(None, ge=1)
    min_word_length: int = Field(3, ge=1, le=32)
    sample_pixels: bool = Field(
        False,
        description="Sample pixel darkness at box centres on redacted PDF (requires redacted_pdf_path).",
    )
    auto_prune_suspicious: bool = Field(
        False,
        description="Remove prunable suspicious short/OCR-fragment rows and write pruned CSV.",
    )
    pruned_output_path: Optional[str] = Field(
        None,
        description="Output path for pruned CSV when auto_prune_suspicious is true.",
    )


class AgentVerifyRedactionResponse(BaseModel):
    status: str
    gradio_api_name: str = "verify_redaction_coverage"
    coverage_pass: bool
    coverage_pass_strict: bool
    coverage_pass_with_cleanup: bool
    pruned_csv_path: Optional[str] = None
    prune_log: Optional[Dict[str, Any]] = None
    report: Dict[str, Any]


class AgentWordLevelOcrSearchRequest(BaseModel):
    ocr_words_csv_path: str = Field(
        ..., description="Path to *_ocr_results_with_words_*.csv"
    )
    search_text: str = Field(..., min_length=3, max_length=500)
    similarity_threshold: float = Field(1.0, ge=0.0, le=1.0)
    use_regex: bool = False
    review_csv_path: Optional[str] = Field(
        None,
        description="Optional *_review_file.csv to flag whether each hit is covered by a box.",
    )


class AgentWordLevelOcrSearchResponse(BaseModel):
    status: str
    gradio_api_name: str = "word_level_ocr_text_search"
    result: Dict[str, Any]


def _merge_redact_direct_mode(body: AgentRedactDocumentRequest) -> dict[str, Any]:
    from cli_redact import get_cli_default_args_dict

    merged: dict[str, Any] = get_cli_default_args_dict()
    merged["task"] = "redact"
    merged["input_file"] = [_path_must_be_allowed_file(p) for p in body.input_files]

    if body.instruction is not None:
        merged["custom_llm_instructions"] = body.instruction
    if body.output_dir is not None:
        # Output folders may not exist yet (CLI will create). Still constrain to allowed roots.
        merged["output_dir"] = _path_must_be_allowed_directory(
            body.output_dir, must_exist=False
        )
    if body.input_dir is not None:
        # Input dir should exist if provided.
        merged["input_dir"] = _path_must_be_allowed_directory(
            body.input_dir, must_exist=True
        )
    if body.ocr_method is not None:
        merged["ocr_method"] = body.ocr_method
    if body.pii_detector is not None:
        merged["pii_detector"] = body.pii_detector

    if body.overrides:
        allowed = set(merged.keys())
        for key, value in body.overrides.items():
            if key not in allowed:
                raise HTTPException(
                    status_code=400,
                    detail=f"Unknown override key '{key}'. Must be a known CLI argument name.",
                )
            merged[key] = value

    return merged


def _run_cli_main(direct: dict[str, Any], gradio_api_name: str) -> AgentTaskResponse:
    from cli_redact import main as cli_main

    buf = io.StringIO()
    old_stdout = sys.stdout
    try:
        sys.stdout = buf
        cli_main(direct_mode_args=direct)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e)) from e
    finally:
        sys.stdout = old_stdout

    log_excerpt = buf.getvalue()
    if len(log_excerpt) > 8000:
        log_excerpt = log_excerpt[-8000:]

    return AgentTaskResponse(
        status="completed",
        gradio_api_name=gradio_api_name,
        task=str(direct.get("task", "")),
        output_dir=str(direct.get("output_dir", "")),
        input_dir=str(direct.get("input_dir", "")),
        message="cli_redact.main finished; see log_excerpt for console output",
        log_excerpt=log_excerpt or None,
    )


@router.post(
    "/redact_document",
    response_model=AgentTaskResponse,
    summary="redact_document (Gradio api_name)",
    description=(
        "Matches Gradio ``api_name='redact_document'``. "
        "``python cli_redact.py --task redact --input_file ...``. "
        "Optional ``instruction`` maps to ``custom_llm_instructions``. "
        "OCR modes: 'Local OCR' | 'AWS Textract' | 'Local text'. "
        "Specific local OCR engines are set via ``overrides.chosen_local_ocr_model`` "
        f"(for example: {LOCAL_OCR_MODEL_OPTIONS}). "
        "PII methods should use configured labels shown on the request schema."
    ),
)
def post_redact_document(
    body: AgentRedactDocumentRequest,
    _: None = Depends(_optional_agent_api_key),
) -> AgentTaskResponse:
    direct = _merge_redact_direct_mode(body)
    return _run_cli_main(direct, "redact_document")


@router.post(
    "/redact_data",
    response_model=AgentTaskResponse,
    summary="redact_data (Gradio api_name)",
    description=(
        "Matches Gradio ``api_name='redact_data'``. Same CLI ``redact`` task as "
        "/redact_document; use CSV/XLSX/DOCX paths for tabular/Word flows. "
        "OCR modes: 'Local OCR' | 'AWS Textract' | 'Local text'. "
        "Specific local OCR engines are set via ``overrides.chosen_local_ocr_model`` "
        f"(for example: {LOCAL_OCR_MODEL_OPTIONS}). "
        "PII methods should use configured labels shown on the request schema."
    ),
)
def post_redact_data(
    body: AgentRedactDataRequest,
    _: None = Depends(_optional_agent_api_key),
) -> AgentTaskResponse:
    direct = _merge_redact_direct_mode(body)
    return _run_cli_main(direct, "redact_data")


@router.post(
    "/tasks/redact",
    response_model=AgentTaskResponse,
    summary="Legacy: same as /redact_document",
    description="Deprecated alias; prefer POST /agent/redact_document.",
    deprecated=True,
    include_in_schema=True,
)
def post_tasks_redact_legacy(
    body: AgentRedactDocumentRequest,
    _: None = Depends(_optional_agent_api_key),
) -> AgentTaskResponse:
    direct = _merge_redact_direct_mode(body)
    return _run_cli_main(direct, "redact_document")


class AgentFindDuplicatePagesRequest(BaseModel):
    input_files: list[str] = Field(..., min_length=1)
    similarity_threshold: Optional[float] = None
    min_word_count: Optional[int] = None
    min_consecutive_pages: Optional[int] = None
    greedy_match: Optional[bool] = None
    combine_pages: Optional[bool] = None
    overrides: Optional[dict[str, Any]] = None


@router.post(
    "/find_duplicate_pages",
    response_model=AgentTaskResponse,
    summary="find_duplicate_pages (Gradio api_name)",
    description="``cli_redact --task deduplicate --duplicate_type pages``.",
)
def post_find_duplicate_pages(
    body: AgentFindDuplicatePagesRequest,
    _: None = Depends(_optional_agent_api_key),
) -> AgentTaskResponse:
    from cli_redact import get_cli_default_args_dict

    merged = get_cli_default_args_dict()
    merged["task"] = "deduplicate"
    merged["duplicate_type"] = "pages"
    merged["input_file"] = [_path_must_be_allowed_file(p) for p in body.input_files]
    if body.similarity_threshold is not None:
        merged["similarity_threshold"] = body.similarity_threshold
    if body.min_word_count is not None:
        merged["min_word_count"] = body.min_word_count
    if body.min_consecutive_pages is not None:
        merged["min_consecutive_pages"] = body.min_consecutive_pages
    if body.greedy_match is not None:
        merged["greedy_match"] = "True" if body.greedy_match else "False"
    if body.combine_pages is not None:
        merged["combine_pages"] = "True" if body.combine_pages else "False"
    if body.overrides:
        allowed = set(merged.keys())
        for k, v in body.overrides.items():
            if k not in allowed:
                raise HTTPException(400, f"Unknown override key: {k}")
            merged[k] = v
    return _run_cli_main(merged, "find_duplicate_pages")


class AgentFindDuplicateTabularRequest(BaseModel):
    input_files: list[str] = Field(..., min_length=1)
    text_columns: Optional[list[str]] = None
    similarity_threshold: Optional[float] = None
    min_word_count: Optional[int] = None
    overrides: Optional[dict[str, Any]] = None


@router.post(
    "/find_duplicate_tabular",
    response_model=AgentTaskResponse,
    summary="find_duplicate_tabular (Gradio api_name)",
)
def post_find_duplicate_tabular(
    body: AgentFindDuplicateTabularRequest,
    _: None = Depends(_optional_agent_api_key),
) -> AgentTaskResponse:
    from cli_redact import get_cli_default_args_dict

    merged = get_cli_default_args_dict()
    merged["task"] = "deduplicate"
    merged["duplicate_type"] = "tabular"
    merged["input_file"] = [_path_must_be_allowed_file(p) for p in body.input_files]
    if body.text_columns is not None:
        merged["text_columns"] = body.text_columns
    if body.similarity_threshold is not None:
        merged["similarity_threshold"] = body.similarity_threshold
    if body.min_word_count is not None:
        merged["min_word_count"] = body.min_word_count
    if body.overrides:
        allowed = set(merged.keys())
        for k, v in body.overrides.items():
            if k not in allowed:
                raise HTTPException(400, f"Unknown override key: {k}")
            merged[k] = v
    return _run_cli_main(merged, "find_duplicate_tabular")


class AgentSummariseDocumentRequest(BaseModel):
    input_files: list[str] = Field(..., min_length=1)
    summarisation_inference_method: Optional[str] = None
    summarisation_format: Optional[str] = None
    summarisation_context: Optional[str] = None
    summarisation_additional_instructions: Optional[str] = None
    overrides: Optional[dict[str, Any]] = None


@router.post(
    "/summarise_document",
    response_model=AgentTaskResponse,
    summary="summarise_document (Gradio api_name)",
)
def post_summarise_document(
    body: AgentSummariseDocumentRequest,
    _: None = Depends(_optional_agent_api_key),
) -> AgentTaskResponse:
    from cli_redact import get_cli_default_args_dict

    merged = get_cli_default_args_dict()
    merged["task"] = "summarise"
    merged["input_file"] = [_path_must_be_allowed_file(p) for p in body.input_files]
    if body.summarisation_inference_method is not None:
        merged["summarisation_inference_method"] = body.summarisation_inference_method
    if body.summarisation_format is not None:
        merged["summarisation_format"] = body.summarisation_format
    if body.summarisation_context is not None:
        merged["summarisation_context"] = body.summarisation_context
    if body.summarisation_additional_instructions is not None:
        merged["summarisation_additional_instructions"] = (
            body.summarisation_additional_instructions
        )
    if body.overrides:
        allowed = set(merged.keys())
        for k, v in body.overrides.items():
            if k not in allowed:
                raise HTTPException(400, f"Unknown override key: {k}")
            merged[k] = v
    return _run_cli_main(merged, "summarise_document")


class AgentCombineReviewPdfsRequest(BaseModel):
    input_files: list[str] = Field(..., min_length=2)
    output_dir: Optional[str] = None


@router.post(
    "/combine_review_pdfs",
    response_model=AgentTaskResponse,
    summary="combine_review_pdfs (Gradio api_name)",
)
def post_combine_review_pdfs(
    body: AgentCombineReviewPdfsRequest,
    _: None = Depends(_optional_agent_api_key),
) -> AgentTaskResponse:
    from cli_redact import get_cli_default_args_dict

    merged = get_cli_default_args_dict()
    merged["task"] = "combine_review_pdfs"
    merged["input_file"] = [_path_must_be_allowed_file(p) for p in body.input_files]
    if body.output_dir is not None:
        merged["output_dir"] = _path_must_be_allowed_directory(body.output_dir)
    return _run_cli_main(merged, "combine_review_pdfs")


class _NamedPath:
    """merge_csv_files expects objects with a .name attribute (Gradio file-like)."""

    __slots__ = ("name",)

    def __init__(self, path: str) -> None:
        self.name = path


class AgentCombineReviewCsvsRequest(BaseModel):
    input_files: list[str] = Field(..., min_length=1)
    output_dir: Optional[str] = Field(
        None, description="Defaults to config OUTPUT_FOLDER"
    )


class AgentApplyReviewRedactionsRequest(BaseModel):
    """Headless parity with Gradio ``api_name='apply_review_redactions'`` (prepare + apply)."""

    pdf_path: str = Field(
        ...,
        description="Path to the source PDF under allowed roots.",
    )
    review_csv_path: str = Field(
        ...,
        description=(
            "Path to the review plan CSV; basename must contain '_review_file' "
            "(e.g. mydoc_review_file.csv)."
        ),
    )
    output_dir: Optional[str] = Field(
        None,
        description="Output directory (created if missing); defaults to OUTPUT_FOLDER.",
    )
    input_dir: Optional[str] = Field(
        None,
        description="Input/working directory for page images; defaults to INPUT_FOLDER.",
    )
    text_extract_method: Optional[str] = Field(
        None,
        description="OCR/text mode passed to prepare (defaults to CLI ocr_method).",
    )
    efficient_ocr: Optional[bool] = Field(
        None,
        description="If set, overrides EFFICIENT_OCR for the prepare step.",
    )


@router.post(
    "/combine_review_csvs",
    response_model=AgentTaskResponse,
    summary="combine_review_csvs (Gradio api_name)",
    description="Uses tools.helper_functions.merge_csv_files (not cli_redact).",
)
def post_combine_review_csvs(
    body: AgentCombineReviewCsvsRequest,
    _: None = Depends(_optional_agent_api_key),
) -> AgentTaskResponse:
    from tools.helper_functions import merge_csv_files

    paths = [_NamedPath(_path_must_be_allowed_file(p)) for p in body.input_files]
    out_dir = body.output_dir or OUTPUT_FOLDER
    out_dir_resolved = _path_must_be_allowed_directory(str(out_dir), must_exist=True)
    sep = "/" if not out_dir_resolved.endswith(("/", "\\")) else ""
    out_files = merge_csv_files(paths, output_folder=out_dir_resolved + sep)
    return AgentTaskResponse(
        status="completed",
        gradio_api_name="combine_review_csvs",
        task="combine_review_csvs",
        output_dir=out_dir_resolved,
        input_dir="",
        message="merge_csv_files completed",
        output_paths=out_files,
    )


class AgentExportReviewRedactionOverlayRequest(BaseModel):
    """Agent JSON body for the same overlay render as Gradio ``api_name='page_redaction_review_image'``."""

    page_image_path: str = Field(
        ...,
        description="Path to page raster (PNG/JPEG) used as underlay; must be under allowed roots.",
    )
    boxes: List[Dict[str, Any]] = Field(
        ...,
        min_length=1,
        description="Annotator-style boxes: label, color, xmin, ymin, xmax, ymax (normalized 0–1).",
    )
    page_number: int = Field(
        1, ge=1, description="1-based page index for the output filename."
    )
    doc_base_name: str = Field(
        "review",
        description="Basename for output file (e.g. document name without extension).",
    )
    review_df_records: Optional[List[Dict[str, Any]]] = Field(
        None,
        description="Optional rows (include at least 'label') for stable label→line-pattern mapping.",
    )
    label_abbrev_chars: Optional[int] = Field(
        None,
        ge=0,
        le=24,
        description="Draw this many leading characters of each label on the image; omit to use REVIEW_OVERLAY_LABEL_ABBREV_CHARS from config (0 = off).",
    )


class AgentExportReviewPageOcrVisualisationRequest(BaseModel):
    """Agent JSON body for the same OCR visualisation as Gradio ``api_name='page_ocr_review_image'``."""

    page_image_path: str = Field(
        ...,
        description="Path to page raster (PNG/JPEG) used as underlay; must be under allowed roots.",
    )
    ocr_results: Dict[str, Any] = Field(
        ...,
        description="Word-level OCR results dict (line_key -> {words:[{text, bounding_box, conf, ...}]}).",
    )
    page_number: int = Field(
        1, ge=1, description="1-based page index (used for naming)."
    )
    doc_base_name: str = Field(
        "review",
        description="Basename for output file (e.g. document name without extension).",
    )


@router.post(
    "/export_review_redaction_overlay",
    response_model=AgentTaskResponse,
    summary="export_review_redaction_overlay (Agent API; Gradio api_name: page_redaction_review_image)",
    description=(
        "Renders hollow redaction outlines and a top-right legend on the page image; "
        "writes ``redaction_overlay/{doc_base_name}_page{n}_redaction_overlay.jpg`` under OUTPUT_FOLDER "
        "(scaled per REVIEW_OVERLAY_MAX_PIXELS, JPEG capped by REVIEW_OVERLAY_MAX_FILE_BYTES). "
        "Uses ``tools.redaction_review.visualise_review_redaction_boxes``."
    ),
)
def post_export_review_redaction_overlay(
    body: AgentExportReviewRedactionOverlayRequest,
    _: None = Depends(_optional_agent_api_key),
) -> AgentTaskResponse:
    import pandas as pd

    from tools.redaction_review import visualise_review_redaction_boxes

    img_path = _path_must_be_allowed_file(body.page_image_path)
    annotator: dict[str, Any] = {"image": img_path, "boxes": body.boxes}
    review_df = (
        pd.DataFrame(body.review_df_records)
        if body.review_df_records
        else pd.DataFrame()
    )
    out_folder_abs = os.path.realpath(
        os.path.abspath(os.path.expanduser(str(OUTPUT_FOLDER)))
    )
    if not validate_path_safety(out_folder_abs):
        raise HTTPException(status_code=400, detail="Unsafe OUTPUT_FOLDER path")
    _must_be_under_allowed_roots(out_folder_abs, str(out_folder_abs))
    try:
        Path(out_folder_abs).mkdir(parents=True, exist_ok=True)
    except OSError:
        raise HTTPException(status_code=500, detail="Could not create OUTPUT_FOLDER")
    out_folder = out_folder_abs

    path = visualise_review_redaction_boxes(
        annotator,
        review_df=review_df,
        output_folder=out_folder,
        page_number=body.page_number,
        doc_base_name=body.doc_base_name,
        label_abbrev_chars=body.label_abbrev_chars,
    )
    if not path:
        raise HTTPException(
            status_code=500,
            detail=(
                "Could not produce overlay PNG (invalid image/boxes or write failed). "
                "Ensure boxes are valid and the image loads."
            ),
        )
    return AgentTaskResponse(
        status="completed",
        gradio_api_name="export_review_redaction_overlay",
        task="export_review_redaction_overlay",
        output_dir=out_folder,
        input_dir="",
        message="Redaction overlay PNG written",
        output_paths=[path],
    )


@router.post(
    "/export_review_page_ocr_visualisation",
    response_model=AgentTaskResponse,
    summary="export_review_page_ocr_visualisation (Agent API; Gradio api_name: page_ocr_review_image)",
    description=(
        "Renders a per-page OCR visualisation using tools.file_redaction.visualise_ocr_words_bounding_boxes; "
        "writes under OUTPUT_FOLDER/review_ocr_visualisations/."
    ),
)
def post_export_review_page_ocr_visualisation(
    body: AgentExportReviewPageOcrVisualisationRequest,
    _: None = Depends(_optional_agent_api_key),
) -> AgentTaskResponse:
    from PIL import Image

    from tools.file_redaction import visualise_ocr_words_bounding_boxes

    img_path = _path_must_be_allowed_file(body.page_image_path)

    out_folder_abs = os.path.realpath(
        os.path.abspath(os.path.expanduser(str(OUTPUT_FOLDER)))
    )
    if not validate_path_safety(out_folder_abs):
        raise HTTPException(status_code=400, detail="Unsafe OUTPUT_FOLDER path")
    _must_be_under_allowed_roots(out_folder_abs, str(out_folder_abs))
    try:
        Path(out_folder_abs).mkdir(parents=True, exist_ok=True)
    except OSError:
        raise HTTPException(status_code=500, detail="Could not create OUTPUT_FOLDER")
    out_folder = out_folder_abs

    safe_base = str(body.doc_base_name or "review")
    image_name = f"{safe_base}_page{int(body.page_number)}.png"
    log_paths: list[str] = []
    try:
        log_paths = visualise_ocr_words_bounding_boxes(
            Image.open(img_path).convert("RGB"),
            body.ocr_results,
            image_name=image_name,
            output_folder=out_folder,
            visualisation_folder="review_ocr_visualisations",
            add_legend=True,
            log_files_output_paths=log_paths,
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e)) from e

    if not log_paths:
        raise HTTPException(
            status_code=500,
            detail="Could not produce OCR visualisation (invalid image/ocr_results or write failed).",
        )
    out_path = log_paths[-1]
    return AgentTaskResponse(
        status="completed",
        gradio_api_name="export_review_page_ocr_visualisation",
        task="export_review_page_ocr_visualisation",
        output_dir=out_folder,
        input_dir="",
        message="OCR visualisation written",
        output_paths=[out_path],
    )


def _gradio_only(api_name: str, detail: str) -> JSONResponse:
    return JSONResponse(
        status_code=501,
        content={
            "gradio_api_name": api_name,
            "detail": detail,
            "hint": (
                "This flow is Gradio-session stateful. Call the named route on the "
                "Gradio HTTP API, not /agent."
            ),
            "gradio_http": {
                "discover_schema": "GET /gradio_api/info",
                "start_call": f"POST /gradio_api/call/{api_name}",
                "request_body_shape": '{"data": [<args in schema order>]}',
                "poll": f"GET /gradio_api/call/{api_name}/{{event_id}}",
            },
            "gradio_client_notes": [
                "Pass api_name explicitly; do not rely on inferring the endpoint from "
                "Python function names (large Blocks apps will look ambiguous).",
                "If predict() still cannot resolve the route, open GET /gradio_api/info "
                "and use the numeric fn_index with gradio_client, or call the HTTP "
                "endpoints directly.",
                "The length of data must match the parameter list for this deployment; "
                "copy order and types from /gradio_api/info.",
            ],
        },
    )


@router.post("/load_and_prepare_documents_or_data")
def post_load_and_prepare_documents_or_data() -> JSONResponse:
    return _gradio_only(
        "load_and_prepare_documents_or_data",
        "Preparation uses Gradio session state and prepare_image_or_pdf_with_efficient_ocr; no single CLI task.",
    )


@router.post(
    "/apply_review_redactions",
    response_model=AgentTaskResponse,
    summary="apply_review_redactions (Gradio api_name)",
    description=(
        "Runs prepare_image_or_pdf_with_efficient_ocr([pdf, review_csv]) then "
        "apply_redactions_to_review_df_and_files — same core pipeline as the Review tab, "
        "without Gradio session state. Requires paths under allowed roots."
    ),
)
def post_apply_review_redactions(
    body: AgentApplyReviewRedactionsRequest,
    _: None = Depends(_optional_agent_api_key),
) -> AgentTaskResponse:
    from tools.simplified_api import run_apply_review_redactions

    pdf = _path_must_be_allowed_file(body.pdf_path)
    csv = _path_must_be_allowed_file(body.review_csv_path)
    out_dir: str | None = None
    if body.output_dir is not None:
        out_dir = _path_must_be_allowed_directory(body.output_dir, must_exist=False)
    in_dir: str | None = None
    if body.input_dir is not None:
        in_dir = _path_must_be_allowed_directory(body.input_dir, must_exist=False)

    try:
        result = run_apply_review_redactions(
            pdf_path=pdf,
            review_csv_path=csv,
            output_dir=out_dir,
            input_dir=in_dir,
            text_extract_method=body.text_extract_method,
            efficient_ocr=body.efficient_ocr,
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"apply_review_redactions failed: {e}",
        ) from e

    return AgentTaskResponse(
        status="completed",
        gradio_api_name="apply_review_redactions",
        task="apply_review_redactions",
        output_dir=result["output_dir"],
        input_dir=result["input_dir"],
        message=result["message"],
        output_paths=result.get("output_paths"),
    )


@router.post(
    "/verify_redaction_coverage",
    response_model=AgentVerifyRedactionResponse,
    summary="verify_redaction_coverage (Pass 1 programmatic QA)",
)
def post_verify_redaction_coverage(
    body: AgentVerifyRedactionRequest,
    _: None = Depends(_optional_agent_api_key),
) -> AgentVerifyRedactionResponse:
    from tools.simplified_api import run_verify_redaction_coverage

    review = _path_must_be_allowed_file(body.review_csv_path)
    ocr_words = _path_must_be_allowed_file(body.ocr_words_csv_path)
    redacted = None
    if body.redacted_pdf_path:
        redacted = _path_must_be_allowed_file(body.redacted_pdf_path)
    try:
        report, pruned_csv_path, prune_log = run_verify_redaction_coverage(
            review_csv_path=review,
            ocr_words_csv_path=ocr_words,
            must_redact=body.must_redact,
            must_not_redact=body.must_not_redact,
            redacted_pdf_path=redacted,
            total_pages=body.total_pages,
            min_word_length=body.min_word_length,
            sample_pixels=body.sample_pixels,
            auto_prune_suspicious=body.auto_prune_suspicious,
            pruned_output_path=body.pruned_output_path,
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e
    except Exception as e:
        raise HTTPException(
            status_code=500, detail=f"verify_redaction_coverage failed: {e}"
        ) from e
    return AgentVerifyRedactionResponse(
        status="completed",
        coverage_pass=bool(report.get("pass_strict", report.get("pass"))),
        coverage_pass_strict=bool(report.get("pass_strict", report.get("pass"))),
        coverage_pass_with_cleanup=bool(report.get("pass_with_cleanup")),
        pruned_csv_path=pruned_csv_path,
        prune_log=prune_log,
        report=report,
    )


@router.post(
    "/word_level_ocr_text_search",
    response_model=AgentWordLevelOcrSearchResponse,
    summary="word_level_ocr_text_search (headless OCR CSV search)",
)
def post_word_level_ocr_text_search(
    body: AgentWordLevelOcrSearchRequest,
    _: None = Depends(_optional_agent_api_key),
) -> AgentWordLevelOcrSearchResponse:
    from tools.simplified_api import run_word_level_ocr_text_search_api

    ocr_words = _path_must_be_allowed_file(body.ocr_words_csv_path)
    review = None
    if body.review_csv_path:
        review = _path_must_be_allowed_file(body.review_csv_path)
    try:
        result = run_word_level_ocr_text_search_api(
            ocr_words_csv_path=ocr_words,
            search_text=body.search_text,
            similarity_threshold=body.similarity_threshold,
            use_regex=body.use_regex,
            review_csv_path=review,
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e)) from e
    except Exception as e:
        raise HTTPException(
            status_code=500, detail=f"word_level_ocr_text_search failed: {e}"
        ) from e
    return AgentWordLevelOcrSearchResponse(status="completed", result=result)


@router.get("/operations")
def list_operations() -> dict[str, Any]:
    return {
        "gradio_api_names": list(GRADIO_API_NAMES),
        "gradio_session_state_endpoints": {
            "description": (
                "These api_name values are exposed on the Gradio HTTP API but return "
                "501 on /agent because they depend on in-memory Gradio state."
            ),
            "discover_schema": "GET /gradio_api/info",
            "call_pattern": 'POST /gradio_api/call/<api_name> with JSON body {"data": [...]}',
            "names": [
                "load_and_prepare_documents_or_data",
            ],
        },
        "routes": [
            {
                "gradio_api_name": "redact_document",
                "method": "POST",
                "path": "/agent/redact_document",
                "implementation": "cli_redact task redact",
                "notes": {
                    "ocr_method": [
                        "Local OCR",
                        "AWS Textract",
                        "Local text",
                    ],
                    "chosen_local_ocr_model_override": LOCAL_OCR_MODEL_OPTIONS,
                    "pii_detector_recommended": [
                        LOCAL_PII_OPTION,
                        AWS_PII_OPTION,
                        AWS_LLM_PII_OPTION,
                        INFERENCE_SERVER_PII_OPTION,
                        LOCAL_TRANSFORMERS_LLM_PII_OPTION,
                        "None",
                    ],
                },
            },
            {
                "gradio_api_name": "redact_data",
                "method": "POST",
                "path": "/agent/redact_data",
                "implementation": "cli_redact task redact",
                "notes": {
                    "ocr_method": [
                        "Local OCR",
                        "AWS Textract",
                        "Local text",
                    ],
                    "chosen_local_ocr_model_override": LOCAL_OCR_MODEL_OPTIONS,
                    "pii_detector_recommended": [
                        LOCAL_PII_OPTION,
                        AWS_PII_OPTION,
                        AWS_LLM_PII_OPTION,
                        INFERENCE_SERVER_PII_OPTION,
                        LOCAL_TRANSFORMERS_LLM_PII_OPTION,
                        "None",
                    ],
                },
            },
            {
                "gradio_api_name": "find_duplicate_pages",
                "method": "POST",
                "path": "/agent/find_duplicate_pages",
                "implementation": "cli_redact deduplicate pages",
            },
            {
                "gradio_api_name": "find_duplicate_tabular",
                "method": "POST",
                "path": "/agent/find_duplicate_tabular",
                "implementation": "cli_redact deduplicate tabular",
            },
            {
                "gradio_api_name": "summarise_document",
                "method": "POST",
                "path": "/agent/summarise_document",
                "implementation": "cli_redact task summarise",
            },
            {
                "gradio_api_name": "combine_review_pdfs",
                "method": "POST",
                "path": "/agent/combine_review_pdfs",
                "implementation": "cli_redact combine_review_pdfs",
            },
            {
                "gradio_api_name": "export_review_redaction_overlay",
                "method": "POST",
                "path": "/agent/export_review_redaction_overlay",
                "implementation": "visualise_review_redaction_boxes",
            },
            {
                "gradio_api_name": "export_review_page_ocr_visualisation",
                "method": "POST",
                "path": "/agent/export_review_page_ocr_visualisation",
                "implementation": "visualise_ocr_words_bounding_boxes",
            },
            {
                "gradio_api_name": "combine_review_csvs",
                "method": "POST",
                "path": "/agent/combine_review_csvs",
                "implementation": "helper merge_csv_files",
            },
            {
                "gradio_api_name": "load_and_prepare_documents_or_data",
                "method": "POST",
                "path": "/agent/load_and_prepare_documents_or_data",
                "implementation": "not_implemented_http",
            },
            {
                "gradio_api_name": "apply_review_redactions",
                "method": "POST",
                "path": "/agent/apply_review_redactions",
                "implementation": "tools.simplified_api.run_apply_review_redactions",
            },
            {
                "gradio_api_name": "verify_redaction_coverage",
                "method": "POST",
                "path": "/agent/verify_redaction_coverage",
                "implementation": "tools.verify_redaction_coverage.verify_redaction_coverage",
                "notes": {
                    "purpose": "Pass 1 programmatic QA — pass_strict (policy), pass_with_cleanup (+ suspicious rows), optional prune and text/pixel checks.",
                    "must_redact": "list of regex strings",
                    "must_not_redact": "list of regex strings",
                    "auto_prune_suspicious": "remove short OCR-fragment rows before reporting",
                    "pages_flagged_for_vlm": "policy/visual failures only",
                    "pages_needing_csv_cleanup": "suspicious rows — prune, not VLM",
                    "leak_likely_causes": "per-page hints when text_layer_leaks (coord_not_normalized, missing_page_boxes, etc.) — not a broken /review_apply",
                },
            },
            {
                "gradio_api_name": "word_level_ocr_text_search",
                "method": "POST",
                "path": "/agent/word_level_ocr_text_search",
                "implementation": "tools.verify_redaction_coverage.run_word_level_ocr_text_search",
            },
        ],
    }


@router.get("/health")
def agent_health() -> dict[str, str]:
    return {"status": "ok", "service": "agent"}