yau-plant-assistant/api/documents.py

"""The document library screens: upload, review, approve, withdraw, restore.

Served under /documents, NOT /docs. FastAPI's Swagger UI already owns /docs and
the customer wants to keep it; two different things under one prefix with two
different access policies is the kind of thing that gets misread during a later
edit.

WHERE THIS RUNS AND WHY. These screens are served by ai-api on
api.yokogawa.tech, not by ai-web. ai-web is on ai.yokogawa.tech, which since
2026-08-28 admits only the SCADA console and passes through no Authelia at all,
so it has no identity to record. Publishers come in on api.yokogawa.tech, where
the forward-auth headers still arrive. Putting the screens where the identity
already is avoids reopening that routing question.

HOW THIS DIFFERS FROM THE DESIGN, all of it deliberate and all of it recorded
in BUILD-AI-CONTAINERS.md S14:

  - No ai-docs-worker. Conversion, chunking and embedding happen inside the
    request. The design puts publication behind a component with no HTTP
    surface; here the same process holds both database roles, so the boundary
    is enforced by db/005's trigger and by which connection each function opens,
    rather than by deployment. A large upload therefore blocks its own request
    rather than queueing - acceptable at this size, and the reason status
    values like `scanning` and `ingesting` are passed through rather than
    lingered in.
  - Conversion is text extraction, not layout parsing, and cannot read a
    scan. See convert.py for what that costs.
  - Files are NOT moved into /datadisk/ai-docs. ai-api has no write access to
    the document tree and is not getting any. The inbox is the store, and
    doc_chunks.source_file points into it. The consequence to know about:
    `ai-ingest --all` walks /datadisk/ai-docs and will not see anything
    published this way, so the two paths must not be used on the same document.
"""

from __future__ import annotations

import hashlib
import html
import logging
import os
import re
import uuid
from datetime import date
from pathlib import Path

import psycopg
from fastapi import APIRouter, Form, HTTPException, Request, UploadFile
from fastapi.responses import HTMLResponse, RedirectResponse

import chunking
import convert
from config import settings
from identity import Actor, DEMO_GROUPS, actor, is_publisher, require_publisher

log = logging.getLogger("api.documents")

router = APIRouter(prefix="/documents", tags=["documents"])

DOC_TYPES = ("procedure", "manual", "rationalisation", "design")

# doc_actions.reason has a CHECK of length >= 10. Mirrored here so the person
# gets a sentence explaining why, instead of a database error.
MIN_REASON = 10

# Anything that is not a plain filename. The stored name is generated from the
# upload_id anyway; this only keeps the ORIGINAL name printable on screen.
_UNSAFE = re.compile(r"[^A-Za-z0-9._ -]")


# --- plumbing ---------------------------------------------------------------


def _connect(role: str) -> psycopg.Connection:
    """Open a connection as one of the two document roles.

    `role` is spelled out at every call site. "uploads" may write the queue and
    may set superseded = TRUE; "ingest" is the only one that may write chunks
    or make a document citable again. Choosing the wrong one here is the whole
    security boundary, so it is never defaulted.
    """
    return psycopg.connect(settings().docs_dsn(role), connect_timeout=10)


def _who(request: Request, declared_name: str | None = None) -> Actor:
    return actor(request, declared_name)


def _embed(texts: list[str]) -> list[list[float]]:
    from openai import AzureOpenAI

    cfg = settings()
    if not (cfg.azure_openai_endpoint and cfg.azure_openai_api_key
            and cfg.embed_deployment):
        raise HTTPException(
            status_code=503,
            detail=(
                "no embedding model is configured, so nothing can be published. "
                "Set AZURE_OPENAI_* and EMBED_DEPLOYMENT."
            ),
        )
    client = AzureOpenAI(
        azure_endpoint=cfg.azure_openai_endpoint,
        api_key=cfg.azure_openai_api_key,
        api_version=cfg.azure_openai_api_version,
    )
    vectors: list[list[float]] = []
    for i in range(0, len(texts), 64):
        response = client.embeddings.create(
            model=cfg.embed_deployment, input=texts[i : i + 64]
        )
        vectors.extend(item.embedding for item in response.data)
    return vectors


def _upload_dir(upload_id: str) -> Path:
    return Path(settings().docs_inbox) / upload_id


# --- HTML -------------------------------------------------------------------
#
# Hand-written rather than templated. Six screens do not earn a template engine,
# a templates directory and another pinned dependency, and keeping the markup
# next to the handler means a reviewer can see what a form posts without
# opening a second file. Everything interpolated goes through e().


def e(value: object) -> str:
    return html.escape("" if value is None else str(value))


def _page(title: str, body: str, who: Actor | None = None) -> HTMLResponse:
    banner = ""
    if who is not None and who.is_demo:
        # The whole point of demo mode being visible. A person looking at this
        # screen must not believe the names on it were verified.
        banner = (
            '<div class="warn"><b>Demo identity.</b> The name below is typed in, '
            'not verified by sign-in. Every change is recorded as '
            f'<code>demo:{e(who.display)}</code> so it can never be mistaken for '
            'an authenticated action.</div>'
        )
    return HTMLResponse(f"""<!doctype html>
<html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>{e(title)} — WRPS Document Library</title>
<style>
 :root {{ color-scheme: light dark; }}
 body {{ font: 15px/1.5 system-ui, sans-serif; margin: 0 auto; padding: 1.5rem;
        max-width: 60rem; }}
 h1 {{ font-size: 1.3rem; }} h2 {{ font-size: 1.05rem; margin-top: 2rem; }}
 table {{ border-collapse: collapse; width: 100%; margin: .5rem 0 1.5rem; }}
 th, td {{ text-align: left; padding: .4rem .6rem; border-bottom: 1px solid #8884;
          vertical-align: top; }}
 th {{ font-weight: 600; font-size: .85rem; text-transform: uppercase;
      letter-spacing: .03em; opacity: .7; }}
 .warn {{ border-left: 4px solid #c60; background: #c6600018; padding: .6rem .9rem;
         margin: 0 0 1.2rem; }}
 .danger {{ border-left-color: #c00; background: #c0000018; }}
 .muted {{ opacity: .65; }}
 pre {{ background: #8881; padding: .8rem; overflow-x: auto; max-height: 26rem;
       white-space: pre-wrap; }}
 label {{ display: block; margin: .6rem 0 .15rem; font-weight: 600;
         font-size: .85rem; }}
 input, select, textarea {{ font: inherit; padding: .35rem; width: 100%;
                           max-width: 28rem; box-sizing: border-box; }}
 button {{ font: inherit; padding: .4rem .9rem; margin-top: .8rem; cursor: pointer; }}
 nav a {{ margin-right: 1rem; }}
 code {{ font-size: .9em; }}
</style></head><body>
<nav><a href="/documents">Library</a><a href="/docs">API reference</a></nav>
<h1>{e(title)}</h1>
{banner}
{body}
</body></html>""")


def _name_field(who_default: str = "") -> str:
    """The demo identity prompt. Only rendered in demo mode."""
    if settings().doc_identity_mode != "demo":
        return ""
    return (
        '<label for="declared_name">Your name (recorded against this action)</label>'
        f'<input id="declared_name" name="declared_name" required maxlength="64" '
        f'value="{e(who_default)}">'
    )


# --- reads ------------------------------------------------------------------


def _fetch(conn: psycopg.Connection, sql: str, args: tuple = ()) -> list[tuple]:
    with conn.cursor() as cur:
        cur.execute(sql, args)
        return cur.fetchall()


@router.get("", response_class=HTMLResponse)
@router.get("/", response_class=HTMLResponse)
def library(request: Request) -> HTMLResponse:
    """The one screen: what is live, what is waiting, what has been withdrawn.

    Readable by anyone who got through the edge. Only the actions are
    restricted, and the restriction is enforced in the handlers - hiding a
    button proves nothing, which is why the API is what checks.
    """
    cfg = settings()
    demo = cfg.doc_identity_mode == "demo"
    # In demo mode there is nobody to identify until they type a name, so the
    # listing must not 401. Build a placeholder purely for the banner.
    who = (
        Actor(stored="", display="", name=None, email=None,
              groups=DEMO_GROUPS, verified=False)
        if demo else _who(request)
    )

    with _connect("uploads") as conn:
        live = _fetch(conn, """
            SELECT doc_number, revision, doc_type, effective_date,
                   chunk_count, ingested_at
              FROM live_documents ORDER BY doc_number, revision""")
        queue = _fetch(conn, """
            SELECT upload_id, original_filename, status, uploaded_by,
                   uploaded_at, error
              FROM doc_uploads
             WHERE status NOT IN ('published','rejected')
             ORDER BY uploaded_at DESC""")
        gone = _fetch(conn, """
            SELECT doc_number, revision, doc_type, chunk_count,
                   withdrawn_at, withdrawn_by, reason
              FROM withdrawn_documents ORDER BY doc_number, revision""")

    rows = "".join(
        f"<tr><td><code>{e(d)}</code></td><td>{e(r)}</td><td>{e(t)}</td>"
        f"<td>{e(ed)}</td><td>{e(n)}</td>"
        f"<td><form method=post action='/documents/withdraw'>"
        f"<input type=hidden name=doc_number value='{e(d)}'>"
        f"<input type=hidden name=revision value='{e(r)}'>"
        f"<button>Withdraw…</button></form></td></tr>"
        for d, r, t, ed, n, _ in live
    ) or "<tr><td colspan=6 class=muted>Nothing is published.</td></tr>"

    qrows = "".join(
        f"<tr><td>{e(f)}</td><td>{e(s)}</td><td>{e(u)}</td><td>{e(at)}</td>"
        f"<td>{e(err) or ''}</td>"
        f"<td><a href='/documents/review/{e(uid)}'>Review</a></td></tr>"
        for uid, f, s, u, at, err in queue
    ) or "<tr><td colspan=6 class=muted>Nothing waiting.</td></tr>"

    grows = "".join(
        f"<tr><td><code>{e(d)}</code></td><td>{e(r)}</td><td>{e(t)}</td>"
        f"<td>{e(n)}</td><td>{e(at)}</td><td>{e(by)}</td><td>{e(why)}</td>"
        f"<td><form method=post action='/documents/restore'>"
        f"<input type=hidden name=doc_number value='{e(d)}'>"
        f"<input type=hidden name=revision value='{e(r)}'>"
        f"<button>Restore…</button></form></td></tr>"
        for d, r, t, n, at, by, why in gone
    ) or "<tr><td colspan=8 class=muted>Nothing withdrawn.</td></tr>"

    return _page("Document library", f"""
<h2>Add a document</h2>
<form method=post action="/documents/upload" enctype="multipart/form-data">
  <label for=file>File (PDF, Word, Excel, Markdown or text — max {cfg.max_upload_mb} MB)</label>
  <input id=file type=file name=file required>
  <label for=proposed_doc_type>Type</label>
  <select id=proposed_doc_type name=proposed_doc_type required>
    {''.join(f'<option value="{t}">{t}</option>' for t in DOC_TYPES)}
  </select>
  <label for=uploader_note>Note (optional) — what changed, and why you are adding it</label>
  <input id=uploader_note name=uploader_note maxlength=500>
  {_name_field()}
  <button>Upload and convert</button>
</form>
<p class=muted>The file is converted to text and shown to you for review.
Nothing can be cited until somebody confirms its document number, revision and
effective date.</p>

<h2>Waiting for review</h2>
<table><tr><th>File<th>Status<th>Uploaded by<th>When<th>Error<th></tr>{qrows}</table>

<h2>Published — citable now</h2>
<table><tr><th>Document<th>Rev<th>Type<th>Effective<th>Chunks<th></tr>{rows}</table>

<h2>Withdrawn</h2>
<p class=muted>Not cited in any answer. The chunks and the audit trail are kept,
so "why did the assistant stop citing this?" still has an answer.</p>
<table><tr><th>Document<th>Rev<th>Type<th>Chunks<th>When<th>By<th>Reason<th></tr>{grows}</table>
""", who)


# --- upload -----------------------------------------------------------------


@router.post("/upload")
async def upload(
    request: Request,
    file: UploadFile,
    proposed_doc_type: str = Form(...),
    uploader_note: str = Form(""),
    declared_name: str = Form(""),
) -> RedirectResponse:
    """Take a file, convert it, and park it for review.

    Anyone who reaches this endpoint may upload - that is deliberate, and it is
    the one action that is not restricted to publishers. Uploading changes
    nothing an operator can see. APPROVING does, and that is what is gated.
    """
    cfg = settings()
    who = _who(request, declared_name)

    if proposed_doc_type not in DOC_TYPES:
        raise HTTPException(status_code=400, detail="unknown document type")

    data = await file.read()
    if not data:
        raise HTTPException(status_code=400, detail="the file is empty")
    if len(data) > cfg.max_upload_mb * 1024 * 1024:
        raise HTTPException(
            status_code=413,
            detail=f"larger than the {cfg.max_upload_mb} MB limit",
        )

    upload_id = str(uuid.uuid4())
    original = _UNSAFE.sub("_", os.path.basename(file.filename or "upload"))[:200]
    folder = _upload_dir(upload_id)
    folder.mkdir(parents=True, exist_ok=True)
    stored = folder / original
    stored.write_bytes(data)

    digest = hashlib.sha256(data).hexdigest()

    # Insert BEFORE converting, so a conversion that crashes still leaves a row
    # naming the file and the person. A failure with no record is the one
    # outcome that teaches nobody anything.
    with _connect("uploads") as conn:
        with conn.cursor() as cur:
            cur.execute("""
                INSERT INTO doc_uploads (upload_id, status, original_filename,
                    stored_path, content_type, size_bytes, sha256,
                    uploaded_by, uploaded_by_name, uploader_note,
                    proposed_doc_type)
                VALUES (%s,'scanning',%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
                (upload_id, file.filename or original, str(stored),
                 file.content_type, len(data), digest,
                 who.stored, who.name, uploader_note or None, proposed_doc_type))
        conn.commit()

        try:
            converted = convert.convert(original, data)
        except convert.ConversionError as exc:
            with conn.cursor() as cur:
                cur.execute(
                    "UPDATE doc_uploads SET status='failed', error=%s "
                    "WHERE upload_id=%s", (str(exc), upload_id))
            conn.commit()
            return RedirectResponse("/documents", status_code=303)

        (folder / "converted.md").write_text(converted.markdown, encoding="utf-8")
        header = chunking.extract_header(converted.markdown)

        with conn.cursor() as cur:
            cur.execute("""
                UPDATE doc_uploads
                   SET status='awaiting_review', page_count=%s, preview_text=%s,
                       detected_doc_number=%s, detected_revision=%s,
                       detected_effective_date=%s, error=NULL
                 WHERE upload_id=%s""",
                (converted.page_count, converted.markdown[:20000],
                 header.doc_number, header.revision, header.effective_date,
                 upload_id))
        conn.commit()

    return RedirectResponse(f"/documents/review/{upload_id}", status_code=303)


# --- review -----------------------------------------------------------------


@router.get("/review/{upload_id}", response_class=HTMLResponse)
def review(request: Request, upload_id: str) -> HTMLResponse:
    demo = settings().doc_identity_mode == "demo"
    who = (
        Actor(stored="", display="", name=None, email=None,
              groups=DEMO_GROUPS, verified=False)
        if demo else _who(request)
    )

    with _connect("uploads") as conn:
        rows = _fetch(conn, """
            SELECT original_filename, status, proposed_doc_type,
                   detected_doc_number, detected_revision,
                   detected_effective_date, preview_text, error, uploaded_by,
                   uploader_note
              FROM doc_uploads WHERE upload_id = %s""", (upload_id,))
        if not rows:
            raise HTTPException(status_code=404, detail="no such upload")
        (fname, status, dtype, dnum, drev, ddate, preview, error, by,
         note) = rows[0]
        live = _fetch(conn, """
            SELECT revision, effective_date, chunk_count FROM live_documents
             WHERE doc_number = %s ORDER BY revision""", (dnum,)) if dnum else []

    if status == "published":
        return _page("Already published", "<p>This upload is live. "
                     "<a href='/documents'>Back to the library</a>.</p>", who)
    if status == "failed":
        return _page("Conversion failed", f"""
<div class="warn danger"><b>{e(fname)} could not be converted.</b>
<p>{e(error)}</p></div>
<p>Nothing was added to the library. <a href="/documents">Back</a>.</p>""", who)

    # What the reviewer is about to supersede, shown BEFORE they tick the box.
    supersede_note = "".join(
        f"<li>Revision <b>{e(r)}</b>, effective {e(d)}, {e(n)} chunks</li>"
        for r, d, n in live
    )
    supersede_block = (
        f"<div class='warn'><b>{e(dnum)} is already in the library:</b>"
        f"<ul>{supersede_note}</ul>"
        "Ticking supersede withdraws those revisions when this one is "
        "published.</div>" if live else ""
    )

    return _page(f"Review — {fname}", f"""
<p class=muted>Uploaded by {e(by)}. {e(note) or ''}</p>
{supersede_block}
<h2>Converted text — this is what the assistant will read</h2>
<p class=muted>The original file is never used to answer a question. If the
text below is wrong or garbled, reject it: approving it puts this text in front
of an operator.</p>
<pre>{e(preview)}</pre>

<h2>Confirm the document</h2>
<div class="warn"><b>A wrong revision on a procedure is a safety issue, not a
data-quality one.</b> The values below were guessed by a regular expression.
Check them against the document itself.</div>
<form method=post action="/documents/review/{e(upload_id)}/approve">
  <label for=confirmed_doc_type>Type</label>
  <select id=confirmed_doc_type name=confirmed_doc_type required>
    {''.join(f'<option value="{t}"{" selected" if t == dtype else ""}>{t}</option>'
             for t in DOC_TYPES)}
  </select>
  <label for=confirmed_doc_number>Document number</label>
  <input id=confirmed_doc_number name=confirmed_doc_number required value="{e(dnum)}">
  <label for=confirmed_revision>Revision</label>
  <input id=confirmed_revision name=confirmed_revision required value="{e(drev)}">
  <label for=confirmed_effective_date>Effective date (YYYY-MM-DD)</label>
  <input id=confirmed_effective_date name=confirmed_effective_date required
         type=date value="{e(ddate)}">
  <label><input type=checkbox name=supersede_previous value=yes
    {"checked" if live else ""}> Withdraw the other revisions of this document</label>
  <label><input type=checkbox name=reference_data_checked value=yes required>
    I understand this changes only what the assistant can <b>cite</b>. Tag
    metadata, alarm setpoints and the Cube models are unchanged, and every
    numeric answer still comes from those.</label>
  {_name_field()}
  <button>Approve and publish</button>
</form>

<h2>Reject</h2>
<form method=post action="/documents/review/{e(upload_id)}/reject">
  <label for=review_note>Reason (required)</label>
  <textarea id=review_note name=review_note rows=2 required
            minlength="{MIN_REASON}"></textarea>
  {_name_field()}
  <button>Reject</button>
</form>
""", who)


@router.post("/review/{upload_id}/reject")
def reject(
    request: Request,
    upload_id: str,
    review_note: str = Form(...),
    declared_name: str = Form(""),
) -> RedirectResponse:
    who = require_publisher(_who(request, declared_name))
    if len(review_note.strip()) < MIN_REASON:
        raise HTTPException(
            status_code=400,
            detail=f"give a reason of at least {MIN_REASON} characters",
        )
    with _connect("uploads") as conn:
        with conn.cursor() as cur:
            cur.execute("""
                UPDATE doc_uploads SET status='rejected', review_note=%s,
                       reviewed_by=%s, reviewed_by_name=%s, reviewed_at=now()
                 WHERE upload_id=%s AND status='awaiting_review'""",
                (review_note.strip(), who.stored, who.name, upload_id))
            if cur.rowcount == 0:
                raise HTTPException(
                    status_code=409, detail="not awaiting review any more")
        conn.commit()
    return RedirectResponse("/documents", status_code=303)


@router.post("/review/{upload_id}/approve")
def approve(
    request: Request,
    upload_id: str,
    confirmed_doc_type: str = Form(...),
    confirmed_doc_number: str = Form(...),
    confirmed_revision: str = Form(...),
    confirmed_effective_date: date = Form(...),
    reference_data_checked: str = Form(""),
    supersede_previous: str = Form(""),
    declared_name: str = Form(""),
) -> RedirectResponse:
    """Confirm the header, then chunk, embed and publish.

    This is the only path that writes doc_chunks, and the only one that needs
    the ingest role. Everything before it is reversible; this is the step that
    puts text in front of an operator.
    """
    who = require_publisher(_who(request, declared_name))
    if confirmed_doc_type not in DOC_TYPES:
        raise HTTPException(status_code=400, detail="unknown document type")
    if not reference_data_checked:
        # The database does not enforce this one - it is an acknowledgement,
        # not a fact. Refusing here is what makes it a question rather than
        # decoration.
        raise HTTPException(
            status_code=400,
            detail="confirm you understand reference data is unchanged",
        )

    supersede = bool(supersede_previous)

    with _connect("uploads") as conn:
        rows = _fetch(conn, """
            SELECT status, stored_path FROM doc_uploads WHERE upload_id=%s""",
            (upload_id,))
        if not rows:
            raise HTTPException(status_code=404, detail="no such upload")
        status, stored_path = rows[0]
        if status != "awaiting_review":
            raise HTTPException(
                status_code=409, detail=f"upload is {status}, not awaiting review")

        with conn.cursor() as cur:
            cur.execute("""
                UPDATE doc_uploads
                   SET status='approved', confirmed_doc_type=%s,
                       confirmed_doc_number=%s, confirmed_revision=%s,
                       confirmed_effective_date=%s, supersede_previous=%s,
                       reference_data_checked=TRUE, reviewed_by=%s,
                       reviewed_by_name=%s, reviewed_at=now()
                 WHERE upload_id=%s""",
                (confirmed_doc_type, confirmed_doc_number.strip(),
                 confirmed_revision.strip(), confirmed_effective_date,
                 supersede, who.stored, who.name, upload_id))
        conn.commit()

    markdown = (_upload_dir(upload_id) / "converted.md").read_text(encoding="utf-8")
    header = chunking.extract_header(markdown)

    # --- chunk ------------------------------------------------------------
    with _connect("ingest") as conn:
        equipment = [r[0] for r in _fetch(conn, "SELECT equipment_id FROM equipment")]

        records: list[tuple] = []
        for page, title, body in chunking.split_sections(markdown):
            for chunk in chunking.chunk_section(body, confirmed_doc_type):
                records.append((page, title, chunk,
                                chunking.link_equipment(chunk, equipment)))

        if not records:
            with _connect("uploads") as uconn:
                with uconn.cursor() as cur:
                    cur.execute(
                        "UPDATE doc_uploads SET status='failed', error=%s "
                        "WHERE upload_id=%s",
                        ("the converted text produced no chunks", upload_id))
                uconn.commit()
            raise HTTPException(status_code=422, detail="nothing to publish")

        vectors = _embed([r[2] for r in records])

        superseded_count = 0
        with conn.cursor() as cur:
            # Replace every earlier publication of THIS DOCUMENT REVISION, not
            # just of this file.
            #
            # source_file carries the upload_id, which is new on every upload,
            # so matching on it alone made re-publishing ADDITIVE: approving
            # WRPS-CTL-001 rev A twice left 38 live chunks, the same passage
            # retrievable and citable twice over. It was invisible on screen,
            # because live_documents groups by (doc_number, revision) and only
            # the chunk count moved.
            #
            # Deleting by (doc_number, revision) also clears any superseded
            # chunks of the same revision left by an earlier withdraw. That is
            # correct: those are stale copies of what is being republished, not
            # history. The history is doc_actions, which records who withdrew
            # what and why, and which nothing can delete from.
            cur.execute(
                "DELETE FROM doc_chunks WHERE source_file=%s"
                "    OR (doc_number=%s AND revision=%s)",
                (stored_path, confirmed_doc_number.strip(),
                 confirmed_revision.strip()))
            replaced = cur.rowcount

            if supersede:
                cur.execute("""
                    UPDATE doc_chunks SET superseded=TRUE
                     WHERE doc_number=%s AND revision <> %s AND superseded=FALSE""",
                    (confirmed_doc_number.strip(), confirmed_revision.strip()))
                superseded_count = cur.rowcount

            for (page, title, chunk, equip), vector in zip(records, vectors):
                cur.execute("""
                    INSERT INTO doc_chunks (source_file, doc_type, doc_number,
                        revision, effective_date, superseded, equipment_id, page,
                        section_title, chunk_text, doc_title, authorising_role,
                        embedding)
                    VALUES (%s,%s,%s,%s,%s,FALSE,%s,%s,%s,%s,%s,%s,%s)""",
                    (stored_path, confirmed_doc_type,
                     confirmed_doc_number.strip(), confirmed_revision.strip(),
                     confirmed_effective_date, equip, page, title, chunk,
                     header.title, header.authorising_role,
                     # str(), not the list: pgvector takes its text input form,
                     # and this matches how ingest.py binds it. Passing the
                     # list itself fails at bind time.
                     str(vector)))
        conn.commit()

    with _connect("uploads") as conn:
        with conn.cursor() as cur:
            cur.execute("""
                UPDATE doc_uploads SET status='published', chunk_count=%s,
                       superseded_count=%s, published_source_file=%s
                 WHERE upload_id=%s""",
                (len(records), superseded_count, stored_path, upload_id))
        conn.commit()

    # `replaced` is worth logging on its own: a non-zero value means this
    # document revision was already in the library and has been overwritten,
    # which is invisible on screen.
    log.info("published %s rev %s: %d chunks, %d replaced, %d superseded, by %s",
             confirmed_doc_number, confirmed_revision, len(records),
             replaced, superseded_count, who.stored)
    return RedirectResponse("/documents", status_code=303)


# --- withdraw and restore ---------------------------------------------------


# response_model=None: the return annotation is a UNION of two Response
# subclasses (the confirmation page, or the redirect after acting) and
# FastAPI would otherwise try to build a Pydantic response model from it
# and fail at import. Caught by test_the_router_can_be_mounted.
@router.post("/withdraw", response_class=HTMLResponse, response_model=None)
def withdraw(
    request: Request,
    doc_number: str = Form(...),
    revision: str = Form(...),
    reason: str = Form(""),
    declared_name: str = Form(""),
) -> HTMLResponse | RedirectResponse:
    """Stop citing a document. Immediate, reversible, recorded.

    Posted twice: the first time from the library with no reason, which renders
    the confirmation form; the second with a reason, which acts. A withdrawal
    without a stated reason is not something to make easy.
    """
    if not reason.strip():
        demo = settings().doc_identity_mode == "demo"
        who = (Actor(stored="", display="", name=None, email=None,
                     groups=DEMO_GROUPS, verified=False)
               if demo else _who(request))
        return _page("Withdraw a document", f"""
<div class="warn"><b>{e(doc_number)} revision {e(revision)}</b> will stop being
cited from the next question asked. The chunks and the audit trail are kept, and
it can be restored.</div>
<form method=post action="/documents/withdraw">
  <input type=hidden name=doc_number value="{e(doc_number)}">
  <input type=hidden name=revision value="{e(revision)}">
  <label for=reason>Why (required, at least {MIN_REASON} characters)</label>
  <textarea id=reason name=reason rows=3 required minlength="{MIN_REASON}"></textarea>
  {_name_field()}
  <button>Withdraw</button>
</form>
<p><a href="/documents">Cancel</a></p>""", who)

    who = require_publisher(_who(request, declared_name))
    if len(reason.strip()) < MIN_REASON:
        raise HTTPException(
            status_code=400,
            detail=f"give a reason of at least {MIN_REASON} characters",
        )

    action_id = str(uuid.uuid4())
    with _connect("uploads") as conn:
        with conn.cursor() as cur:
            # Written BEFORE the act, completed after, so an interruption
            # leaves evidence rather than silence.
            cur.execute("""
                INSERT INTO doc_actions (action_id, action, status, doc_number,
                    revision, actor, actor_name, actor_groups, reason)
                VALUES (%s,'withdraw','pending',%s,%s,%s,%s,%s,%s)""",
                (action_id, doc_number, revision, who.stored, who.name,
                 who.groups, reason.strip()))
            conn.commit()

            cur.execute("""
                UPDATE doc_chunks SET superseded=TRUE
                 WHERE doc_number=%s AND revision=%s AND superseded=FALSE""",
                (doc_number, revision))
            affected = cur.rowcount

            # No file move: ai-api has no write access to the document tree.
            # The database flip is what stops citation, which is the part
            # anybody is waiting for; file_moved_to stays NULL and says so.
            cur.execute("""
                UPDATE doc_actions SET status='complete', chunks_affected=%s,
                       completed_at=now() WHERE action_id=%s""",
                (affected, action_id))
        conn.commit()

    log.info("withdrew %s rev %s: %d chunks, by %s",
             doc_number, revision, affected, who.stored)
    return RedirectResponse("/documents", status_code=303)


# response_model=None: the return annotation is a UNION of two Response
# subclasses (the confirmation page, or the redirect after acting) and
# FastAPI would otherwise try to build a Pydantic response model from it
# and fail at import. Caught by test_the_router_can_be_mounted.
@router.post("/restore", response_class=HTMLResponse, response_model=None)
def restore(
    request: Request,
    doc_number: str = Form(...),
    revision: str = Form(...),
    reason: str = Form(""),
    declared_name: str = Form(""),
) -> HTMLResponse | RedirectResponse:
    """Make a withdrawn document citable again.

    Uses the INGEST role, not the uploads one: db/005 has a trigger that refuses
    to let the web role set superseded = FALSE. Anything that makes a document
    citable goes through the role that writes chunks.

    Refused while another revision of the same document is live - restoring the
    old revision of a procedure alongside the new one is the failure this whole
    project exists to avoid.
    """
    if not reason.strip():
        demo = settings().doc_identity_mode == "demo"
        who = (Actor(stored="", display="", name=None, email=None,
                     groups=DEMO_GROUPS, verified=False)
               if demo else _who(request))
        return _page("Restore a document", f"""
<div class="warn"><b>{e(doc_number)} revision {e(revision)}</b> will become
citable again from the next question asked.</div>
<form method=post action="/documents/restore">
  <input type=hidden name=doc_number value="{e(doc_number)}">
  <input type=hidden name=revision value="{e(revision)}">
  <label for=reason>Why (required, at least {MIN_REASON} characters)</label>
  <textarea id=reason name=reason rows=3 required minlength="{MIN_REASON}"></textarea>
  {_name_field()}
  <button>Restore</button>
</form>
<p><a href="/documents">Cancel</a></p>""", who)

    who = require_publisher(_who(request, declared_name))
    if len(reason.strip()) < MIN_REASON:
        raise HTTPException(
            status_code=400,
            detail=f"give a reason of at least {MIN_REASON} characters",
        )

    action_id = str(uuid.uuid4())
    with _connect("ingest") as conn:
        live = _fetch(conn, """
            SELECT revision FROM live_documents
             WHERE doc_number=%s AND revision <> %s""", (doc_number, revision))
        if live:
            raise HTTPException(
                status_code=409,
                detail=(
                    f"revision {live[0][0]} of {doc_number} is live. Withdraw it "
                    f"first - two live revisions of one document is the failure "
                    f"this refuses to create."
                ),
            )

        with conn.cursor() as cur:
            cur.execute("""
                INSERT INTO doc_actions (action_id, action, status, doc_number,
                    revision, actor, actor_name, actor_groups, reason)
                VALUES (%s,'restore','pending',%s,%s,%s,%s,%s,%s)""",
                (action_id, doc_number, revision, who.stored, who.name,
                 who.groups, reason.strip()))
            conn.commit()

            cur.execute("""
                UPDATE doc_chunks SET superseded=FALSE
                 WHERE doc_number=%s AND revision=%s AND superseded=TRUE""",
                (doc_number, revision))
            affected = cur.rowcount

            cur.execute("""
                UPDATE doc_actions SET status='complete', chunks_affected=%s,
                       completed_at=now() WHERE action_id=%s""",
                (affected, action_id))
        conn.commit()

    log.info("restored %s rev %s: %d chunks, by %s",
             doc_number, revision, affected, who.stored)
    return RedirectResponse("/documents", status_code=303)