fix(engine): drop tracebacks from warning logs on handled fail-open paths

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:]:

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)

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,
  )

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:

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:

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:

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()

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: