Add validity to the document

AGENTS.md

@@ -62,6 +62,7 @@ Core modules (all under `pipeline/`):
 - `document_repository.py` — `DocumentRepository`, a domain layer over `db.py` for document reads; `db.py` stays one-function-per-query with no domain knowledge
 - `network_constants.py` — fixed values sent on the network (catalog/resource ids, topics, languages, Beckn version). The file v2 edits when the AI layer starts deriving them
 - `catalog_builder.py` — pure builder mapping a document's knowledge kind (`advisory`/`scheme`) to the single `OnDemand` Beckn catalog announced for that kind; returns `None` for any other kind. No env, no I/O
+- `document_validity.py` — pure module owning Document Validity: the period a document's chunks answer searches in, the upload-day start with no end by default, and the clock injection point. Shared by `api.py` (stamping the default at upload), `scheme_catalog.py` (validating a reviewer's edit), `activities.py` (stamping chunk payloads) and `vector_store/qdrant_store.py` (filtering at search time). `db.py` stores the dates it is handed and defaults nothing
 - `discovery_publish_service.py` — `DiscoveryPublishService`: owns the Publish to Network env vars and the HTTP call. Deliberately Temporal-free
 - `models.py` — Pydantic models, including `DocumentStage` enum and `PIPELINE_STAGES` (the stepper-UI stage list)
 - `config.py` — `Config` dataclass reading env vars, with defaults

CONTEXT.md

@@ -28,5 +28,9 @@ _Avoid_: "catalog" alone.
 What a document *is* to the network — `advisory`, `scheme`, `video`, or an operator-entered slug — held in `documents.document_kind`. Asserted by a reviewer during the pipeline, never inferred from the file, and absent until then (every document starts as the default `document`). Only `advisory` and `scheme` map to a Network Catalog Envelope; the rest publish nothing.
 _Avoid_: "document type" — collides with `source_type`/`canonical_input_type`, which describe the input *format* (pdf, spreadsheet). Also avoid saying an advisory is "uploaded": what is uploaded is a file, which only becomes an advisory when someone classifies it.
 
+**Document Validity**:
+The period a document's chunks answer searches in, held as `documents.valid_from` / `documents.valid_to` and stamped onto every chunk's vector payload as `start_date` / `end_date`. Defaulted on upload to start that day with no end, meaning it never expires; a reviewer can narrow either end in the same form that sets the Knowledge Kind, behind the `VITE_DOCUMENT_VALIDITY_ENABLED` flag. Both ends are inclusive, and search filters on it so an expired or not-yet-started document is never answered from. Chunks ingested before it existed carry neither date and stay searchable — see `docs/ADR/0005-document-validity-filters-search.md`.
+_Avoid_: "expiry" — there is a start as well as an end. Also avoid calling it the document's lifetime: the document stays in the console, editable and reingestable, once its period has passed; only its search answers stop.
+
 **network_visible**:
 An operator-controlled flag on a document/scheme (not a pipeline stage) that gates whether it's exposed to other BAPs through the pull-based Scheme Catalog snapshot. Independent of Publish to Network.

README.md

@@ -516,7 +516,7 @@ Translation data is surfaced through the page model and review endpoints.
 - `GET /documents/{workflow_id}/qdrant`
 - `GET /documents/{workflow_id}/qdrant/chunks`
 - `POST /documents/{workflow_id}/reingest`
-- `POST /search`
+- `POST /search` — only answers from documents whose validity period covers today; `valid_on` asks about another day and `include_expired` drops the filter (see `docs/openapi-search.yaml`)
 - `GET /indexes/summary`
 - `GET /indexes/{index_name}/settings`
 - `GET /indexes/{index_name}/stats`

docs/ADR/0005-document-validity-filters-search.md

@@ -0,0 +1,39 @@
+# Document validity filters search
+
+Context: a chunk, once ingested, answered searches forever. Agricultural content does not work that way — a rabi sowing advisory is wrong advice in June, and a scheme circular superseded last year is worse than no answer. Nothing in the pipeline could express "this stopped being true", so the only way to retire content was to disable the whole document, losing it from the operator console too.
+
+A short-lived predecessor (ADR 0005, since removed along with its code) put a *catalog* validity window on the Network Catalog Envelope at the prod gate. That window was kind-level and about whether this provider still claims to serve a knowledge kind at all; it never spoke to whether a given document's content was current, which is what this ADR is about.
+
+Decision:
+- **A document-level concept, owned in one place.** `documents.valid_from` / `documents.valid_to`, stamped onto every chunk's Qdrant payload as `start_date` / `end_date`, owned by a pure module `pipeline/document_validity.py` that holds the rules so the API, the ingest activity and the vector store never re-derive them.
+- **Defaulted on upload, not asked for.** The upload endpoints stamp the upload day as the start, with no end. An uploader is not asked a question they cannot answer, and no document reaches the index without a period. An end nobody chose would retire content on a date nobody meant, so the default expires never and narrowing it is a deliberate act. The default is computed in `api.py` and passed down: `db.upsert_document` stores the dates verbatim on INSERT and has no opinion on what they should be, keeping `db.py` free of domain knowledge as the rest of that layer is. A row created around the upload path (a script, a backfill) therefore has no period, which search reads as "no stated period" — the same as the pre-validity corpus.
+- **Confirmed or moved by the reviewer, in the form that already exists.** The dates sit beside the Knowledge Kind selector and travel on the same `PATCH /documents/{id}/scheme-metadata`. Choosing the period is part of classifying a document, not a separate errand, and the reviewer sees real prefilled dates rather than an empty field. Either end may be sent alone; the other is taken from what is stored, so moving the end date does not require restating the start.
+- **Validated like a bad kind, not like a bad type.** `document_validity.parse_period` raises, `apply_scheme_metadata` lets it through as a `ValueError`, and the endpoint answers 400 — the same shape a bad `document_kind` gets, not a Pydantic 422. A rejected period stores nothing, kind included.
+- **Both ends inclusive.** A document uploaded this morning starts today and must answer this afternoon; an end date names the last day it answers rather than the first day it does not. The requirement was written as "today greater than start and less than end", but exclusive bounds would hide a document on its own first day, which is the default every document gets.
+- **Search filters on it by default**, as `(start_date missing OR start_date <= today) AND (end_date missing OR end_date >= today)`. Per-end rather than one window clause, so a partially-dated point behaves sensibly instead of vanishing.
+- **Undated chunks stay searchable.** The missing-field branches are the whole reason the rule is shaped this way: every point already in Qdrant carries no dates, and a change that silently emptied the live index would rightly be called a regression. Such a document picks up a period the next time it is ingested — anchored on its own upload day, not on today, so reingesting a two-year-old document does not quietly extend its life by another year.
+- **The clock is injected.** `QdrantVectorStore(clock=...)` decides what "today" means, defaulting to the real clock. Tests pin a day instead of writing fixtures relative to whenever they run, and `POST /search` exposes `valid_on` (answer as of another day) and `include_expired` (drop the filter) so an operator can inspect what an expired document still holds. Both default to the safe behaviour: a caller that asks for nothing gets today's valid documents only.
+- **Document-scoped reads ignore validity.** `list_by_doc_id`, `delete_by_doc_id` and `delete_chunk` pass no date: an expired chunk is still that document's chunk, and a purge that skipped expired chunks would orphan them in the index.
+- **Indexed as Qdrant `DATETIME`**, created idempotently on every `ensure_collection` including one that already exists — a collection created before the field existed has no index for it, and ingest is the one path that reliably runs against every live collection. The filter was verified to work unindexed too, so a collection still being backfilled filters correctly, just more slowly.
+
+Known limitation, accepted: expiry is enforced at **read** time, not by a sweeper. An expired document's vectors stay in Qdrant, still costing storage and still visible to anything that queries Qdrant directly without this filter (including `include_expired`). That is the right trade for now — deleting on expiry would make un-expiring a document a reingest — but a caller bypassing `/search` is not protected.
+
+Alternatives considered:
+- **Reuse the catalog validity window from ADR 0005.** Rejected at the time, and moot since that feature was removed: those dates defaulted to today/today and carried a kind-level meaning on the wire, so reusing them would have expired every document the day it was approved.
+- **Collect the dates at the prod gate.** Rejected: DEV search is filtered too, and the prod gate is reached long after ingestion — a document would be searchable with no period for most of its life.
+- **Store epoch integers instead of `YYYY-MM-DD` strings.** Rejected: Qdrant's `DatetimeRange` accepts date-only strings (verified against a live instance), and an operator reading a chunk payload can see `2027-03-31` rather than decoding a number.
+- **Filter after retrieval, in the API.** Rejected: it silently shrinks the candidate set below `top_k`, so an index full of expired documents would return a short page of results rather than the valid ones further down.
+- **Ask the uploader for the dates.** Rejected: the uploader is often not the person who knows how long content holds, and blocking upload on it would slow the common case for a value the reviewer sets anyway.
+- **Default to no end date (never expires).** Rejected: it makes expiry opt-in, so content would go stale by default — the same failure this ADR exists to fix.
+
+## Amendment — open-ended by default, and gated in the UI (2026-09-25)
+
+The original decision gave every upload a one-year end date. Business revised that: a document should stay answerable until someone decides otherwise, so the default is now **start at the upload day, no end**. `end_date is None` means the document never expires, and the search filter already read a missing end that way — the `(end_date missing OR end_date >= today)` clause was written for the pre-validity corpus and covers this unchanged, which is why no filter logic moved.
+
+Consequences worth naming:
+- `period_from_row` now treats the **start** as what decides whether a period exists. A row with a start and a NULL end rebuilds as an open-ended period rather than as "no period", which is the shape every upload now stores.
+- A chunk with no expiry carries **no** `end_date` key in its Qdrant payload rather than a null one, so the missing-field branch matches.
+- `add_years` and `DEFAULT_VALIDITY_YEARS` existed only to derive the old default end. Nothing derives one any more, so they were removed rather than left as dead code.
+- Clearing an end date is how a reviewer returns a document to never-expiring: a blank end is a real answer, not a missing one.
+
+The reviewer-facing date fields sit behind `VITE_DOCUMENT_VALIDITY_ENABLED` (default off), following the `VITE_AUTH_ENABLED` pattern. The flag gates **only the entry fields** — the backend stamps a period on every upload and search filters on it regardless — so enabling it exposes an existing capability rather than switching one on. With it off, classifying a document sends no dates at all and leaves the stored period untouched.

docs/openapi-search.yaml

@@ -237,6 +237,22 @@ components:
           type: boolean
           description: If true, also return the pre-cap/pre-rerank candidate list as raw_hits.
           default: false
+        include_expired:
+          type: boolean
+          description: |
+            Drop the document-validity filter, returning chunks whose period
+            has ended or not yet started. An operator tool for inspecting what
+            an expired document still holds — not for a caller serving an end
+            user, who should never be answered from an expired document.
+          default: false
+        valid_on:
+          type: string
+          format: date
+          description: |
+            Answer as of this calendar day (`YYYY-MM-DD`) instead of today —
+            what search would have returned, or will return, on that date. A
+            malformed value returns 400. Ignored when include_expired is true.
+          default: "(today)"
 
     SearchHit:
       type: object
@@ -296,6 +312,18 @@ components:
         chunk_number:
           type: integer
           description: Alias of chunk_num, set when chunk_num is present.
+        start_date:
+          type: string
+          format: date
+          description: |
+            First day this chunk's document answers searches. Absent on chunks
+            ingested before validity existed, which are treated as current.
+        end_date:
+          type: string
+          format: date
+          description: |
+            Last day this chunk's document answers searches (inclusive).
+            Absent on chunks ingested before validity existed.
         section:
           type: string
         token_count:

pipeline/activities.py

@@ -23,7 +23,7 @@
 from temporalio import activity
 from temporalio.exceptions import ApplicationError
 
-from . import scheme_catalog
+from . import document_validity, scheme_catalog
 from .catalog_builder import normalize_document_kind
 from .chunking import chunk_pages, load_chunking_config
 from .discovery_publish_service import DiscoveryPublishService
@@ -641,6 +641,8 @@ def _prepare_records(
     scheme_code: str | None = None,
     scheme_name: str | None = None,
     scheme_aliases: list[str] | None = None,
+    valid_from: str | None = None,
+    valid_to: str | None = None,
 ) -> list[dict]:
     metadata = _get_doc_metadata(filename)
     resolved_instance = _normalize_instance(instance)
@@ -672,6 +674,10 @@ def _prepare_records(
         else []
     )
     instance_name = instance_display_name(resolved_instance)
+    # One period for the whole document, resolved before the loop: every chunk
+    # of a document expires together, and re-deriving it per chunk would let a
+    # midnight boundary split one document across two periods.
+    validity = document_validity.period_from_row(valid_from, valid_to)
 
     records = []
     for chunk in chunks:
@@ -716,6 +722,13 @@ def _prepare_records(
             "quality_score": float(quality_score) if str(quality_score).strip().replace(".", "", 1).isdigit() else 0.0,
             "priority_rank": float(priority_rank) if str(priority_rank).strip().replace(".", "", 1).isdigit() else 0.0,
         }
+        if validity is not None:
+            record["start_date"] = validity.start_date
+            # Omitted, not null, when the document never expires: the search
+            # filter's "end_date missing" branch is what keeps it answerable
+            # forever, and _record_payload drops None anyway.
+            if validity.end_date is not None:
+                record["end_date"] = validity.end_date
         if is_scheme:
             record["scheme_code"] = (scheme_code or "").strip().lower()
             record["scheme_name"] = resolved_scheme_name
@@ -751,6 +764,23 @@ def _scheme_fields_from_doc(doc: dict | None) -> dict:
     }
 
 
+def _validity_fields_from_doc(doc: dict | None) -> dict:
+    """Extract the validity kwargs for `_prepare_records` from a documents row.
+
+    A row with no stored period is stamped with the default anchored on its
+    upload day - start there, no end - so every chunk written from here on
+    carries a start. Old points already in the index keep none until their
+    document is reingested, which is what keeps them searchable meanwhile.
+    """
+    doc = doc or {}
+    period = document_validity.period_from_row(
+        doc.get("valid_from"), doc.get("valid_to")
+    )
+    if period is None:
+        period = document_validity.period_from_upload_timestamp(doc.get("created_at"))
+    return {"valid_from": period.start_date, "valid_to": period.end_date}
+
+
 def prepare_ingestion_records(
     document_id: str,
     filename: str,
@@ -810,6 +840,10 @@ def _passage_schema_definition(use_tensor_prefix_field: bool = True) -> dict:
         {"name": "priority_rank", "type": "float", "features": ["filter"]},
         {"name": "text", "type": "text", "features": ["lexical_search"]},
         {"name": "priority", "type": "float", "features": ["score_modifier", "filter"]},
+        # Document validity - filtered on at search time, so the day a chunk
+        # starts and stops answering travels with the chunk itself.
+        {"name": "start_date", "type": "date", "features": ["filter"]},
+        {"name": "end_date", "type": "date", "features": ["filter"]},
     ]
     if use_tensor_prefix_field:
         all_fields.append({"name": "text_for_embedding", "type": "text"})
@@ -1196,6 +1230,7 @@ async def prepare_for_ingestion(
         name_en=name_en,
         description=description,
         instance=(doc or {}).get("instance"),
+        **_validity_fields_from_doc(doc),
     )
     activity.logger.info(f"Prepared {len(records)} records")
     return records
@@ -1305,6 +1340,7 @@ async def promote_document_to_prod_qdrant(
         workflow_id=workflow_id,
         instance=doc.get("instance"),
         **scheme_kwargs,
+        **_validity_fields_from_doc(doc),
     )
     activity.logger.info(
         "Promoting %s records to PROD Qdrant collection %s (kind=%s)",
@@ -1373,6 +1409,7 @@ async def ingest_document_from_db(
         workflow_id=workflow_id,
         instance=doc.get("instance"),
         **_scheme_fields_from_doc(doc),
+        **_validity_fields_from_doc(doc),
     )
     payload_path = _write_json_temp(records)
     try:
@@ -1419,15 +1456,14 @@ async def ingest_document_from_db(
 async def publish_catalog_to_network(workflow_id: str, transaction_id: str) -> dict:
     """POST a catalog/publish envelope to the Discovery Service and record the exchange.
 
-    The catalog depends on the document's knowledge kind, which is read here
-    rather than passed in as an activity argument - that keeps both workflow
-    call sites, and any workflow already in flight, untouched.
+    The catalog depends on the document's knowledge kind, read here rather than
+    passed in as an activity argument - that keeps every workflow call site, and
+    any workflow already in flight, untouched.
     """
     from . import db
 
-    document_kind = normalize_document_kind(
-        DocumentRepository().get_document_kind(workflow_id)
-    )
+    repository = DocumentRepository()
+    document_kind = normalize_document_kind(repository.get_document_kind(workflow_id))
 
     service = DiscoveryPublishService()
     result = await asyncio.to_thread(