diff --git a/docs/design_index.md b/docs/design_index.md index 0dbd67db..8ac10015 100644 --- a/docs/design_index.md +++ b/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) | diff --git a/docs/engineering/task_semantic_retrieval.md b/docs/engineering/task_semantic_retrieval.md new file mode 100644 index 00000000..074b075b --- /dev/null +++ b/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. diff --git a/scripts/evaluate_task_semantic_retrieval.py b/scripts/evaluate_task_semantic_retrieval.py new file mode 100644 index 00000000..6c66bc51 --- /dev/null +++ b/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() diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py index 0086f6e1..197e600a 100644 --- a/src/backend/integrations/internal_mcp/catalogue.py +++ b/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( diff --git a/src/backend/server/routes/task_routes.py b/src/backend/server/routes/task_routes.py index 1cc1fc22..fba72900 100644 --- a/src/backend/server/routes/task_routes.py +++ b/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"), diff --git a/src/backend/services/task_management_service.py b/src/backend/services/task_management_service.py index 6633bc85..9c42076e 100644 --- a/src/backend/services/task_management_service.py +++ b/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 {} + ), } diff --git a/src/backend/services/task_semantic_index_service.py b/src/backend/services/task_semantic_index_service.py new file mode 100644 index 00000000..c3eb8b95 --- /dev/null +++ b/src/backend/services/task_semantic_index_service.py @@ -0,0 +1,352 @@ +"""Semantic task retrieval over canonical, actor-visible snapshots. + +The SQLite index is disposable derived storage, never an authority for task +text or visibility. It stores only vectors and revision hashes. Every search +reads and filters canonical tasks before consulting the index. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import math +import os +from pathlib import Path +import sqlite3 +import time +from typing import Any, Mapping + +from ..security import access_control +from ..db.repositories.concepts_repository import ConceptsRepository +from .rag_service import get_rag_service + +logger = logging.getLogger(__name__) +DOCUMENT_VERSION = "task_document.v1" +DOCUMENT_FIELDS = ( + "task_concept_id", + "title", + "description", + "status", + "priority", + "assignee_concept_id", + "created_by_concept_id", + "organisation_concept_id", + "project_concept_id", + "collection_concept_ids", + "labels", + "components", + "task_source_id", + "external_references", + "reference_code", + "created_at", + "updated_at", + "start_date", + "due_date", +) +BATCH_SIZE = 32 + + +def build_task_document(task: Mapping[str, Any]) -> dict[str, Any]: + """Preserve task identity, context and provenance without promoting text to policy.""" + if not task.get("task_concept_id"): + raise ValueError("Task identity is required") + fields = {key: task.get(key) for key in DOCUMENT_FIELDS} + text = json.dumps(fields, sort_keys=True, ensure_ascii=False, default=str) + revision = hashlib.sha256((DOCUMENT_VERSION + text).encode()).hexdigest() + return { + "id": task["task_concept_id"], + "text": text, + "revision": revision, + "schema_version": DOCUMENT_VERSION, + "source": "canonical_task_service", + } + + +def _scope() -> str: + # Scope is derived from the server actor, never query filter identifiers. + return json.dumps( + [ + access_control.get_effective_user_concept_id(), + access_control.get_effective_organisation_concept_id(), + ] + ) + + +def _index_path() -> Path: + return Path( + os.environ.get( + "VON_TASK_SEMANTIC_INDEX_PATH", "data/task_semantic_index.sqlite3" + ) + ) + + +def _connect() -> sqlite3.Connection: + path = _index_path() + path.parent.mkdir(parents=True, exist_ok=True) + # Vectors are private derived data even though raw task text is not stored. + fd = os.open(path, os.O_CREAT | os.O_RDWR, 0o600) + os.close(fd) + connection = sqlite3.connect(path, timeout=10) + try: + connection.execute( + "CREATE TABLE IF NOT EXISTS task_vectors (" + "scope TEXT NOT NULL, task_id TEXT NOT NULL, revision TEXT NOT NULL, " + "model TEXT NOT NULL, vector TEXT NOT NULL, touched REAL NOT NULL, " + "PRIMARY KEY(scope, task_id))" + ) + except Exception: + connection.close() + raise + return connection + + +def clear_task_semantic_index() -> int: + """Discard the current trusted actor's derived cache; next search rebuilds it.""" + connection = _connect() + try: + with connection: + return connection.execute( + "DELETE FROM task_vectors WHERE scope = ?", (_scope(),) + ).rowcount + finally: + connection.close() + + +def _normalise_vector(value: Any) -> list[float]: + vector = [float(component) for component in value] + if not vector or not all(math.isfinite(component) for component in vector): + raise ValueError("Invalid embedding") + magnitude = math.sqrt(sum(component * component for component in vector)) + if not math.isfinite(magnitude) or magnitude == 0: + raise ValueError("Invalid embedding norm") + return [component / magnitude for component in vector] + + +def rank_tasks( + tasks: list[dict[str, Any]], query: str +) -> tuple[list[dict[str, Any]], dict[str, Any]]: + """Index current filtered tasks and rank by cosine similarity. + + No cached text or cached candidate IDs may enter the result. Replacing or + pruning a concurrent cache entry affects cache efficiency only: each search + retains the vectors matching its own canonical snapshot in memory. + """ + started = time.monotonic() + documents = [build_task_document(task) for task in tasks] + diagnostics: dict[str, Any] = { + "status": "ready", + "document_version": DOCUMENT_VERSION, + "candidates": len(tasks), + "cache_hits": 0, + "embedded": 0, + "ranking": "cosine", + "source": "canonical_task_service", + } + if not tasks: + clear_task_semantic_index() + diagnostics["elapsed_ms"] = round((time.monotonic() - started) * 1000, 2) + return [], diagnostics + + service = get_rag_service() + # Pin the actual embedder for this operation; a runtime setting change must + # not mix vector spaces between the query and successive document batches. + capture = getattr(service, "_capture_embedding_runtime", None) + if not callable(capture): + raise RuntimeError("Task indexing requires a versioned embedding runtime") + summary, model, _ = capture("task_semantic_index") + model_key = json.dumps(summary["embedding_signature"], sort_keys=True) + diagnostics["embedding_version"] = hashlib.sha256(model_key.encode()).hexdigest() + query_vector = _normalise_vector(model.get_query_embedding(query)) + vectors: dict[str, list[float]] = {} + scope = _scope() + connection = _connect() + try: + cached = { + row[0]: row[1:] + for row in connection.execute( + "SELECT task_id, revision, model, vector FROM task_vectors WHERE scope = ?", + (scope,), + ) + } + pending = [] + for document in documents: + row = cached.get(document["id"]) + if row and row[0] == document["revision"] and row[1] == model_key: + try: + vector = _normalise_vector(json.loads(row[2])) + if len(vector) != len(query_vector): + raise ValueError("Embedding dimension changed") + vectors[document["id"]] = vector + diagnostics["cache_hits"] += 1 + continue + except (ValueError, TypeError): + pass # Corrupt individual rows are recoverable cache misses. + pending.append(document) + for start in range(0, len(pending), BATCH_SIZE): + batch = pending[start : start + BATCH_SIZE] + embeddings = model.get_text_embedding_batch([doc["text"] for doc in batch]) + if len(embeddings) != len(batch): + raise ValueError("Embedding batch incomplete") + with connection: + for document, embedding in zip(batch, embeddings): + vector = _normalise_vector(embedding) + if len(vector) != len(query_vector): + raise ValueError("Embedding dimension mismatch") + vectors[document["id"]] = vector + connection.execute( + "INSERT OR REPLACE INTO task_vectors VALUES (?, ?, ?, ?, ?, ?)", + ( + scope, + document["id"], + document["revision"], + model_key, + json.dumps(vector), + time.time(), + ), + ) + diagnostics["embedded"] += 1 + # The cache represents the current filtered snapshot. Filter changes may + # evict reusable rows, but deleted/revoked tasks cannot remain candidates. + access_control.invalidate_current_access_evaluator() + if ConceptsRepository.collection() is None: + raise RuntimeError( + "Canonical task storage unavailable during access recheck" + ) + current_ids = set() + candidate_ids = list(vectors) + for start in range(0, len(candidate_ids), 200): + current_ids.update( + row["concept_id"] + for row in ConceptsRepository.find( + {"concept_id": {"$in": candidate_ids[start : start + 200]}}, + projection={"concept_id": 1, "_id": 0}, + ) + ) + diagnostics["access_recheck_excluded"] = len(vectors) - len(current_ids) + with connection: + connection.executemany( + "DELETE FROM task_vectors WHERE scope = ? AND task_id = ?", + [ + (scope, task_id) + for task_id in set(cached) | set(vectors) + if task_id not in current_ids + ], + ) + results = [] + for task, document in zip(tasks, documents): + if document["id"] not in current_ids: + continue + score = sum(a * b for a, b in zip(query_vector, vectors[document["id"]])) + results.append( + { + **task, + "semantic_score": round(score, 8), + "retrieval_text": document["text"], + "indexed_revision": document["revision"], + } + ) + results.sort( + key=lambda task: (-task["semantic_score"], task["task_concept_id"]) + ) + diagnostics["elapsed_ms"] = round((time.monotonic() - started) * 1000, 2) + logger.info( + "Task semantic retrieval candidates=%d embedded=%d cache_hits=%d elapsed_ms=%s", + len(tasks), + diagnostics["embedded"], + diagnostics["cache_hits"], + diagnostics["elapsed_ms"], + ) + return results, diagnostics + finally: + connection.close() + + +def search_semantic_tasks( + *, query: str | None, filters: Mapping[str, Any] +) -> dict[str, Any]: + from .task_management_service import InvalidTaskDataError, _search_tasks + + if not isinstance(query, str) or not query.strip(): + raise InvalidTaskDataError("Semantic search requires a non-empty query") + if any(key.startswith("_") for key in filters): + raise InvalidTaskDataError("Internal search parameters are not accepted") + actor, source = access_control.get_effective_user_concept_id_with_source() + if access_control.is_bypass_enabled() or ( + actor + and source + not in { + access_control.AUTHENTICATED_SESSION_ACTOR_SOURCE, + access_control.TRUSTED_IN_PROCESS_ACTOR_SOURCE, + } + ): + raise InvalidTaskDataError("Semantic task search requires trusted actor scope") + organisation = ( + access_control.get_effective_organisation_concept_id() if actor else None + ) + with ( + access_control.override_current_actor(actor, organisation), + access_control.force_access_control_enforcement(), + ): + try: + if ConceptsRepository.collection() is None: + raise RuntimeError("Canonical task storage unavailable") + return _search_tasks(query=None, _semantic_query=query.strip(), **filters) + except InvalidTaskDataError: + raise + except Exception as exc: + # Never log provider exception bodies: they may echo private input. + logger.warning( + "Task semantic retrieval unavailable (%s)", type(exc).__name__ + ) + result = _search_tasks(query=query, **filters) + result["semantic_retrieval"] = { + "status": "degraded", + "ranking": "lexical", + "error_code": "task_semantic_index_unavailable", + "recovery": "Retry semantic search; an operator can rebuild the disposable local index.", + } + return result + + +def reindex_semantic_tasks() -> dict[str, Any]: + """Rebuild all currently visible tasks through the normal canonical read path. + + Internal maintenance entry point for a trusted actor-bound operator. No + canonical task or shared vector namespace is mutated. + """ + if ( + not access_control.get_effective_user_concept_id() + or access_control.is_bypass_enabled() + ): + raise ValueError("Reindexing requires a bound actor") + clear_task_semantic_index() + result = search_semantic_tasks( + query="task index recovery", filters={"limit": 1, "bulk_visibility": "include"} + ) + return dict(result["semantic_retrieval"]) + + +def assemble_task_context( + tasks: list[dict[str, Any]], *, character_budget: int = 12000 +) -> list[dict[str, Any]]: + """Bound RAG context while preserving exact citation identity and truncation.""" + context = [] + remaining = max(0, character_budget) + for task in tasks: + if remaining == 0: + break + text = task["retrieval_text"] + excerpt = text[:remaining] + context.append( + { + "citation": task["task_concept_id"], + "text": excerpt, + "score": task["semantic_score"], + "source": "canonical_task_service", + "truncated": len(excerpt) < len(text), + "content_role": "retrieved_data", + } + ) + remaining -= len(excerpt) + return context diff --git a/tests/backend/test_task_semantic_index_service.py b/tests/backend/test_task_semantic_index_service.py new file mode 100644 index 00000000..4aeeda13 --- /dev/null +++ b/tests/backend/test_task_semantic_index_service.py @@ -0,0 +1,352 @@ +"""Isolated end-to-end task retrieval; no model requests or live databases.""" + +import json +import sqlite3 +from pathlib import Path + +import mongomock +import pytest +from flask import Flask + +from src.backend.security import access_control as access +from src.backend.services import task_management_service as tasks +from src.backend.services import task_semantic_index_service as semantic + + +class FixtureEmbedder: + """Known vector geometry tests retrieval plumbing, not neural model quality.""" + + def __init__(self): + self.texts = [] + self.fail = False + self.on_embed = None + + def get_query_embedding(self, text): + if self.fail: + raise RuntimeError("private provider input must not be logged") + return [1.0, 0.0] + + def get_text_embedding_batch(self, texts): + self.texts.extend(texts) + if self.on_embed: + self.on_embed() + return [[1.0, 0.0] if "airfare" in text else [0.0, 1.0] for text in texts] + + +@pytest.fixture +def environment(monkeypatch, tmp_path): + collection = mongomock.MongoClient().db.concepts + monkeypatch.setattr(tasks.ConceptsRepository, "collection", lambda: collection) + monkeypatch.setattr(access, "get_concepts_collection", lambda: collection) + rows = {} + monkeypatch.setattr( + tasks, + "get_texts_for_concepts", + lambda ids, **kw: {key: rows.get(key, []) for key in ids}, + ) + model = FixtureEmbedder() + signature = {"provider": "fixture", "model": "fixture-v1"} + + class Runtime: + def _capture_embedding_runtime(self, namespace): + return {"embedding_signature": dict(signature)}, model, {} + + monkeypatch.setattr(semantic, "get_rag_service", Runtime) + monkeypatch.setenv("VON_TASK_SEMANTIC_INDEX_PATH", str(tmp_path / "index.sqlite3")) + + def add( + key, title, description="", actor=None, org=None, status="pending", project=None + ): + identity = "#V#" + key + relationships = {"is_an_instance_of": [tasks.TASK_SPECIFICATION_TYPE_ID]} + if actor: + relationships["#V#specific_to_user"] = [actor] + if org: + relationships["#V#specific_to_organisation"] = [org] + collection.insert_one( + { + "concept_id": identity, + "relationships": relationships, + "metadata": {"project_concept_id": project}, + } + ) + rows[identity] = [ + {"predicate": "#V#hasName", "text": title}, + {"predicate": "#V#hasDescription", "text": description}, + {"predicate": "#V#hasTaskStatus", "text": status}, + ] + return identity + + return collection, rows, model, signature, add + + +def search(**kwargs): + return tasks.search_tasks( + query="organise transport", search_mode="semantic", **kwargs + ) + + +def test_semantic_paraphrase_filters_pagination_and_context(environment): + _, _, model, _, add = environment + add("minutes", "Summarise lab meeting", project="#V#project") + winner = add( + "travel", "Arrange conference flights", "Book airfare", project="#V#project" + ) + add( + "done", + "Arrange old travel", + "airfare", + status="completed", + project="#V#project", + ) + add("other", "Other trip", "airfare", project="#V#other") + with access.override_current_actor("#V#alice"): + result = search(project_concept_id="#V#project", statuses=["pending"], limit=1) + assert result["tasks"][0]["task_concept_id"] == winner + assert result["total"] == 2 + assert result["semantic_retrieval"]["embedded"] == 2 + assert len(model.texts) == 2 + assert result["rag_context"][0]["citation"] == winner + assert result["rag_context"][0]["content_role"] == "retrieved_data" + page = search( + project_concept_id="#V#project", statuses=["pending"], limit=1, offset=1 + ) + assert page["tasks"][0]["task_concept_id"] == "#V#minutes" + assert page["semantic_retrieval"]["cache_hits"] == 2 + assert tasks.search_tasks(query="organise transport")["count"] == 0 + + +def test_revision_change_refreshes_without_timestamp_change(environment): + _, rows, model, _, add = environment + identity = add("travel", "Travel", "Book airfare") + with access.override_current_actor("#V#alice"): + initial = search() + assert search()["semantic_retrieval"]["cache_hits"] == 1 + rows[identity][1]["text"] = "Cancelled; retain meeting notes instead" + updated = search() + assert updated["semantic_retrieval"]["embedded"] == 1 + assert ( + initial["tasks"][0]["indexed_revision"] + != updated["tasks"][0]["indexed_revision"] + ) + assert "Cancelled" in updated["rag_context"][0]["text"] + assert len(model.texts) == 2 + + +def test_model_version_and_corrupt_vector_recover(environment): + _, _, _, signature, add = environment + add("travel", "Travel", "airfare") + with access.override_current_actor("#V#alice"): + search() + signature["model"] = "fixture-v2" + assert search()["semantic_retrieval"]["embedded"] == 1 + with sqlite3.connect(semantic._index_path()) as connection: + connection.execute("UPDATE task_vectors SET vector = 'invalid json'") + assert search()["semantic_retrieval"]["embedded"] == 1 + receipt = semantic.reindex_semantic_tasks() + assert receipt["embedded"] == 1 + assert receipt["cache_hits"] == 0 + + +def test_actor_organisation_and_anonymous_boundaries(environment): + _, _, model, _, add = environment + add("private", "Private travel", "airfare", actor="#V#alice") + add("org", "Organisation travel", "airfare", org="#V#org") + add("public", "Public minutes") + with access.override_current_actor("#V#alice", "#V#org"): + assert search()["count"] == 3 + with access.override_current_actor("#V#bob", "#V#other"): + result = search(assignee_concept_id="#V#alice") + assert result["count"] == 0 # Payload filter never supplies authority. + result = search() + assert [task["task_concept_id"] for task in result["tasks"]] == ["#V#public"] + assert result["semantic_retrieval"]["cache_hits"] == 0 + # An anonymous caller cannot acquire organisation visibility from an org selection. + with access.override_current_actor(None, "#V#org"): + assert search()["count"] == 1 + assert sum("Private travel" in text for text in model.texts) == 1 + + +@pytest.mark.parametrize("delete", [False, True]) +def test_permission_revocation_and_deletion_remove_warm_results(environment, delete): + collection, _, _, _, add = environment + identity = add("travel", "Private travel", "airfare", actor="#V#alice") + with access.override_current_actor("#V#alice"): + assert search()["count"] == 1 + if delete: + collection.delete_one({"concept_id": identity}) + else: + collection.update_one( + {"concept_id": identity}, + {"$set": {"relationships.#V#specific_to_user": ["#V#bob"]}}, + ) + result = search() + assert result["count"] == 0 + assert result["rag_context"] == [] + with sqlite3.connect(semantic._index_path()) as connection: + assert ( + connection.execute("SELECT count(*) FROM task_vectors").fetchone()[0] + == 0 + ) + + +def test_revoked_during_embedding_is_excluded(environment): + collection, _, model, _, add = environment + identity = add("travel", "Private travel", "airfare", actor="#V#alice") + model.on_embed = lambda: collection.delete_one({"concept_id": identity}) + with access.override_current_actor("#V#alice"): + result = search() + assert result["count"] == 0 + assert result["semantic_retrieval"]["access_recheck_excluded"] == 1 + + +def test_provider_failure_returns_explicit_lexical_fallback_without_private_error( + environment, caplog +): + _, _, model, _, add = environment + add("travel", "organise transport", "airfare") + model.fail = True + with access.override_current_actor("#V#alice"): + result = search() + assert result["semantic_retrieval"]["status"] == "degraded" + assert result["count"] == 1 + assert "private provider input" not in caplog.text + assert "rag_context" not in result + + +@pytest.mark.parametrize("embedding", [[0, 0], [float("nan"), 1], [1], []]) +def test_invalid_embedding_does_not_produce_semantic_success(environment, embedding): + _, _, model, _, add = environment + add("travel", "Travel", "airfare") + model.get_text_embedding_batch = lambda texts: [embedding for _ in texts] + with access.override_current_actor("#V#alice"): + assert search()["semantic_retrieval"]["status"] == "degraded" + + +def test_rest_path_uses_server_session_and_current_canonical_tasks(environment): + from src.backend.server.routes.task_routes import task_bp + + _, _, _, _, add = environment + add("alice", "Private flights", "airfare", actor="#V#alice") + add("bob", "Other flights", "airfare", actor="#V#bob") + app = Flask(__name__) + app.secret_key = "fixture-only" + app.register_blueprint(task_bp, url_prefix="/api/tasks") + client = app.test_client() + with client.session_transaction() as session: + session["user_concept_id"] = "#V#alice" + response = client.get( + "/api/tasks/search", + query_string={"query": "organise transport", "search_mode": "semantic"}, + ) + assert response.status_code == 200 + payload = response.get_json() + assert payload["semantic_retrieval"]["status"] == "ready" + assert [task["task_concept_id"] for task in payload["tasks"]] == ["#V#alice"] + + +def test_context_budget_and_document_provenance(): + task = { + "task_concept_id": "#V#travel", + "description": "x" * 20000, + "external_references": {"jira": {"external_id": "ABC-1"}}, + } + document = semantic.build_task_document(task) + assert "ABC-1" in document["text"] + context = semantic.assemble_task_context( + [{**task, "retrieval_text": document["text"], "semantic_score": 1}] + ) + assert len(context[0]["text"]) == 12000 + assert context[0]["truncated"] is True + + +def test_rejects_invalid_modes_internal_parameters_and_bypass(environment): + with pytest.raises(tasks.InvalidTaskDataError): + tasks.search_tasks(search_mode="invented") + with pytest.raises(tasks.InvalidTaskDataError): + tasks.search_tasks(_semantic_query="private") + with pytest.raises(tasks.InvalidTaskDataError): + tasks.search_tasks(search_mode="semantic", query=" ") + with access.bypass_access_control(), pytest.raises(tasks.InvalidTaskDataError): + search() + + +def test_evaluation_fixture_has_explicit_relevance_judgements(): + fixture = json.loads( + (Path(__file__).parents[1] / "fixtures/task_retrieval/cases.json").read_text() + ) + ids = {task["id"] for task in fixture["tasks"]} + assert all(set(case["relevant"]) <= ids for case in fixture["queries"]) + + +def test_evaluation_measures_semantic_and_lexical_baselines(): + from scripts.evaluate_task_semantic_retrieval import evaluate + + cases = { + "tasks": [ + {"id": "travel", "title": "Arrange flights", "description": "airfare"}, + {"id": "minutes", "title": "Record discussion", "description": "meeting"}, + ], + "queries": [{"query": "organise transport", "relevant": ["travel"]}], + } + receipt = evaluate(cases, FixtureEmbedder(), k=1) + assert receipt["aggregate"]["semantic"] == { + "recall_at_k": 1, + "reciprocal_rank": 1, + "ndcg_at_k": 1, + } + assert receipt["aggregate"]["lexical"]["recall_at_k"] == 0 + assert receipt["index_build_ms"] >= 0 + + +def test_mcp_task_search_returns_citation_context(environment): + from src.backend.integrations.internal_mcp.catalogue import _task_search + + _, _, _, _, add = environment + identity = add("travel", "Arrange flights", "airfare", actor="#V#alice") + with access.override_current_actor("#V#alice"): + result = _task_search(query="organise transport", search_mode="semantic") + assert result["success"] is True + assert result["rag_context"][0]["citation"] == identity + + +def test_partial_batch_failure_preserves_reusable_batches(environment): + _, _, model, _, add = environment + for index in range(40): + add(f"task_{index}", "Arrange flights", "airfare") + original = model.get_text_embedding_batch + calls = 0 + + def fail_second_batch(texts): + nonlocal calls + calls += 1 + if calls == 2: + raise RuntimeError("temporary embedding outage") + return original(texts) + + model.get_text_embedding_batch = fail_second_batch + with access.override_current_actor("#V#alice"): + assert search()["semantic_retrieval"]["status"] == "degraded" + model.get_text_embedding_batch = original + receipt = search()["semantic_retrieval"] + assert receipt["cache_hits"] == 32 + assert receipt["embedded"] == 8 + + +def test_index_file_contains_no_task_text_and_uses_private_mode(environment): + _, _, _, _, add = environment + add("travel", "Private conference title", "Private airfare details") + with access.override_current_actor("#V#alice"): + search() + path = semantic._index_path() + assert b"Private conference title" not in path.read_bytes() + assert b"Private airfare details" not in path.read_bytes() + assert path.stat().st_mode & 0o777 == 0o600 + + +def test_missing_storage_is_degraded_not_authoritative_empty(environment, monkeypatch): + monkeypatch.setattr(tasks.ConceptsRepository, "collection", lambda: None) + with access.override_current_actor("#V#alice"): + result = search() + assert result["semantic_retrieval"]["status"] == "degraded" + assert not semantic._index_path().exists() diff --git a/tests/fixtures/task_retrieval/cases.json b/tests/fixtures/task_retrieval/cases.json new file mode 100644 index 00000000..5746df06 --- /dev/null +++ b/tests/fixtures/task_retrieval/cases.json @@ -0,0 +1,15 @@ +{ + "description": "Synthetic ordinary research-administration queries; relevance judgements are fixtures, not measured model quality.", + "tasks": [ + {"id": "travel", "title": "Arrange conference flights", "description": "Book return airfare and accommodation for the research symposium."}, + {"id": "minutes", "title": "Summarise lab meeting", "description": "Record decisions and action items from the weekly research discussion."}, + {"id": "review", "title": "Review experiment results", "description": "Compare replication measurements with the published findings."}, + {"id": "invoice", "title": "Reconcile supplier invoice", "description": "Check the equipment purchase receipt against the laboratory budget."} + ], + "queries": [ + {"query": "organise transport to the academic event", "relevant": ["travel"]}, + {"query": "what did the team agree to do", "relevant": ["minutes"]}, + {"query": "check whether the study reproduced", "relevant": ["review"]}, + {"query": "verify the bill for lab supplies", "relevant": ["invoice"]} + ] +}