From 65bc4887e9fa269d290897af35adca87b7ba8999 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 24 Oct 2026 22:51:52 -0400 Subject: [PATCH 1/2] Every command failure pickles across a worker Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_013ejYrMLnLWah8wCDDcb57V From e39a920a4d9f4cc39da2f54832e389089df42fd3 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Sat, 24 Oct 2026 22:52:22 -0400 Subject: [PATCH 2/2] Move the failure pickling to CommandFailure GitCommandError takes (repository, run), so a raised one unpickled in a ForkedProcessPool parent called __init__(message), died with TypeError and surfaced as BrokenProcessPool with the git message lost. CoordinationError already rebuilt itself from class, args and state; that __reduce__ now lives on the family owner, so every member pickles, and the test walks the family. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_013ejYrMLnLWah8wCDDcb57V --- src/agent_comms/coordination_errors.py | 12 ++++--- tests/test_command_failure_pickles.py | 43 ++++++++++++++++++++++++++ 2 files changed, 50 insertions(+), 5 deletions(-) create mode 100644 tests/test_command_failure_pickles.py diff --git a/src/agent_comms/coordination_errors.py b/src/agent_comms/coordination_errors.py index a58ad79410..f8e9d25056 100644 --- a/src/agent_comms/coordination_errors.py +++ b/src/agent_comms/coordination_errors.py @@ -12,7 +12,13 @@ class DrainDeferred(Exception): # noqa: N818 - capability, not an outcome class CommandFailure(Exception): # noqa: N818 - capability, not an outcome - """A failure a command reports to its caller as a failed result instead of crashing.""" + """A failure a command reports to its caller as a failed result instead of crashing. + + It crosses a worker's pipe as its class, args and state, never through a member's own + ``__init__``, whose parameters are not its args, and keeps its original cause.""" + + def __reduce__(self): + return __newobj__, (type(self), *self.args), self.__dict__ | {"__cause__": self.__cause__} class DrainUnavailable(DrainDeferred): @@ -22,10 +28,6 @@ class DrainUnavailable(DrainDeferred): class CoordinationError(RuntimeError, DrainUnavailable, CommandFailure): """Base class for fail-loud coordinator errors.""" - def __reduce__(self): - """Keep the original cause when an observation sends this error over its pipe.""" - return __newobj__, (type(self), *self.args), self.__dict__ | {"__cause__": self.__cause__} - class DeferredRead(Exception): # noqa: N818 - capability, not an outcome """A read refused only for now, for currentness or contention; retrying may answer it.""" diff --git a/tests/test_command_failure_pickles.py b/tests/test_command_failure_pickles.py new file mode 100644 index 0000000000..f077a1e352 --- /dev/null +++ b/tests/test_command_failure_pickles.py @@ -0,0 +1,43 @@ +"""Every command failure crosses a worker's pipe whole, whatever its constructor takes.""" + +import importlib +import pickle +import pkgutil +import subprocess + +import pytest + +import agent_comms +from agent_comms.coordination_errors import CommandFailure +from agent_comms.repo import GitCommandError + + +def members(family: type) -> set[type]: + return {member for child in family.__subclasses__() for member in {child, *members(child)}} + + +def declared() -> list[type[CommandFailure]]: + """The family once every module declaring a member is imported.""" + for module in pkgutil.walk_packages(agent_comms.__path__, f"{agent_comms.__name__}."): + if ".skills." not in module.name: + importlib.import_module(module.name) + return sorted(members(CommandFailure), key=lambda member: member.__qualname__) + + +@pytest.mark.parametrize("member", declared(), ids=lambda member: member.__qualname__) +def test_each_member_unpickles_with_its_message_state_and_cause(member) -> None: + failure = member.__new__(member, "git exited 128") + failure.detail = "kept" + failure.__cause__ = ValueError("root") + copy = pickle.loads(pickle.dumps(failure)) + assert type(copy) is member + assert copy.args == ("git exited 128",) + assert copy.detail == "kept" + assert copy.__cause__.args == ("root",) + + +def test_a_raised_git_failure_keeps_its_git_message() -> None: + run = subprocess.CompletedProcess(("git",), 128, b"", b"fatal: bad object\n") + copy = pickle.loads(pickle.dumps(GitCommandError("/repo", run))) + assert str(copy) == "Git in /repo: fatal: bad object" + assert (copy.returncode, copy.stderr) == (128, "fatal: bad object\n")