fix(wac): one failure record for tasks and steps, in every round (#10368)

* fix(wac): hand a caught task and step failure the same shape in every round Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * refactor(wac): decide the failure record once, server-side Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(wac): leave a legacy SDK's failure marker untouched, and ship wacError to jsr Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(wac): carry a step's custom error fields, and bound the stack in bytes Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(wac): keep a step's extra fields serializable and bounded Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(wac): record a non-Error throw the way a task records it Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(wac): guard the last unguarded throw site in the step marker Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(wac): make failure reporting non-throwing on both clients Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(wac): take the step traceback the way the executor takes it Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(wac): contain the reads that happen before a failure is checkpointed Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(wac): fall back to the checkpointed marker, not the live one Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(wac): keep non-finite fields and hostile proxies out of the checkpoint path Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix(wac): keep the snapshot that passed the serialization probe Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * chore(wac): keep the failure-record module's surface to what is used Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-18 16:02:10 +00:00 · 2026-07-28 13:15:24 +02:00
parent 8d684c0b23
commit aeaea57ca1
17 changed files with 1143 additions and 136 deletions
@@ -976,13 +976,21 @@ class TestRaisingInlineStepIsCheckpointed:
    on the never-resolving future forever.
    """

+    # A failed child job reports `{"error": {"name", "message", "stack"}}`, and a
+    # failed step() has to be indistinguishable from it. The stack is a traceback
+    # string, asserted separately.
    MARKER = {
        "__wmill_error": True,
        "message": "boom",
        "step_key": "risky",
-        "result": {"error": "boom", "type": "ValueError"},
+        "result": {"error": {"name": "ValueError", "message": "boom"}},
    }

+    @staticmethod
+    def _without_stack(marker: dict) -> dict:
+        error = {k: v for k, v in marker["result"]["error"].items() if k != "stack"}
+        return {**marker, "result": {**marker["result"], "error": error}}
+
    @staticmethod
    def _boom():
        raise ValueError("boom")
@@ -1003,36 +1011,205 @@ class TestRaisingInlineStepIsCheckpointed:
        r = _run_workflow(self._wf(), {}, {"x": 5})
        assert r["type"] == "inline_checkpoint"
        assert r["key"] == "risky"
-        assert r["result"] == self.MARKER
+        assert self._without_stack(r["result"]) == self.MARKER
+        stack = r["result"]["result"]["error"]["stack"]
+        # Frames only and the SDK's own `result = fn()` frame dropped, the way
+        # the python executor formats a failed job's stack.
+        assert 'raise ValueError("boom")' in stack
+        assert "result = fn()" not in stack
+        assert not stack.startswith("Traceback")
+
+    def test_custom_exception_attributes_survive_under_extra(self):
+        """A failed child job reports custom attributes under ``error.extra``;
+        a step dropping them would make the same exception carry less depending
+        on how it was run."""
+        from wmill.client import _step_error_marker
+
+        class HttpError(ValueError):
+            def __init__(self):
+                super().__init__("429")
+                self.code = 429
+
+        error = _step_error_marker("k", HttpError())["result"]["error"]
+        assert error["extra"] == {"code": 429}
+        assert error["name"] == "HttpError"
+        assert "extra" not in _step_error_marker("k", ValueError("plain"))["result"]["error"]
+
+    def test_unserializable_attributes_do_not_cost_the_fast_path(self):
+        """The fast-path POST serializes strictly, so an exception holding a
+        live object — ``resp.raise_for_status()`` is the common one — must not
+        make the marker unserializable and drop the step onto the slow path."""
+        import json as _json
+
+        from wmill.client import _step_error_marker
+
+        class Boom(Exception):
+            def __init__(self):
+                super().__init__("boom")
+                self.response = object()
+                self.status = 429
+
+        marker = _step_error_marker("k", Boom())
+        _json.dumps(marker)  # raises if an attribute leaked through unserialized
+        assert marker["result"]["error"]["extra"]["status"] == 429
+
+    def test_an_exception_whose_str_raises_still_reports(self):
+        """Every coercion of the user's exception runs inside the ``except``
+        reporting it, so one that raises would replace their failure with an
+        unrelated one and leave the step uncheckpointed."""
+        import json as _json
+
+        from wmill.client import _step_error_marker
+
+        class Hostile(Exception):
+            def __str__(self):
+                raise RuntimeError("cannot be rendered")
+
+        marker = _step_error_marker("k", Hostile())
+        _json.dumps(marker)
+        assert marker["result"]["error"]["name"] == "Hostile"
+        assert "unrepresentable" in marker["result"]["error"]["message"]
+
+        # ...including one that makes reading its own traceback raise
+        class HostileTraceback(Exception):
+            def __getattribute__(self, item):
+                if item == "__traceback__":
+                    raise RuntimeError("no traceback for you")
+                return super().__getattribute__(item)
+
+        marker = _step_error_marker("k", HostileTraceback())
+        _json.dumps(marker)
+        assert marker["result"]["error"]["name"] == "HostileTraceback"
+
+        # ...or reading its own attributes
+        class HostileDict(Exception):
+            def __getattribute__(self, item):
+                if item == "__dict__":
+                    raise RuntimeError("no attributes for you")
+                return super().__getattribute__(item)
+
+        marker = _step_error_marker("k", HostileDict())
+        _json.dumps(marker)
+        assert marker["result"]["error"]["name"] == "HostileDict"
+
+        # A float is serializable, so `default=` never sees NaN — it would go out
+        # as a bare `NaN` literal, which is not JSON and which the backend
+        # rejects, so the step could not be checkpointed at all.
+        class NotFinite(Exception):
+            def __init__(self):
+                super().__init__("nan")
+                self.value = float("nan")
+                self.limit = float("inf")
+
+        marker = _step_error_marker("k", NotFinite())
+        _json.dumps(marker, allow_nan=False)
+        assert marker["result"]["error"]["extra"] == {"value": "NaN", "limit": "Infinity"}

    def test_fast_path_posts_error_and_raises_the_replay_exception(self, monkeypatch):
        """The default path: the checkpoint is POSTed and the workflow body gets
        the same ``TaskError`` a replay rebuilds from the marker — raising the
        original ``ValueError`` here would make ``except ValueError:`` catch on
-        this run and miss on the next one."""
+        this run and miss on the next one. ``except`` is control flow, so
+        anything a handler can branch on has to be identical in both rounds."""
        _set_inline_fast_path_env(monkeypatch)

-        stub = _StubInlineClient()
+        class _EchoingStub(_StubInlineClient):
+            """The endpoint normalizes the failure before storing it and echoes
+            back what it stored. The echo deliberately differs from what was
+            posted, so the assertions below can tell which copy was raised from."""
+
+            async def post(self, url, content=None):
+                await super().post(url, content=content)
+                stored = {**self.posted[-1]["result"], "message": "normalized by the backend"}
+
+                class _Response:
+                    # the endpoint answers with a JSON body; a backend predating
+                    # the echo answers without one, which is how the client tells
+                    # "no echo" from "an echo it could not read"
+                    headers = {"content-type": "application/json"}
+
+                    def raise_for_status(self):
+                        pass
+
+                    def json(self):
+                        return {"failure": stored}
+
+                return _Response()
+
+        stub = _EchoingStub()
        posted = stub.posted

        async def run():
            ctx = WorkflowCtx({})
            ctx._inline_http_client = stub
-            with pytest.raises(TaskError, match="boom") as live:
+            with pytest.raises(TaskError) as live:
                await ctx._run_inline_step("risky", self._boom)
-            # ...and the replay of that very checkpoint raises the same thing.
-            replayed = WorkflowCtx({"completed_steps": {"risky": self.MARKER}})
-            with pytest.raises(TaskError, match="boom") as replay:
+            # The live round raised from the record the backend stored, not from
+            # the marker it posted: that is what keeps the two rounds identical
+            # even if the SDK and the backend ever build a record differently.
+            stored = {**posted[0]["result"], "message": "normalized by the backend"}
+            assert str(live.value) == "normalized by the backend"
+
+            # ...and the replay of that very record raises the same thing.
+            replayed = WorkflowCtx({"completed_steps": {"risky": stored}})
+            with pytest.raises(TaskError) as replay:
                await replayed._run_inline_step("risky", self._boom)
            assert type(live.value) is type(replay.value)
            assert live.value.args == replay.value.args
-            assert live.value.result == replay.value.result == self.MARKER["result"]
-            assert isinstance(live.value.__cause__, ValueError)
+            assert live.value.result == replay.value.result == stored["result"]
+            assert live.value.step_key == replay.value.step_key == "risky"
+            # A step has no child job to name, and nothing hangs off __cause__:
+            # a replay has no original exception to chain, so neither round does.
+            assert live.value.child_job_id is replay.value.child_job_id is None
+            assert live.value.__cause__ is replay.value.__cause__ is None

        asyncio.run(run())
        assert len(posted) == 1
        assert posted[0]["key"] == "risky"
-        assert posted[0]["result"] == self.MARKER
+        assert self._without_stack(posted[0]["result"]) == self.MARKER
+
+    def test_a_missing_echo_is_not_the_same_as_an_unreadable_one(self, monkeypatch):
+        """A backend predating the echo answers without a JSON body and the
+        locally checkpointed marker stands in. A JSON body that will not parse
+        means the stored record exists but is unknown, so the round has to end
+        and let the next one read whatever the backend actually kept."""
+        _set_inline_fast_path_env(monkeypatch)
+
+        def _client(headers, json_impl):
+            class _Response:
+                def raise_for_status(self):
+                    pass
+
+            _Response.headers = headers
+            _Response.json = json_impl
+
+            class _Client(_StubInlineClient):
+                async def post(self, url, content=None):
+                    await super().post(url, content=content)
+                    return _Response()
+
+            return _Client()
+
+        def _boom_json(self):
+            raise ValueError("not json")
+
+        async def run():
+            # no JSON body: the fast path still completes, raising the failure
+            ctx = WorkflowCtx({})
+            ctx._inline_http_client = _client({}, _boom_json)
+            with pytest.raises(TaskError):
+                await ctx._run_inline_step("risky", self._boom)
+
+            # a JSON body that will not parse: fall through to the suspend path
+            ctx = WorkflowCtx({})
+            ctx._inline_http_client = _client(
+                {"content-type": "application/json"}, _boom_json
+            )
+            with pytest.raises(_StepSuspend) as suspend:
+                await ctx._run_inline_step("risky", self._boom)
+            assert suspend.value.dispatch_info["key"] == "risky"
+
+        asyncio.run(run())

    def test_replay_reraises_and_does_not_hang(self):
        checkpoint = {
@@ -2654,6 +2654,8 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]:

 import asyncio as _asyncio
 import contextvars as _contextvars
+import sys as _sys
+import traceback as _traceback


 def _assert_usable_step_key(key: str, what: str) -> None:
@@ -2687,21 +2689,62 @@ class _StepFailure(BaseException):


 class TaskError(Exception):
-    """Raised when a WAC task step failed.
+    """Raised when a WAC ``task`` or ``step`` failed.

    Attributes:
        step_key: The checkpoint key of the failed step.
-        child_job_id: The UUID of the failed child job.
-        result: The error result from the child job.
+        child_job_id: The UUID of the failed child job, or ``None`` for a
+            ``step()``, which runs in the workflow job and has no child job.
+        result: ``{"error": {"name", "message", "stack"?, "extra"?}}`` — the
+            same shape whether a task or a step failed. ``name`` and ``message``
+            are always present; ``stack`` only when the failure had a traceback,
+            and ``extra`` only when it carried custom fields of its own, dropped
+            with ``extra_omitted: True`` beside it when too large to checkpoint.
    """

-    def __init__(self, message: str, *, step_key: str = "", child_job_id: str = "", result=None):
+    def __init__(self, message: str, *, step_key: str = "", child_job_id: Optional[str] = None, result=None):
        super().__init__(message)
        self.step_key = step_key
        self.child_job_id = child_job_id
        self.result = result


+def _safe_str(o) -> str:
+    """``str()`` on the failing side's own object, which can raise in turn — a
+    detached ORM row, a proxy over a closed connection, an ``__str__`` that
+    itself fails. Every coercion here runs inside the ``except`` that is
+    reporting the user's failure, so an escape would replace their error with an
+    unrelated one and skip the checkpoint entirely."""
+    try:
+        return str(o)
+    except Exception:
+        return f"<unrepresentable {type(o).__name__}>"
+
+
+def _step_error_stack(exc: BaseException) -> str:
+    """The traceback of a failed ``step()`` body, formatted the way the python
+    executor formats a failed job's: frames only, and the frame that called into
+    the user's code dropped. Here that first frame is ``_run_inline_step``'s own
+    ``result = fn()``, the counterpart of the generated wrapper frame the
+    executor strips, so a step's stack and a task's stack read alike.
+
+    Taken from ``sys.exc_info()`` the way the executor takes it, falling back to
+    the attribute: an exception overriding ``__getattribute__`` makes reading
+    ``__traceback__`` raise, and this runs inside the ``except`` reporting the
+    user's failure, so an escape would lose both their error and the checkpoint.
+    """
+    tb = _sys.exc_info()[2]
+    if tb is None:
+        try:
+            tb = exc.__traceback__
+        except Exception:
+            return ""
+    try:
+        return "".join(_traceback.format_tb(tb)[1:]).strip()
+    except Exception:
+        return ""
+
+
 def _json_round_trip(value):
    """Put a value through the checkpoint's encoding without checkpointing it, so
    the paths that never persist anything still hand back the shape the ones that
@@ -2711,23 +2754,69 @@ def _json_round_trip(value):

 def _step_error_marker(key: str, exc: BaseException) -> dict:
    """Serialize a failed ``step()`` body into the ``__wmill_error`` marker that
-    task failures also use, so it can be stored in ``completed_steps``."""
+    task failures also use, so it can be stored in ``completed_steps``.
+
+    The marker's final shape is decided by the backend (``wac_failure_record``),
+    which normalizes task failures through the same function; what is built here
+    is the raw material plus the envelope the backend recognizes."""
+    error = {"name": type(exc).__name__, "message": _safe_str(exc)}
+    stack = _step_error_stack(exc)
+    if stack:
+        error["stack"] = stack
+    # Custom attributes go under ``extra``, the same key the python executor uses
+    # for a failed child job, so an exception carrying e.g. a ``code`` keeps it
+    # whether it failed as a task or as a step.
+    #
+    # Coerced through ``default=str`` the way the executor writes its own error:
+    # the fast-path POST serializes strictly, and the commonest failing step
+    # there is — ``resp.raise_for_status()``, whose ``__dict__`` holds a request
+    # and a response object — would otherwise fail to serialize and silently
+    # drop every such failure onto the slow suspend-and-replay path.
+    # ``getattr``'s default only swallows ``AttributeError``; an exception
+    # overriding ``__getattribute__`` raises whatever it likes from this read.
+    try:
+        extra = getattr(exc, "__dict__", None)
+    except Exception:
+        extra = None
+    if extra:
+        # ``default`` is where json hands back the objects it cannot represent,
+        # and ``str()`` on a detached ORM row or a proxy over a closed
+        # connection raises in turn. This runs inside the ``except`` that is
+        # reporting the user's failure, so an escape here would replace their
+        # error with an unrelated one and skip the checkpoint entirely.
+        # Narrow: what json raises for something it cannot represent. A broader
+        # catch would hide a mistake in this function as a silently missing
+        # field, which is how it read before.
+        try:
+            # ``parse_constant`` catches the one thing ``default`` cannot: a
+            # float is serializable, so ``NaN``/``Infinity`` pass through as
+            # bare literals that are not JSON and that the backend's extractor
+            # rejects — taking the whole checkpoint down with them. Kept as
+            # their text rather than dropped, so the attribute still says
+            # something.
+            error["extra"] = json.loads(
+                json.dumps(extra, default=_safe_str), parse_constant=lambda c: c
+            )
+        except (TypeError, ValueError, RecursionError):
+            pass
    return {
        "__wmill_error": True,
-        "message": str(exc),
+        "message": _safe_str(exc),
        "step_key": key,
-        "result": {"error": str(exc), "type": type(exc).__name__},
+        "result": {"error": error},
    }


-def _step_error_from_marker(marker: dict, name: str) -> TaskError:
-    """Rebuild the exception a failed step raises. Both the run that produced the
-    failure and every later replay go through here, so a workflow's ``except``
-    clauses see the same type either way."""
+def _task_error_from_marker(marker: dict, fallback_message: str) -> TaskError:
+    """Rebuild the exception a failed task or step raises. The run that produced
+    the failure and every later replay go through here: ``except`` is control
+    flow, ``@workflow`` re-runs its body from the top every round, so a handler
+    that branches on the failure it caught must be handed the same thing in
+    every round or it dispatches different tasks on the way back."""
    return TaskError(
-        marker.get("message", f"Step '{name}' failed"),
+        marker.get("message") or fallback_message,
        step_key=marker.get("step_key", ""),
-        child_job_id=marker.get("child_job_id", ""),
+        child_job_id=marker.get("child_job_id"),
        result=marker.get("result"),
    )

@@ -2792,12 +2881,7 @@ class WorkflowCtx:
        if key in self._completed:
            val = self._completed[key]
            if isinstance(val, dict) and val.get("__wmill_error"):
-                raise TaskError(
-                    val.get("message", f"Task '{name}' failed"),
-                    step_key=val.get("step_key", ""),
-                    child_job_id=val.get("child_job_id", ""),
-                    result=val.get("result"),
-                )
+                raise _task_error_from_marker(val, f"Task '{name}' failed")
            return self._resolved(val)

        if self._executing_key is not None:
@@ -2904,7 +2988,7 @@ class WorkflowCtx:
        if key in self._completed:
            val = self._completed[key]
            if isinstance(val, dict) and val.get("__wmill_error"):
-                raise _step_error_from_marker(val, name)
+                raise _task_error_from_marker(val, f"Step '{name}' failed")
            return val

        if self._executing_key is not None:
@@ -2916,16 +3000,25 @@ class WorkflowCtx:
        t0 = _time_mod.monotonic()
        # A raised step still has to reach ``completed_steps``, or a replay with
        # ``_executing_key`` set finds nothing recorded and parks forever on the
-        # ``_asyncio.Future()`` above. ``_StepSuspend`` and ``CancelledError`` are
-        # ``BaseException``, so they pass through untouched.
-        step_error: Optional[Exception] = None
+        # ``_asyncio.Future()`` above. The control-flow signals (``_StepSuspend``,
+        # ``_StepFailure``) and ``CancelledError`` are ``BaseException``, so they
+        # pass through untouched.
+        step_failed = False
        try:
            result = fn()
            if _asyncio.iscoroutine(result):
                result = await result
        except Exception as _exc:
-            step_error = _exc
+            step_failed = True
            result = _step_error_marker(key, _exc)
+            # The failure is reported as a value from here on, so nothing else
+            # prints the traceback. Without this a step that fails and is never
+            # caught leaves a job log whose deepest frame is inside this client.
+            print(f"--- WAC: {key} failed ---")
+            print(f"{type(_exc).__name__}: {_safe_str(_exc)}")
+            _step_stack = result["result"]["error"].get("stack")
+            if _step_stack:
+                print(_step_stack)
        duration_ms = int((_time_mod.monotonic() - t0) * 1000)

        # Fast path: POST the delta to the new per-job API endpoint and return
@@ -2943,6 +3036,7 @@ class WorkflowCtx:
        _token = os.environ.get("WM_TOKEN")
        if _fast_path_enabled and _job_id and _workspace and _base and _token:
            _fast_path_ok = False
+            _stored_failure = None
            _replay_result = None
            try:
                # ``default=str`` is the encoder the worker wrapper uses on the
@@ -2978,6 +3072,20 @@ class WorkflowCtx:
                        content=_payload,
                    )
                    _resp.raise_for_status()
+                    if step_failed:
+                        # The backend normalizes the failure before storing it,
+                        # and hands back what it stored. Raising from that, not
+                        # from the marker posted above, is what makes this round
+                        # and every replay read the same record even if the two
+                        # sides ever disagree about how to build one.
+                        #
+                        # A backend predating the echo answers without a JSON
+                        # body, and the round-tripped marker below stands in. A
+                        # JSON body that will not parse is different: the record
+                        # may already be committed and its content is unknown,
+                        # so let it raise and take the suspend path instead.
+                        if "json" in _resp.headers.get("content-type", ""):
+                            _stored_failure = (_resp.json() or {}).get("failure")
                _fast_path_ok = True
            except Exception as _e:
                logger.info(
@@ -2987,12 +3095,21 @@ class WorkflowCtx:
                )
                # fall through to the legacy suspend path
            if _fast_path_ok:
-                # Raise what a replay would rebuild from the marker, never the
+                # Raise what a replay would rebuild from the record, never the
                # original: a replay cannot reconstruct the original type, so
                # raising it here would make ``except ValueError:`` catch on this
-                # run and miss on the next. ``__cause__`` is for tracebacks only.
-                if step_error is not None:
-                    raise _step_error_from_marker(result, name) from step_error
+                # run and miss on the next. Nothing is chained onto
+                # ``__cause__`` for the same reason — the traceback a replay can
+                # still show is in ``result["error"]["stack"]``. ``_stored_failure``
+                # is None against a backend that predates the echoed record,
+                # which is what ``_replay_result`` below stands in for.
+                if step_failed:
+                    # ``_replay_result``, not ``result``: the fallback has to be
+                    # what the checkpoint holds, so the round that ran the body
+                    # reads what every replay of it will.
+                    raise _task_error_from_marker(
+                        _stored_failure or _replay_result, f"Step '{name}' failed"
+                    )
                # Return the round trip of what was checkpointed, never the
                # in-memory value: handing back the live object would let the
                # round that ran the body branch on a type — tuple, datetime —