diff --git a/.env.example b/.env.example index f35d586..d579c8a 100644 --- a/.env.example +++ b/.env.example @@ -35,6 +35,31 @@ CHAT_DEPLOYMENT= # flagship — final prose only CHEAP_DEPLOYMENT= # nano/mini — classifier, entities, tool selection EMBED_DEPLOYMENT= # text-embedding-3-small +# --- Documents / upload UI (Phase 9) ----------------------------------------- +# Two roles, and the split IS the safety boundary. db/005 carries a trigger that +# stops uploads_rw setting superseded = FALSE, so the web path can make a +# document less visible and never more. ingest_rw is the only one that writes +# chunks or restores a document. +UPLOADS_DB_USER=uploads_rw +UPLOADS_DB_PASSWORD= +INGEST_DB_USER=ingest_rw +INGEST_DB_PASSWORD= +DOCS_INBOX=/inbox +MAX_UPLOAD_MB=25 + +# authelia | demo +# authelia the actor is Remote-User from the forward-auth headers. The design. +# demo the actor is TYPED ON THE FORM. Self-asserted and unverified - +# exactly what the design forbids. Rows are written as `demo:` +# with actor_groups = 'DEMO-UNVERIFIED' so they can never be mistaken +# for authenticated ones, and every screen says so. +DOC_IDENTITY_MODE=authelia +# Who may approve, withdraw or restore. Comma-separated. EMPTY MEANS NOBODY and +# every mutating endpoint 403s - that is the intended failure direction. This +# stands in for the AD group AI_DocPublishers; swapping to the group later is a +# config change, not a code change. +DOC_PUBLISHERS= + # --- no-LLM stub mode -------------------------------------------------------- # OFF for anything real. With it on, no model is called: the class comes from # the caller instead of the classifier and the prose is a fixed placeholder. diff --git a/BUILD-AI-CONTAINERS.md b/BUILD-AI-CONTAINERS.md index 77a0286..04ae16c 100644 --- a/BUILD-AI-CONTAINERS.md +++ b/BUILD-AI-CONTAINERS.md @@ -724,6 +724,14 @@ local time changes the answer. - Uploaded documents are not scanned for malware — there is no ClamAV on this host. Extension, magic bytes and size only, on a host that also runs the demo PLC - `ai-api` trusts Authelia's `Remote-User` / `Remote-Groups` headers because nothing outside the `proxy` network can reach it. Any container on `proxy` could forge them; that assumption is exactly as strong as the no-published-ports rule +**Phase 9 as built, 2026-08-28.** The document screens are live at `api.yokogawa.tech/documents` and diverge from §16 in five ways. All five are reversible and none needed anything from outside the project; each is a demo affordance, not a design improvement. + +- **Identity is self-asserted.** `DOC_IDENTITY_MODE=demo`: the actor is typed on the form, not taken from `Remote-User`, which is exactly what §16 forbids. Rows are written as `demo:` with `actor_groups = 'DEMO-UNVERIFIED'` and every screen says so, precisely so that a self-asserted row stays tellable from an authenticated one after real auth goes on — `doc_actions` is a table nothing can delete from, so an ambiguity there is permanent. **The publisher list is one name, `admin`, with no password**, standing in for `AI_DocPublishers`; anyone who reaches the page can claim it. Swap `DOC_IDENTITY_MODE=authelia` and `DOC_PUBLISHERS` for the AD group and this is closed. +- **No `ai-docs-worker`.** Conversion, chunking and embedding run inside the HTTP request, and the same process holds both `uploads_rw` and `ingest_rw`. `db/005`'s trigger still stops the web role un-withdrawing anything, so the boundary holds — but it is now a code boundary rather than a deployment one, and a large upload blocks its own request. +- **Text extraction, not document parsing.** pypdf, python-docx and openpyxl instead of Docling, because Docling pulls torch and lin001 must not build or run that. No layout, no table structure, and **scans cannot be read at all** — they are refused rather than stored empty. Tolerable only because a person reads the converted text before it can be cited. `api/convert.py` is the single file to change. +- **Chunking is duplicated** between `api/chunking.py` and `ingest/ingest.py`, because they live in different images. They must stay identical or the same document chunks differently depending on who loaded it. `api/tests/test_documents.py` locks the rule that matters; it cannot see drift in `ingest.py`. +- **Published files stay in `/datadisk/ai-docs-inbox`** and are never moved into `/datadisk/ai-docs`. `ai-api` has no write access to the document tree. Consequence: `ai-ingest --all` cannot see anything published through the UI, so **the two ingest paths must not be used on the same document**. + Production closes these in the order: network segmentation → secrets → SQL guardrails → document control integration → HA. Phase 9 makes document control integration the more urgent of those, not less: once operators can add documents, the assistant's document set drifts from the controlled set faster. --- diff --git a/README.md b/README.md index 38d014e..6b0c75c 100644 --- a/README.md +++ b/README.md @@ -284,6 +284,28 @@ engineer with access to `imh`, not something this script can decide. ### 8. Phase 9 — operator document upload +**BUILT AND LIVE 2026-08-28, ahead of Phase 8, at the customer's direction.** +The screens are at `https://api.yokogawa.tech/documents` — served by `ai-api`, +not `ai-web`, because `ai.yokogawa.tech` is now SCADA-only and carries no +identity at all. Upload → convert → review → approve, plus withdraw and +restore. The pool screen was explicitly descoped. + +Files are converted to text with pypdf / python-docx / openpyxl and the +converted text is shown to the reviewer before approval — the raw file is never +what the assistant reads, and a bad conversion is meant to be caught by eye. +**Scanned documents cannot be read** and are refused rather than stored empty. + +Two things to know before trusting it. **Identity is self-asserted**: the +publisher is a typed name checked against a one-entry list (`admin`) with no +password, so anyone who reaches the page can claim it. Rows are marked +`demo:` / `DEMO-UNVERIFIED` so they stay distinguishable from +authenticated ones later. And **the two ingest paths must not be used on the +same document** — files published through the UI stay in the inbox and +`ai-ingest --all` cannot see them. Full list of divergences in +[`BUILD-AI-CONTAINERS.md`](BUILD-AI-CONTAINERS.md) §14. + +The design below is what §16 specifies, and remains the target. + **After Phase 8 passes, not before.** When the PLC logic or the SCADA program changes, a new document is issued and the assistant is wrong about the plant until it is ingested. Today that needs SSH to a live shared host. Phase 9 puts diff --git a/api/Dockerfile b/api/Dockerfile index cb46569..ab41929 100644 --- a/api/Dockerfile +++ b/api/Dockerfile @@ -11,6 +11,12 @@ WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir --requirement requirements.txt +# The document-upload dependencies go in a SECOND layer, so changing them does +# not invalidate the one above. Four pure-Python wheels, a few megabytes, no +# compiler. See requirements-docs.txt for what belongs here and what does not. +COPY requirements-docs.txt . +RUN pip install --no-cache-dir --requirement requirements-docs.txt + COPY . . USER appuser diff --git a/api/chunking.py b/api/chunking.py new file mode 100644 index 0000000..5660abf --- /dev/null +++ b/api/chunking.py @@ -0,0 +1,174 @@ +"""Sectioning, chunking and header extraction — MIRRORED FROM ingest/ingest.py. + +WHY THIS FILE IS A COPY, WHICH IS NORMALLY THE WRONG ANSWER. + +Documents reach `doc_chunks` by two routes now: `ai-ingest` at a terminal, and +the upload screen in this API. **They must chunk identically.** If they do not, +the same document ingested by the two paths produces different chunks, +different embeddings and different retrieval behaviour — and the difference +would show up as an assistant that answers a question correctly or not +depending on who loaded the document, which is close to undiagnosable. + +The functions below are therefore mirrored from `ingest/ingest.py` with the +logic unchanged. They are not imported from it because `ingest.py` lives in a +different image, with a different and heavier dependency set. +A shared module needs both images built from a common context, which is a +deployment change to a live host and not one to make in the same stroke as a +new feature. + +**If you change chunking, change it in BOTH files.** `api/tests/test_chunking.py` +locks the behaviour that matters — the refusal to split a numbered step +sequence — so a drift in that rule fails the suite. It cannot detect drift in +`ingest.py`, which is the residual risk and is recorded in +BUILD-AI-CONTAINERS.md S14. + +Mirrored: MARKDOWN_HEADING, STEP_RE, the header regexes, approx_tokens, +split_sections, chunk_section, link_equipment, extract_header, parse_date. +NOT mirrored: parse_document, and everything that touches the disk. +""" + +from __future__ import annotations + +import os +import re +from dataclasses import dataclass +from datetime import date, datetime + +CHUNK_TOKEN_TARGET = int(os.environ.get("CHUNK_TOKEN_TARGET", "800")) + +MARKDOWN_HEADING = re.compile(r"^(#{1,6})\s+(.*\S)\s*$") + +DOC_NUMBER_RE = re.compile(r"\b(WRPS-[A-Z]{2,4}-\d{3,4})\b") +REVISION_RE = re.compile( + r"\b(?:rev(?:ision)?|issue)[\s.:]*([A-Z0-9]{1,4})\b", re.IGNORECASE +) +DATE_RE = re.compile( + r"\b(?:effective|issued|approved)[\s\w]{0,12}?[:\s]\s*" + r"(\d{1,2}[/-]\d{1,2}[/-]\d{2,4}|\d{4}-\d{2}-\d{2}|" + r"\d{1,2}\s+\w+\s+\d{4})\b", + re.IGNORECASE, +) +TITLE_RE = re.compile(r"^\s*title[\s:]+(.+\S)\s*$", re.IGNORECASE | re.MULTILINE) +AUTHORISING_ROLE_RE = re.compile( + r"^[ \t]*authoris(?:ing|ed)[ \t]*(?:role|by)?[ \t]*:[ \t]*(.+\S)[ \t]*$", + re.IGNORECASE | re.MULTILINE, +) + +# A numbered step. Used to refuse to split, not to parse the procedure. +STEP_RE = re.compile(r"^\s*(?:\d+\.|\(\d+\)|step\s+\d+)", re.IGNORECASE | re.MULTILINE) + + +@dataclass(frozen=True) +class Header: + doc_number: str | None + revision: str | None + effective_date: date | None + title: str | None + authorising_role: str | None + + +def parse_date(text: str) -> date | None: + for fmt in ("%d/%m/%Y", "%d-%m-%Y", "%Y-%m-%d", "%d %B %Y", "%d %b %Y", "%d/%m/%y"): + try: + return datetime.strptime(text.strip(), fmt).date() + except ValueError: + continue + return None + + +def extract_header(text: str) -> Header: + """Pull document identity from the first page. ALWAYS confirmed by a human. + + Everything here is a proposal shown on the review screen with the field + already filled in. It is not authority. A wrong revision on a procedure is + a safety issue, and a regex is not a person. + """ + head = text[:4000] + number = DOC_NUMBER_RE.search(head) + revision = REVISION_RE.search(head) + effective = DATE_RE.search(head) + title = TITLE_RE.search(head) + role = AUTHORISING_ROLE_RE.search(head) + return Header( + doc_number=number.group(1) if number else None, + revision=revision.group(1) if revision else None, + effective_date=parse_date(effective.group(1)) if effective else None, + title=title.group(1).strip() if title else None, + authorising_role=role.group(1).strip() if role else None, + ) + + +def approx_tokens(text: str) -> int: + """Rough, and deliberately so - it decides when to split, and the rule that + matters is the one that refuses to.""" + return len(text) // 4 + + +def split_sections(text: str) -> list[tuple[int, str, str]]: + """Markdown -> [(page, section_title, section_text)]. + + Mirrors ingest.parse_markdown, which takes a Path; this takes the text, + because by the time it gets here the upload has already been converted in + memory. + + Page is always 1. A converted document has no pages that survive the + conversion, so a citation to one carries a section title and no page + number. That is honest. Do not invent page numbers to make citations look + uniform. + """ + sections: list[tuple[int, str, list[str]]] = [] + current_title = "(untitled)" + buffer: list[str] = [] + + for line in text.splitlines(): + heading = MARKDOWN_HEADING.match(line) + if heading: + if buffer: + sections.append((1, current_title, buffer)) + current_title = heading.group(2).strip() + buffer = [] + elif line.strip(): + buffer.append(line.rstrip()) + + if buffer: + sections.append((1, current_title, buffer)) + + return [(page, title, "\n".join(body)) for page, title, body in sections] + + +def chunk_section(text: str, doc_type: str) -> list[str]: + """Split a section, unless splitting it would break a step sequence. + + For procedures the rule is absolute: a section containing numbered steps is + emitted whole, however long it is. An oversized chunk costs tokens. Half a + procedure costs more than that. + """ + if approx_tokens(text) <= CHUNK_TOKEN_TARGET: + return [text] + + if doc_type == "procedure" and STEP_RE.search(text): + return [text] + + chunks: list[str] = [] + buffer: list[str] = [] + for paragraph in text.split("\n\n"): + candidate = "\n\n".join(buffer + [paragraph]) + if buffer and approx_tokens(candidate) > CHUNK_TOKEN_TARGET: + chunks.append("\n\n".join(buffer)) + buffer = [paragraph] + else: + buffer.append(paragraph) + if buffer: + chunks.append("\n\n".join(buffer)) + return chunks + + +def link_equipment(text: str, equipment_ids: list[str]) -> str | None: + """Tie a chunk to equipment when it is unambiguously about one thing. + + Two different units mentioned means no link, not a guess - a chunk linked + to the wrong pump is worse than one linked to nothing, because retrieval + filtering will then hide it from the pump it actually describes. + """ + found = {eid for eid in equipment_ids if eid.lower() in text.lower()} + return found.pop() if len(found) == 1 else None diff --git a/api/config.py b/api/config.py index 498a807..b6fa8d5 100644 --- a/api/config.py +++ b/api/config.py @@ -53,6 +53,52 @@ class Settings(BaseModel): query_timeout_seconds: int = 30 max_output_tokens: int = 1200 + # --- Documents (Phase 9) ----------------------------------------------- + # Two extra roles, and the split is the safety boundary, not bookkeeping. + # + # uploads_rw the queue, and withdraw. db/005 carries a trigger that + # refuses to let this role set superseded = FALSE, so it can + # make a document LESS visible and never more. + # ingest_rw writes doc_chunks: approve and restore. Anything that makes + # a document citable uses this one. + # + # In the design these are two components - ai-api and ai-docs-worker, the + # second with no HTTP surface. There is no worker here, so both connections + # live in this process. That is a real deviation and it is recorded in + # BUILD-AI-CONTAINERS.md S14: the trigger still blocks the web role, but the + # process holding uploads_rw also holds ingest_rw, so the separation is now + # a code boundary rather than a deployment one. Splitting the worker out + # later is a config change and a compose file, not a redesign. + uploads_db_user: str = "uploads_rw" + uploads_db_password: str = "" + ingest_db_user: str = "ingest_rw" + ingest_db_password: str = "" + + # Where uploaded files land before review. Never inside /datadisk/ai-docs: + # that folder means "the documents this plant runs on", and an unreviewed + # upload is not one of those. + docs_inbox: str = "/inbox" + max_upload_mb: int = 25 + + # --- Document identity ------------------------------------------------- + # "authelia" - the actor is Remote-User from the forward-auth headers, and + # a missing header is a 401. This is the design. + # "demo" - the actor is TYPED BY THE PERSON on the form. Self-asserted, + # unverified, and exactly what the design forbids ("identity + # comes from the headers, never from the request body"). + # + # Demo mode is off unless asked for, the screens carry a banner saying the + # name is unverified, and every row it writes is stored as `demo:` + # with actor_groups = 'DEMO-UNVERIFIED'. That marking is the point: when + # real auth goes on, a self-asserted audit row must still be tellable from + # an authenticated one. Without it they are indistinguishable forever. + doc_identity_mode: str = "authelia" + # Who may change anything. In the design this is the AD group + # AI_DocPublishers; with no AD group available it is a name list, swapped + # for the group later with one config change. Empty means nobody, and the + # API fails closed - approve, withdraw and restore all 403. + doc_publishers: tuple[str, ...] = () + # --- Cube -------------------------------------------------------------- cubejs_api_url: str = "http://cube:4000/cubejs-api/v1" cubejs_api_secret: str = "" @@ -69,10 +115,27 @@ class Settings(BaseModel): f"@{self.pghost}:{self.pgport}/{self.pgdatabase}" ) + def docs_dsn(self, role: str) -> str: + """DSN for one of the document roles. Never log the result. + + `role` is "uploads" or "ingest" - spelled out at every call site rather + than defaulted, because picking the wrong one is the difference between + a web request that can withdraw a document and one that can publish it. + """ + user, password = { + "uploads": (self.uploads_db_user, self.uploads_db_password), + "ingest": (self.ingest_db_user, self.ingest_db_password), + }[role] + return ( + f"postgresql://{user}:{password}" + f"@{self.pghost}:{self.pgport}/{self.pgdatabase}" + ) + def redacted(self) -> dict[str, object]: """Safe to log and safe to return from /healthz.""" secret = {"pgpassword", "azure_openai_api_key", "cubejs_api_secret", - "langfuse_secret_key"} + "langfuse_secret_key", "uploads_db_password", + "ingest_db_password"} return { k: ("set" if v else "unset") if k in secret else v for k, v in self.model_dump().items() @@ -108,6 +171,18 @@ def settings() -> Settings: max_rows_returned=int(env.get("MAX_ROWS_RETURNED", "5000")), query_timeout_seconds=int(env.get("QUERY_TIMEOUT_SECONDS", "30")), max_output_tokens=int(env.get("MAX_OUTPUT_TOKENS", "1200")), + uploads_db_user=env.get("UPLOADS_DB_USER", "uploads_rw"), + uploads_db_password=env.get("UPLOADS_DB_PASSWORD", ""), + ingest_db_user=env.get("INGEST_DB_USER", "ingest_rw"), + ingest_db_password=env.get("INGEST_DB_PASSWORD", ""), + docs_inbox=env.get("DOCS_INBOX", "/inbox"), + max_upload_mb=int(env.get("MAX_UPLOAD_MB", "25")), + doc_identity_mode=env.get("DOC_IDENTITY_MODE", "authelia").lower(), + # Comma-separated. Blank entries dropped so a trailing comma in an env + # file cannot silently authorise "". + doc_publishers=tuple( + n.strip() for n in env.get("DOC_PUBLISHERS", "").split(",") if n.strip() + ), cubejs_api_url=env.get("CUBEJS_API_URL", "http://cube:4000/cubejs-api/v1"), cubejs_api_secret=env.get("CUBEJS_API_SECRET", ""), langfuse_host=env.get("LANGFUSE_HOST", "http://langfuse:3000"), diff --git a/api/convert.py b/api/convert.py new file mode 100644 index 0000000..62d86e6 --- /dev/null +++ b/api/convert.py @@ -0,0 +1,217 @@ +"""Uploaded file -> Markdown, before anything is chunked or embedded. + +WHY THERE IS A CONVERSION STEP AT ALL. The raw PDF is never what the assistant +reads - it never was. Retrieval only ever sees `doc_chunks.chunk_text`. What +this module adds is that the extracted text becomes a thing a person can LOOK +AT before approving it. If a table comes out as garbage, the reviewer sees the +garbage and rejects the document, instead of an operator discovering it months +later inside a citation. + +WHAT THESE CONVERTERS DO AND DO NOT DO. They extract text, not layout. A +multi-column page interleaves. A merged-cell spreadsheet flattens. A scanned +page yields nothing at all, and is refused rather than stored empty. + +That limitation is acceptable HERE, and only here, because a person reads the +converted text before it can be cited - the same safety net the design already +required for the document header. It would not be acceptable in a pipeline that +published without review. + +REPLACING THEM. Everything below the `convert()` boundary is swappable: return +Markdown from bytes, raise ConversionError when you cannot. Nothing outside this +module knows which library did the work, so a better converter is a change to +this file alone. If the real documents turn out to be scans, that is the +conversation to have - OCR is the missing capability, and nothing else here +changes. +""" + +from __future__ import annotations + +import io +import re +from dataclasses import dataclass + +# A scanned page yields a handful of stray characters, not nothing - so an +# emptiness check has to have a floor above zero. Below this, across the whole +# document, we refuse rather than store a blank document that a reviewer might +# approve without noticing there is nothing in it. +_MIN_CHARS = 200 + +SUPPORTED = {".pdf", ".docx", ".xlsx", ".xlsm", ".md", ".txt"} + + +class ConversionError(Exception): + """Conversion failed, or produced something not worth reviewing.""" + + +@dataclass(frozen=True) +class Converted: + markdown: str + converter: str # recorded on the upload row - which code produced this + page_count: int | None + + +def _clean(text: str) -> str: + """Collapse the whitespace damage that text extraction always leaves. + + Not cosmetic: runs of blank lines and trailing spaces change where a chunker + splits, so the same document converted twice should look the same. + """ + text = text.replace("\r\n", "\n").replace("\r", "\n") + text = re.sub(r"[ \t]+\n", "\n", text) + text = re.sub(r"\n{3,}", "\n\n", text) + return text.strip() + + +def _from_pdf(data: bytes) -> Converted: + from pypdf import PdfReader + + try: + reader = PdfReader(io.BytesIO(data)) + except Exception as exc: + raise ConversionError(f"not a readable PDF: {exc}") from exc + + if reader.is_encrypted: + # Refuse rather than guess at an empty password. A locked document that + # silently converts to nothing is the worst outcome here. + raise ConversionError( + "this PDF is encrypted - remove the protection and upload it again" + ) + + parts: list[str] = [] + for number, page in enumerate(reader.pages, start=1): + try: + text = page.extract_text() or "" + except Exception: + # One bad page must not lose the other ninety. Mark it so the + # reviewer can see exactly what is missing. + text = "" + parts.append(f"\n\n\n") + if text.strip(): + parts.append(f"\n\n\n\n{text}") + + return Converted( + markdown=_clean("".join(parts)), + converter="pypdf", + page_count=len(reader.pages), + ) + + +def _from_docx(data: bytes) -> Converted: + import docx + + try: + document = docx.Document(io.BytesIO(data)) + except Exception as exc: + raise ConversionError(f"not a readable .docx: {exc}") from exc + + lines: list[str] = [] + for para in document.paragraphs: + text = para.text.strip() + if not text: + continue + # Word's built-in heading styles are the one piece of structure that + # survives reliably, and headings matter: section_title is what a + # citation falls back to. + style = (para.style.name or "").lower() if para.style else "" + if style.startswith("heading"): + level = "".join(c for c in style if c.isdigit()) or "1" + lines.append(f"{'#' * min(int(level), 6)} {text}") + elif style.startswith("title"): + lines.append(f"# {text}") + else: + lines.append(text) + + for index, table in enumerate(document.tables, start=1): + lines.append(f"\n") + for row in table.rows: + cells = [c.text.strip().replace("|", "\\|") for c in row.cells] + lines.append("| " + " | ".join(cells) + " |") + + return Converted( + markdown=_clean("\n\n".join(lines)), + converter="python-docx", + page_count=None, + ) + + +def _from_xlsx(data: bytes) -> Converted: + from openpyxl import load_workbook + + try: + # read_only keeps a large workbook from being held in memory twice, and + # data_only takes the cached VALUE of a formula rather than the formula + # text. A cell reading "=SUM(B2:B9)" is not something to embed - and if + # the workbook was never opened in Excel there is no cached value, so + # that cell comes through empty. Say so in the output rather than + # letting it look like a blank cell. + book = load_workbook(io.BytesIO(data), read_only=True, data_only=True) + except Exception as exc: + raise ConversionError(f"not a readable spreadsheet: {exc}") from exc + + lines: list[str] = [] + for sheet in book.worksheets: + lines.append(f"## {sheet.title}") + for row in sheet.iter_rows(values_only=True): + cells = ["" if v is None else str(v).strip().replace("|", "\\|") + for v in row] + if not any(cells): + continue + lines.append("| " + " | ".join(cells) + " |") + lines.append("") + book.close() + + return Converted( + markdown=_clean("\n".join(lines)), + converter="openpyxl", + page_count=None, + ) + + +def _from_text(data: bytes) -> Converted: + try: + text = data.decode("utf-8") + except UnicodeDecodeError: + text = data.decode("latin-1") + return Converted(markdown=_clean(text), converter="passthrough", page_count=None) + + +_CONVERTERS = { + ".pdf": _from_pdf, + ".docx": _from_docx, + ".xlsx": _from_xlsx, + ".xlsm": _from_xlsx, + ".md": _from_text, + ".txt": _from_text, +} + + +def convert(filename: str, data: bytes) -> Converted: + """Convert an uploaded file to Markdown, or raise ConversionError. + + Dispatch is on the extension, which is a claim the uploader made about the + file. The libraries below all fail loudly on a mismatch, so a .docx renamed + to .pdf raises rather than producing plausible rubbish. + """ + suffix = "." + filename.rsplit(".", 1)[-1].lower() if "." in filename else "" + handler = _CONVERTERS.get(suffix) + if handler is None: + raise ConversionError( + f"{suffix or 'no extension'} is not supported - " + f"expected one of {', '.join(sorted(SUPPORTED))}" + ) + + result = handler(data) + + if len(result.markdown) < _MIN_CHARS: + # The common cause by far is a scanned document: a picture of text, + # which text extraction cannot read. Name that explicitly, because + # "conversion produced nothing" sends somebody looking for a bug in the + # upload instead of at the file. + raise ConversionError( + f"only {len(result.markdown)} characters of text came out of this " + f"file. If it is a scan or a photograph, no text can be extracted " + f"from it - it needs OCR, which this converter does not do. " + f"Nothing has been added to the library." + ) + + return result diff --git a/api/documents.py b/api/documents.py new file mode 100644 index 0000000..0d70442 --- /dev/null +++ b/api/documents.py @@ -0,0 +1,789 @@ +"""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 = ( + '
Demo identity. The name below is typed in, ' + 'not verified by sign-in. Every change is recorded as ' + f'demo:{e(who.display)} so it can never be mistaken for ' + 'an authenticated action.
' + ) + return HTMLResponse(f""" + + +{e(title)} — WRPS Document Library + + +

{e(title)}

+{banner} +{body} +""") + + +def _name_field(who_default: str = "") -> str: + """The demo identity prompt. Only rendered in demo mode.""" + if settings().doc_identity_mode != "demo": + return "" + return ( + '' + f'' + ) + + +# --- 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"{e(d)}{e(r)}{e(t)}" + f"{e(ed)}{e(n)}" + f"
" + f"" + f"" + f"
" + for d, r, t, ed, n, _ in live + ) or "Nothing is published." + + qrows = "".join( + f"{e(f)}{e(s)}{e(u)}{e(at)}" + f"{e(err) or ''}" + f"Review" + for uid, f, s, u, at, err in queue + ) or "Nothing waiting." + + grows = "".join( + f"{e(d)}{e(r)}{e(t)}" + f"{e(n)}{e(at)}{e(by)}{e(why)}" + f"
" + f"" + f"" + f"
" + for d, r, t, n, at, by, why in gone + ) or "Nothing withdrawn." + + return _page("Document library", f""" +

Add a document

+
+ + + + + + + {_name_field()} + +
+

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.

+ +

Waiting for review

+{qrows}
FileStatusUploaded byWhenError
+ +

Published — citable now

+{rows}
DocumentRevTypeEffectiveChunks
+ +

Withdrawn

+

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.

+{grows}
DocumentRevTypeChunksWhenByReason
+""", 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", "

This upload is live. " + "Back to the library.

", who) + if status == "failed": + return _page("Conversion failed", f""" +
{e(fname)} could not be converted. +

{e(error)}

+

Nothing was added to the library. Back.

""", who) + + # What the reviewer is about to supersede, shown BEFORE they tick the box. + supersede_note = "".join( + f"
  • Revision {e(r)}, effective {e(d)}, {e(n)} chunks
    • " + for r, d, n in live + ) + supersede_block = ( + f"
    {e(dnum)} is already in the library:" + f"
      {supersede_note}
    " + "Ticking supersede withdraws those revisions when this one is " + "published.
    " if live else "" + ) + + return _page(f"Review — {fname}", f""" +

    Uploaded by {e(by)}. {e(note) or ''}

    +{supersede_block} +

    Converted text — this is what the assistant will read

    +

    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.

    +
    {e(preview)}
    + +

    Confirm the document

    +
    A wrong revision on a procedure is a safety issue, not a +data-quality one. The values below were guessed by a regular expression. +Check them against the document itself.
    +
    + + + + + + + + + + + {_name_field()} + +
    + +

    Reject

    +
    + + + {_name_field()} + +
    +""", 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 any earlier ingest of this exact file, so re-publishing + # cannot double the chunks. + cur.execute("DELETE FROM doc_chunks WHERE source_file=%s", (stored_path,)) + + 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() + + log.info("published %s rev %s: %d chunks, %d superseded, by %s", + confirmed_doc_number, confirmed_revision, len(records), + 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""" +
    {e(doc_number)} revision {e(revision)} will stop being +cited from the next question asked. The chunks and the audit trail are kept, and +it can be restored.
    +
    + + + + + {_name_field()} + +
    +

    Cancel

    """, 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""" +
    {e(doc_number)} revision {e(revision)} will become +citable again from the next question asked.
    +
    + + + + + {_name_field()} + +
    +

    Cancel

    """, 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) diff --git a/api/identity.py b/api/identity.py new file mode 100644 index 0000000..88ed268 --- /dev/null +++ b/api/identity.py @@ -0,0 +1,169 @@ +"""Who is acting, and whether they are allowed to. + +Two questions, deliberately separate: + + actor() WHO. In `authelia` mode this is Remote-User from the + forward-auth headers and nothing else. A missing header is a + 401, never an anonymous fallback - getting to ai-api without + passing Authelia is not a state in which to accept a change to + the document set. + + require_publisher() WHETHER. A name list today (DOC_PUBLISHERS), the AD + group AI_DocPublishers later. Empty list means nobody, and every + mutating endpoint 403s. Fail closed. + +DEMO MODE. `DOC_IDENTITY_MODE=demo` takes the actor from a form field instead. +That is precisely what the design forbids - "identity comes from Authelia's +forwarded headers, never from the request body" - and it is here because the +customer asked for a demo with no password. Three things make it survivable: + + 1. It is off unless asked for. + 2. Every screen carries a banner saying the name is unverified. + 3. Rows are written as `demo:` with actor_groups = 'DEMO-UNVERIFIED'. + +(3) is the one that matters in a year. When real auth goes on, the audit trail +still has to distinguish a name somebody typed from a name Authelia proved. If +demo rows were written as bare names they would be indistinguishable from real +ones forever, and `doc_actions` is a table nothing can delete from - so the +ambiguity would be permanent. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from fastapi import HTTPException, Request + +from config import settings + +# Authelia's forward-auth subrequest returns these; Caddy's shared `authelia` +# snippet on lin001 copies all four upstream (checked 2026-08-27). They are +# trusted ONLY because nothing outside the proxy network can reach this app. +# Any container on `proxy` could forge them - that assumption is exactly as +# strong as the no-published-ports rule. +_USER_HEADER = "Remote-User" +_NAME_HEADER = "Remote-Name" +_EMAIL_HEADER = "Remote-Email" +_GROUPS_HEADER = "Remote-Groups" + +DEMO_GROUPS = "DEMO-UNVERIFIED" + + +@dataclass(frozen=True) +class Actor: + """The person making a change, as it will be written to doc_actions.""" + + # What goes in doc_actions.actor / doc_uploads.*_by. Carries the `demo:` + # prefix in demo mode - do not strip it anywhere on the write path. + stored: str + # What to show on screen. No prefix. + display: str + name: str | None + email: str | None + groups: str + verified: bool + + @property + def is_demo(self) -> bool: + return not self.verified + + +def _header(request: Request, key: str) -> str | None: + value = request.headers.get(key) + return value.strip() if value and value.strip() else None + + +def actor(request: Request, declared_name: str | None = None) -> Actor: + """Resolve who is acting, or raise 401. + + `declared_name` is the form field, and it is ignored outside demo mode - + passing it in `authelia` mode must not be able to change the recorded + actor, which is the whole point of taking identity from the headers. + """ + cfg = settings() + remote_user = _header(request, _USER_HEADER) + + if cfg.doc_identity_mode != "demo": + if not remote_user: + raise HTTPException( + status_code=401, + detail=( + "no Remote-User header - this request did not pass through " + "Authelia. Refusing to record an anonymous document change." + ), + ) + return Actor( + stored=remote_user, + display=remote_user, + name=_header(request, _NAME_HEADER), + email=_header(request, _EMAIL_HEADER), + groups=_header(request, _GROUPS_HEADER) or "", + verified=True, + ) + + # --- demo mode --------------------------------------------------------- + typed = (declared_name or "").strip() + if not typed: + raise HTTPException( + status_code=400, + detail="a name is required - it is recorded against this change", + ) + if len(typed) > 64: + raise HTTPException(status_code=400, detail="name too long") + + # Record the Authelia identity too WHEN THERE IS ONE. On + # api.yokogawa.tech there usually is, because that hostname is still gated + # even though this mode ignores it for authorisation. It costs nothing and + # gives a cross-check: a demo row whose header disagrees with the typed + # name is worth a second look. + groups = DEMO_GROUPS + if remote_user: + groups = f"{DEMO_GROUPS}; authelia={remote_user}" + + return Actor( + stored=f"demo:{typed}", + display=typed, + name=typed, + email=_header(request, _EMAIL_HEADER), + groups=groups, + verified=False, + ) + + +def is_publisher(who: Actor) -> bool: + """May this actor change the document set? + + Compared against the DISPLAY name, not the stored one - the `demo:` prefix + is a provenance marker on the audit row, not part of anybody's identity. + Case-insensitive: "Admin" and "admin" are the same person, and a demo that + fails on capitalisation teaches nothing. + """ + allowed = {n.casefold() for n in settings().doc_publishers} + if not allowed: + return False + return who.display.casefold() in allowed + + +def require_publisher(who: Actor) -> Actor: + """403 unless this actor may act. Called by EVERY mutating endpoint. + + This is the check that actually enforces. The Authelia rule for the + document paths is the other one, and the UI hiding a button is neither - + the design is explicit that a hidden button proves nothing, and the gate + item for this phase is verified by calling the API directly as a + non-publisher. + """ + if not is_publisher(who): + if not settings().doc_publishers: + raise HTTPException( + status_code=403, + detail=( + "no publishers are configured, so nothing may be changed. " + "Set DOC_PUBLISHERS." + ), + ) + raise HTTPException( + status_code=403, + detail=f"{who.display} is not a publisher", + ) + return who diff --git a/api/main.py b/api/main.py index 389f074..0c0fb5c 100644 --- a/api/main.py +++ b/api/main.py @@ -21,6 +21,7 @@ from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field import agent +import documents from config import settings from contracts import ContractViolation from guardrails import GuardrailViolation @@ -95,6 +96,12 @@ def _langfuse(): return None +# The document library screens. Mounted under /documents, NOT /docs - FastAPI's +# Swagger UI already owns /docs and the customer keeps it. Two things under one +# prefix with two different access policies is what gets misread later. +app.include_router(documents.router) + + class AskRequest(BaseModel): question: str = Field(min_length=3, max_length=1000) question_class: str | None = Field( diff --git a/api/requirements-docs.txt b/api/requirements-docs.txt new file mode 100644 index 0000000..dd9a37a --- /dev/null +++ b/api/requirements-docs.txt @@ -0,0 +1,18 @@ +# Document upload (Phase 9). SEPARATE FROM requirements.txt ON PURPOSE. +# +# Installed in its own layer, AFTER the main one. A change here therefore does +# not invalidate the cached layer carrying fastapi, langgraph, langfuse and the +# rest - so adding or bumping a document dependency costs a few megabytes of +# pure-Python wheels instead of re-resolving the whole tree on a 2 vCPU host +# that also runs the demo plant's PLC. +# +# All four are PURE PYTHON: no compiler runs, no native extension is built. +# +# These extract text, not layout, and cannot read a scan. api/convert.py says +# what that costs and why review is what makes it acceptable. Anything heavier - +# in particular anything pulling torch - does not belong in this file; it +# belongs in an image built somewhere other than lin001. +python-multipart==0.0.20 +pypdf==5.1.0 +python-docx==1.1.2 +openpyxl==3.1.5 diff --git a/api/tests/test_documents.py b/api/tests/test_documents.py new file mode 100644 index 0000000..59f267a --- /dev/null +++ b/api/tests/test_documents.py @@ -0,0 +1,217 @@ +"""The document-library rules that must hold without a database or a network. + +Same principle as the contract tests: the rules that matter are in Python, so +they can be exercised without an API key, a database or a running host. What is +covered here is what would be expensive to discover on the host and impossible +to discover from `docker ps`. +""" + +from __future__ import annotations + +import pytest +from fastapi import HTTPException + +import chunking +import convert +import identity +from config import Settings + + +class _Req: + """The two attributes identity.actor() actually reads.""" + + def __init__(self, headers: dict[str, str] | None = None): + self.headers = headers or {} + + +def _use(monkeypatch, **kwargs): + """Give identity.py a Settings built for this one test. + + The real settings() is lru_cached off the process environment, which is no + use here - each test needs a different identity mode and publisher list. + monkeypatch restores the original at teardown, so nothing leaks between + tests and the cache is never touched. + """ + cfg = Settings(**kwargs) + monkeypatch.setattr(identity, "settings", lambda: cfg) + return cfg + + +# --- chunking --------------------------------------------------------------- +# Mirrored from ingest/ingest.py. If this drifts, the same document ingested by +# the two paths produces different chunks - see the header of api/chunking.py. + + +def test_a_numbered_step_sequence_in_a_procedure_is_never_split(): + steps = "\n\n".join(f"{n}. Do the {n}th thing. " + "x" * 400 + for n in range(1, 30)) + assert chunking.approx_tokens(steps) > chunking.CHUNK_TOKEN_TARGET + assert len(chunking.chunk_section(steps, "procedure")) == 1 + + +def test_the_same_oversized_text_IS_split_when_it_is_not_a_procedure(): + # The refusal is scoped to procedures on purpose. A manual has no steps to + # break, and keeping it whole would just cost tokens. + prose = "\n\n".join("Some prose about the station. " + "x" * 400 + for _ in range(30)) + assert len(chunking.chunk_section(prose, "manual")) > 1 + + +def test_sections_split_on_headings_and_keep_their_titles(): + sections = chunking.split_sections( + "# Purpose\nWhy this exists.\n\n## Scope\nWhat it covers.\n") + assert [t for _, t, _ in sections] == ["Purpose", "Scope"] + + +def test_equipment_is_linked_only_when_it_is_unambiguous(): + ids = ["PU-301", "PU-302"] + assert chunking.link_equipment("Isolate PU-301 before work.", ids) == "PU-301" + # Two units mentioned is not a tie to break - a chunk linked to the wrong + # pump is hidden from the pump it actually describes. + assert chunking.link_equipment("PU-301 and PU-302 share a header.", ids) is None + + +def test_header_extraction_is_a_proposal_and_may_find_nothing(): + found = chunking.extract_header( + "Title: Wet Well Interlock Bypass\n" + "WRPS-OPS-014 Revision 3\n" + "Effective: 01/03/2026\n" + "Authorising role: Station Maintenance Supervisor\n") + assert found.doc_number == "WRPS-OPS-014" + assert found.revision == "3" + assert found.authorising_role == "Station Maintenance Supervisor" + + empty = chunking.extract_header("A document with no header at all.") + assert empty.doc_number is None and empty.revision is None + + +# --- conversion ------------------------------------------------------------- + + +def test_a_document_with_almost_no_extractable_text_is_refused(): + """A scan converts to nothing. Refusing is the point. + + Storing it would put a blank document in front of a reviewer who might + approve it without noticing there is nothing in it. + """ + with pytest.raises(convert.ConversionError) as exc: + convert.convert("scan.md", b"# Title\n") + assert "OCR" in str(exc.value) + + +def test_an_unsupported_extension_is_refused_by_name(): + with pytest.raises(convert.ConversionError) as exc: + convert.convert("drawing.dwg", b"x" * 5000) + assert ".dwg" in str(exc.value) + + +def test_plain_text_passes_through_and_keeps_its_headings(): + body = "# Purpose\n\n" + "This station has three pumps. " * 40 + result = convert.convert("notes.md", body.encode()) + assert result.converter == "passthrough" + assert result.markdown.startswith("# Purpose") + + +# --- identity --------------------------------------------------------------- + + +def test_authelia_mode_refuses_a_request_with_no_remote_user(monkeypatch): + """Reaching ai-api without passing Authelia is not a state in which to + accept a change to the document set.""" + _use(monkeypatch, doc_identity_mode="authelia") + with pytest.raises(HTTPException) as exc: + identity.actor(_Req()) + assert exc.value.status_code == 401 + + +def test_authelia_mode_ignores_a_name_supplied_in_the_body(monkeypatch): + """The form field must not be able to change the recorded actor. That is + the entire reason identity comes from the headers.""" + _use(monkeypatch, doc_identity_mode="authelia") + who = identity.actor(_Req({"Remote-User": "cliu"}), declared_name="someone-else") + assert who.stored == "cliu" + assert who.verified is True + + +def test_demo_mode_marks_every_row_so_it_can_never_pass_as_authenticated(monkeypatch): + _use(monkeypatch, doc_identity_mode="demo") + who = identity.actor(_Req(), declared_name="admin") + assert who.stored == "demo:admin" + assert who.display == "admin" + assert who.groups == identity.DEMO_GROUPS + assert who.verified is False + + +def test_demo_mode_records_the_authelia_user_alongside_the_typed_name(monkeypatch): + """A demo row whose header disagrees with the typed name is worth a look.""" + _use(monkeypatch, doc_identity_mode="demo") + who = identity.actor(_Req({"Remote-User": "cliu"}), declared_name="admin") + assert "authelia=cliu" in who.groups + + +def test_demo_mode_still_requires_a_name(monkeypatch): + _use(monkeypatch, doc_identity_mode="demo") + with pytest.raises(HTTPException) as exc: + identity.actor(_Req(), declared_name=" ") + assert exc.value.status_code == 400 + + +# --- authorisation ---------------------------------------------------------- + + +def test_with_no_publishers_configured_nobody_may_change_anything(monkeypatch): + """Fail closed. An empty list must not mean 'everyone'.""" + _use(monkeypatch, doc_identity_mode="demo", doc_publishers=()) + who = identity.actor(_Req(), declared_name="admin") + with pytest.raises(HTTPException) as exc: + identity.require_publisher(who) + assert exc.value.status_code == 403 + + +def test_a_non_publisher_is_refused_even_though_authelia_let_them_in(monkeypatch): + _use(monkeypatch, doc_identity_mode="demo", doc_publishers=("admin",)) + with pytest.raises(HTTPException) as exc: + identity.require_publisher(identity.actor(_Req(), declared_name="dan")) + assert exc.value.status_code == 403 + + +def test_the_publisher_check_ignores_capitalisation(monkeypatch): + _use(monkeypatch, doc_identity_mode="demo", doc_publishers=("admin",)) + assert identity.is_publisher(identity.actor(_Req(), declared_name="Admin")) + + +def test_the_demo_prefix_is_provenance_and_not_part_of_the_identity(monkeypatch): + """`demo:admin` is admin. The prefix marks how we know, not who they are - + if it were matched, demo mode would authorise nobody.""" + _use(monkeypatch, doc_identity_mode="demo", doc_publishers=("admin",)) + who = identity.actor(_Req(), declared_name="admin") + assert who.stored == "demo:admin" + assert identity.is_publisher(who) + + +# --- the router itself ------------------------------------------------------ + + +def test_the_router_can_be_mounted(): + """Import documents.py and mount it on a real FastAPI app. + + This exists because it did not, and the first deployment crash-looped on + import: `withdraw` and `restore` return a UNION of two Response subclasses, + and FastAPI tried to build a Pydantic response model from the annotation. + Nothing in the rest of this file imports documents.py, so every unit test + passed against code that could not start. + + A route-table assertion is not the point - reaching the assertion is. The + failure mode this catches is at import time. + """ + from fastapi import FastAPI + + import documents + + app = FastAPI() + app.include_router(documents.router) + + paths = {r.path for r in app.routes} + for expected in ("/documents", "/documents/upload", "/documents/withdraw", + "/documents/restore", "/documents/review/{upload_id}"): + assert expected in paths, f"{expected} is not mounted" diff --git a/compose/ai-compose.yml b/compose/ai-compose.yml index 7521f53..a5b46ae 100644 --- a/compose/ai-compose.yml +++ b/compose/ai-compose.yml @@ -144,15 +144,21 @@ services: # gets. Deliberately not /datadisk/ai-docs: a file that has been uploaded # but not yet approved must not be visible to `ai-ingest --all`. # - # COMMENTED OUT UNTIL PHASE 9, on the same principle as caddy/ai-routes.caddy: - # add each piece at the phase that needs it. Docker creates a missing bind - # source as a ROOT-OWNED directory, and ai-api does not run as root - so - # deploying this before the directory exists gives you an inbox the API - # cannot write to, on the growing disk of a live shared host. - # Create it first, then uncomment: - # sudo install -d -o 10002 -g 10002 /datadisk/ai-docs-inbox - # volumes: - # - /datadisk/ai-docs-inbox:/inbox + # ENABLED 2026-08-28. Docker creates a missing bind source as a ROOT-OWNED + # directory and ai-api does not run as root, so the directory must exist + # with the right owner BEFORE this comes up: + # sudo install -d -o 10001 -g 10001 /datadisk/ai-docs-inbox + # + # 10001, not the 10002 this comment used to say. 10002 is the ingest + # image's user; there is no ai-docs-worker in this build, so the process + # writing the inbox is ai-api's appuser at 10001. Check the Dockerfile + # rather than this line if it ever moves. + # + # /datadisk, not /: the root disk is 62 GB and uploads accumulate. And NOT + # /datadisk/ai-docs - a file that has been uploaded but not yet approved + # must not be visible to `ai-ingest --all`. + volumes: + - /datadisk/ai-docs-inbox:/inbox healthcheck: test: ["CMD", "python", "-m", "app_healthcheck"] interval: 30s