diff --git a/config/portfolio-catalog.yaml b/config/portfolio-catalog.yaml
index fe8b6dc5..1fc8e58f 100644
--- a/config/portfolio-catalog.yaml
+++ b/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
diff --git a/src/notion_export.py b/src/notion_export.py
index 27275fc5..e87833c3 100644
--- a/src/notion_export.py
+++ b/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
 
 
diff --git a/src/notion_sync.py b/src/notion_sync.py
index 93f152a9..f1b1008f 100644
--- a/src/notion_sync.py
+++ b/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
diff --git a/src/operator_os_seam_linter.py b/src/operator_os_seam_linter.py
index 48941653..5e29fb80 100644
--- a/src/operator_os_seam_linter.py
+++ b/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],
     *,
diff --git a/src/portfolio_catalog.py b/src/portfolio_catalog.py
index 629f9c4e..4e46cd13 100644
--- a/src/portfolio_catalog.py
+++ b/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")),
diff --git a/src/portfolio_truth_publish.py b/src/portfolio_truth_publish.py
index 8a68edff..9e0fd8ab 100644
--- a/src/portfolio_truth_publish.py
+++ b/src/portfolio_truth_publish.py
@@ -10,7 +10,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 from pathlib import Path
-from typing import Iterator
+from typing import Iterable, Iterator
 
 from src.github_security_coverage import (
     DEFAULT_ATTENTION_STATES,
@@ -234,6 +234,7 @@ def _portfolio_truth_publication_lock(latest_path: Path) -> Iterator[None]:
 
 _REPO_ROOT = Path(__file__).resolve().parents[1]
 _CONFIG_DIR = _REPO_ROOT / "config"
+_PUBLISH_JOURNAL_NAME = ".portfolio-truth-publish-journal.json"
 
 
 def _build_project_registry_json(snapshot, *, include_notion: bool) -> str:
@@ -402,6 +403,16 @@ def _publish_portfolio_truth_locked(
         else 24
     )
     latest_path = truth_latest_path(output_dir)
+    project_registry_path = output_dir / "project-registry.json"
+    _recover_interrupted_publication(
+        output_dir,
+        allowed_targets={
+            latest_path,
+            registry_output,
+            portfolio_report_output,
+            project_registry_path,
+        },
+    )
     notion_context_fallback = (
         load_prior_notion_context(latest_path) if allow_empty_notion else None
     )
@@ -455,7 +466,6 @@ def _publish_portfolio_truth_locked(
     )
     latest_name = latest_path.name
     snapshot_json = json.dumps(build_result.snapshot.to_dict(), indent=2) + "\n"
-    project_registry_path = output_dir / "project-registry.json"
     project_registry_json = _build_project_registry_json(
         build_result.snapshot, include_notion=include_notion
     )
@@ -488,8 +498,12 @@ def _publish_portfolio_truth_locked(
         project_registry_path: True,
     }
     temp_files = {path: _stage_text(path, content) for path, content in targets.items()}
-    originals = {path: (path.read_text() if path.exists() else None) for path in targets}
-    published: list[Path] = []
+    backups = {
+        path: (_stage_bytes(path, path.read_bytes()) if path.exists() else None)
+        for path in targets
+    }
+    journal_path = output_dir / _PUBLISH_JOURNAL_NAME
+    _write_publish_journal(journal_path, temp_files=temp_files, backups=backups)
 
     # This live guard is intentionally separate from the snapshot's shared
     # evaluation clock: it catches receipt replacement or expiry before writes.
@@ -513,25 +527,24 @@ def _publish_portfolio_truth_locked(
                 _verify_prior_security_evidence_current(prior_security_evidence)
             for path, staged in temp_files.items():
                 if path in {registry_output, portfolio_report_output} and not changed[path]:
-                    staged.unlink(missing_ok=True)
                     continue
                 staged.replace(path)
-                published.append(path)
-    except Exception as exc:
-        for path in reversed(published):
-            original = originals[path]
-            if original is None:
-                path.unlink(missing_ok=True)
-            else:
-                path.write_text(original)
-        for staged in temp_files.values():
-            staged.unlink(missing_ok=True)
+                _fsync_directory(path.parent)
+    except BaseException as exc:
+        _recover_interrupted_publication(
+            output_dir,
+            allowed_targets={
+                latest_path,
+                registry_output,
+                portfolio_report_output,
+                project_registry_path,
+            },
+        )
         if isinstance(exc, (SecurityCoverageError, ValueError)):
             raise PortfolioTruthPublishError(str(exc)) from exc
         raise
 
-    for staged in temp_files.values():
-        staged.unlink(missing_ok=True)
+    _cleanup_publish_transaction(journal_path, temp_files.values(), backups.values())
 
     collision_summary = build_result.snapshot.source_summary["checkout_collisions"]
     return PortfolioTruthPublishResult(
@@ -587,9 +600,153 @@ def _stage_text(target: Path, content: str) -> Path:
         "w", delete=False, dir=target.parent, suffix=f".{target.name}.tmp"
     ) as handle:
         handle.write(content)
+        handle.flush()
+        os.fsync(handle.fileno())
+        return Path(handle.name)
+
+
+def _stage_bytes(target: Path, content: bytes) -> Path:
+    target.parent.mkdir(parents=True, exist_ok=True)
+    with tempfile.NamedTemporaryFile(
+        "wb", delete=False, dir=target.parent, suffix=f".{target.name}.bak"
+    ) as handle:
+        handle.write(content)
+        handle.flush()
+        os.fsync(handle.fileno())
         return Path(handle.name)
 
 
+def _fsync_directory(path: Path) -> None:
+    descriptor = os.open(path, os.O_RDONLY)
+    try:
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+
+
+def _write_publish_journal(
+    journal_path: Path,
+    *,
+    temp_files: dict[Path, Path],
+    backups: dict[Path, Path | None],
+) -> None:
+    journal_targets: list[dict[str, object]] = []
+    for target in temp_files:
+        backup = backups[target]
+        journal_targets.append(
+            {
+                "target": str(target.resolve()),
+                "staged": str(temp_files[target].resolve()),
+                "backup": str(backup.resolve()) if backup is not None else None,
+                "existed": backup is not None,
+            }
+        )
+    payload = {
+        "schema": "PortfolioTruthPublishJournalV1",
+        "targets": journal_targets,
+    }
+    staged_journal = _stage_text(
+        journal_path, json.dumps(payload, indent=2, sort_keys=True) + "\n"
+    )
+    staged_journal.replace(journal_path)
+    _fsync_directory(journal_path.parent)
+
+
+def _allowed_recovery_target(
+    target: Path, *, output_dir: Path, allowed_targets: set[Path]
+) -> bool:
+    resolved = target.resolve()
+    if resolved in {path.resolve() for path in allowed_targets}:
+        return True
+    return (
+        resolved.parent == output_dir.resolve()
+        and resolved.name.startswith("portfolio-truth-")
+        and resolved.name.endswith(".json")
+        and resolved.name != "portfolio-truth-latest.json"
+    )
+
+
+def _recover_interrupted_publication(
+    output_dir: Path, *, allowed_targets: set[Path]
+) -> None:
+    journal_path = output_dir / _PUBLISH_JOURNAL_NAME
+    if not journal_path.exists():
+        return
+    try:
+        payload = json.loads(journal_path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise PortfolioTruthPublishError(
+            f"Cannot recover interrupted publication: invalid journal {journal_path}"
+        ) from exc
+    if payload.get("schema") != "PortfolioTruthPublishJournalV1":
+        raise PortfolioTruthPublishError(
+            f"Cannot recover interrupted publication: unsupported journal {journal_path}"
+        )
+    rows = payload.get("targets")
+    if not isinstance(rows, list):
+        raise PortfolioTruthPublishError(
+            f"Cannot recover interrupted publication: malformed journal {journal_path}"
+        )
+
+    staged_paths: list[Path] = []
+    backup_paths: list[Path] = []
+    for row in rows:
+        if not isinstance(row, dict):
+            raise PortfolioTruthPublishError(
+                f"Cannot recover interrupted publication: malformed journal {journal_path}"
+            )
+        target = Path(str(row.get("target", "")))
+        staged = Path(str(row.get("staged", "")))
+        backup_value = row.get("backup")
+        backup = Path(str(backup_value)) if backup_value else None
+        if not _allowed_recovery_target(
+            target, output_dir=output_dir, allowed_targets=allowed_targets
+        ):
+            raise PortfolioTruthPublishError(
+                f"Cannot recover interrupted publication target outside contract: {target}"
+            )
+        if staged.resolve().parent != target.resolve().parent:
+            raise PortfolioTruthPublishError(
+                f"Cannot recover interrupted publication: invalid staged path for {target}"
+            )
+        if backup is not None and backup.resolve().parent != target.resolve().parent:
+            raise PortfolioTruthPublishError(
+                f"Cannot recover interrupted publication: invalid backup path for {target}"
+            )
+        if backup is None:
+            target.unlink(missing_ok=True)
+        else:
+            if not backup.exists():
+                raise PortfolioTruthPublishError(
+                    f"Cannot recover interrupted publication: backup missing for {target}"
+                )
+            restored = _stage_bytes(target, backup.read_bytes())
+            restored.replace(target)
+            backup_paths.append(backup)
+        staged_paths.append(staged)
+        _fsync_directory(target.parent)
+
+    _cleanup_publish_transaction(journal_path, staged_paths, backup_paths)
+
+
+def _cleanup_publish_transaction(
+    journal_path: Path,
+    staged_paths: Iterable[Path],
+    backup_paths: Iterable[Path | None],
+) -> None:
+    # The journal is the recovery authority. Retire and fsync it before deleting
+    # backups so a process death during cleanup can leave only harmless orphans,
+    # never a live journal that points at already-deleted recovery material.
+    journal_path.unlink(missing_ok=True)
+    if journal_path.parent.exists():
+        _fsync_directory(journal_path.parent)
+    for path in staged_paths:
+        path.unlink(missing_ok=True)
+    for backup_path in backup_paths:
+        if backup_path is not None:
+            backup_path.unlink(missing_ok=True)
+
+
 def _content_changed(path: Path, content: str) -> bool:
     if not path.exists():
         return True
diff --git a/tests/test_notion_export.py b/tests/test_notion_export.py
index b9e05c06..0305a872 100644
--- a/tests/test_notion_export.py
+++ b/tests/test_notion_export.py
@@ -141,6 +141,28 @@ def test_normalized_alias_match_resolves_spacing_and_case(self):
         assert _lookup_project_mapping("MCPAudit", mapping)["localProjectId"] == "mcp-id"
 
 
+def test_lookup_project_mapping_rejects_ambiguous_normalized_aliases() -> None:
+    mapping = {
+        "Foo Bar": {"localProjectId": "space"},
+        "Foo-Bar": {"localProjectId": "dash"},
+        "Foo_Bar": {"localProjectId": "underscore"},
+    }
+
+    assert _lookup_project_mapping("foo.bar", mapping) is None
+
+
+def test_lookup_project_mapping_accepts_aliases_for_the_same_page() -> None:
+    mapping = {
+        "GitHub Repo Auditor": {"localProjectId": "same-page", "sourceId": "one"},
+        "GithubRepoAuditor": {"localProjectId": "same-page", "sourceId": "two"},
+    }
+
+    resolved = _lookup_project_mapping("github repo auditor", mapping)
+
+    assert resolved is not None
+    assert resolved["localProjectId"] == "same-page"
+
+
 class TestBiggestDrag:
     def test_finds_lowest(self):
         audit = _make_report()["audits"][0]
diff --git a/tests/test_notion_sync_integrity.py b/tests/test_notion_sync_integrity.py
new file mode 100644
index 00000000..0dcaa337
--- /dev/null
+++ b/tests/test_notion_sync_integrity.py
@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import pytest
+
+from src import notion_sync
+
+
+@dataclass
+class Response:
+    status_code: int
+    payload: dict
+
+    def json(self) -> dict:
+        return self.payload
+
+
+def test_existing_event_enumeration_fails_after_partial_page(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    responses = iter(
+        [
+            Response(
+                200,
+                {
+                    "results": [
+                        {
+                            "properties": {
+                                "Event Key": {
+                                    "rich_text": [
+                                        {"text": {"content": "audit::report::one"}}
+                                    ]
+                                }
+                            }
+                        }
+                    ],
+                    "has_more": True,
+                    "next_cursor": "page-2",
+                },
+            ),
+            Response(503, {}),
+        ]
+    )
+    monkeypatch.setattr(
+        notion_sync,
+        "_notion_request",
+        lambda *args, **kwargs: next(responses),
+    )
+    monkeypatch.setattr(notion_sync.time, "sleep", lambda _: None)
+
+    with pytest.raises(RuntimeError, match="before completion"):
+        notion_sync._query_existing_event_keys("events", "token", "version")
diff --git a/tests/test_operator_os_seam_linter.py b/tests/test_operator_os_seam_linter.py
index 99a955e7..25435c31 100644
--- a/tests/test_operator_os_seam_linter.py
+++ b/tests/test_operator_os_seam_linter.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import hashlib
 import json
 import sqlite3
 from datetime import UTC, datetime
@@ -51,6 +52,10 @@ def _passing_paths(tmp_path: Path) -> tuple[Path, list[Path]]:
 def _refresh_truth_for_cli(path: Path) -> None:
     payload = json.loads(path.read_text())
     payload["generated_at"] = datetime.now(UTC).isoformat()
+    catalog = Path(__file__).parents[1] / "config" / "portfolio-catalog.yaml"
+    payload["inputs"] = {
+        "catalog": {"sha256": hashlib.sha256(catalog.read_bytes()).hexdigest()}
+    }
     path.write_text(json.dumps(payload))
 
 
@@ -137,6 +142,47 @@ def test_fresh_artifact_passes(tmp_path: Path) -> None:
     assert result.passed
 
 
+def test_recent_artifact_fails_when_catalog_source_hash_differs(tmp_path: Path) -> None:
+    truth, markdown = _passing_paths(tmp_path)
+    catalog = tmp_path / "portfolio-catalog.yaml"
+    catalog.write_text("repos: {}\n", encoding="utf-8")
+    payload = json.loads(truth.read_text())
+    payload["inputs"] = {"catalog": {"sha256": "0" * 64}}
+    truth.write_text(json.dumps(payload))
+
+    result = lint_operator_os_seams(
+        truth_path=truth,
+        markdown_paths=markdown,
+        catalog_path=catalog,
+        now=NOW,
+    )
+
+    assert not result.passed
+    finding = next(item for item in result.findings if item.check == "artifact_freshness")
+    assert "different catalog content" in finding.violation
+
+
+def test_recent_artifact_fails_when_catalog_source_hash_is_absent(tmp_path: Path) -> None:
+    truth, markdown = _passing_paths(tmp_path)
+    catalog = tmp_path / "portfolio-catalog.yaml"
+    catalog.write_text("repos: {}\n", encoding="utf-8")
+    payload = json.loads(truth.read_text())
+    payload["inputs"] = {"catalog": {"sha256": None}}
+    truth.write_text(json.dumps(payload))
+
+    result = lint_operator_os_seams(
+        truth_path=truth,
+        markdown_paths=markdown,
+        catalog_path=catalog,
+        now=NOW,
+    )
+
+    assert not result.passed
+    finding = next(item for item in result.findings if item.check == "artifact_freshness")
+    assert finding.level == "fail"
+    assert "not source-bound" in finding.violation
+
+
 def test_stale_artifact_fails(tmp_path: Path) -> None:
     truth, markdown = _passing_paths(tmp_path)
     _write_truth(truth, generated_at="2026-07-01T00:00:00+00:00")
diff --git a/tests/test_portfolio_catalog.py b/tests/test_portfolio_catalog.py
index b71a37ef..ef76840d 100644
--- a/tests/test_portfolio_catalog.py
+++ b/tests/test_portfolio_catalog.py
@@ -66,6 +66,46 @@ def test_load_portfolio_catalog_accepts_defaults_and_repo_entries(tmp_path: Path
     assert catalog["repos"]["repob"]["doctor_standard"] == ""
 
 
+def test_load_portfolio_catalog_rejects_non_utf8(tmp_path: Path) -> None:
+    path = tmp_path / "portfolio-catalog.yaml"
+    path.write_bytes(b"repos:\n  bad: \xff\n")
+
+    catalog = load_portfolio_catalog(path)
+
+    assert catalog["repos"] == {}
+    assert "UTF-8" in catalog["errors"][0]
+    assert "offset" in catalog["errors"][0]
+
+
+def test_load_portfolio_catalog_rejects_duplicate_repo_key(tmp_path: Path) -> None:
+    path = tmp_path / "portfolio-catalog.yaml"
+    path.write_text(
+        "repos:\n  RepoA:\n    owner: first\n  RepoA:\n    owner: second\n",
+        encoding="utf-8",
+    )
+
+    catalog = load_portfolio_catalog(path)
+
+    assert catalog["repos"] == {}
+    assert "duplicate mapping key 'RepoA'" in catalog["errors"][0]
+
+
+def test_load_portfolio_catalog_rejects_invalid_group_order(tmp_path: Path) -> None:
+    path = tmp_path / "portfolio-catalog.yaml"
+    path.write_text(
+        "groups:\n"
+        "  active:\n"
+        "    path_prefixes: [active]\n"
+        "    order: first\n",
+        encoding="utf-8",
+    )
+
+    catalog = load_portfolio_catalog(path)
+
+    assert catalog["groups"] == {}
+    assert "order must be an integer" in catalog["errors"][0]
+
+
 def test_load_portfolio_catalog_indexes_repo_aliases(tmp_path: Path):
     path = tmp_path / "portfolio-catalog.yaml"
     path.write_text(
diff --git a/tests/test_portfolio_truth_publish_integrity.py b/tests/test_portfolio_truth_publish_integrity.py
new file mode 100644
index 00000000..0deb2d89
--- /dev/null
+++ b/tests/test_portfolio_truth_publish_integrity.py
@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+import src.portfolio_truth_publish as publish_mod
+
+
+@pytest.mark.parametrize("replaced_count", [1, 2, 3, 4, 5])
+def test_publish_journal_recovers_process_death_after_each_replacement(
+    tmp_path: Path, replaced_count: int
+) -> None:
+    output_dir = tmp_path / "output"
+    output_dir.mkdir()
+    snapshot = output_dir / "portfolio-truth-2026-07-17T120000Z.json"
+    latest = output_dir / "portfolio-truth-latest.json"
+    registry = tmp_path / "project-registry.md"
+    report = tmp_path / "PORTFOLIO-AUDIT-REPORT.md"
+    project_registry = output_dir / "project-registry.json"
+    targets = [snapshot, latest, registry, report, project_registry]
+    prior = {
+        snapshot: None,
+        latest: "old latest\n",
+        registry: "old registry\n",
+        report: "old report\n",
+        project_registry: "old project registry\n",
+    }
+    for path, content in prior.items():
+        if content is not None:
+            path.write_text(content, encoding="utf-8")
+
+    staged = {
+        path: publish_mod._stage_text(path, f"new generation {index}\n")
+        for index, path in enumerate(targets)
+    }
+    backups = {
+        path: (
+            publish_mod._stage_bytes(path, path.read_bytes())
+            if path.exists()
+            else None
+        )
+        for path in targets
+    }
+    journal = output_dir / publish_mod._PUBLISH_JOURNAL_NAME
+    publish_mod._write_publish_journal(
+        journal, temp_files=staged, backups=backups
+    )
+    for path in targets[:replaced_count]:
+        staged[path].replace(path)
+
+    publish_mod._recover_interrupted_publication(
+        output_dir,
+        allowed_targets={latest, registry, report, project_registry},
+    )
+
+    for path, content in prior.items():
+        if content is None:
+            assert not path.exists()
+        else:
+            assert path.read_text(encoding="utf-8") == content
+    assert not journal.exists()
+    assert not any(path.exists() for path in staged.values())
+    assert not any(path is not None and path.exists() for path in backups.values())
+
+
+def test_publish_cleanup_retires_journal_before_deleting_backups(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    journal = tmp_path / publish_mod._PUBLISH_JOURNAL_NAME
+    staged = tmp_path / "staged.tmp"
+    backup = tmp_path / "target.bak"
+    for path in (journal, staged, backup):
+        path.write_text("data", encoding="utf-8")
+
+    removed: list[Path] = []
+    original_unlink = Path.unlink
+
+    def recording_unlink(path: Path, *args, **kwargs) -> None:
+        removed.append(path)
+        original_unlink(path, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "unlink", recording_unlink)
+
+    publish_mod._cleanup_publish_transaction(journal, [staged], [backup])
+
+    assert removed == [journal, staged, backup]