fix: route sync output_invalid through _fail_run

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

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"