"""Batch membership and tenant scope — what one click actually covers. Two questions, both settled 2026-09-07 and both recorded in KNOWLEDGE_SURFACE_PLAN.md: 1. **Which documents?** The caller's own (`documents.user_id = caller`), filtered to the types the pipeline can actually parse. The company-wide alternative is deferred, not dropped — hence `ingest_scope()` below, which is the single place that answer lives. 2. **Whose knowledge is it?** The company's. Rows carry `scope_id` (the tenant) beside `user_id` (the uploader), so switching membership to company-wide later needs a different WHERE and no backfill. ⚠️ `scope_id` must never enter the `entity_id` derivation. `ids.py` hashes `doc_id` + content and nothing else; keeping it that way is what makes a future re-point an `UPDATE … SET scope_id` — a value migration — rather than a re-key that would lose every expert approval. """ from __future__ import annotations import re from typing import Literal from sqlalchemy import select from src.db.postgres.connection import AsyncSessionLocal from src.db.postgres.models import Document, User from src.middlewares.logging import get_logger logger = get_logger("knowledge_selector") # Only these reach the pipeline. Six of the eight live `documents` rows are # xlsx/csv — those belong to the structured-data path and are skipped with a # reason rather than silently ignored. PARSEABLE_TYPES = frozenset({"pdf", "docx"}) IngestScope = Literal["user", "company"] # The single flip point (N10f). Batch membership and the diff baseline must move # together: selecting per user while diffing company-wide would classify a # user's candidates against approvals they cannot see — an asymmetry that is # invisible when testing with one user. INGEST_SCOPE: IngestScope = "user" def ingest_scope() -> IngestScope: """Whose documents one click covers. `user` today, `company` later.""" return INGEST_SCOPE def normalise_company(company: str | None) -> str | None: """Fold a free-text company name into a usable tenant key. Interim only: dedorch has no companies table, so the tenant key is derived from `users.company`, which is nullable free text with no id. Casefolding and collapsing whitespace absorbs the common variations; it does not save us from a genuine typo forking a knowledge base, which is why the contract asks Harry for a real `companies` table (Q6). """ if not company: return None folded = re.sub(r"\s+", " ", company).strip().casefold() return folded or None async def resolve_scope_id(user_id: str) -> str: """The tenant key for a user: their normalised company, else the user. Falls back to `user:` rather than raising, so a user with no company recorded still gets a private, well-formed scope instead of sharing one with every other company-less user. """ async with AsyncSessionLocal() as session: result = await session.execute(select(User.company).where(User.id == user_id)) row = result.first() company = normalise_company(row[0] if row else None) if company: return company logger.info( "scope_id_fallback", extra={"user_id": user_id, "reason": "no company on user row"}, ) return f"user:{user_id}" async def select_documents( user_id: str, document_ids: list[str] | None = None ) -> tuple[list[Document], list[tuple[Document, str]]]: """Pick the batch. Returns (selected, [(document, skip_reason), …]). An explicit `document_ids` list is still filtered by owner and type — a caller naming someone else's document, or a spreadsheet, gets it skipped with a reason rather than honoured. `documents.status` is deliberately NOT filtered on: it tracks the separate pgvector processing path, and a row exists only once `blob_name` is set, so it says nothing about whether this pipeline can read the file. A blob that turns out to be unreadable fails that one document and settles the job `partial`. """ async with AsyncSessionLocal() as session: stmt = select(Document).where(Document.user_id == user_id) if document_ids: stmt = stmt.where(Document.id.in_(document_ids)) result = await session.execute(stmt) rows = list(result.scalars().all()) selected: list[Document] = [] skipped: list[tuple[Document, str]] = [] for doc in rows: file_type = (doc.file_type or "").strip().lower().lstrip(".") if file_type not in PARSEABLE_TYPES: skipped.append((doc, f"unsupported_type:{file_type or 'unknown'}")) continue selected.append(doc) # A named id that matched nothing is a caller error worth surfacing: it is # either a typo or another user's document, and both look like silence. if document_ids: found = {d.id for d in rows} for missing in [d for d in document_ids if d not in found]: logger.info( "ingest_document_not_found", extra={"user_id": user_id, "document_id": missing}, ) return selected, skipped