fix: preserve terminal failures when result publication fails

src/backtest_engine/production.py

@@ -65,6 +65,14 @@ def _required(environ: Mapping[str, str], name: str) -> str:
     return value
 
 
+def _required_uuid(environ: Mapping[str, str], name: str) -> str:
+    value = _required(environ, name)
+    try:
+        return str(uuid.UUID(value))
+    except ValueError as exc:
+        raise ConfigurationError(f"{name} must be a UUID") from exc
+
+
 def service_endpoint(environ: Mapping[str, str], service: str) -> str | None:
     """Resolve an emulator endpoint without coupling S3 and SQS together.
 
@@ -892,6 +900,7 @@ def _feature_object_reader(environ: Mapping[str, str]) -> S3VersionedFeatureObje
 def orchestrator_job_handler(
     environ: Mapping[str, str] = os.environ,
 ) -> OrchestratorJobHandler:
+    correlation_id = _required_uuid(environ, "BACKTEST_WORKER_CORRELATION_ID")
     policy = load_runtime_policy(Path(_required(environ, "BACKTEST_RUNTIME_POLICY_FILE")))
 
     engine = _engine(environ)
@@ -919,5 +928,5 @@ def orchestrator_job_handler(
         risk_limits=policy.risk_limits,
         runtime=BasicPlanRuntime(),
         wall_clock=lambda: datetime.now(UTC),
-        correlation_id=_required(environ, "BACKTEST_WORKER_CORRELATION_ID"),
+        correlation_id=correlation_id,
     )

src/backtest_engine/wiring.py

@@ -53,6 +53,7 @@
 
 from __future__ import annotations
 
+import logging
 import uuid
 from collections.abc import Callable, Iterator, Mapping, Sequence
 from contextlib import contextmanager
@@ -181,6 +182,9 @@
 )
 
 
+_LOGGER = logging.getLogger(__name__)
+
+
 __all__ = [
     "API_REQUIRED_ENV",
     "COST_MODEL_VERSION",
@@ -1486,16 +1490,25 @@ def __call__(self, job: Mapping[str, Any], context: JobContext) -> JobOutcome:
         try:
             binding = self.bind(envelope, context)
         except JobNotSatisfiable as exc:
-            self._publish(
-                envelope,
-                self._correlation_id,
-                status="FAILED",
-                delivery_attempt=context.receive_count,
-                failedAt=_utc_text(self._wall_clock()),
-                attempt=context.attempt_number,
-                failureCode=exc.reason_code,
-                retryable=False,
-            )
+            try:
+                self._publish(
+                    envelope,
+                    self._correlation_id,
+                    status="FAILED",
+                    delivery_attempt=context.receive_count,
+                    failedAt=_utc_text(self._wall_clock()),
+                    attempt=context.attempt_number,
+                    failureCode=exc.reason_code,
+                    retryable=False,
+                )
+            except Exception:
+                _LOGGER.exception(
+                    "terminal backtest result publish failed run_id=%s attempt=%s reason_code=%s; "
+                    "preserving permanent failure",
+                    envelope.run_id,
+                    context.attempt_number,
+                    exc.reason_code,
+                )
             return JobOutcome(JobResult.PERMANENT_FAILURE, reason_code=exc.reason_code)
 
         started_at = self._wall_clock()

tests/test_production.py

@@ -11,6 +11,7 @@
 
 import pytest
 
+import backtest_engine.production as production
 from backtest_engine.api import RESULT_INGEST_SCOPE
 from backtest_engine.backtest_request_intake import RequestLane
 from backtest_engine.production import (
@@ -25,6 +26,7 @@
     SqsExecutionJobQueue,
     api_authenticator,
     load_execution_policy_catalog,
+    orchestrator_job_handler,
     service_endpoint,
 )
 
@@ -33,6 +35,18 @@
 BOT_ID = UUID("bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb")
 
 
+def test_worker_correlation_id_is_rejected_before_other_worker_dependencies_are_built() -> None:
+    with pytest.raises(ConfigurationError, match="BACKTEST_WORKER_CORRELATION_ID must be a UUID"):
+        orchestrator_job_handler({"BACKTEST_WORKER_CORRELATION_ID": "i-07a6870a8c4c199dc"})
+
+
+def test_worker_correlation_id_is_normalized_to_the_result_event_uuid_format() -> None:
+    assert production._required_uuid(  # type: ignore[attr-defined]
+        {"BACKTEST_WORKER_CORRELATION_ID": "AAAAAAAA-AAAA-4AAA-8AAA-AAAAAAAAAAAA"},
+        "BACKTEST_WORKER_CORRELATION_ID",
+    ) == "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"
+
+
 def test_service_specific_aws_endpoint_overrides_the_legacy_shared_endpoint() -> None:
     environment = {
         "AWS_ENDPOINT_URL": "http://legacy:4566",

tests/test_wiring.py

@@ -15,6 +15,8 @@
 from __future__ import annotations
 
 import copy
+import logging
+import uuid
 from datetime import UTC, date, datetime
 from decimal import ROUND_HALF_UP, Decimal, localcontext
 from fractions import Fraction
@@ -53,6 +55,7 @@
     EXECUTION_MODEL_VERSION,
     BasicPlanReplayFactory,
     ExecutionModelEngine,
+    JobEnvelope,
     JobNotSatisfiable,
     OrchestratorJobHandler,
     WiringError,
@@ -436,6 +439,88 @@ def publish(self, event: Any, *, delivery_attempt: int) -> None:
     assert sink.events[0]["deliveryAttempt"] == 3
 
 
+def _terminal_failure_envelope() -> JobEnvelope:
+    return JobEnvelope(
+        run_id=uuid.UUID("55555555-5555-4555-8555-555555555555"),
+        bot_id=uuid.UUID("00000000-0000-4000-8000-0000000000b1"),
+        owner_account_id=uuid.UUID("66666666-6666-4666-8666-666666666666"),
+        idempotency_key="terminal-publish-failure",
+        input_bundle_id=uuid.UUID("00000000-0000-4000-8000-0000000000b2"),
+        input_bundle_fingerprint="sha256:" + "1" * 64,
+        execution_policy_version="official-backtest-policy-v1",
+        compiled_plan_checksum="sha256:" + "2" * 64,
+        dataset_manifest_id=uuid.UUID("00000000-0000-4000-8000-0000000000b3"),
+        expected_dataset_hash="sha256:" + "3" * 64,
+        expected_snapshot_hash="sha256:" + "4" * 64,
+        datasets=(),
+        feature_materializations=(),
+        evaluation_period_id=None,
+        input_set_hash=None,
+    )
+
+
+def _terminal_failure_handler() -> OrchestratorJobHandler:
+    handler = object.__new__(OrchestratorJobHandler)
+    handler._wall_clock = lambda: COMPLETED_AT
+    handler._correlation_id = "77777777-7777-4777-8777-777777777777"
+
+    def unsatisfiable(*_args: Any) -> None:
+        raise JobNotSatisfiable("compiled plan is not resolvable", reason_code="REQUIRED_INPUT_UNAVAILABLE")
+
+    handler.bind = unsatisfiable  # type: ignore[method-assign]
+    return handler
+
+
+def test_unsatisfiable_binding_publishes_the_terminal_failure(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class Sink:
+        def __init__(self) -> None:
+            self.events: list[dict[str, Any]] = []
+
+        def publish(self, event: Any, *, delivery_attempt: int) -> None:
+            self.events.append(dict(event) | {"deliveryAttempt": delivery_attempt})
+
+    envelope = _terminal_failure_envelope()
+    sink = Sink()
+    handler = _terminal_failure_handler()
+    handler._sink = sink
+    context = JobContext("execution-key", 1, 1, "message-1", "worker-1")
+    monkeypatch.setattr(JobEnvelope, "parse", classmethod(lambda _cls, _job: envelope))
+
+    outcome = handler({}, context)
+
+    assert outcome.result is JobResult.PERMANENT_FAILURE
+    assert outcome.reason_code == "REQUIRED_INPUT_UNAVAILABLE"
+    assert sink.events[0]["status"] == "FAILED"
+    assert sink.events[0]["failureCode"] == "REQUIRED_INPUT_UNAVAILABLE"
+    assert sink.events[0]["retryable"] is False
+    assert sink.events[0]["metadata"]["correlationId"] == "77777777-7777-4777-8777-777777777777"
+
+
+def test_terminal_binding_failure_survives_its_failure_event_publish_failing(
+    caplog: pytest.LogCaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    envelope = _terminal_failure_envelope()
+    handler = _terminal_failure_handler()
+
+    def unavailable_sink(*_args: Any, **_kwargs: Any) -> None:
+        raise OSError("result sink unavailable")
+
+    handler._publish = unavailable_sink  # type: ignore[method-assign]
+    context = JobContext("execution-key", 1, 1, "message-1", "worker-1")
+    monkeypatch.setattr(JobEnvelope, "parse", classmethod(lambda _cls, _job: envelope))
+
+    with caplog.at_level(logging.ERROR):
+        outcome = handler({}, context)
+
+    assert outcome.result is JobResult.PERMANENT_FAILURE
+    assert outcome.reason_code == "REQUIRED_INPUT_UNAVAILABLE"
+    assert "result sink unavailable" in caplog.text
+    assert "REQUIRED_INPUT_UNAVAILABLE" in caplog.text
+
+
 # ==========================================================================
 # The evaluation window comes from the pinned dataset
 # ==========================================================================