From 90699a1b231d6fee28d2800bd61597cdfd8e618e Mon Sep 17 00:00:00 2001
From: Rick Gorman <rickgorman@users.noreply.github.com>
Date: Thu, 20 Aug 2026 22:31:30 -0700
Subject: [PATCH] fix: route sync output_invalid through _fail_run

The async tick path already retries missing declared outputs via
_fail_run + exit rules. The sync _process_launch_result path marked
FAILED and halted the job immediately, so a loop-on-output_invalid
rule never fired. Match the async path.

Reproduce: uv run pytest tests/test_engine.py::TestSyncOutputInvalidExitRules -q
On unpatched master that test fails with job FAILED after one empty artifact.
---
 src/stepwise/engine.py | 16 +++++--------
 tests/test_engine.py   | 52 ++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 58 insertions(+), 10 deletions(-)

diff --git a/src/stepwise/engine.py b/src/stepwise/engine.py
index b4f60849..07f289ee 100644
--- a/src/stepwise/engine.py
+++ b/src/stepwise/engine.py
@@ -2785,18 +2785,14 @@ def _process_launch_result(
                     if not validation_error:
                         validation_error = self._check_artifact_size(step_def, result.envelope)
                     if validation_error:
-                        run.status = StepRunStatus.FAILED
-                        run.error = validation_error
+                        # Match the async tick path: route through _fail_run so
+                        # exit rules can retry on output_invalid instead of
+                        # always halting the job.
                         run.result = result.envelope
                         run.pid = None
-                        run.completed_at = _now()
-                        self.store.save_run(run)
-                        self._emit(job.id, STEP_FAILED, {
-                            "step": step_name,
-                            "attempt": attempt,
-                            "error": validation_error,
-                        }, job=job)
-                        self._halt_job(job, run)
+                        self._fail_run(job, run, step_def,
+                                       error=validation_error,
+                                       error_category="output_invalid")
                     else:
                         run.result = result.envelope
                         run.executor_state = result.executor_state
diff --git a/tests/test_engine.py b/tests/test_engine.py
index f687e75d..76563010 100644
--- a/tests/test_engine.py
+++ b/tests/test_engine.py
@@ -1252,3 +1252,55 @@ def test_exit_rule_eval_error_logged(self, engine, store, caplog):
                     for r in caplog.records), (
             f"Expected warning about 'check' rule, got: {[r.message for r in caplog.records]}"
         )
+
+
+class TestSyncOutputInvalidExitRules:
+    """Sync launch must route missing outputs through _fail_run.
+
+    The async tick path already does this. The sync `_process_launch_result`
+    path used to mark FAILED and `_halt_job` immediately, so exit rules
+    that retry on `output_invalid` never fired. Repro: a callable that
+    returns {} on the first attempt (declared output `status` missing)
+    and a valid artifact on retry.
+    """
+
+    def test_missing_output_retries_via_exit_rule(self):
+        calls = {"n": 0}
+
+        def flaky_emit(inputs):
+            calls["n"] += 1
+            if calls["n"] == 1:
+                return {}
+            return {"status": "ok"}
+
+        register_step_fn("flaky_emit", flaky_emit)
+
+        engine = make_engine()
+        w = WorkflowDefinition(steps={
+            "emit": StepDefinition(
+                name="emit", outputs=["status"],
+                executor=ExecutorRef("callable", {"fn_name": "flaky_emit"}),
+                exit_rules=[
+                    ExitRule("retry_invalid", "field_match", {
+                        "field": "error_category",
+                        "value": "output_invalid",
+                        "action": "loop",
+                        "target": "emit",
+                        "max_iterations": 5,
+                    }, priority=10),
+                ],
+            ),
+        })
+
+        job = engine.create_job("sync-output-invalid", w)
+        engine.start_job(job.id)
+        job = engine.get_job(job.id)
+
+        assert job.status == JobStatus.COMPLETED, (
+            f"expected retry-then-complete, got {job.status.value}; "
+            f"calls={calls['n']} runs={[r.status.value for r in engine.get_runs(job.id, 'emit')]}"
+        )
+        assert calls["n"] == 2
+        runs = engine.get_runs(job.id, "emit")
+        assert [r.status for r in runs] == [StepRunStatus.FAILED, StepRunStatus.COMPLETED]
+        assert runs[0].error_category == "output_invalid"