feat(core): retry a failed operation in place

[aaaaahaaaaa]

What

Phase 1 of the retry system: an operation that fails is re-executed in place, within a declared attempt budget, without the run ever seeing the intermediate failure.

Implements the retry spec sections 3 and 4, following the phase 1 plan. Core only: nothing in the platform reads a policy yet.

Why

Production runs fail daily on transient vendor faults, and the only retry that exists is manual and whole-run. The cheapest level that can heal such a failure is the operation itself, seconds later, in the same run.

The shape

RetryPolicy (new interloper/retry/ package, exported as il.RetryPolicy) carries numbers only: max_attempts, delay, backoff, max_delay, jitter. Whether a given error is worth another attempt is behaviour, not configuration, so it lives on Operation.retryable(error) next to Operation.failure(), and exception types never enter serialized config.

The policy is declared on the component whose unit it governs, which is what keeps one budget from meaning three things:

declared on	governs
Operation.retry	that operation's execution
Source.retry	the default its assets inherit
Job.retry	that job's runs (read by phase 2)

A source's policy reaches its assets through Source._resolve, the existing one-line-per-default mechanism alongside dataset, normalizer and materialization_strategy. Both decorators accept retry= with no change, since they split **overrides against the class's fields.

For the reviewer

Nothing retries yet. There is no instance-wide default, by design, so behaviour is unchanged until a component declares a policy. That is deliberate and recorded as a follow-up in the spec, with the two questions it needs answered first.

A retried attempt is not a verdict. Only an exhausted or declined failure calls mark_failed, so a node still working through its attempts cancels no dependents and does not trip fail_fast. OPERATION_RETRIED records the intermediate attempts and OPERATION_FAILED keeps meaning exhausted. That is the same rule the hooks will follow one level up in phase 2.

Attempts are their own event rows. RunState._operation_event_id is a uuid5 of (run_id, component_id, event_type), so without a change attempt 2's events would collide with attempt 1's and dedup away. The attempt now joins the key, and is left out when it is 1, so every id written before retries existed is unchanged and the host/child dedup still works.

The multi-process worker owns its loop and none of the reporting. Only the child sees the failures, but the parent holds the RunState and emits every event, so the errors of retried attempts travel back in the result tuple (now six elements) and the parent replays them through mark_retried. Counters and events therefore match what actually happened in the child.

Two test-surface notes: two existing job tests assert the exact config_schema property set, which legitimately grows by retry; and I added an end-to-end test through a real process pool beyond the plan, because without it the parent's replay path was only ever exercised with an empty list.

Verification

uv run ruff check, uv run ty check and uv run pytest all pass: 2883 passed, 3 skipped, up 28 tests.

Unrelated and pre-existing: test_worker_adopts_and_releases_the_parent_span_context fails when its file is run in isolation, tripping on assert "traceparent" in metadata before _worker is called. It is the known core global-state issue where telemetry is only initialised under the broader suite, and it passes in the runner directory and the full suite.

Next

Phase 2 (platform) makes a failed run queue its own next attempt, moves backfill accounting onto stacks, and gates hooks on a stack's verdict.

By Digitl

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!