feat: persist evidence-bound product relations

backend/app/main.py

@@ -548,6 +548,21 @@ def _can_see_post(account: CurrentAccount, post: asyncpg.Record) -> bool:
     )
 
 
+def _can_see_product_relation_target(
+    account: CurrentAccount, relation: asyncpg.Record
+) -> bool:
+    """Apply ABAC to the normalized target's own evidence post."""
+    return source_post_visible(
+        {
+            "visibility_code": relation["target_visibility_code"],
+            "corporate_entity_id": relation["target_corporate_entity_id"],
+            "process_unit_id": relation["target_process_unit_id"],
+        },
+        account.corporate_entity_ids,
+        account.process_unit_ids,
+    )
+
+
 def _is_synthetic_demo_member(member: dict[str, Any], demo_entity_ids: set[str]) -> bool:
     """Identify one pure seed row without hiding real rows sharing its entity."""
     return bool(demo_entity_ids) and member["corporate_entity_id"] in demo_entity_ids and not bool(
@@ -1742,6 +1757,44 @@ async def read_post(
             "order by mention.mention_ordinal",
             post_id,
         )
+        # Safe SQL: the eligibility predicate is an immutable schema fragment; post id is bound.
+        product_relation_rows = await conn.fetch(  # nosemgrep: python.lang.security.audit.sqli.asyncpg-sqli.asyncpg-sqli
+            "select relation.mention_ordinal, relation.relation_type_code, "
+            "'operations_fact' as target_kind_code, "
+            "'operations_fact:' || relation.case_kind_code || ':' || relation.fact_ordinal::text as target_id, "
+            "fact.value_text as target_label, relation.evidence_text, "
+            "relation.evidence_post_id, evidence_post.visibility_code, "
+            "evidence_post.corporate_entity_id, evidence_post.process_unit_id, "
+            "target_evidence_post.visibility_code as target_visibility_code, "
+            "target_evidence_post.corporate_entity_id as target_corporate_entity_id, "
+            "target_evidence_post.process_unit_id as target_process_unit_id "
+            "from product_operations_fact_relation relation "
+            "join operations_case_fact fact on fact.post_id = relation.post_id "
+            "and fact.case_kind_code = relation.case_kind_code "
+            "and fact.fact_ordinal = relation.fact_ordinal "
+            "join source_post evidence_post on evidence_post.post_id = relation.evidence_post_id "
+            "join source_post target_evidence_post on target_evidence_post.post_id = fact.evidence_post_id "
+            "where relation.post_id = $1 and "
+            f"{SOURCE_POST_ELIGIBILITY_SQL.format(alias='evidence_post')} and "
+            f"{SOURCE_POST_ELIGIBILITY_SQL.format(alias='target_evidence_post')} "
+            "union all "
+            "select relation.mention_ordinal, relation.relation_type_code, "
+            "'project' as target_kind_code, 'project:' || relation.project_key as target_id, "
+            "project.project_name as target_label, relation.evidence_text, "
+            "relation.evidence_post_id, evidence_post.visibility_code, "
+            "evidence_post.corporate_entity_id, evidence_post.process_unit_id, "
+            "evidence_post.visibility_code as target_visibility_code, "
+            "evidence_post.corporate_entity_id as target_corporate_entity_id, "
+            "evidence_post.process_unit_id as target_process_unit_id "
+            "from product_project_relation relation "
+            "join post_project_mention project on project.post_id = relation.post_id "
+            "and project.project_key = relation.project_key "
+            "join source_post evidence_post on evidence_post.post_id = relation.evidence_post_id "
+            "where relation.post_id = $1 and "
+            f"{SOURCE_POST_ELIGIBILITY_SQL.format(alias='evidence_post')} "
+            "order by mention_ordinal, target_kind_code, target_id",
+            post_id,
+        )
         known_at = None
         if as_of_clock is not None:
             known_at = await fetch_known_at_revision(conn, post_id, as_of_clock)
@@ -1758,6 +1811,20 @@ async def read_post(
                 "product_level_code": item["product_level_code"],
                 "evidence_text": item["evidence_text"],
                 "evidence_post_id": item["evidence_post_id"],
+                "relations": [
+                    {
+                        "relation_type_code": relation["relation_type_code"],
+                        "target_kind_code": relation["target_kind_code"],
+                        "target_id": relation["target_id"],
+                        "target_label": relation["target_label"],
+                        "evidence_text": relation["evidence_text"],
+                        "evidence_post_id": relation["evidence_post_id"],
+                    }
+                    for relation in product_relation_rows
+                    if relation["mention_ordinal"] == item["mention_ordinal"]
+                    and _can_see_post(account, relation)
+                    and _can_see_product_relation_target(account, relation)
+                ],
             }
             for item in product_rows
             if _can_see_post(account, item)

backend/app/operations_case_ingestion.py

@@ -38,6 +38,10 @@ async def persist_operations_cases(
 ) -> None:
     """Atomically replace one post's normalized case analysis."""
     async with conn.transaction():
+        # Product-to-fact evidence is valid only for the exact normalized
+        # target rows it was extracted against. Removing the owning analysis
+        # first prevents unchanged replacement values from bypassing a rerun.
+        await conn.execute("delete from post_product_analysis where post_id = $1", post_id)
         await conn.execute(
             "delete from operations_case_analysis where post_id = $1", post_id
         )

backend/app/operations_dashboard.py

@@ -275,6 +275,28 @@ async def fetch_operations_dashboard(
         """,
         *args,
     )
+    product_relation_rows = await conn.fetch(
+        f"""
+        select relation.post_id, relation.case_kind_code, relation.fact_ordinal,
+               relation.relation_type_code, mention.extracted_product_name,
+               catalog.canonical_product_name, relation.evidence_text,
+               relation.evidence_post_id
+          from product_operations_fact_relation relation
+          join post_product_mention mention
+            on mention.post_id = relation.post_id
+           and mention.mention_ordinal = relation.mention_ordinal
+          left join product_catalog catalog
+            on catalog.product_catalog_id = mention.product_catalog_id
+          join source_post post on post.post_id = relation.post_id
+          join source_post evidence_post on evidence_post.post_id = relation.evidence_post_id
+         where {visible}
+           and {visible_evidence}
+           and ($5::boolean is false or relation.case_kind_code = 'external_information')
+         order by relation.post_id, relation.case_kind_code, relation.fact_ordinal,
+                  relation.mention_ordinal
+        """,
+        *args,
+    )
     missing_rows = await conn.fetch(
         f"""
         select missing.post_id, missing.case_kind_code, missing.fact_type_code
@@ -330,7 +352,23 @@ async def fetch_operations_dashboard(
         if external_only
         else await _fetch_topic_context_dashboard(conn, visible, args[:4])
     )
-    facts: dict[tuple[str, str], list[dict[str, str]]] = {}
+    product_relations: dict[tuple[str, str, int], list[dict[str, str]]] = {}
+    for row in product_relation_rows:
+        relation_key = (
+            str(row["post_id"]),
+            row["case_kind_code"],
+            int(row["fact_ordinal"]),
+        )
+        product_relations.setdefault(relation_key, []).append(
+            {
+                "relation_type_code": row["relation_type_code"],
+                "product_name": row["canonical_product_name"]
+                or row["extracted_product_name"],
+                "evidence_text": row["evidence_text"],
+                "evidence_post_id": str(row["evidence_post_id"]),
+            }
+        )
+    facts: dict[tuple[str, str], list[dict[str, Any]]] = {}
     for row in fact_rows:
         key = (str(row["post_id"]), row["case_kind_code"])
         projected_fact = {
@@ -342,6 +380,12 @@ async def fetch_operations_dashboard(
             "ontology_class_iri": str(LW.OperationsCaseFact),
             "provenance_relation_iri": PROV_WAS_DERIVED_FROM,
         }
+        related_products = product_relations.get(
+            (str(row["post_id"]), row["case_kind_code"], int(row["fact_ordinal"])),
+            [],
+        )
+        if related_products:
+            projected_fact["product_relations"] = related_products
         target_kind = row["relation_target_kind_code"]
         if target_kind in EXTERNAL_RELATION_TARGETS:
             target_label, target_class, predicate = EXTERNAL_RELATION_TARGETS[target_kind]

backend/app/post_content_worker.py

@@ -52,6 +52,7 @@
 from lineageweave.product_semantics import (
     ContextualOrchestratorProductExtractionClient,
     ProductEvidenceSource,
+    ProductRelationTarget,
     product_analysis_input_sha256,
 )
 
@@ -224,7 +225,36 @@ async def _persist_product_analysis_if_needed(
         for source in operation_sources
         if source.post_id == post_id
     )
-    input_digest = product_analysis_input_sha256(sources)
+    async with pool.acquire() as conn:
+        operation_rows = await conn.fetch(
+            "select case_kind_code, fact_ordinal, fact_type_code, value_text "
+            "from operations_case_fact where post_id = $1 "
+            "order by case_kind_code, fact_ordinal",
+            post_id,
+        )
+        project_rows = await conn.fetch(
+            "select project_key, project_name from post_project_mention "
+            "where post_id = $1 order by project_key",
+            post_id,
+        )
+    targets = tuple(
+        ProductRelationTarget(
+            f"operations_fact:{row['case_kind_code']}:{row['fact_ordinal']}",
+            "operations_fact",
+            f"{row['fact_type_code']}: {row['value_text']}",
+            (post_id, str(row["case_kind_code"]), str(row["fact_ordinal"])),
+        )
+        for row in operation_rows
+    ) + tuple(
+        ProductRelationTarget(
+            f"project:{row['project_key']}",
+            "project",
+            str(row["project_name"]),
+            (post_id, str(row["project_key"])),
+        )
+        for row in project_rows
+    )
+    input_digest = product_analysis_input_sha256(sources, targets)
     async with pool.acquire() as conn:
         already_persisted = bool(
             await conn.fetchval(
@@ -241,16 +271,19 @@ async def _persist_product_analysis_if_needed(
     client = ContextualOrchestratorProductExtractionClient(
         orchestrator_base_url, orchestrator_api_key
     )
-    mentions = await asyncio.to_thread(client.extract, sources)
+    extraction = await asyncio.to_thread(
+        client.extract, sources, targets, session_id=session_id
+    )
     async with pool.acquire() as conn:
-        resolved = await resolve_product_mentions(conn, mentions)
+        resolved = await resolve_product_mentions(conn, extraction.mentions)
         await persist_product_mentions(
             conn,
             post_id,
             source_body_digest,
             input_digest,
             session_id,
             resolved,
+            extraction,
         )
 
 

backend/app/post_summary_ingestion.py

@@ -323,6 +323,10 @@ async def _replace_summary_projection(
     """Write one atomic replacement using pre-resolved shared identities."""
     # Summary replacement owns only R&R projections. Keyman mentions remain
     # independent and are combined only by the graph read/derivation view.
+    # Product-to-project evidence belongs to the exact normalized project
+    # target set. Invalidate it before replacing those targets so a deleted
+    # relation cannot be mistaken for an already-complete analysis.
+    await conn.execute("delete from post_product_analysis where post_id = $1", post_id)
     await conn.execute(
         "delete from post_summary_person_mention where post_id = $1",
         post_id,

backend/app/product_semantic_ingestion.py

@@ -5,6 +5,7 @@
 from typing import Any, Protocol
 
 from lineageweave.product_semantics import (
+    ProductExtraction,
     ProductMention,
     ResolvedProductMention,
     normalize_product_alias,
@@ -52,6 +53,7 @@ async def persist_product_mentions(
     analysis_input_sha256: str,
     orchestrator_session_id: str,
     mentions: tuple[ResolvedProductMention, ...],
+    extraction: ProductExtraction | None = None,
 ) -> None:
     """Atomically replace one exact post's product analysis projection."""
     async with conn.transaction():
@@ -81,3 +83,43 @@ async def persist_product_mentions(
                 mention.evidence_post_id,
                 mention.evidence_input_sha256,
             )
+        if extraction is None:
+            return
+        for relation in extraction.relations:
+            if relation.target_kind_code == "operations_fact":
+                target_post_id, case_kind_code, fact_ordinal = relation.target_locator
+                if target_post_id != post_id:
+                    raise ValueError("product relation target is outside the focal post")
+                await conn.execute(
+                    "insert into product_operations_fact_relation "
+                    "(post_id, mention_ordinal, case_kind_code, fact_ordinal, "
+                    "relation_type_code, evidence_text, evidence_post_id, evidence_input_sha256) "
+                    "values ($1, $2, $3, $4, $5, $6, $7, $8)",
+                    post_id,
+                    relation.mention_ordinal,
+                    case_kind_code,
+                    int(fact_ordinal),
+                    relation.relation_type_code,
+                    relation.evidence_text,
+                    relation.evidence_post_id,
+                    relation.evidence_input_sha256,
+                )
+            elif relation.target_kind_code == "project":
+                target_post_id, project_key = relation.target_locator
+                if target_post_id != post_id:
+                    raise ValueError("product relation target is outside the focal post")
+                await conn.execute(
+                    "insert into product_project_relation "
+                    "(post_id, mention_ordinal, project_key, relation_type_code, "
+                    "evidence_text, evidence_post_id, evidence_input_sha256) "
+                    "values ($1, $2, $3, $4, $5, $6, $7)",
+                    post_id,
+                    relation.mention_ordinal,
+                    project_key,
+                    relation.relation_type_code,
+                    relation.evidence_text,
+                    relation.evidence_post_id,
+                    relation.evidence_input_sha256,
+                )
+            else:  # pragma: no cover - parser owns the closed vocabulary
+                raise ValueError("unsupported product relation target kind")

backend/tests/test_api.py

@@ -2019,6 +2019,34 @@ def test_post_detail_returns_authorized_product_evidence(
                     "Synthetic evidence", seeded_db["public_post_id"], "c" * 64,
                 ),
             )
+            cur.execute(
+                "insert into post_project_mention "
+                "(post_id, project_key, project_name, evidence_text, confidence, "
+                "ontology_iri, extraction_method) values (%s, %s, %s, %s, %s, %s, %s) "
+                "on conflict (post_id, project_key) do nothing",
+                (
+                    seeded_db["public_post_id"],
+                    "synthetic-product-project",
+                    "Synthetic Product Project",
+                    "Synthetic evidence",
+                    1,
+                    "https://contextualwisdomlab.github.io/LineageWeave/ontology#Project",
+                    "synthetic_fixture",
+                ),
+            )
+            cur.execute(
+                "insert into product_project_relation "
+                "(post_id, mention_ordinal, project_key, relation_type_code, "
+                "evidence_text, evidence_post_id, evidence_input_sha256) "
+                "values (%s, 0, %s, 'used_by_project', %s, %s, %s)",
+                (
+                    seeded_db["public_post_id"],
+                    "synthetic-product-project",
+                    "Synthetic evidence",
+                    seeded_db["public_post_id"],
+                    "c" * 64,
+                ),
+            )
             cur.execute(
                 "insert into post_product_mention "
                 "(post_id, mention_ordinal, extracted_product_name, "
@@ -2049,6 +2077,14 @@ def test_post_detail_returns_authorized_product_evidence(
         "product_level_code": "product_model",
         "evidence_text": "Synthetic evidence",
         "evidence_post_id": seeded_db["public_post_id"],
+        "relations": [{
+            "relation_type_code": "used_by_project",
+            "target_kind_code": "project",
+            "target_id": "synthetic-product-project",
+            "target_label": "Synthetic Product Project",
+            "evidence_text": "Synthetic evidence",
+            "evidence_post_id": seeded_db["public_post_id"],
+        }],
     }]