From f14b27eca8b5d3cd9a0746885a700b97f5bcaa88 Mon Sep 17 00:00:00 2001
From: James Wiesebron <jameswiesebron@khanacademy.org>
Date: Mon, 27 Apr 2026 12:07:33 -0700
Subject: [PATCH 1/2] =?UTF-8?q?Fix=20#2134:=20structured=20audit=20events?=
 =?UTF-8?q?=20for=20plan=E2=86=92contract=20populate?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The populate step had six possible outcomes (success + five silent
early-returns + one outer catch-all) and all of them logged generic
prose messages, so when #1931's contract was committed with phases:[]
we could not tell from logs which path fired.

Replace the prose log calls with two stable event names:
- contract_phases_populated (success)
- contract_phases_ingest_failed with reason= discriminator:
  egg_contracts_unavailable, no_draft_path, plan_draft_missing,
  contract_load_failed, parse_failed, unexpected_exception.

Keeps the _safe wrapper's exception-swallowing policy (#1890): a
populate failure must not block the HITL gate. The audit event is
the canonical signal.

Adds orchestrator/tests/test_populate_contract_audit_events.py
covering all 7 events plus a regression that asserts a known-good
plan populates contract.phases[] when run through _populate_contract_from_plan_safe.
---
 orchestrator/routes/pipelines.py              |  37 ++-
 .../test_populate_contract_audit_events.py    | 314 ++++++++++++++++++
 2 files changed, 342 insertions(+), 9 deletions(-)
 create mode 100644 orchestrator/tests/test_populate_contract_audit_events.py

diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py
index 44587526cf..fd3036359c 100644
--- a/orchestrator/routes/pipelines.py
+++ b/orchestrator/routes/pipelines.py
@@ -10997,8 +10997,9 @@ def _populate_contract_from_plan_safe(
         _populate_contract_from_plan(repo_path, pipeline_id, pipeline_mode, issue_number)
     except Exception as pop_err:
         logger.warning(
-            "Failed to populate contract from plan (continuing)",
+            "contract_phases_ingest_failed",
             pipeline_id=pipeline_id,
+            reason="unexpected_exception",
             error=str(pop_err),
         )
 
@@ -11017,25 +11018,41 @@ def _populate_contract_from_plan(
     try:
         from egg_contracts.loader import load_contract, save_contract
     except ImportError:
-        logger.warning("egg_contracts not available, skipping contract population")
+        logger.warning(
+            "contract_phases_ingest_failed",
+            pipeline_id=pipeline_id,
+            reason="egg_contracts_unavailable",
+        )
         return
 
     # Resolve draft path
     draft_rel = _get_draft_path("plan", issue_number=issue_number, pipeline_id=pipeline_id)
     if not draft_rel:
-        logger.warning("No draft path for plan phase", pipeline_id=pipeline_id)
+        logger.warning(
+            "contract_phases_ingest_failed",
+            pipeline_id=pipeline_id,
+            reason="no_draft_path",
+        )
         return
 
     plan_path = repo_path / draft_rel
     if not plan_path.exists():
-        logger.warning("Plan draft not found, skipping contract population", path=str(plan_path))
+        logger.warning(
+            "contract_phases_ingest_failed",
+            pipeline_id=pipeline_id,
+            reason="plan_draft_missing",
+            path=str(plan_path),
+        )
         return
 
     try:
         contract = load_contract(pipeline_id, repo_path)
-    except Exception:
+    except Exception as load_err:
         logger.warning(
-            "Contract not found for pipeline, skipping population", pipeline_id=pipeline_id
+            "contract_phases_ingest_failed",
+            pipeline_id=pipeline_id,
+            reason="contract_load_failed",
+            error=str(load_err),
         )
         return
 
@@ -11047,8 +11064,9 @@ def _populate_contract_from_plan(
 
         if not result.success:
             logger.warning(
-                "Plan parsing failed, skipping contract population",
+                "contract_phases_ingest_failed",
                 pipeline_id=pipeline_id,
+                reason="parse_failed",
                 error=result.error,
             )
             return
@@ -11084,7 +11102,7 @@ def _populate_contract_from_plan(
             save_contract(contract, repo_path)
             task_count = sum(len(p.tasks) for p in contract.phases)
             logger.info(
-                "Contract populated from plan",
+                "contract_phases_populated",
                 pipeline_id=pipeline_id,
                 phase_count=len(contract.phases),
                 task_count=task_count,
@@ -11093,8 +11111,9 @@ def _populate_contract_from_plan(
 
     except Exception as e:
         logger.warning(
-            "Failed to populate contract from plan",
+            "contract_phases_ingest_failed",
             pipeline_id=pipeline_id,
+            reason="unexpected_exception",
             error=str(e),
         )
 
diff --git a/orchestrator/tests/test_populate_contract_audit_events.py b/orchestrator/tests/test_populate_contract_audit_events.py
new file mode 100644
index 0000000000..75375feada
--- /dev/null
+++ b/orchestrator/tests/test_populate_contract_audit_events.py
@@ -0,0 +1,314 @@
+"""Audit events for ``_populate_contract_from_plan`` (issue #2134).
+
+The populate step has six possible outcomes (success + five silent
+early-returns + one outer catch-all).  Each outcome must emit a
+structured log event with a stable name and a discriminator field so a
+recurrence of the #1931 empty-phases incident can be diagnosed from
+logs alone.
+
+Verified outcomes:
+
+* ``contract_phases_populated`` — happy path
+* ``contract_phases_ingest_failed`` with ``reason``:
+  - ``egg_contracts_unavailable``
+  - ``no_draft_path``
+  - ``plan_draft_missing``
+  - ``contract_load_failed``
+  - ``parse_failed``
+  - ``unexpected_exception`` (inner catch-all + outer ``_safe`` wrapper)
+
+structlog output bypasses pytest's ``caplog`` fixture in this codebase
+(see ``test_decisions_routes.py``), so we patch the module-level
+``logger`` and inspect ``call_args_list`` directly.
+"""
+
+from __future__ import annotations
+
+import sys
+import textwrap
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# Match the heavy-dependency mocking pattern from test_diagnostic_logging_1633.py
+_docker_mock = MagicMock()
+sys.modules.setdefault("docker", _docker_mock)
+sys.modules.setdefault("docker.errors", _docker_mock.errors)
+sys.modules.setdefault("docker.types", _docker_mock.types)
+
+
+SAMPLE_PLAN = textwrap.dedent("""\
+    # Plan: Test plan
+
+    ## Implementation
+
+    ### Phase 1: Implement
+
+    Body.
+
+    ```yaml
+    # yaml-tasks
+    pr:
+      title: "Test PR"
+      description: "Test description"
+    phases:
+      - id: 1
+        name: Implement
+        goal: Test goal
+        tasks:
+          - id: TASK-1-1
+            description: "Task one"
+            acceptance: "Done"
+            files:
+              - src/x.py
+    ```
+""")
+
+
+def _populate_calls(mock_logger):
+    """Return all logger calls from the populate function (any level)."""
+    return (
+        mock_logger.info.call_args_list
+        + mock_logger.warning.call_args_list
+        + mock_logger.error.call_args_list
+    )
+
+
+def _ingest_failed_calls(mock_logger):
+    """Return calls whose first positional arg is the failure event name."""
+    return [
+        c
+        for c in mock_logger.warning.call_args_list
+        if c.args and c.args[0] == "contract_phases_ingest_failed"
+    ]
+
+
+class TestSuccessEvent:
+    """Happy path emits ``contract_phases_populated``."""
+
+    def test_emits_populated_event_on_success(self, tmp_path):
+        """A populated plan draft + contract produces the success event."""
+        from egg_contracts.loader import create_contract
+        from routes.pipelines import _populate_contract_from_plan
+
+        pipeline_id = "pipeline-success"
+
+        create_contract(pipeline_id=pipeline_id, title="Test", repo_root=tmp_path)
+
+        drafts_dir = tmp_path / ".egg-state" / "drafts"
+        drafts_dir.mkdir(parents=True, exist_ok=True)
+        (drafts_dir / f"{pipeline_id}-plan.md").write_text(SAMPLE_PLAN)
+
+        with patch("routes.pipelines.logger") as mock_logger:
+            _populate_contract_from_plan(tmp_path, pipeline_id, "local")
+
+        success_calls = [
+            c
+            for c in mock_logger.info.call_args_list
+            if c.args and c.args[0] == "contract_phases_populated"
+        ]
+        assert len(success_calls) == 1, (
+            f"Expected one contract_phases_populated event, got {_populate_calls(mock_logger)}"
+        )
+
+        kwargs = success_calls[0].kwargs
+        assert kwargs["pipeline_id"] == pipeline_id
+        assert kwargs["phase_count"] == 1
+        assert kwargs["task_count"] == 1
+        assert kwargs["has_pr_metadata"] is True
+
+        # No failure events should have been emitted.
+        assert _ingest_failed_calls(mock_logger) == []
+
+
+class TestFailureEvents:
+    """Each silent early-return path emits a discriminated failure event."""
+
+    def test_egg_contracts_unavailable(self, tmp_path):
+        """ImportError on egg_contracts.loader emits the matching reason."""
+        from routes.pipelines import _populate_contract_from_plan
+
+        # Force ``from egg_contracts.loader import ...`` to raise ImportError
+        # by setting the module to None in sys.modules.
+        with (
+            patch.dict(sys.modules, {"egg_contracts.loader": None}),
+            patch("routes.pipelines.logger") as mock_logger,
+        ):
+            _populate_contract_from_plan(tmp_path, "pipeline-no-loader", "local")
+
+        failures = _ingest_failed_calls(mock_logger)
+        assert len(failures) == 1
+        assert failures[0].kwargs["reason"] == "egg_contracts_unavailable"
+        assert failures[0].kwargs["pipeline_id"] == "pipeline-no-loader"
+
+    def test_no_draft_path(self, tmp_path):
+        """``_get_draft_path`` returning None emits no_draft_path."""
+        from routes.pipelines import _populate_contract_from_plan
+
+        with (
+            patch("routes.pipelines._get_draft_path", return_value=None),
+            patch("routes.pipelines.logger") as mock_logger,
+        ):
+            _populate_contract_from_plan(tmp_path, "pipeline-no-path", "local")
+
+        failures = _ingest_failed_calls(mock_logger)
+        assert len(failures) == 1
+        assert failures[0].kwargs["reason"] == "no_draft_path"
+        assert failures[0].kwargs["pipeline_id"] == "pipeline-no-path"
+
+    def test_plan_draft_missing(self, tmp_path):
+        """Plan file not on disk emits plan_draft_missing with the path."""
+        from routes.pipelines import _populate_contract_from_plan
+
+        with patch("routes.pipelines.logger") as mock_logger:
+            _populate_contract_from_plan(tmp_path, "pipeline-no-draft", "local")
+
+        failures = _ingest_failed_calls(mock_logger)
+        assert len(failures) == 1
+        assert failures[0].kwargs["reason"] == "plan_draft_missing"
+        assert failures[0].kwargs["pipeline_id"] == "pipeline-no-draft"
+        assert "pipeline-no-draft-plan.md" in failures[0].kwargs["path"]
+
+    def test_contract_load_failed(self, tmp_path):
+        """``load_contract`` raising emits contract_load_failed with the error."""
+        from routes.pipelines import _populate_contract_from_plan
+
+        # Plan draft must exist so we get past the plan_draft_missing gate.
+        drafts_dir = tmp_path / ".egg-state" / "drafts"
+        drafts_dir.mkdir(parents=True, exist_ok=True)
+        (drafts_dir / "pipeline-bad-contract-plan.md").write_text(SAMPLE_PLAN)
+
+        with (
+            patch(
+                "egg_contracts.loader.load_contract",
+                side_effect=RuntimeError("contract corrupt"),
+            ),
+            patch("routes.pipelines.logger") as mock_logger,
+        ):
+            _populate_contract_from_plan(tmp_path, "pipeline-bad-contract", "local")
+
+        failures = _ingest_failed_calls(mock_logger)
+        assert len(failures) == 1
+        assert failures[0].kwargs["reason"] == "contract_load_failed"
+        assert "contract corrupt" in failures[0].kwargs["error"]
+
+    def test_parse_failed(self, tmp_path):
+        """A plan whose yaml-tasks block doesn't parse emits parse_failed."""
+        from egg_contracts.loader import create_contract
+        from routes.pipelines import _populate_contract_from_plan
+
+        pipeline_id = "pipeline-bad-parse"
+        create_contract(pipeline_id=pipeline_id, title="Test", repo_root=tmp_path)
+
+        # Force parse_plan to return success=False.
+        fake_result = MagicMock()
+        fake_result.success = False
+        fake_result.error = "yaml-tasks block missing required field 'phases'"
+
+        drafts_dir = tmp_path / ".egg-state" / "drafts"
+        drafts_dir.mkdir(parents=True, exist_ok=True)
+        (drafts_dir / f"{pipeline_id}-plan.md").write_text("# Plan with no yaml-tasks\n")
+
+        with (
+            patch("egg_contracts.plan_parser.parse_plan", return_value=fake_result),
+            patch("routes.pipelines.logger") as mock_logger,
+        ):
+            _populate_contract_from_plan(tmp_path, pipeline_id, "local")
+
+        failures = _ingest_failed_calls(mock_logger)
+        assert len(failures) == 1
+        assert failures[0].kwargs["reason"] == "parse_failed"
+        assert "yaml-tasks" in failures[0].kwargs["error"]
+
+    def test_unexpected_exception_inner_catch(self, tmp_path):
+        """Anything raising inside the parse block is caught with unexpected_exception."""
+        from egg_contracts.loader import create_contract
+        from routes.pipelines import _populate_contract_from_plan
+
+        pipeline_id = "pipeline-explode"
+        create_contract(pipeline_id=pipeline_id, title="Test", repo_root=tmp_path)
+
+        drafts_dir = tmp_path / ".egg-state" / "drafts"
+        drafts_dir.mkdir(parents=True, exist_ok=True)
+        (drafts_dir / f"{pipeline_id}-plan.md").write_text(SAMPLE_PLAN)
+
+        with (
+            patch(
+                "egg_contracts.plan_parser.parse_plan",
+                side_effect=RuntimeError("parser exploded"),
+            ),
+            patch("routes.pipelines.logger") as mock_logger,
+        ):
+            _populate_contract_from_plan(tmp_path, pipeline_id, "local")
+
+        failures = _ingest_failed_calls(mock_logger)
+        assert len(failures) == 1
+        assert failures[0].kwargs["reason"] == "unexpected_exception"
+        assert "parser exploded" in failures[0].kwargs["error"]
+
+
+class TestSafeWrapper:
+    """``_populate_contract_from_plan_safe`` is the outer backstop."""
+
+    def test_safe_wrapper_emits_unexpected_exception_on_inner_raise(self, tmp_path):
+        """If the inner function raises, the wrapper emits unexpected_exception."""
+        from routes.pipelines import _populate_contract_from_plan_safe
+
+        with (
+            patch(
+                "routes.pipelines._populate_contract_from_plan",
+                side_effect=RuntimeError("inner blew up"),
+            ),
+            patch("routes.pipelines.logger") as mock_logger,
+        ):
+            _populate_contract_from_plan_safe(tmp_path, "pipeline-wrap", "local")
+
+        failures = _ingest_failed_calls(mock_logger)
+        assert len(failures) == 1
+        assert failures[0].kwargs["reason"] == "unexpected_exception"
+        assert failures[0].kwargs["pipeline_id"] == "pipeline-wrap"
+        assert "inner blew up" in failures[0].kwargs["error"]
+
+    def test_safe_wrapper_does_not_propagate(self, tmp_path):
+        """Inner failures must not escape — HITL gate must remain reachable (#1890)."""
+        from routes.pipelines import _populate_contract_from_plan_safe
+
+        with patch(
+            "routes.pipelines._populate_contract_from_plan",
+            side_effect=RuntimeError("boom"),
+        ):
+            # No assertion — just must not raise.
+            _populate_contract_from_plan_safe(tmp_path, "pipeline-quiet", "local")
+
+
+class TestRegressionEmptyPhases:
+    """Direct regression for the #1931 incident referenced by #2134.
+
+    A pipeline whose plan has a known-good yaml-tasks block, fed
+    through the same wrapper used at the post-plan persist step, must
+    leave ``contract.phases`` populated.
+    """
+
+    def test_known_good_plan_populates_phases(self, tmp_path):
+        from egg_contracts.loader import create_contract, load_contract
+        from routes.pipelines import _populate_contract_from_plan_safe
+
+        pipeline_id = "pipeline-1931-regression"
+        create_contract(pipeline_id=pipeline_id, title="Test", repo_root=tmp_path)
+
+        drafts_dir = tmp_path / ".egg-state" / "drafts"
+        drafts_dir.mkdir(parents=True, exist_ok=True)
+        (drafts_dir / f"{pipeline_id}-plan.md").write_text(SAMPLE_PLAN)
+
+        _populate_contract_from_plan_safe(tmp_path, pipeline_id, "local")
+
+        contract = load_contract(pipeline_id, tmp_path)
+        assert len(contract.phases) == 1, (
+            "Contract phases must not be empty after populate when plan "
+            "yaml-tasks parses cleanly — see issue #2134 / #1931."
+        )
+        assert len(contract.phases[0].tasks) == 1
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])

From 097e5e729dd5bbf276bd7970759cef83fc319960 Mon Sep 17 00:00:00 2001
From: "egg-reviewer[bot]"
 <261018737+egg-reviewer[bot]@users.noreply.github.com>
Date: Mon, 27 Apr 2026 19:23:54 +0000
Subject: [PATCH 2/2] Address review: distinguish unexpected_exception layers +
 capture #1931 empty-result case
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Four observability refinements from PR #2150 review:

1. Inner vs outer unexpected_exception: add source="parse_save" (inner
   catch in _populate_contract_from_plan) vs source="safe_wrapper"
   (outer _populate_contract_from_plan_safe) so an operator hitting
   the discriminator knows which layer raised.
2. Preserve tracebacks: add exc_info=True to both unexpected_exception
   sites (and the phases.py endpoint catch-all) — str(e) alone hides
   where the failure actually came from.
3. Empty-result discriminator: when parse_plan returns success=True
   but yields no phases and no PR metadata (the #1931 failure mode),
   emit contract_phases_ingest_failed with reason="empty_result"
   instead of staying silent. Closes the diagnosability gap the PR
   description called out as a follow-up.
4. Rename the phases.py populate_contract endpoint catch-all log from
   the prose 'Failed to populate contract' to the structured event
   name contract_populate_endpoint_failed so grep across the codebase
   no longer finds two unrelated events for the same string.

Test updates:
- test_unexpected_exception_inner_catch now asserts source="parse_save"
  + exc_info=True.
- test_safe_wrapper_emits_unexpected_exception_on_inner_raise now
  asserts source="safe_wrapper" + exc_info=True.
- New test_empty_result_emits_discriminator covers the #1931 case
  (parse success with no phases / no PR metadata).
---
 orchestrator/routes/phases.py                 |  3 +-
 orchestrator/routes/pipelines.py              | 14 +++++
 .../test_populate_contract_audit_events.py    | 59 ++++++++++++++++++-
 3 files changed, 74 insertions(+), 2 deletions(-)

diff --git a/orchestrator/routes/phases.py b/orchestrator/routes/phases.py
index 522e6533c7..3074138966 100644
--- a/orchestrator/routes/phases.py
+++ b/orchestrator/routes/phases.py
@@ -1048,9 +1048,10 @@ def populate_contract(pipeline_id: str) -> tuple[Response, int]:
         )
     except Exception as e:
         logger.error(
-            "Failed to populate contract",
+            "contract_populate_endpoint_failed",
             pipeline_id=pipeline_id,
             error=str(e),
+            exc_info=True,
         )
         return make_error_response(
             f"Failed to populate contract: {e}",
diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py
index fd3036359c..6f056008af 100644
--- a/orchestrator/routes/pipelines.py
+++ b/orchestrator/routes/pipelines.py
@@ -11000,7 +11000,9 @@ def _populate_contract_from_plan_safe(
             "contract_phases_ingest_failed",
             pipeline_id=pipeline_id,
             reason="unexpected_exception",
+            source="safe_wrapper",
             error=str(pop_err),
+            exc_info=True,
         )
 
 
@@ -11108,13 +11110,25 @@ def _populate_contract_from_plan(
                 task_count=task_count,
                 has_pr_metadata=contract.pr is not None,
             )
+        else:
+            # Parse succeeded but yielded neither phases nor PR metadata —
+            # this is the #1931 failure mode (empty contract with no error).
+            # Emit a discriminator so the gap is visible in audit logs.
+            logger.warning(
+                "contract_phases_ingest_failed",
+                pipeline_id=pipeline_id,
+                reason="empty_result",
+                warning_count=len(result.warnings),
+            )
 
     except Exception as e:
         logger.warning(
             "contract_phases_ingest_failed",
             pipeline_id=pipeline_id,
             reason="unexpected_exception",
+            source="parse_save",
             error=str(e),
+            exc_info=True,
         )
 
 
diff --git a/orchestrator/tests/test_populate_contract_audit_events.py b/orchestrator/tests/test_populate_contract_audit_events.py
index 75375feada..e3e537319d 100644
--- a/orchestrator/tests/test_populate_contract_audit_events.py
+++ b/orchestrator/tests/test_populate_contract_audit_events.py
@@ -15,7 +15,10 @@
   - ``plan_draft_missing``
   - ``contract_load_failed``
   - ``parse_failed``
-  - ``unexpected_exception`` (inner catch-all + outer ``_safe`` wrapper)
+  - ``empty_result`` — parse succeeded but yielded no phases / no PR
+    metadata (the #1931 failure mode)
+  - ``unexpected_exception`` with ``source="parse_save"`` (inner catch)
+    or ``source="safe_wrapper"`` (outer ``_safe`` wrapper)
 
 structlog output bypasses pytest's ``caplog`` fixture in this codebase
 (see ``test_decisions_routes.py``), so we patch the module-level
@@ -244,7 +247,57 @@ def test_unexpected_exception_inner_catch(self, tmp_path):
         failures = _ingest_failed_calls(mock_logger)
         assert len(failures) == 1
         assert failures[0].kwargs["reason"] == "unexpected_exception"
+        # source distinguishes inner catch from the outer _safe wrapper.
+        assert failures[0].kwargs["source"] == "parse_save"
         assert "parser exploded" in failures[0].kwargs["error"]
+        # Tracebacks must be preserved for unexpected exceptions.
+        assert failures[0].kwargs.get("exc_info") is True
+
+    def test_empty_result_emits_discriminator(self, tmp_path):
+        """Parse success with no phases and no PR metadata emits empty_result.
+
+        This is the #1931 failure mode: ``parse_plan`` returns
+        ``success=True`` but ``to_contract_phases()`` yields ``[]`` and
+        ``pr_title`` is None — the contract stays empty.  Without this
+        event the gap is invisible to operators.
+        """
+        from egg_contracts.loader import create_contract
+        from routes.pipelines import _populate_contract_from_plan
+
+        pipeline_id = "pipeline-empty-parse"
+        create_contract(pipeline_id=pipeline_id, title="Test", repo_root=tmp_path)
+
+        drafts_dir = tmp_path / ".egg-state" / "drafts"
+        drafts_dir.mkdir(parents=True, exist_ok=True)
+        (drafts_dir / f"{pipeline_id}-plan.md").write_text("# Plan\n")
+
+        # Force parse_plan to return success with no phases or PR metadata.
+        fake_result = MagicMock()
+        fake_result.success = True
+        fake_result.warnings = []
+        fake_result.to_contract_phases.return_value = []
+        fake_result.pr_title = None
+        fake_result.pr_description = None
+        fake_result.pr_test_plan = None
+        fake_result.pr_manual_steps = None
+
+        with (
+            patch("egg_contracts.plan_parser.parse_plan", return_value=fake_result),
+            patch("routes.pipelines.logger") as mock_logger,
+        ):
+            _populate_contract_from_plan(tmp_path, pipeline_id, "local")
+
+        failures = _ingest_failed_calls(mock_logger)
+        assert len(failures) == 1
+        assert failures[0].kwargs["reason"] == "empty_result"
+        assert failures[0].kwargs["pipeline_id"] == pipeline_id
+        # Success event must NOT be emitted in this case.
+        success_calls = [
+            c
+            for c in mock_logger.info.call_args_list
+            if c.args and c.args[0] == "contract_phases_populated"
+        ]
+        assert success_calls == []
 
 
 class TestSafeWrapper:
@@ -266,8 +319,12 @@ def test_safe_wrapper_emits_unexpected_exception_on_inner_raise(self, tmp_path):
         failures = _ingest_failed_calls(mock_logger)
         assert len(failures) == 1
         assert failures[0].kwargs["reason"] == "unexpected_exception"
+        # source distinguishes the outer wrapper from the inner catch.
+        assert failures[0].kwargs["source"] == "safe_wrapper"
         assert failures[0].kwargs["pipeline_id"] == "pipeline-wrap"
         assert "inner blew up" in failures[0].kwargs["error"]
+        # Tracebacks must be preserved for unexpected exceptions.
+        assert failures[0].kwargs.get("exc_info") is True
 
     def test_safe_wrapper_does_not_propagate(self, tmp_path):
         """Inner failures must not escape — HITL gate must remain reachable (#1890)."""