feat(dashboard): complete governed semantic evidence paths

ARCHITECTURE.md

@@ -15,17 +15,15 @@ repos in the ecosystem:
   trajectories, uncertainty-quantified estimates) is
   [TEPP](https://github.com/ContextualWisdomLab/TEPP)'s job.
 
-This is why the org-wide rule that mathematical/psychometrics computation
-layers must be Rust with GPU + CPU multithreading does not apply to this
-repo: LineageWeave does no such computation. Its heaviest per-request work
-is fusing a handful of `[0, 1]` channel scores over a bounded candidate
-window (`reconstruct.DEFAULT_CANDIDATE_WINDOW`, default 50) -- a scheduling
-and orchestration problem, not a numerical-estimation one. If a future
-version added real statistical inference (e.g. estimating thread-assignment
-uncertainty), that layer would move into TEPP rather than being built here,
-consistent with the dependency direction the ecosystem's own architecture
-docs already establish (`psychometrics-commons`'s TRD explicitly forbids a
-downstream product from reimplementing a measurement engine's model).
+ADR 0208 fixes the end state: LineageWeave retains wire validation,
+authorization, provenance persistence, and UI projection only. The current
+Python IRT/report, residual-map, similarity, graph-ranking, and fusion paths
+are explicitly inventoried migration debt rather than evidence that this
+repository owns their mathematics. They move by construct to TEPP,
+fast-mlsirm, or RankWeave after versioned Rust CPU/GPU owner contracts pass
+recovery/equivalence checks; affected product paths fail closed during each
+cutover rather than substituting a local estimate. See
+`docs/doctoring/python-mathematical-compute-boundary-audit.md`.
 
 ## Data flow
 

backend/app/analysis_run_start.py

@@ -391,6 +391,21 @@ async def _persist_tepp_result(
                         anchor["criterion_validity_status"],
                         anchor["validated_pair_count"],
                     )
+                    await conn.execute(
+                        """
+                        update lineage_channel_weight
+                           set anchor_method_code = 'tepp_lineage_criterion_v1'
+                         where estimation_run_id = $1
+                           and estimation_method_code = 'mls2plm_expected_information'
+                           and source_snapshot_sha256 = $2
+                           and knowledge_cutoff = $3
+                           and sample_pair_count = $4
+                        """,
+                        estimation_run_id,
+                        anchor["source_snapshot_sha256"],
+                        anchor_cutoff,
+                        anchor["validated_pair_count"],
+                    )
     except (asyncpg.PostgresError, TypeError, ValueError):
         return False
     return True

backend/app/global_ask_queue.py

@@ -29,6 +29,7 @@
 from fastapi import HTTPException, status
 
 from lineageweave.ask_delivery import build_ask_delivery
+from lineageweave.embedding_client import EmbeddingClient, NullEmbeddingClient
 from lineageweave.http_client import HttpClientError
 from lineageweave.observability import record_server_failure
 from lineageweave.post_chat import (
@@ -213,6 +214,7 @@ async def compute_global_ask_answer(
     process_unit_ids: set[str],
     process_scope_limited: bool,
     chat_client: PostChatClient,
+    embedding_client: EmbeddingClient | None = None,
 ) -> dict[str, Any]:
     """Assemble one complete Ask answer payload from authorized evidence.
 
@@ -243,6 +245,7 @@ def can_see(row: asyncpg.Record) -> bool:
                 process_unit_ids,
                 question=question_text,
                 today=today,
+                embedding_client=embedding_client,
             )
     except Exception as exc:
         log_internal_fault("global_ask", exc)
@@ -346,6 +349,7 @@ async def process_global_ask_job(
     *,
     job_id: str,
     chat_factory: Callable[[], PostChatClient],
+    embedding_factory: Callable[[], EmbeddingClient] = NullEmbeddingClient,
 ) -> None:
     """Claim, answer, and settle one Ask job.
 
@@ -392,6 +396,7 @@ async def process_global_ask_job(
                 process_unit_ids=process_unit_ids,
                 process_scope_limited=process_scope_limited,
                 chat_client=chat_client,
+                embedding_client=embedding_factory(),
             ),
             timeout=JOB_DEADLINE_SECONDS,
         )
@@ -505,6 +510,7 @@ async def consume_global_ask_stream_once(
     *,
     last_id: str,
     chat_factory: Callable[[], PostChatClient],
+    embedding_factory: Callable[[], EmbeddingClient] = NullEmbeddingClient,
     limiter: asyncio.Semaphore | None = None,
     tasks: set[asyncio.Task] | None = None,
 ) -> str:
@@ -524,12 +530,21 @@ async def consume_global_ask_stream_once(
             job_id = str(fields.get("global_ask_job_id", "")).strip()
             if job_id:
                 if limiter is None:
-                    await process_global_ask_job(pool, job_id=job_id, chat_factory=chat_factory)
+                    await process_global_ask_job(
+                        pool,
+                        job_id=job_id,
+                        chat_factory=chat_factory,
+                        embedding_factory=embedding_factory,
+                    )
                 else:
                     await limiter.acquire()
                     task = asyncio.create_task(
                         _process_and_release(
-                            pool, job_id=job_id, chat_factory=chat_factory, limiter=limiter
+                            pool,
+                            job_id=job_id,
+                            chat_factory=chat_factory,
+                            embedding_factory=embedding_factory,
+                            limiter=limiter,
                         )
                     )
                     if tasks is not None:
@@ -544,11 +559,17 @@ async def _process_and_release(
     *,
     job_id: str,
     chat_factory: Callable[[], PostChatClient],
+    embedding_factory: Callable[[], EmbeddingClient],
     limiter: asyncio.Semaphore,
 ) -> None:
     """Run one dispatched job and free its concurrency slot afterwards."""
     try:
-        await process_global_ask_job(pool, job_id=job_id, chat_factory=chat_factory)
+        await process_global_ask_job(
+            pool,
+            job_id=job_id,
+            chat_factory=chat_factory,
+            embedding_factory=embedding_factory,
+        )
     finally:
         limiter.release()
 
@@ -568,6 +589,7 @@ async def run_global_ask_worker(
     pool: asyncpg.Pool,
     *,
     chat_factory: Callable[[], PostChatClient],
+    embedding_factory: Callable[[], EmbeddingClient] = NullEmbeddingClient,
 ) -> None:
     """Run the at-least-once Ask consumer with periodic queued-row recovery."""
     last_id = await _stream_tail(client)
@@ -586,6 +608,7 @@ async def run_global_ask_worker(
                     pool,
                     last_id=last_id,
                     chat_factory=chat_factory,
+                    embedding_factory=embedding_factory,
                     limiter=limiter,
                     tasks=tasks,
                 )

backend/app/lineage_ingestion.py

@@ -221,14 +221,14 @@ async def load_estimated_channel_weights(
     sets: dict[str, list] = {}
     for row in all_rows:
         sets.setdefault(row["channel_set_code"], []).append(row)
-    rows = next(
-        (
-            candidate
-            for candidate in sets.values()
-            if {row["channel_code"] for row in candidate} == active_channels
-        ),
-        [],
-    )
+    matching_sets = [
+        candidate
+        for candidate in sets.values()
+        if {row["channel_code"] for row in candidate} == active_channels
+    ]
+    if len(matching_sets) != 1:
+        return None
+    rows = matching_sets[0]
     persisted = {row["channel_code"]: float(row["weight_value"]) for row in rows}
     if not persisted or set(persisted) != active_channels:
         return None

backend/app/main.py

@@ -87,6 +87,7 @@
 from lineageweave.post_summary import ContextualOrchestratorPostSummaryClient, NullPostSummaryClient
 from lineageweave.relation_verification import NullRelationVerificationClient, SearxngRelationVerificationClient
 from lineageweave.semantic_hints import customer_hint_trust, format_semantic_hints
+from lineageweave.similar_voc import ContextualOrchestratorSimilarVocAnalysisClient
 from lineageweave.ontology import LW
 from lineageweave.rankweave_client import build_rankweave_client
 from lineageweave.naruon_calendar_workspace import (
@@ -220,6 +221,8 @@
 
 _POST_READ = "post_read"
 _POST_ADMIN = "post_admin"
+_SIMILAR_VOC_PAGE_SIZE = 8
+_SIMILAR_VOC_REQUEST_TIMEOUT_SECONDS = 180.0
 
 
 @asynccontextmanager
@@ -272,6 +275,7 @@
                 chat_factory=lambda: _post_chat_client(
                     timeout=load_settings().orchestrator_answer_timeout_seconds
                 ),
+                embedding_factory=_embedding_client,
             )
         )
         app.state.global_ask_worker = global_ask_worker
@@ -482,6 +486,18 @@
     )
 
 
+def _similar_voc_client():
+    """Live semantic-pair client, or ``None`` when inference is unavailable."""
+    settings = load_settings()
+    if not (settings.orchestrator_base_url and settings.orchestrator_api_key):
+        return None
+    return ContextualOrchestratorSimilarVocAnalysisClient(
+        base_url=settings.orchestrator_base_url,
+        api_key=settings.orchestrator_api_key,
+        timeout=_SIMILAR_VOC_REQUEST_TIMEOUT_SECONDS,
+    )
+
+
 def _rankweave_client():
     """In-process RankWeave unless RANKWEAVE_DISABLED=1 (ADR 0024)."""
     return build_rankweave_client(disabled=load_settings().rankweave_disabled)
@@ -756,7 +772,11 @@
     async with pool.acquire() as conn:
         try:
             return await fetch_operations_dashboard(
-                conn, account.corporate_entity_ids, period_start, period_end
+                conn,
+                account.corporate_entity_ids,
+                account.process_unit_ids,
+                period_start,
+                period_end,
             )
         except ValueError as exc:
             raise HTTPException(status.HTTP_422_UNPROCESSABLE_CONTENT, str(exc)) from exc
@@ -1754,7 +1774,8 @@
         # Safe SQL: the eligibility predicate is an immutable schema fragment; post id is bound.
         row = await conn.fetchrow(  # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli
             """
-            select source_post.post_id, source_post.post_title, source_post.voc_type_code,
+            select source_post.post_id, source_post.post_title, source_post.post_body,
+                   source_post.voc_type_code,
                    source_post.visibility_code, source_post.corporate_entity_id,
                    source_post.process_unit_id, source_post.created_at, source_post.author_account_id,
                    source_post.source_process_unit_code, source_post.source_author_code,
@@ -1776,6 +1797,95 @@
     return row
 
 
+@app.get("/api/posts/{post_id}/similar-voc")
+async def read_similar_voc(
+    post_id: str,
+    offset: int = Query(0, ge=0),
+    account: CurrentAccount = Depends(get_current_account),
+    pool: asyncpg.Pool = Depends(get_pool),
+) -> dict[str, Any]:
+    """Return authorized, semantically adjudicated prior VOC evidence.
+
+    Persisted ``repeat_issue`` classifications narrow the candidate corpus
+    without lexical matching. contextual-orchestrator then establishes each
+    pair; event time orders the display and is not a relevance score.
+    """
+    focal = await _load_visible_post(post_id, account, pool)
+    client = _similar_voc_client()
+    if client is None:
+        raise HTTPException(
+            status.HTTP_503_SERVICE_UNAVAILABLE,
+            "similar VOC inference is unavailable; configure contextual-orchestrator and retry",
+        )
+    async with pool.acquire() as conn:
+        rows = await conn.fetch(
+            f"""
+            select post.post_id, post.post_title, post.post_body,
+                   post.visibility_code, post.corporate_entity_id, post.process_unit_id,
+                   coalesce(post.event_occurred_at, post.created_at) as occurred_at
+              from operations_case_classification classification
+              join source_post post on post.post_id = classification.post_id
+             where classification.case_kind_code = 'repeat_issue'
+               and post.post_id <> $1
+               and post.post_body <> ''
+               and (post.visibility_code = 'public'
+                    or (post.corporate_entity_id::text = any($2::text[])
+                        and (cardinality($3::text[]) = 0
+                             or post.process_unit_id::text = any($3::text[]))))
+               and {SOURCE_POST_ELIGIBILITY_SQL.format(alias='post')}
+             order by coalesce(post.event_occurred_at, post.created_at) desc, post.post_id
+            offset $4 limit $5
+            """,
+            post_id,
+            list(account.corporate_entity_ids),
+            list(account.process_unit_ids),
+            offset,
+            _SIMILAR_VOC_PAGE_SIZE + 1,
+        )
+    candidates = [row for row in rows[:_SIMILAR_VOC_PAGE_SIZE] if _can_see_post(account, row)]
+
+    async def _adjudicate(candidate: asyncpg.Record):
+        with use_llm_metadata(build_post_llm_metadata(post_id, focal)):
+            return await asyncio.to_thread(
+                client.analyze,
+                focal["post_title"],
+                focal["post_body"],
+                str(candidate["post_id"]),
+                candidate["post_title"],
+                candidate["post_body"],
+            )
+
+    try:
+        results = await asyncio.wait_for(
+            asyncio.gather(*(_adjudicate(candidate) for candidate in candidates), return_exceptions=True),
+            timeout=_SIMILAR_VOC_REQUEST_TIMEOUT_SECONDS,
+        )
+    except TimeoutError:
+        results = ()
+    items = []
+    for candidate, evidence in zip(candidates, results):
+        if evidence is None or isinstance(evidence, BaseException):
+            continue
+        items.append(
+            {
+                "post_id": evidence.candidate_post_id,
+                "post_title": candidate["post_title"],
+                "issue_summary": evidence.issue_summary,
+                "focal_evidence_text": evidence.focal_evidence_text,
+                "candidate_evidence_text": evidence.candidate_evidence_text,
+                "customer_cohort_text": evidence.customer_cohort_text,
+                "action_history": evidence.action_history,
+                "occurred_at": candidate["occurred_at"].isoformat(),
+            }
+        )
+    return {
+        "items": items,
+        "next_offset": offset + _SIMILAR_VOC_PAGE_SIZE
+        if len(rows) > _SIMILAR_VOC_PAGE_SIZE
+        else None,
+    }
+
+
 async def _load_post_semantic_hints(conn: asyncpg.Connection, post_id: str) -> str:
     """Render author, business-unit, sales-pool, and customer hints without treating them as proof."""
     rows = await conn.fetch(

backend/app/operations_case_ingestion.py

@@ -2,9 +2,9 @@
 
 from __future__ import annotations
 
-import hashlib
 from typing import Any, Protocol
 
+from backend.app.post_content_queue import source_body_sha256
 from lineageweave.operations_case_analysis import OperationsCase
 
 
@@ -23,8 +23,8 @@ async def executemany(self, query: str, args: list[tuple[object, ...]]) -> Any:
 
 
 def source_body_digest(body: str) -> str:
-    """Return the digest that binds inference to an exact source body."""
-    return hashlib.sha256(body.encode("utf-8")).hexdigest()
+    """Return the digest that binds inference to an exact focal source body."""
+    return source_body_sha256(body)
 
 
 async def persist_operations_cases(
@@ -40,22 +40,24 @@ async def persist_operations_cases(
         await conn.execute(
             "insert into operations_case_analysis (post_id, source_body_sha256, orchestrator_session_id) values ($1, $2, $3)",
             post_id,
-            source_body_digest(source_body),
+            source_body_sha256(source_body),
             orchestrator_session_id,
         )
         for case in cases:
             await conn.execute(
-                "insert into operations_case_classification (post_id, case_kind_code, summary_text, evidence_text) values ($1, $2, $3, $4)",
+                "insert into operations_case_classification (post_id, case_kind_code, summary_text, evidence_text, evidence_post_id, evidence_input_sha256) values ($1, $2, $3, $4, $5, $6)",
                 post_id,
                 case.case_kind_code,
                 case.summary_text,
                 case.evidence_text,
+                case.evidence_post_id,
+                case.evidence_input_sha256,
             )
             if case.facts:
                 await conn.executemany(
-                    "insert into operations_case_fact (post_id, case_kind_code, fact_ordinal, fact_type_code, value_text, evidence_text) values ($1, $2, $3, $4, $5, $6)",
+                    "insert into operations_case_fact (post_id, case_kind_code, fact_ordinal, fact_type_code, value_text, evidence_text, evidence_post_id, evidence_input_sha256) values ($1, $2, $3, $4, $5, $6, $7, $8)",
                     [
-                        (post_id, case.case_kind_code, ordinal, fact.fact_type_code, fact.value_text, fact.evidence_text)
+                        (post_id, case.case_kind_code, ordinal, fact.fact_type_code, fact.value_text, fact.evidence_text, fact.evidence_post_id, fact.evidence_input_sha256)
                         for ordinal, fact in enumerate(case.facts)
                     ],
                 )