feat: persist evidence-bound occupational constructs

CHANGELOG.md

@@ -13,6 +13,10 @@ All notable changes to this project are documented here. Format follows
   behaviors, and FJA worker functions distinct. Record-to-construct links
   require a reified evidence span and PROV-O derivation/time; unsupported
   DPT-to-psychology crosswalks and local scores remain unavailable (ADR 0248).
+- Versioned occupational construct vocabularies and semantic-unit assertions
+  now persist in normalized tables. Database and application validation require
+  same-Post verbatim evidence, and authorized Post detail exposes provenance
+  without internal identifiers or numerical scores (ADR 0249).
 
 - The DOT/FJA Data/People/Things worker-function taxonomy is now published
   in the canonical ontology: all 24 worker functions carry the official

backend/app/main.py

@@ -141,6 +141,9 @@
     post_content_is_complete,
     publish_post_content_event,
 )
+from backend.app.occupational_construct_ingestion import (
+    load_occupational_construct_assertions,
+)
 from backend.app.post_content_worker import run_post_content_worker
 from backend.app.post_eligibility import SOURCE_POST_ELIGIBILITY_SQL, source_post_visible
 from backend.app.post_evaluation_ingestion import (
@@ -1654,13 +1657,17 @@ async def read_post(
         project_evidence = await _load_project_evidence(
             conn, post_id, row["source_project_code"], row["source_project_name"]
         )
+        occupational_construct_assertions = await load_occupational_construct_assertions(
+            conn, 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)
     payload = {
         **_serialize_post(row, labels),
         "post_body": row["post_body"],
         "project_evidence": project_evidence,
+        "occupational_construct_assertions": occupational_construct_assertions,
     }
     if known_at is not None:
         payload["known_at"] = known_at

backend/app/occupational_construct_ingestion.py

@@ -0,0 +1,207 @@
+"""Persist evidence-bound occupational constructs under ADR 0249."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+from urllib.parse import urlsplit
+
+
+CONSTRUCT_FAMILIES = frozenset(
+    {
+        "cognitive_ability",
+        "work_style",
+        "work_activity",
+        "affective_reaction",
+        "performance_behavior",
+    }
+)
+TRUTH_STATUS_CODES = frozenset(
+    {
+        "truth_authoritative",
+        "truth_observed",
+        "truth_inferred",
+        "truth_proposed",
+        "truth_superseded",
+        "truth_rejected",
+    }
+)
+
+
+def _https_iri(value: str, field_name: str) -> str:
+    """Return one normalized HTTPS IRI or reject untrusted input."""
+    normalized = value.strip()
+    parsed = urlsplit(normalized)
+    if parsed.scheme != "https" or not parsed.netloc or parsed.username or parsed.password:
+        raise ValueError(f"{field_name} must be an absolute credential-free HTTPS IRI")
+    return normalized
+
+
+@dataclass(frozen=True)
+class ConstructVocabulary:
+    """One immutable external vocabulary release."""
+
+    vocabulary_iri: str
+    version_label: str
+    license_iri: str
+    attribution_text: str
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "vocabulary_iri", _https_iri(self.vocabulary_iri, "vocabulary_iri"))
+        object.__setattr__(self, "license_iri", _https_iri(self.license_iri, "license_iri"))
+        if not self.version_label.strip() or not self.attribution_text.strip():
+            raise ValueError("vocabulary version and attribution must be non-empty")
+
+
+@dataclass(frozen=True)
+class OccupationalConstruct:
+    """One versioned external occupational construct."""
+
+    vocabulary: ConstructVocabulary
+    construct_iri: str
+    family_code: str
+    preferred_label: str
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "construct_iri", _https_iri(self.construct_iri, "construct_iri"))
+        if self.family_code not in CONSTRUCT_FAMILIES:
+            raise ValueError(f"unsupported occupational construct family {self.family_code!r}")
+        if not self.preferred_label.strip():
+            raise ValueError("construct preferred label must be non-empty")
+
+
+@dataclass(frozen=True)
+class OccupationalConstructAssertion:
+    """One construct assertion bound to a verbatim semantic-unit span."""
+
+    post_content_unit_id: str
+    unit_text: str
+    construct: OccupationalConstruct
+    evidence_text: str
+    truth_status_code: str
+    extraction_method: str
+
+    def __post_init__(self) -> None:
+        for name, value in (
+            ("post_content_unit_id", self.post_content_unit_id),
+            ("unit_text", self.unit_text),
+            ("evidence_text", self.evidence_text),
+            ("truth_status_code", self.truth_status_code),
+            ("extraction_method", self.extraction_method),
+        ):
+            if not value.strip():
+                raise ValueError(f"{name} must be non-empty")
+        if self.evidence_text not in self.unit_text:
+            raise ValueError("construct evidence must be a verbatim semantic-unit span")
+        if self.truth_status_code not in TRUTH_STATUS_CODES:
+            raise ValueError(f"unsupported ontology truth status {self.truth_status_code!r}")
+
+
+async def persist_occupational_construct_assertions(
+    conn: Any,
+    post_id: str,
+    orchestrator_session_id: str,
+    assertions: tuple[OccupationalConstructAssertion, ...],
+) -> None:
+    """Atomically replace one Post's construct assertions and shared registry rows."""
+    if not post_id.strip() or not orchestrator_session_id.strip():
+        raise ValueError("post and orchestrator session identifiers must be non-empty")
+    async with conn.transaction():
+        await conn.execute(
+            "delete from post_occupational_construct_assertion where post_id = $1",
+            post_id,
+        )
+        for assertion in assertions:
+            vocabulary = assertion.construct.vocabulary
+            vocabulary_id = await conn.fetchval(
+                """
+                insert into occupational_construct_vocabulary
+                    (vocabulary_iri, version_label, license_iri, attribution_text)
+                values ($1, $2, $3, $4)
+                on conflict (vocabulary_iri, version_label) do update set
+                    vocabulary_iri = excluded.vocabulary_iri
+                where occupational_construct_vocabulary.license_iri = excluded.license_iri
+                  and occupational_construct_vocabulary.attribution_text = excluded.attribution_text
+                returning vocabulary_id
+                """,
+                vocabulary.vocabulary_iri,
+                vocabulary.version_label,
+                vocabulary.license_iri,
+                vocabulary.attribution_text,
+            )
+            if vocabulary_id is None:
+                raise ValueError("vocabulary metadata conflicts with its immutable version")
+            construct_id = await conn.fetchval(
+                """
+                insert into occupational_construct
+                    (vocabulary_id, construct_iri, construct_family_code, preferred_label)
+                values ($1, $2, $3, $4)
+                on conflict (vocabulary_id, construct_iri) do update set
+                    construct_iri = excluded.construct_iri
+                where occupational_construct.construct_family_code = excluded.construct_family_code
+                  and occupational_construct.preferred_label = excluded.preferred_label
+                returning construct_id
+                """,
+                vocabulary_id,
+                assertion.construct.construct_iri,
+                assertion.construct.family_code,
+                assertion.construct.preferred_label,
+            )
+            if construct_id is None:
+                raise ValueError("construct metadata conflicts with its immutable vocabulary version")
+            await conn.execute(
+                """
+                insert into post_occupational_construct_assertion
+                    (post_id, post_content_unit_id, construct_id, evidence_text,
+                     truth_status_code, extraction_method, orchestrator_session_id)
+                values ($1, $2, $3, $4, $5, $6, $7)
+                """,
+                post_id,
+                assertion.post_content_unit_id,
+                construct_id,
+                assertion.evidence_text,
+                assertion.truth_status_code,
+                assertion.extraction_method,
+                orchestrator_session_id,
+            )
+
+
+async def load_occupational_construct_assertions(
+    conn: Any, post_id: str
+) -> list[dict[str, object]]:
+    """Load assertions only after the caller has authorized their Post."""
+    rows = await conn.fetch(
+        """
+        select construct.construct_iri, construct.construct_family_code,
+               construct.preferred_label, vocabulary.vocabulary_iri,
+               vocabulary.version_label, assertion.evidence_text,
+               assertion.truth_status_code, assertion.extraction_method,
+               assertion.generated_at, unit.unit_index
+          from post_occupational_construct_assertion assertion
+          join occupational_construct construct on construct.construct_id = assertion.construct_id
+          join occupational_construct_vocabulary vocabulary
+            on vocabulary.vocabulary_id = construct.vocabulary_id
+          join post_content_unit unit
+            on unit.post_content_unit_id = assertion.post_content_unit_id
+         where assertion.post_id = $1
+         order by unit.unit_index, construct.construct_family_code,
+                  construct.preferred_label, construct.construct_iri
+        """,
+        post_id,
+    )
+    return [
+        {
+            "construct_iri": row["construct_iri"],
+            "construct_family_code": row["construct_family_code"],
+            "preferred_label": row["preferred_label"],
+            "vocabulary_iri": row["vocabulary_iri"],
+            "vocabulary_version": row["version_label"],
+            "evidence_text": row["evidence_text"],
+            "truth_status_code": row["truth_status_code"],
+            "extraction_method": row["extraction_method"],
+            "generated_at": row["generated_at"],
+            "unit_index": row["unit_index"],
+            "provenance": "post_occupational_construct_assertion.evidence_text",
+        }
+        for row in rows
+    ]

backend/tests/test_api.py

@@ -231,6 +231,14 @@
     / "migrations"
     / "0183_source_post_event_occurred_at.sql"
 )
+_ONTOLOGY_TRUTH_STATUS_MIGRATION = (
+    Path(__file__).resolve().parents[2] / "migrations" / "0175_ontology_truth_status.sql"
+)
+_OCCUPATIONAL_CONSTRUCT_MIGRATION = (
+    Path(__file__).resolve().parents[2]
+    / "migrations"
+    / "0238_occupational_construct_assertion.sql"
+)
 
 
 def _postgres_available() -> bool:
@@ -383,12 +391,16 @@ def seeded_db(demo_analyst_token):
             cur.execute(_LEFTOVER_MAP_COVERAGE_MIGRATION.read_text())
             cur.execute(_GLOBAL_ASK_JOB_MIGRATION.read_text())
             cur.execute(_GLOBAL_ASK_SCOPE_MIGRATION.read_text())
-            cur.execute(_GLOBAL_ASK_EVIDENCE_SEARCH_MIGRATION.read_text())
+            for statement in _GLOBAL_ASK_EVIDENCE_SEARCH_MIGRATION.read_text().split(";\n\n"):
+                if statement.strip():
+                    cur.execute(statement + ";")
             cur.execute(_GLOBAL_ASK_KNOWLEDGE_CUTOFF_MIGRATION.read_text())
             cur.execute(_GLOBAL_ASK_PUBLIC_VERIFICATION_MIGRATION.read_text())
             cur.execute(_EVENT_OCCURRED_AT_MIGRATION.read_text())
+            cur.execute(_ONTOLOGY_TRUTH_STATUS_MIGRATION.read_text())
             cur.execute(_LEFTOVER_MAP_AXIS_MIGRATION.read_text())
             cur.execute(_CHANNEL_EVIDENCE_MIGRATION.read_text())
+            cur.execute(_OCCUPATIONAL_CONSTRUCT_MIGRATION.read_text())
             cur.execute(_LEFTOVER_MAP_UNEXPLAINED_MIGRATION.read_text())
             cur.execute(_LEFTOVER_MAP_CROSS_SHARE_MIGRATION.read_text())
             cur.execute(_LEFTOVER_MAP_RECONSTRUCTION_MIGRATION.read_text())
@@ -1945,6 +1957,76 @@ def test_post_detail_exposes_explicit_and_semantic_project_evidence(
     assert listed_post["project_evidence"][0]["provenance"] == "post_project_mention.evidence_text"
 
 
+def test_post_detail_exposes_evidence_bound_occupational_construct(
+    client, demo_analyst_token, seeded_db
+) -> None:
+    """The authorized detail projection preserves its exact synthetic evidence."""
+    conn = psycopg2.connect(seeded_db["dsn"])
+    try:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                insert into post_content_unit
+                    (post_id, unit_index, unit_kind_code, unit_text)
+                values (%s, 90, 'plain_text', %s)
+                returning post_content_unit_id
+                """,
+                (
+                    seeded_db["public_post_id"],
+                    "Synthetic record requires oral comprehension.",
+                ),
+            )
+            unit_id = cur.fetchone()[0]
+            cur.execute(
+                """
+                insert into occupational_construct_vocabulary
+                    (vocabulary_iri, version_label, license_iri, attribution_text)
+                values (%s, '31.0', %s, 'Synthetic O*NET attribution')
+                returning vocabulary_id
+                """,
+                (
+                    "https://www.onetcenter.org/database.html",
+                    "https://creativecommons.org/licenses/by/4.0/",
+                ),
+            )
+            vocabulary_id = cur.fetchone()[0]
+            cur.execute(
+                """
+                insert into occupational_construct
+                    (vocabulary_id, construct_iri, construct_family_code, preferred_label)
+                values (%s, %s, 'cognitive_ability', 'Oral Comprehension')
+                returning construct_id
+                """,
+                (vocabulary_id, "https://data.onetcenter.org/element/1.A.1.a.1"),
+            )
+            construct_id = cur.fetchone()[0]
+            cur.execute(
+                """
+                insert into post_occupational_construct_assertion
+                    (post_id, post_content_unit_id, construct_id, evidence_text,
+                     truth_status_code, extraction_method, orchestrator_session_id)
+                values (%s, %s, %s, 'oral comprehension', 'truth_inferred',
+                        'contextual_orchestrator_structured', 'synthetic-session')
+                """,
+                (seeded_db["public_post_id"], unit_id, construct_id),
+            )
+        conn.commit()
+    finally:
+        conn.close()
+
+    response = client.get(
+        f"/api/posts/{seeded_db['public_post_id']}",
+        headers={"Authorization": f"Bearer {demo_analyst_token}"},
+    )
+    assert response.status_code == 200
+    assertions = response.json()["occupational_construct_assertions"]
+    assert assertions[0]["preferred_label"] == "Oral Comprehension"
+    assert assertions[0]["evidence_text"] == "oral comprehension"
+    assert assertions[0]["provenance"] == (
+        "post_occupational_construct_assertion.evidence_text"
+    )
+
+
 def test_post_detail_as_of_returns_the_cutoff_known_body(
     client, demo_analyst_token, seeded_db
 ) -> None: