fix: preserve and log backtest retry reasons

DEPLOYMENT.md

@@ -113,6 +113,7 @@ The worker also requires:
 | --- | --- |
 | `BACKTEST_DATABASE_URL` | Shared PostgreSQL URL. |
 | `BACKTEST_WORKER_ID` | Stable instance identity for attempt evidence. |
+| `BACKTEST_LOG_LEVEL` | Optional Python logging level name; defaults to `INFO`. Invalid names fail startup. Retry transitions include run, attempt, receive count, message, and reason identifiers without payloads or credentials. |
 | `BACKTEST_RESULTS_BUCKET`, `BACKTEST_RESULTS_PREFIX` | Same result store as the API. |
 | `BACKTEST_MARKET_DATA_BUCKET` | Immutable market-data input bucket. |
 | `BACKTEST_MARKET_DATA_CACHE` | Writable private cache directory, normally `/tmp/idea2strategy-market-data`. |
@@ -123,6 +124,11 @@ The worker also requires:
 | `BACKTEST_RUNTIME_POLICY_FILE` | Read-only versioned attempt, microstructure, fractional eligibility, and risk-limit document. |
 | `BACKTEST_WORKER_CORRELATION_ID` | Deployment-operation correlation identifier. |
 
+Every retry closes the current `backtest.run_attempts` row with
+`terminal_reason_code=RETRY_RELEASED` and preserves the handler or orchestrator reason in
+`failure_code`; the owning `backtest.runs` row remains non-terminal so a later delivery may create
+the next fenced attempt.
+
 AWS credentials are resolved by the SDK credential chain. On EC2, use the
 instance profile; do not inject access keys. `AWS_REGION` selects the region and
 `AWS_ENDPOINT_URL` is only for a local S3/SQS emulator. When local S3 and SQS

src/backtest_engine/wiring.py

@@ -782,10 +782,22 @@ def heartbeat(self, key: str, claim: ExecutionClaim, *, lease_duration: timedelt
         with self._persistence.unit_of_work() as uow:
             uow.attempts.heartbeat_fenced(attempt_id, claim_token, lease_duration=lease_duration)
 
-    def release(self, key: str, *, now: datetime, claim: ExecutionClaim | None = None) -> None:
+    def release(
+        self,
+        key: str,
+        *,
+        now: datetime,
+        claim: ExecutionClaim | None = None,
+        reason_code: str | None = None,
+    ) -> None:
         attempt_id, claim_token = self._claim_ids(claim)
         with self._persistence.unit_of_work() as uow:
-            uow.attempts.release_fenced(attempt_id, claim_token, terminal_reason_code="RETRY_RELEASED")
+            uow.attempts.release_fenced(
+                attempt_id,
+                claim_token,
+                terminal_reason_code="RETRY_RELEASED",
+                failure_code=reason_code or "RETRY_REQUESTED",
+            )
 
     def finish(
         self,

src/backtest_engine/worker.py

@@ -174,7 +174,14 @@ def claim(
 
     def heartbeat(self, key: str, claim: ExecutionClaim, *, lease_duration: timedelta) -> None: ...
 
-    def release(self, key: str, *, now: datetime, claim: ExecutionClaim | None = None) -> None: ...
+    def release(
+        self,
+        key: str,
+        *,
+        now: datetime,
+        claim: ExecutionClaim | None = None,
+        reason_code: str | None = None,
+    ) -> None: ...
 
     def finish(
         self,
@@ -236,7 +243,14 @@ def claim(
     def heartbeat(self, key: str, claim: ExecutionClaim, *, lease_duration: timedelta) -> None:
         return None
 
-    def release(self, key: str, *, now: datetime, claim: ExecutionClaim | None = None) -> None:
+    def release(
+        self,
+        key: str,
+        *,
+        now: datetime,
+        claim: ExecutionClaim | None = None,
+        reason_code: str | None = None,
+    ) -> None:
         """Hand a retryable attempt back so the next delivery can re-claim it."""
         with self._lock:
             record = self._records.pop(key, None)
@@ -487,9 +501,24 @@ def _handle(self, message: Mapping[str, Any]) -> HandledMessage:
         if outcome.result is JobResult.RETRY:
             # Release the CAS record so the redelivery is a *new* attempt, then
             # make the message immediately visible again.
-            self._store.release(key, now=self._clock(), claim=claim)
+            reason_code = outcome.reason_code or "RETRY_REQUESTED"
+            _LOGGER.info(
+                "backtest job retry released run_id=%s attempt=%s receive_count=%s "
+                "reason_code=%s message_id=%s",
+                run_id,
+                context.attempt_number,
+                context.receive_count,
+                reason_code,
+                message_id,
+            )
+            self._store.release(
+                key,
+                now=self._clock(),
+                claim=claim,
+                reason_code=reason_code,
+            )
             self._return_to_queue(receipt)
-            return HandledMessage(message_id, MessageDisposition.RETURNED, outcome.reason_code, key)
+            return HandledMessage(message_id, MessageDisposition.RETURNED, reason_code, key)
 
         self._store.finish(key, ExecutionRecordStatus.FAILED, now=self._clock(), claim=claim)
         return self._dead_letter(message, receipt, outcome.reason_code or "PERMANENT_FAILURE", message_id, key)
@@ -509,6 +538,13 @@ def _invoke(
         try:
             return self._handler(job, context)
         except Exception as exc:
+            _LOGGER.exception(
+                "backtest job handler raised run_id=%s attempt=%s receive_count=%s message_id=%s",
+                job.get("backtestRunId"),
+                context.attempt_number,
+                context.receive_count,
+                context.message_id,
+            )
             return JobOutcome(JobResult.RETRY, reason_code=f"HANDLER_ERROR:{type(exc).__name__}")
         finally:
             done.set()
@@ -817,6 +853,20 @@ def _next_wait_seconds(self) -> float:
     BacktestLane.COMPETITION: 1,
 }
 
+_LOG_FORMAT = "%(asctime)s %(levelname)s %(name)s %(message)s"
+
+
+def _configure_logging(environ: Mapping[str, str]) -> str:
+    """Configure the dedicated console process before any worker components start."""
+    level_name = environ.get("BACKTEST_LOG_LEVEL", "INFO").strip().upper() or "INFO"
+    level = logging.getLevelNamesMapping().get(level_name)
+    if not isinstance(level, int):
+        raise WorkerConfigurationError(
+            f"BACKTEST_LOG_LEVEL must be a Python logging level name, got {level_name!r}"
+        )
+    logging.basicConfig(level=level, format=_LOG_FORMAT, force=True)
+    return level_name
+
 
 def _config_from_env(environ: Mapping[str, str]) -> WorkerConfig:
     missing = [name for name in _REQUIRED_ENV if not environ.get(name)]
@@ -975,11 +1025,18 @@ def _runtime_sqs_client(environ: Mapping[str, str]) -> Any:
 
 
 def run() -> None:
+    log_level = _configure_logging(os.environ)
     lane_mode = any(os.environ.get(f"BACKTEST_{lane.value.upper()}_QUEUE_URL") for lane in BacktestLane)
     if lane_mode:
         configs, lane_limits, global_limit = _lane_configs_from_env(os.environ)
     else:
         config = _config_from_env(os.environ)
+    _LOGGER.info(
+        "backtest worker starting worker_id=%s lane_mode=%s log_level=%s",
+        os.environ.get("BACKTEST_WORKER_ID", ""),
+        lane_mode,
+        log_level,
+    )
     handler: JobHandler = load_factory(os.environ["BACKTEST_JOB_HANDLER"], "BACKTEST_JOB_HANDLER")
     # No in-memory fallback. `InMemoryExecutionKeyStore` is a process-local dictionary;
     # a deployment that got it by leaving one variable unset would silently lose the

tests/test_lane_scheduler.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import logging
 import threading
 import time
 from collections import deque
@@ -421,10 +422,28 @@ def test_scheduler_acknowledges_only_after_durable_finish() -> None:
     assert client.deleted == [message["ReceiptHandle"]]
 
 
-def test_scheduler_retry_releases_the_claim_and_returns_message_to_its_lane() -> None:
+def test_scheduler_retry_releases_the_claim_and_returns_message_to_its_lane(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
     message = _message(BacktestLane.CUSTOM, 2)
     client = OneMessageSqs(message)
-    store = InMemoryExecutionKeyStore()
+
+    class RetryReasonStore(InMemoryExecutionKeyStore):
+        released_reason_code: str | None = None
+
+        def release(
+            self,
+            key: str,
+            *,
+            now: datetime,
+            claim: Any = None,
+            reason_code: str | None = None,
+        ) -> None:
+            self.released_reason_code = reason_code
+            super().release(key, now=now, claim=claim, reason_code=reason_code)
+
+    store = RetryReasonStore()
+    caplog.set_level(logging.INFO, logger="backtest_engine.worker")
     worker = BacktestWorker(
         client=client,
         config=WorkerConfig(
@@ -453,3 +472,41 @@ def test_scheduler_retry_releases_the_claim_and_returns_message_to_its_lane() ->
     assert [item.disposition for item in completed] == [MessageDisposition.RETURNED]
     assert client.visibility_changes == [0]
     assert client.deleted == []
+    assert store.released_reason_code == "TRANSIENT_DEPENDENCY"
+    assert "reason_code=TRANSIENT_DEPENDENCY" in caplog.text
+
+
+def test_handler_exception_emits_traceback_and_a_stable_retry_reason(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    message = _message(BacktestLane.BASIC, 3)
+    client = OneMessageSqs(message)
+
+    def fail(_job: Mapping[str, Any], _context: Any) -> JobOutcome:
+        raise OSError("simulated data read failure")
+
+    worker = BacktestWorker(
+        client=client,
+        config=WorkerConfig(
+            queue_url="https://sqs.local/basic",
+            dead_letter_queue_url="https://sqs.local/basic-dlq",
+            worker_id="lane-worker",
+            max_receive_count=3,
+            visibility_timeout=timedelta(seconds=30),
+            wait_time=timedelta(0),
+            max_messages=1,
+            heartbeat_interval=timedelta(seconds=10),
+        ),
+        handler=fail,
+        store=InMemoryExecutionKeyStore(),
+        clock=lambda: T0,
+    )
+    caplog.set_level(logging.INFO, logger="backtest_engine.worker")
+
+    handled = worker.handle_message(message)
+
+    assert handled.disposition is MessageDisposition.RETURNED
+    assert handled.reason_code == "HANDLER_ERROR:OSError"
+    assert "backtest job handler raised" in caplog.text
+    assert "simulated data read failure" in caplog.text
+    assert "reason_code=HANDLER_ERROR:OSError" in caplog.text

tests/test_wiring_persistence.py

@@ -138,9 +138,26 @@ def test_a_released_attempt_is_closed_and_retry_gets_a_fresh_fence(
     key = worker_execution_key_for(str(run_id), "OFFICIAL_BACKTEST:retried")
     first = store.claim(key, run_id=str(run_id), owner="worker-a", now=T0, lease_duration=LEASE)
 
-    store.release(key, now=T0 + timedelta(seconds=5), claim=first)
+    store.release(
+        key,
+        now=T0 + timedelta(seconds=5),
+        claim=first,
+        reason_code="REQUIRED_DATA_UNAVAILABLE",
+    )
     assert attempt_rows(admin_engine, run_id)[0]["status"] == "FAILED"
     assert store.status(key) is ExecutionRecordStatus.FAILED
+    with admin_engine.connect() as connection:
+        retry_reason = connection.execute(
+            text(
+                "SELECT failure_code, terminal_reason_code "
+                "FROM backtest.run_attempts WHERE id = CAST(:id AS uuid)"
+            ),
+            {"id": first.attempt_id},
+        ).mappings().one()
+    assert dict(retry_reason) == {
+        "failure_code": "REQUIRED_DATA_UNAVAILABLE",
+        "terminal_reason_code": "RETRY_RELEASED",
+    }
 
     reclaimed = store.claim(
         key, run_id=str(run_id), owner="worker-b", now=T0 + timedelta(seconds=6), lease_duration=LEASE

tests/test_worker.py

@@ -15,6 +15,7 @@
 
 import contextlib
 import json
+import logging
 import os
 import threading
 import time
@@ -27,6 +28,7 @@
 import boto3
 import pytest
 
+from backtest_engine import worker as worker_module
 from backtest_engine.worker import (
     WORKER_EXECUTION_KEY_MAX_LENGTH,
     BacktestWorker,
@@ -48,6 +50,38 @@
 T0 = datetime(2026, 1, 1, tzinfo=timezone.utc)
 
 
+@pytest.mark.parametrize(
+    ("environ", "expected_level"),
+    [
+        ({}, logging.INFO),
+        ({"BACKTEST_LOG_LEVEL": "debug"}, logging.DEBUG),
+    ],
+)
+def test_worker_entrypoint_configures_runtime_logging(
+    monkeypatch: pytest.MonkeyPatch,
+    environ: Mapping[str, str],
+    expected_level: int,
+) -> None:
+    configured: list[dict[str, object]] = []
+    monkeypatch.setattr(worker_module.logging, "basicConfig", lambda **values: configured.append(values))
+
+    level_name = worker_module._configure_logging(environ)
+
+    assert level_name == logging.getLevelName(expected_level)
+    assert configured == [
+        {
+            "level": expected_level,
+            "format": "%(asctime)s %(levelname)s %(name)s %(message)s",
+            "force": True,
+        }
+    ]
+
+
+def test_worker_entrypoint_rejects_an_unknown_log_level() -> None:
+    with pytest.raises(WorkerConfigurationError, match="BACKTEST_LOG_LEVEL"):
+        worker_module._configure_logging({"BACKTEST_LOG_LEVEL": "chatty"})
+
+
 def test_sqs_client_uses_the_explicit_runtime_region(monkeypatch: pytest.MonkeyPatch) -> None:
     captured: dict[str, object] = {}