Add semantic task retrieval with scoped embeddings and RAG context

docs/design_index.md

@@ -90,6 +90,7 @@ live.
 | Workflow or orchestration | Relevant vocabulary/semantics in the [VWL manual](engineering/von_workflow_language_manual.md); domain examples and appendices are reference material |
 | Prompts, models, routing, optimisation, or fine-tuning | [prompt programmes and model routing](engineering/prompt_programs_and_model_routing_playbook.md) |
 | Retrieval, memory, RAG, KB growth, or long-horizon state | [agent memory and enduring knowledge](engineering/agent_memory_and_enduring_knowledge.md) |
+| Semantic task search, task embeddings and task RAG context | [Task semantic retrieval](engineering/task_semantic_retrieval.md) |
 | Represented, experience- or discussion-derived learning for role, capability, workflow, tool, acquisition, introspection, message, or task decisions | Draft [represented advice design](engineering/represented_advice_design.md), including Phase 0 subtraction, Phase 0.5 non-active candidates, and the [completed bounded negative learning cycle](engineering/jvnautosci_2720_learning_cycle_2026-09-05.md) under JVNAUTOSCI-2720; then the applicable prompt, memory, evaluation, workflow, or security guidance. The cycle links source-derived formation, experimental exposure, outcome evidence, rejection and subsequent verified absence. It does not establish positive advice efficacy, an ordinary runtime consumer, production activation, represented Arm C or autonomous promotion. |
 | Private Otter archive access, conversation images or research-slide reading | [Private Otter archive and conversation images](engineering/otter_archive_and_conversation_images.md) |
 | Reusing external ontology concepts with source identity and attribution | [KnowKat ontology sources and governed adoption](engineering/knowkat_ontology_sources.md) |

docs/engineering/task_semantic_retrieval.md

@@ -0,0 +1,124 @@
+# Semantic task retrieval
+
+- **Kind:** Bounded implementation and operational reference
+- **Lifecycle:** Active
+- **Authority:** This retrieval path only; canonical task and access services remain authoritative
+- **Owner:** Von maintainers
+- **Reviewed:** 13 September 2026
+- **Review trigger:** Task visibility, embedding runtime or task search changes
+
+`task_search(query="organise transport to the academic event", search_mode="semantic")`
+and `GET /api/tasks/search?search_mode=semantic&query=...` retrieve task meaning
+through the existing task search filters. The default remains lexical search.
+The semantic mode returns ranked tasks, `semantic_retrieval` diagnostics and
+`rag_context` excerpts with task concept IDs for citations. These excerpts are
+retrieved data, not instructions or authority. The existing task-search tool
+delivers this context to Von without another prompt or workflow stage.
+
+## Representation and retrieval
+
+`task_document.v1` serialises identity, title, description, status, priority,
+assignee, creator, organisation, project, collection IDs, labels, components,
+source, external references, reference code and relevant timestamps. Context
+identifiers preserve canonical references; this slice does not expand them to
+project descriptions or ingest comments, attachments, private source archives
+or work products. Tasks remain canonical Vontology-backed operational records;
+their embedding cache is derived storage, not enduring domain knowledge.
+
+Search reads canonical actor-visible tasks and applies the existing filters
+before embedding or ranking. It pins the existing RAG embedding runtime for
+the query and document batches, then ranks by cosine similarity with concept
+ID as a deterministic tie-breaker. Pagination follows ranking. There is no
+unstated relevance threshold: even a weak nearest neighbour can appear, so
+consumers must judge relevance from the returned task and score. RAG context
+is limited to 12,000 text characters, with explicit excerpt truncation.
+
+The index compares document SHA-256 and the RAG provider/model/host signature.
+Unchanged documents reuse vectors; edits, status changes and model changes
+regenerate them. The query vector is always generated in the pinned runtime.
+Full task documents are retained in the response as ordinary task data; only
+vectors, hashes, identifiers and timestamps are stored in SQLite.
+
+## Access, freshness and lifecycle
+
+Semantic search binds the trusted server-session or in-process actor and
+forces canonical visibility enforcement. Assignee/project/organisation query
+filters cannot supply actor identity. Anonymous queries see public tasks only.
+Legacy header-derived identities and access-bypass contexts cannot enter this
+semantic path. There is a second batched canonical visibility/existence read
+after embedding, which excludes tasks deleted or revoked during embedding.
+As with ordinary reads, this is a snapshot, not a serialisable transaction
+against edits occurring after the read.
+
+`VON_TASK_SEMANTIC_INDEX_PATH` defaults to the ignored
+`data/task_semantic_index.sqlite3`. Newly created files use mode 0600. Treat
+vectors as private data and keep the containing runtime directory private.
+The cache is partitioned by the trusted actor and organisation. It contains no
+independent task text, audience grant or candidate authority. Every request
+gets its candidates and content from canonical storage; an old cache row can
+never resurrect a deleted task or grant a revoked audience access.
+
+After successful indexing, the actor partition retains only the current
+filtered snapshot. A filter change can therefore evict useful cached vectors;
+this favours simple recovery and bounded retained data over maximum reuse.
+Deleted/revoked rows disappear from that partition on its next search. An
+inactive actor partition can retain derived vectors until operator cleanup,
+but cannot expose them through retrieval. Concurrent cache writers may cause
+extra embedding work, but cannot replace the vectors retained by an in-flight
+search for its own snapshot.
+
+Trusted operator code may call `clear_task_semantic_index()` in an actor-bound
+context to discard that partition, or `reindex_semantic_tasks()` to rebuild all
+currently visible tasks, including bulk tasks. Neither operation changes task
+meaning or shared RAG namespaces. Ordinary next searches also rebuild missing
+rows, so recovery needs no database migration, scheduler or durable queue.
+For a corrupt SQLite file, an operator may stop its callers, move the disposable
+file aside, and resume: the next search creates a fresh index. Keep this action
+separate from canonical databases. Rolling back the code leaves lexical search
+and canonical task state intact; the derived file can be discarded.
+
+## Failure and observability
+
+Embedding, malformed-vector or local index failures retry through existing
+lexical search and explicitly return `status=degraded`, `ranking=lexical` and
+`error_code=task_semantic_index_unavailable`. They do not claim semantic empty
+results or return old cached text. If canonical lexical reads also fail, the
+existing task error path reports failure. Retry is on demand, and successful
+embedding batches survive later failures. Corrupt individual vector rows are
+treated as cache misses. Logs omit task/query text and provider exception
+bodies; diagnostics report candidate, embedded, cache-hit and access-recheck
+counts, embedding-version hash and indexing/ranking elapsed milliseconds.
+
+## Evidence and operating limits
+
+The tests in `tests/backend/test_task_semantic_index_service.py` exercise the
+real filtered task service, repository visibility filters, local SQLite and
+Flask route with an isolated mock database and deterministic fixture embedder.
+They cover paraphrase geometry versus lexical matching, status/project filters,
+pagination, current text after edits, embedding-version changes, corruption,
+provider failure, cross-actor and cross-organisation reads, public reads,
+revocation, deletion and deletion during embedding. This proves the selected
+plumbing and access boundaries, not the deployed embedding model's quality.
+
+`tests/fixtures/task_retrieval/cases.json` supplies ordinary research and
+administrative tasks with explicit relevance judgements. The optional harness:
+
+```sh
+pdm run python scripts/evaluate_task_semantic_retrieval.py --use-configured-embedder
+```
+
+embeds only the synthetic fixture, does not read or mutate canonical tasks,
+and reports recall@k, reciprocal rank, nDCG@k, index build time, per-query time
+and the embedding signature against the current substring baseline. It requires
+an available configured embedder; running it can incur that provider's charges.
+No live embedding-provider quality or latency result is claimed by fixture tests.
+Access-control acceptance is the route/service test suite, not this harness.
+
+This implementation performs exact vector ranking over matching canonical
+tasks. It reuses the existing batched hydration path and adds a batched access
+read, without per-task database reads. Cold searches embed every matching task;
+warm searches still read canonical task data and rank all candidates. Use
+project/collection filters for large imports. Large-corpus latency and an
+approximate nearest-neighbour index are not established by this delivery.
+No new model-routing policy, task workflow, public deployment or background
+index activation is included.

scripts/evaluate_task_semantic_retrieval.py

@@ -0,0 +1,134 @@
+"""Evaluate synthetic task retrieval without reading or writing canonical tasks.
+
+The CLI requires explicit opt-in before invoking the configured embedding
+provider. Imported evaluate() supports an isolated local/fixture embedder.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+from pathlib import Path
+import sys
+import time
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from src.backend.services.task_semantic_index_service import (  # noqa: E402
+    _normalise_vector,
+    build_task_document,
+)
+
+
+def _metrics(ranked, relevant, k):
+    hits = [identity in relevant for identity in ranked[:k]]
+    ideal = sum(1 / math.log2(i + 2) for i in range(min(k, len(relevant))))
+    return {
+        "recall_at_k": sum(hits) / len(relevant),
+        "reciprocal_rank": next((1 / (i + 1) for i, hit in enumerate(hits) if hit), 0),
+        "ndcg_at_k": sum(hit / math.log2(i + 2) for i, hit in enumerate(hits)) / ideal,
+    }
+
+
+def evaluate(cases, embedder, *, k=3):
+    if k < 1:
+        raise ValueError("k must be positive")
+    started = time.monotonic()
+    documents = [
+        build_task_document(
+            {
+                "task_concept_id": task["id"],
+                "title": task["title"],
+                "description": task["description"],
+            }
+        )
+        for task in cases["tasks"]
+    ]
+    vectors = [
+        _normalise_vector(value)
+        for value in embedder.get_text_embedding_batch(
+            [document["text"] for document in documents]
+        )
+    ]
+    if len(vectors) != len(documents):
+        raise ValueError("Incomplete embeddings")
+    build_ms = (time.monotonic() - started) * 1000
+    observations = []
+    for case in cases["queries"]:
+        query_start = time.monotonic()
+        vector = _normalise_vector(embedder.get_query_embedding(case["query"]))
+        if any(len(candidate) != len(vector) for candidate in vectors):
+            raise ValueError("Embedding dimension mismatch")
+        ranked = sorted(
+            zip(documents, vectors),
+            key=lambda item: (
+                -sum(a * b for a, b in zip(vector, item[1])),
+                item[0]["id"],
+            ),
+        )
+        semantic = [document["id"] for document, _ in ranked]
+        # Fair current task-search baseline: case-insensitive substring match.
+        lexical = [
+            task["id"]
+            for task in cases["tasks"]
+            if any(
+                case["query"].lower() in task[field].lower()
+                for field in ("title", "description")
+            )
+        ]
+        observations.append(
+            {
+                "semantic": _metrics(semantic, set(case["relevant"]), k),
+                "lexical": _metrics(lexical, set(case["relevant"]), k),
+                "query_elapsed_ms": round((time.monotonic() - query_start) * 1000, 2),
+            }
+        )
+    if not observations:
+        raise ValueError("Evaluation queries are required")
+    aggregate = {
+        mode: {
+            key: sum(row[mode][key] for row in observations) / len(observations)
+            for key in observations[0][mode]
+        }
+        for mode in ("semantic", "lexical")
+    }
+    return {
+        "k": k,
+        "task_count": len(documents),
+        "query_count": len(observations),
+        "index_build_ms": round(build_ms, 2),
+        "aggregate": aggregate,
+        "observations": observations,
+        "boundary": "Synthetic relevance evaluation; access correctness is tested separately through canonical task retrieval.",
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--use-configured-embedder", action="store_true")
+    parser.add_argument(
+        "--cases",
+        type=Path,
+        default=REPO_ROOT / "tests/fixtures/task_retrieval/cases.json",
+    )
+    parser.add_argument("--k", type=int, default=3)
+    args = parser.parse_args()
+    if not args.use_configured_embedder:
+        parser.error(
+            "--use-configured-embedder is required to authorise embedding requests"
+        )
+    from src.backend.services.rag_service import get_rag_service
+
+    summary, model, _ = get_rag_service()._capture_embedding_runtime(
+        "task_retrieval_evaluation"
+    )
+    receipt = evaluate(json.loads(args.cases.read_text()), model, k=args.k)
+    receipt["embedding_signature"] = summary["embedding_signature"]
+    print(json.dumps(receipt, indent=2))
+
+
+if __name__ == "__main__":
+    main()

src/backend/integrations/internal_mcp/catalogue.py

@@ -35903,6 +35903,7 @@ def _task_search(**kwargs):
     try:
         result = search_tasks(
             query=kwargs.get("query"),
+            search_mode=kwargs.get("search_mode", "lexical"),
             project_concept_id=kwargs.get("project_concept_id"),
             collection_concept_id=kwargs.get("collection_concept_id"),
             status_filter=kwargs.get("status_filter") or kwargs.get("status"),
@@ -45116,6 +45117,7 @@ def _build_default_catalogue_task_and_workflow_definitions() -> List[MethodDefin
                     "project_concept_id": (str, type(None)),
                     "collection_concept_id": (str, type(None)),
                     "query": (str, type(None)),
+                    "search_mode": str,
                     "status_filter": (str, type(None)),
                     "status": (str, type(None)),
                     "statuses": (list, type(None)),
@@ -45169,7 +45171,9 @@ def _build_default_catalogue_task_and_workflow_definitions() -> List[MethodDefin
             description=(
                 "Search Von's internal task store with rich filters (status, assignee, "
                 "creator, report-to, labels, planning metadata, category/source semantics, "
-                "hierarchy, date ranges, and dependency state)."
+                "hierarchy, date ranges, and dependency state). Set search_mode='semantic' "
+                "with a natural-language query to retrieve related tasks and citation-ready "
+                "RAG context. Inspect semantic_retrieval for embedding failures and lexical fallback."
             ),
         ),
         MethodDefinition(

src/backend/server/routes/task_routes.py

@@ -806,6 +806,7 @@ def search_tasks_route() -> ResponseReturnValue:
 
         result = search_tasks(
             query=request.args.get("query"),
+            search_mode=request.args.get("search_mode", "lexical"),
             project_concept_id=request.args.get("project_concept_id"),
             collection_concept_id=request.args.get("collection_concept_id"),
             status_filter=request.args.get("status_filter"),

src/backend/services/task_management_service.py

@@ -3406,6 +3406,21 @@ def _actor_scoped_task_query_candidates(
 
 
 def search_tasks(
+    *, query: str | None = None, search_mode: str = "lexical", **filters
+) -> Dict[str, Any]:
+    """Search canonical tasks lexically or by actor-scoped semantic relevance."""
+    if any(key.startswith("_") for key in filters):
+        raise InvalidTaskDataError("Internal search parameters are not accepted")
+    if search_mode == "semantic":
+        from .task_semantic_index_service import search_semantic_tasks
+
+        return search_semantic_tasks(query=query, filters=filters)
+    if search_mode != "lexical":
+        raise InvalidTaskDataError("search_mode must be lexical or semantic")
+    return _search_tasks(query=query, **filters)
+
+
+def _search_tasks(
     *,
     query: str | None = None,
     project_concept_id: str | None = None,
@@ -3443,6 +3458,7 @@ def search_tasks(
     bulk_collection_ids: list[str] | str | None = None,
     limit: int = 50,
     offset: int = 0,
+    _semantic_query: str | None = None,
 ) -> Dict[str, Any]:
     """Search tasks with Jira-style filters over Vontology-backed task concepts."""
 
@@ -3523,6 +3539,11 @@ def search_tasks(
                 sort=[("updated_at", -1), ("created_at", -1), ("concept_id", 1)],
             )
         )
+    if _semantic_query is not None:
+        # Do not index placeholder text after a failed canonical text read.
+        query_texts_by_task = get_texts_for_concepts(
+            [doc["concept_id"] for doc in docs]
+        )
     tasks = _build_task_responses(docs, texts_by_task=query_texts_by_task)
 
     status_values: set[str] = set()
@@ -3859,8 +3880,20 @@ def search_tasks(
         bulk_collection_ids=bulk_collection_ids,
     )
     tasks = visibility_payload["tasks"]
+    semantic_diagnostics = None
+    if _semantic_query is not None:
+        from .task_semantic_index_service import assemble_task_context, rank_tasks
+
+        tasks, semantic_diagnostics = rank_tasks(tasks, _semantic_query)
     total = len(tasks)
     paged = tasks[offset : offset + limit]
+    rag_context = None
+    if semantic_diagnostics is not None:
+        rag_context = assemble_task_context(paged)
+        paged = [
+            {key: value for key, value in task.items() if key != "retrieval_text"}
+            for task in paged
+        ]
     return {
         "tasks": paged,
         "total": total,
@@ -3877,6 +3910,14 @@ def search_tasks(
         "total_is_exhaustive": bool(
             query_candidate_prefilter.get("total_is_exhaustive", True)
         ),
+        **(
+            {
+                "semantic_retrieval": semantic_diagnostics,
+                "rag_context": rag_context,
+            }
+            if semantic_diagnostics is not None
+            else {}
+        ),
     }