Repair portfolio data-integrity failure modes

config/portfolio-catalog.yaml

@@ -1370,10 +1370,11 @@ repos:
     category: infrastructure
     tool_provenance: codex
     notes: Drill-verified (dated passing drill) at 2026-07-17 tribunal judgment; tribunal
-      KEEP maps to this manual-only standing-tooling contract.
+      KEEP maps to this manual-only standing-tooling contract. The 2026-07-11
+      disposable real-repository recovery drill also passed, including broken-main
+      rejection, timeout handling, and force-tracked .env blocking.
     maturity_program: maintain
     target_maturity: operating
-    notes: Session 2026-07-11 disposable real-repository recovery drill passed, including broken-main rejection, timeout handling, and force-tracked .env blocking.
   sovereign-intelligence:
     owner: d
     purpose: local evidence-receipt intelligence workspace for repeated scenario comparison and decision support

src/notion_export.py

@@ -110,9 +110,17 @@ def _lookup_project_mapping(name: str, mapping: dict[str, dict]) -> dict | None:
     normalized = _normalize(name)
     if not normalized:
         return None
-    for mapped_name, mapped_project in mapping.items():
-        if _normalize(mapped_name) == normalized:
-            return mapped_project
+    candidates = [
+        mapped_project
+        for mapped_name, mapped_project in mapping.items()
+        if _normalize(mapped_name) == normalized
+    ]
+    destination_ids = {
+        str(candidate.get("localProjectId") or "").strip()
+        for candidate in candidates
+    }
+    if candidates and len(destination_ids) == 1 and "" not in destination_ids:
+        return candidates[0]
     return None
 
 

src/notion_sync.py

@@ -68,7 +68,7 @@ def _query_existing_event_keys(
     token: str,
     version: str,
 ) -> set[str]:
-    """Query existing audit event keys for deduplication."""
+    """Query every existing audit event key or fail closed."""
     keys: set[str] = set()
     start_cursor = None
 
@@ -85,9 +85,17 @@ def _query_existing_event_keys(
 
         resp = _notion_request("POST", f"/databases/{events_db_id}/query", token, version, body)
         if not resp or resp.status_code != 200:
-            break
+            status = getattr(resp, "status_code", "no-response")
+            raise RuntimeError(
+                f"existing-event enumeration failed before completion: status={status}"
+            )
 
-        data = resp.json()
+        try:
+            data = resp.json()
+        except (ValueError, TypeError) as exc:
+            raise RuntimeError("existing-event enumeration returned invalid JSON") from exc
+        if not isinstance(data, dict) or not isinstance(data.get("results", []), list):
+            raise RuntimeError("existing-event enumeration returned an invalid page")
         for page in data.get("results", []):
             props = page.get("properties", {})
             ek = props.get("Event Key", {})
@@ -98,6 +106,10 @@ def _query_existing_event_keys(
         if not data.get("has_more"):
             break
         start_cursor = data.get("next_cursor")
+        if not isinstance(start_cursor, str) or not start_cursor:
+            raise RuntimeError(
+                "existing-event enumeration was partial: has_more without next_cursor"
+            )
         time.sleep(REQUEST_DELAY)
 
     return keys
@@ -201,7 +213,18 @@ def sync_notion_events(
 
     # Query existing event keys for dedup
     print("  Querying existing audit events...", file=sys.stderr)
-    existing_keys = _query_existing_event_keys(events_db_id, token, version)
+    try:
+        existing_keys = _query_existing_event_keys(events_db_id, token, version)
+    except RuntimeError as exc:
+        print(f"  Refusing Notion creates: {exc}", file=sys.stderr)
+        return {
+            "created": 0,
+            "deduped": 0,
+            "updated_projects": 0,
+            "errors": 1,
+            "skipped": True,
+            "reason": "incomplete existing-event enumeration",
+        }
     print(f"  Found {len(existing_keys)} existing audit events.", file=sys.stderr)
 
     created = 0

src/operator_os_seam_linter.py

@@ -139,6 +139,12 @@ def lint_operator_os_seams(
                 identity_since=identity_since,
             )
         )
+        if catalog_path is not None and not contract_shadow:
+            findings.extend(
+                _check_catalog_source_binding(
+                    truth, truth_path=truth_path, catalog_path=catalog_path
+                )
+            )
         if contract_shadow:
             findings.extend(
                 _check_contract_shadow(
@@ -302,6 +308,48 @@ def _load_truth_artifact(
     return data
 
 
+def _check_catalog_source_binding(
+    truth: dict[str, Any], *, truth_path: Path, catalog_path: Path
+) -> list[SeamLintFinding]:
+    """Refuse an unqualified current result unless truth binds the live catalog."""
+    if not catalog_path.is_file():
+        return [
+            SeamLintFinding(
+                check="artifact_freshness",
+                artifact=str(truth_path),
+                violation="current catalog is unavailable",
+                detail=f"catalog={catalog_path}",
+                level="fail",
+            )
+        ]
+    inputs = truth.get("inputs")
+    catalog_input = inputs.get("catalog") if isinstance(inputs, dict) else None
+    declared_hash = (
+        catalog_input.get("sha256") if isinstance(catalog_input, dict) else None
+    )
+    if not isinstance(declared_hash, str) or not declared_hash:
+        return [
+            SeamLintFinding(
+                check="artifact_freshness",
+                artifact=str(truth_path),
+                violation="truth artifact is not source-bound to the current catalog",
+                detail="inputs.catalog.sha256 is absent; freshness is UNKNOWN",
+                level="fail",
+            )
+        ]
+    actual_hash = hashlib.sha256(catalog_path.read_bytes()).hexdigest()
+    if declared_hash != actual_hash:
+        return [
+            SeamLintFinding(
+                check="artifact_freshness",
+                artifact=str(truth_path),
+                violation="truth artifact was produced from different catalog content",
+                detail=f"declared={declared_hash}; actual={actual_hash}",
+            )
+        ]
+    return []
+
+
 def _check_artifact_freshness(
     truth: dict[str, Any],
     *,

src/portfolio_catalog.py

@@ -84,8 +84,43 @@ def load_portfolio_catalog(path: Path | None = None) -> dict[str, Any]:
             "repos": {},
         }
 
+    class UniqueKeyLoader(yaml.SafeLoader):
+        pass
+
+    def construct_unique_mapping(loader: Any, node: Any, deep: bool = False) -> dict[Any, Any]:
+        mapping: dict[Any, Any] = {}
+        for key_node, value_node in node.value:
+            key = loader.construct_object(key_node, deep=deep)
+            if key in mapping:
+                raise yaml.constructor.ConstructorError(
+                    "while constructing a mapping",
+                    node.start_mark,
+                    f"duplicate mapping key {key!r}",
+                    key_node.start_mark,
+                )
+            mapping[key] = loader.construct_object(value_node, deep=deep)
+        return mapping
+
+    UniqueKeyLoader.add_constructor(
+        yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, construct_unique_mapping
+    )
+
     try:
-        loaded = yaml.safe_load(catalog_path.read_text()) or {}
+        raw = catalog_path.read_text(encoding="utf-8")
+        loaded = yaml.load(raw, Loader=UniqueKeyLoader) or {}
+    except UnicodeDecodeError as exc:
+        return {
+            "path": str(catalog_path),
+            "exists": True,
+            "errors": [
+                "Failed to read portfolio catalog as UTF-8: "
+                f"invalid byte at offset {exc.start}."
+            ],
+            "warnings": [],
+            "defaults": {},
+            "groups": {},
+            "repos": {},
+        }
     except yaml.YAMLError as exc:
         return {
             "path": str(catalog_path),
@@ -246,6 +281,18 @@ def _normalize_group_entries(
             raw_value, label=f"Portfolio catalog group '{key}'", warnings=warnings
         )
 
+        raw_order = raw_value.get("order", order)
+        if isinstance(raw_order, bool):
+            errors.append(f"Portfolio catalog group '{key}' order must be an integer.")
+            continue
+        try:
+            normalized_order = int(raw_order)
+        except (TypeError, ValueError):
+            errors.append(
+                f"Portfolio catalog group '{key}' order must be an integer, got {raw_order!r}."
+            )
+            continue
+
         normalized = {
             "group_key": key,
             "label": _safe_text(raw_value.get("label")) or key,
@@ -254,7 +301,7 @@ def _normalize_group_entries(
             or _safe_text(raw_value.get("label"))
             or key,
             "section_note": _safe_text(raw_value.get("section_note")),
-            "order": int(raw_value.get("order", order)),
+            "order": normalized_order,
             "path_prefixes": prefixes,
             "owner": _safe_text(raw_value.get("owner")),
             "team": _safe_text(raw_value.get("team")),