diff --git a/src/conductor/engine/checkpoint.py b/src/conductor/engine/checkpoint.py
index 95f7f027..9deba044 100644
--- a/src/conductor/engine/checkpoint.py
+++ b/src/conductor/engine/checkpoint.py
@@ -295,8 +295,9 @@ def save_checkpoint(
 
             return checkpoint_path
 
-        except Exception:
-            logger.warning("Failed to save checkpoint", exc_info=True)
+        except Exception as exc:
+            logger.warning("Failed to save checkpoint (%s: %s)", type(exc).__name__, exc)
+            logger.debug("Checkpoint save traceback", exc_info=True)
             return None
 
     @staticmethod
@@ -510,8 +511,16 @@ def _delete_periodic_checkpoints(
         """
         try:
             candidates = CheckpointManager._periodic_checkpoints_for_run(workflow_path, run_id)
-        except Exception:
-            logger.warning("Failed to list checkpoints for %s", action, exc_info=True)
+        except Exception as exc:
+            logger.warning(
+                "Failed to list checkpoints for %s (run_id=%s, workflow=%s) (%s: %s)",
+                action,
+                run_id,
+                workflow_path,
+                type(exc).__name__,
+                exc,
+            )
+            logger.debug("Checkpoint listing traceback", exc_info=True)
             return
         # list_checkpoints sorts newest-first, so anything past keep_last is old.
         for cp in candidates[keep_last:]:
diff --git a/src/conductor/engine/validator.py b/src/conductor/engine/validator.py
index a642df72..611ea32d 100644
--- a/src/conductor/engine/validator.py
+++ b/src/conductor/engine/validator.py
@@ -192,12 +192,14 @@ async def validate(
         except asyncio.CancelledError:
             # Interrupt / cancellation must propagate — never silently pass.
             raise
-        except Exception:
+        except Exception as exc:
             logger.warning(
-                "Validator call failed for agent '%s'; treating as pass",
+                "Validator call failed for agent '%s'; treating as pass (%s: %s)",
                 agent.name,
-                exc_info=True,
+                type(exc).__name__,
+                exc,
             )
+            logger.debug("Validator call traceback for agent '%s'", agent.name, exc_info=True)
             return ValidationOutcome(passed=True, errored=True)
 
         passed, issues, parse_ok = self._parse(output.content)
diff --git a/src/conductor/engine/workflow.py b/src/conductor/engine/workflow.py
index 0f85d710..5b6bfeac 100644
--- a/src/conductor/engine/workflow.py
+++ b/src/conductor/engine/workflow.py
@@ -2138,8 +2138,13 @@ def _write_checkpoint(
                         if hasattr(p, "get_session_cwds"):
                             copilot_session_cwds = p.get_session_cwds()  # type: ignore[union-attr]
                         break
-        except Exception:
-            logger.warning("Failed to collect provider session IDs for checkpoint", exc_info=True)
+        except Exception as exc:
+            logger.warning(
+                "Failed to collect provider session IDs for checkpoint (%s: %s)",
+                type(exc).__name__,
+                exc,
+            )
+            logger.debug("Session ID collection traceback", exc_info=True)
             copilot_session_ids = None
             copilot_session_cwds = None
 
@@ -3036,12 +3041,14 @@ def _emit_v(event_type: str, data: dict[str, Any]) -> None:
                 outcome = await validate_coro
         except asyncio.CancelledError:
             raise
-        except Exception:
+        except Exception as exc:
             logger.warning(
-                "Validator call for '%s' timed out or failed; treating as pass",
+                "Validator call for '%s' timed out or failed; treating as pass (%s: %s)",
                 agent.name,
-                exc_info=True,
+                type(exc).__name__,
+                exc,
             )
+            logger.debug("Validator call traceback for '%s'", agent.name, exc_info=True)
             outcome = ValidationOutcome(passed=True, errored=True)
         _v_elapsed = _time.time() - _v_start
 
@@ -3094,15 +3101,23 @@ def _emit_v(event_type: str, data: dict[str, Any]) -> None:
             )
         except asyncio.CancelledError:
             raise
-        except Exception:
+        except Exception as exc:
             # The re-run hit a real failure (provider error, agent timeout, or
             # the retried output failed the agent's output schema). Fail open
             # to the original output, but surface it — otherwise enabling the
             # validator would silently downgrade a hard failure into a quiet
             # one. The original is recorded once by the caller under the
             # primary name; it is NOT also attributed to the validator row.
+            # The warning stays concise because the workflow continues; the
+            # full traceback is kept at debug level for diagnosis (issue #357).
             logger.warning(
-                "Validator re-run failed for '%s'; using original output",
+                "Validator re-run failed for '%s'; using original output (%s: %s)",
+                agent.name,
+                type(exc).__name__,
+                exc,
+            )
+            logger.debug(
+                "Validator re-run traceback for '%s'",
                 agent.name,
                 exc_info=True,
             )
diff --git a/src/conductor/providers/_event_format.py b/src/conductor/providers/_event_format.py
index a2d2c2e6..ef25c9fe 100644
--- a/src/conductor/providers/_event_format.py
+++ b/src/conductor/providers/_event_format.py
@@ -64,7 +64,7 @@ def emit_parse_recovery_event(
             },
         )
     except Exception:
-        logger.warning("Error in event_callback for agent_parse_recovery", exc_info=True)
+        logger.debug("Error in event_callback for agent_parse_recovery", exc_info=True)
 
 
 def format_tool_arguments(arguments: Any, max_length: int = 500) -> str | None:
diff --git a/src/conductor/providers/hermes.py b/src/conductor/providers/hermes.py
index 94fa5a3d..389c7264 100644
--- a/src/conductor/providers/hermes.py
+++ b/src/conductor/providers/hermes.py
@@ -724,7 +724,7 @@ def _fire(callback: EventCallback | None, event: str, data: dict[str, Any]) -> N
     try:
         callback(event, data)
     except Exception:
-        logger.warning("Error in event_callback for %s", event, exc_info=True)
+        logger.debug("Error in event_callback for %s", event, exc_info=True)
 
 
 async def _wait_for_event(event: asyncio.Event) -> None:
diff --git a/tests/test_engine/test_checkpoint.py b/tests/test_engine/test_checkpoint.py
index 2c8f27d7..d902e25e 100644
--- a/tests/test_engine/test_checkpoint.py
+++ b/tests/test_engine/test_checkpoint.py
@@ -16,6 +16,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import os
 import stat
 import sys
@@ -238,6 +239,45 @@ def test_never_raises_on_failure(self, tmp_path: Path) -> None:
 
         assert result is None
 
+    def test_save_checkpoint_failure_logs_concise_warning_and_debug_traceback(
+        self, tmp_path: Path, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """Requirement (issue #357): the handled fail-open path must not print a
+        traceback at WARNING level — a concise warning names the exception, and
+        the full traceback is only available at DEBUG level."""
+        wf = _write_workflow(tmp_path)
+        ctx = _make_context()
+        limits = _make_limits()
+        error = RuntimeError("err")
+
+        fake_dir = tmp_path / "no" / "such" / "dir"
+        with (
+            patch.object(CheckpointManager, "get_checkpoints_dir", return_value=fake_dir),
+            caplog.at_level(logging.DEBUG, logger="conductor.engine.checkpoint"),
+        ):
+            result = CheckpointManager.save_checkpoint(wf, ctx, limits, "a", error, {})
+
+        assert result is None
+
+        warnings = [
+            r
+            for r in caplog.records
+            if r.name == "conductor.engine.checkpoint"
+            and r.levelno == logging.WARNING
+            and "Failed to save checkpoint" in r.getMessage()
+        ]
+        assert len(warnings) == 1
+        assert warnings[0].exc_info is None
+        # The warning names the exception type so the cause is identifiable
+        # without a traceback (e.g. FileNotFoundError from the bad directory).
+        assert "FileNotFoundError" in warnings[0].getMessage()
+        debugs = [
+            r
+            for r in caplog.records
+            if r.name == "conductor.engine.checkpoint" and r.levelno == logging.DEBUG
+        ]
+        assert any(r.exc_info is not None for r in debugs)
+
     def test_handles_non_serializable_inputs(self, tmp_path: Path) -> None:
         wf = _write_workflow(tmp_path)
         ctx = _make_context()
@@ -949,6 +989,45 @@ def test_keep_last_zero_is_noop(self, tmp_path: Path) -> None:
 
         assert len(remaining) == 3
 
+    def test_listing_failure_logs_run_context_and_debug_traceback(
+        self, tmp_path: Path, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """Requirement (issue #357): when checkpoint rotation fails to list, the
+        WARNING must name the run and workflow so the leak is actionable without
+        DEBUG, and the traceback lives only at DEBUG level."""
+        wf = _write_workflow(tmp_path, "name: wf\n")
+
+        with (
+            patch.object(
+                CheckpointManager,
+                "_periodic_checkpoints_for_run",
+                side_effect=OSError("disk gone"),
+            ),
+            caplog.at_level(logging.DEBUG, logger="conductor.engine.checkpoint"),
+        ):
+            CheckpointManager.rotate_periodic_checkpoints(wf, "run-42", keep_last=1)
+
+        warnings = [
+            r
+            for r in caplog.records
+            if r.name == "conductor.engine.checkpoint"
+            and r.levelno == logging.WARNING
+            and "Failed to list checkpoints" in r.getMessage()
+        ]
+        assert len(warnings) == 1
+        assert warnings[0].exc_info is None
+        message = warnings[0].getMessage()
+        assert "run-42" in message
+        assert str(wf) in message
+        assert "OSError" in message
+        assert "disk gone" in message
+        debugs = [
+            r
+            for r in caplog.records
+            if r.name == "conductor.engine.checkpoint" and r.levelno == logging.DEBUG
+        ]
+        assert any(r.exc_info is not None for r in debugs)
+
 
 class TestCleanupPeriodicForRun:
     def test_removes_all_periodic_for_run_keeps_failure(self, tmp_path: Path) -> None:
diff --git a/tests/test_engine/test_validator.py b/tests/test_engine/test_validator.py
index c74a700f..3375faa1 100644
--- a/tests/test_engine/test_validator.py
+++ b/tests/test_engine/test_validator.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import logging
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
@@ -207,6 +208,42 @@ async def test_provider_error_fails_open(self) -> None:
         assert outcome.errored is True
         assert outcome.output is None
 
+    @pytest.mark.asyncio
+    async def test_provider_error_logs_concise_warning_and_debug_traceback(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """Requirement (issue #357): the handled fail-open path must not print a
+        traceback at WARNING level — a concise warning names the exception, and
+        the full traceback is only available at DEBUG level."""
+        agent = _agent()
+        provider = MagicMock()
+        provider.execute = AsyncMock(side_effect=RuntimeError("boom"))
+
+        with caplog.at_level(logging.DEBUG, logger="conductor.engine.validator"):
+            outcome = await OutputValidator().validate(agent, "p", {"summary": "x"}, provider)
+
+        assert outcome.passed is True
+        assert outcome.errored is True
+
+        warnings = [
+            r
+            for r in caplog.records
+            if r.name == "conductor.engine.validator"
+            and r.levelno == logging.WARNING
+            and "Validator call failed" in r.getMessage()
+        ]
+        assert len(warnings) == 1
+        assert warnings[0].exc_info is None
+        assert "RuntimeError" in warnings[0].getMessage()
+        assert "boom" in warnings[0].getMessage()
+        assert "reviewer" in warnings[0].getMessage()
+        debugs = [
+            r
+            for r in caplog.records
+            if r.name == "conductor.engine.validator" and r.levelno == logging.DEBUG
+        ]
+        assert any(r.exc_info is not None for r in debugs)
+
     @pytest.mark.asyncio
     async def test_malformed_output_fails_open(self) -> None:
         agent = _agent()
diff --git a/tests/test_engine/test_validator_integration.py b/tests/test_engine/test_validator_integration.py
index b35976e5..d6926476 100644
--- a/tests/test_engine/test_validator_integration.py
+++ b/tests/test_engine/test_validator_integration.py
@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 from typing import Any
 
 import pytest
@@ -453,7 +454,9 @@ async def exec_fn(*, agent: AgentDef, rendered_prompt: str, **kw: Any) -> AgentO
         assert any(r.input_tokens == 900 and r.output_tokens == 400 for r in vrows)
 
     @pytest.mark.asyncio
-    async def test_rerun_failure_keeps_original_no_double_count_and_emits_event(self) -> None:
+    async def test_rerun_failure_keeps_original_no_double_count_and_emits_event(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
         async def exec_fn(*, agent: AgentDef, rendered_prompt: str, **kw: Any) -> AgentOutput:
             if agent.output and "passed" in agent.output:
                 return AgentOutput(
@@ -471,9 +474,10 @@ async def exec_fn(*, agent: AgentDef, rendered_prompt: str, **kw: Any) -> AgentO
         )
         events: list[tuple[str, dict[str, Any]]] = []
 
-        result = await engine._apply_validator(
-            agent, original, 0.5, {}, executor, None, lambda e, d: events.append((e, d))
-        )
+        with caplog.at_level(logging.DEBUG, logger="conductor.engine.workflow"):
+            result = await engine._apply_validator(
+                agent, original, 0.5, {}, executor, None, lambda e, d: events.append((e, d))
+            )
 
         assert result is original  # original kept on re-run failure
         # Discarded run is NOT recorded when the re-run fails (no double count):
@@ -486,6 +490,27 @@ async def exec_fn(*, agent: AgentDef, rendered_prompt: str, **kw: Any) -> AgentO
         failed = [d for (e, d) in events if e == "agent_validation_failed"]
         assert any(d.get("rerun_errored") for d in failed)
 
+        # Requirement (issue #357): the handled fail-open path must not print a
+        # traceback at WARNING level — a concise warning names the exception, and
+        # the full traceback is only available at DEBUG level.
+        warnings = [
+            r
+            for r in caplog.records
+            if r.name == "conductor.engine.workflow"
+            and r.levelno == logging.WARNING
+            and "Validator re-run failed" in r.getMessage()
+        ]
+        assert len(warnings) == 1
+        assert warnings[0].exc_info is None
+        assert "RuntimeError" in warnings[0].getMessage()
+        assert "rerun boom" in warnings[0].getMessage()
+        debugs = [
+            r
+            for r in caplog.records
+            if r.name == "conductor.engine.workflow" and r.levelno == logging.DEBUG
+        ]
+        assert any(r.exc_info is not None for r in debugs)
+
     @pytest.mark.asyncio
     async def test_partial_rerun_keeps_original(self) -> None:
         async def exec_fn(*, agent: AgentDef, rendered_prompt: str, **kw: Any) -> AgentOutput: