fix: read the etag where the SDK keeps it, and say what the probe is …

app/adapters/detector_readiness.py

@@ -5,10 +5,13 @@
 by repeating a short request rather than by holding one open.
 """
 
+import logging
 import time
 from collections.abc import Awaitable, Callable
 from typing import Any, Protocol
 
+logger = logging.getLogger("screen")
+
 
 class HttpClientLike(Protocol):
     """The subset of an async HTTP client this module uses."""
@@ -62,17 +65,28 @@ async def wait_until_ready(
         True if the detector became ready, False if the deadline passed first.
     """
     started = now()
+    attempt = 0
     while True:
+        attempt += 1
+        reason = "not_serving"
         try:
             ready = await probe()
-        except Exception:  # noqa: BLE001 - any failure means "not ready yet"
+        except Exception as exc:  # noqa: BLE001 - any failure means "not ready yet"
             # The probe is supplied by the caller, so the ways it can fail are
             # not knowable here. A detector that has not started refuses the
             # connection, which is the expected state while it loads rather
             # than a failure to report.
             ready = False
+            reason = type(exc).__name__
+        waited = round(now() - started)
+        context = {"attempt": attempt, "waited_s": waited}
         if ready:
+            logger.info("detector_ready", extra={"context": context})
             return True
-        if now() - started + interval_s > deadline_s:
+        logger.info(
+            "detector_not_ready_yet", extra={"context": {**context, "reason": reason}}
+        )
+        if waited + interval_s > deadline_s:
+            logger.error("detector_never_ready", extra={"context": context})
             return False
         await sleep(interval_s)

app/adapters/job_store_table.py

@@ -171,10 +171,15 @@ async def fail_if_pending(self, job_id: str, error: str) -> bool:
         refused rather than to replace the result.
         """
         try:
-            entity = dict(await self._table.get_entity(job_id, job_id))
+            entity = await self._table.get_entity(job_id, job_id)
         except ResourceNotFoundError:
             return False
-        if entity.get("status") != JobStatus.PENDING.value:
+        if dict(entity).get("status") != JobStatus.PENDING.value:
+            return False
+        # The etag is metadata on the entity, not one of its properties, so it
+        # does not survive being copied into a plain dict.
+        etag = getattr(entity, "metadata", {}).get("etag")
+        if etag is None:
             return False
         try:
             await self._table.update_entity(
@@ -186,7 +191,7 @@ async def fail_if_pending(self, job_id: str, error: str) -> bool:
                     "error": error,
                 },
                 mode=UpdateMode.MERGE,
-                etag=entity.get("etag"),
+                etag=etag,
                 match_condition=MatchConditions.IfNotModified,
             )
         except ResourceModifiedError:

app/worker.py

@@ -41,7 +41,18 @@
 # happen between requests, never inside one.
 DETECTOR_READY_DEADLINE_S = 900.0
 DETECTOR_PROBE_INTERVAL_S = 15.0
-DETECTOR_PROBE_TIMEOUT_S = 30.0
+
+# A probe is an ordinary request, so ingress severs it at 240 seconds like any
+# other. It is set well above a healthy endpoint's response time because the
+# platform holds the request open while it starts a replica for an app scaled to
+# zero: a probe that gives up in seconds can abandon that start before a replica
+# exists, and the endpoint is then never reached however often it is retried.
+DETECTOR_PROBE_TIMEOUT_S = 180.0
+
+# How long the queue hides a message it has handed out. Everything one job can
+# wait for has to fit inside it, or the message is redelivered while the
+# execution holding it is still working.
+VISIBILITY_TIMEOUT_S = 1800.0
 
 
 async def drain(

tests/unit/test_detector_readiness.py

@@ -124,3 +124,37 @@ async def sleep(seconds: float) -> None:
     assert not ready
     assert clock["t"] <= 100 + 30
     assert probes <= 3
+
+
+@pytest.mark.asyncio
+async def test_each_attempt_is_logged():
+    """A silent wait cannot be told apart from a hung one. Each attempt is
+    recorded so the logs show whether the endpoint is being reached at all."""
+    import logging
+
+    events: list[str] = []
+
+    class _Collect(logging.Handler):
+        def emit(self, record: logging.LogRecord) -> None:
+            events.append(record.getMessage())
+
+    handler = _Collect()
+    logger = logging.getLogger("screen")
+    logger.addHandler(handler)
+    previous = logger.level
+    logger.setLevel(logging.INFO)
+    answers = iter([False, True])
+
+    async def probe() -> bool:
+        return next(answers)
+
+    try:
+        await wait_until_ready(
+            probe, deadline_s=60, interval_s=5, sleep=_no_sleep, now=lambda: 0.0
+        )
+    finally:
+        logger.removeHandler(handler)
+        logger.setLevel(previous)
+
+    assert "detector_not_ready_yet" in events
+    assert "detector_ready" in events

tests/unit/test_job_store_table.py

@@ -73,6 +73,16 @@ def test_the_result_is_stored_as_one_json_column():
     assert json.loads(entity["result"])["assessment"]["fit_score"] == 4
 
 
+class _Entity(dict):
+    """Stands in for ``azure.data.tables.TableEntity``.
+
+    The etag lives in ``metadata``, not among the properties, so a plain dict
+    would let the adapter appear to read one that is not there.
+    """
+
+    metadata: dict
+
+
 class _FakeTableClient:
     """Records what the adapter sends, and answers reads from that record.
 
@@ -82,13 +92,14 @@ class _FakeTableClient:
 
     def __init__(self) -> None:
         self.entities: dict[str, dict] = {}
+        self.etags: dict[str, str] = {}
         self._version = 0
         self.on_read: Callable[[], Awaitable[None]] | None = None
 
     def _stamp(self, row_key: str) -> None:
         """Give the row a new etag, as any write to it does."""
         self._version += 1
-        self.entities[row_key]["etag"] = f"W/\"{self._version}\""
+        self.etags[row_key] = f'W/"{self._version}"'
 
     async def upsert_entity(self, entity: dict, **kwargs) -> None:
         """Merge into the stored row, as UpdateMode.MERGE does.
@@ -113,7 +124,7 @@ async def update_entity(self, entity: dict, **kwargs) -> None:
             from azure.core.exceptions import ResourceNotFoundError
 
             raise ResourceNotFoundError("no such entity")
-        if kwargs.get("etag") is not None and kwargs["etag"] != stored.get("etag"):
+        if kwargs.get("etag") != self.etags.get(row_key):
             raise ResourceModifiedError("etag mismatch")
         stored.update(entity)
         self._stamp(row_key)
@@ -123,7 +134,8 @@ async def get_entity(self, partition_key: str, row_key: str) -> dict:
 
         if row_key not in self.entities:
             raise ResourceNotFoundError("no such entity")
-        entity = dict(self.entities[row_key])
+        entity = _Entity(self.entities[row_key])
+        entity.metadata = {"etag": self.etags.get(row_key)}
         if self.on_read is not None:
             # Lets a test interleave another writer between a read and the
             # write that depends on it.

tests/unit/test_worker.py

@@ -154,3 +154,27 @@ async def never_ready() -> bool:
     job = await store.get(job_id)
     assert job is not None
     assert job.status is JobStatus.FAILED
+
+
+def test_the_per_job_budget_fits_inside_one_delivery():
+    """Everything one job can wait for has to fit inside the window the queue
+    hides its message for. Beyond that the message reappears while the
+    execution holding it is still running, and the job is done twice."""
+    from app.config import settings
+    from app.worker import (
+        DETECTOR_PROBE_TIMEOUT_S,
+        DETECTOR_READY_DEADLINE_S,
+        VISIBILITY_TIMEOUT_S,
+    )
+
+    budget = (
+        DETECTOR_READY_DEADLINE_S
+        + settings.llm_guardrail_timeout_s
+        + settings.llm_timeout_s
+    )
+    assert budget < VISIBILITY_TIMEOUT_S
+
+    # A probe is one request, so it is subject to the same ingress limit as any
+    # other. Long enough for the platform to begin starting a replica, short
+    # enough that the platform does not sever it first.
+    assert 60 < DETECTOR_PROBE_TIMEOUT_S < 240