From 51ef4e84ee578bbe713c0e4f10703d02109bff0d Mon Sep 17 00:00:00 2001 From: Genadij Blinov Date: Tue, 4 Aug 2026 01:10:36 +0300 Subject: [PATCH 1/2] fix(engine): drop tracebacks from warning logs on handled fail-open paths When a validator-triggered agent re-run fails, the engine correctly fails open and keeps the original output, but logged the warning with exc_info=True, so the normal CLI output showed a full Python traceback even though the workflow continued successfully. The warning is now concise (exception type and message), with the full traceback kept at DEBUG level. The same pattern is applied to the other handled fail-open paths: the validator call itself (engine/validator.py and workflow.py), session-ID collection for checkpoints, checkpoint save, and checkpoint rotation listing. Event-callback errors in hermes and _event_format move to logger.debug, matching the existing convention in claude_agent_sdk and _pydantic_ai. The regression test now asserts there is no traceback at WARNING level and that it remains available at DEBUG level. Closes #357 --- src/conductor/engine/checkpoint.py | 12 ++++--- src/conductor/engine/validator.py | 8 +++-- src/conductor/engine/workflow.py | 29 ++++++++++++---- src/conductor/providers/_event_format.py | 2 +- src/conductor/providers/hermes.py | 2 +- .../test_engine/test_validator_integration.py | 33 ++++++++++++++++--- 6 files changed, 66 insertions(+), 20 deletions(-) diff --git a/src/conductor/engine/checkpoint.py b/src/conductor/engine/checkpoint.py index 95f7f027..4bdec0f7 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,11 @@ 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 (%s: %s)", action, 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_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: From 54d4a60da9ee223e6697a6ebc505df4e4d57e70a Mon Sep 17 00:00:00 2001 From: Genadij Blinov Date: Tue, 4 Aug 2026 18:14:28 +0300 Subject: [PATCH 2/2] test(engine): pin fail-open logging contract and name run context in checkpoint warnings Address review feedback on PR #367: - Include run_id and workflow_path in the checkpoint rotation/cleanup listing-failure WARNING so a silent checkpoint leak is actionable without turning on DEBUG. - Add caplog regression tests for save_checkpoint and rotate_periodic_checkpoints asserting the new contract: exactly one concise WARNING without exc_info, full traceback only at DEBUG level. - Add the matching caplog test for engine/validator.py's provider-error fail-open path so a future refactor can't quietly reintroduce exc_info=True on the WARNING. --- src/conductor/engine/checkpoint.py | 7 ++- tests/test_engine/test_checkpoint.py | 79 ++++++++++++++++++++++++++++ tests/test_engine/test_validator.py | 37 +++++++++++++ 3 files changed, 122 insertions(+), 1 deletion(-) diff --git a/src/conductor/engine/checkpoint.py b/src/conductor/engine/checkpoint.py index 4bdec0f7..9deba044 100644 --- a/src/conductor/engine/checkpoint.py +++ b/src/conductor/engine/checkpoint.py @@ -513,7 +513,12 @@ def _delete_periodic_checkpoints( candidates = CheckpointManager._periodic_checkpoints_for_run(workflow_path, run_id) except Exception as exc: logger.warning( - "Failed to list checkpoints for %s (%s: %s)", action, type(exc).__name__, exc + "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 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()