From 3e764265799d70e7b1d4335f89c631ac4b968a15 Mon Sep 17 00:00:00 2001 From: inesaranab Date: Tue, 11 Aug 2026 18:24:55 +0200 Subject: [PATCH 1/5] fix: wait for the detector between requests, not inside one The first end-to-end run in production failed after exactly 240 seconds: 13:32:50 guardrail begins 13:36:50 API call failed on attempt 1: stream timeout That is the platform's ingress limit, and the finding is that it applies to internal ingress too. Moving the caller off the HTTP path was not sufficient: the worker still reaches the detector over HTTP, so somebody was still holding a connection open across a 13-minute cold start. llm_guardrail_timeout_s of 900 never applied, because the proxy severs the connection first -- the same way requestTimeout never applied on the public side. The worker now establishes readiness by repeating a short request until the detector answers, then screens against a warm server, where the call returns in seconds. No single request spans the cold start. The probe doubles as the activation, since a request to an app at zero replicas is what starts it. A job taken while the detector stays unreachable is recorded as failed rather than attempted, because attempting it would spend the whole request budget on a call that cannot succeed and end in the same failure. Records in infra/gemma/README.md that the 240s limit is not specific to public ingress, and that cost scales with wake-ups rather than screenings: the same fifty screenings cost about EUR 2 batched and about EUR 50 scattered. --- .codex/config.toml | 2 + .codex/hooks.json | 48 ++++++++++++++ .codex/hooks/conventions_reminder.py | 49 ++++++++++++++ .codex/hooks/response_style.py | 34 ++++++++++ app/adapters/detector_readiness.py | 70 +++++++++++++++++++ app/worker.py | 49 +++++++++++++- infra/gemma/README.md | 41 ++++++++++++ tests/unit/test_detector_readiness.py | 96 +++++++++++++++++++++++++++ tests/unit/test_worker.py | 25 +++++++ 9 files changed, 411 insertions(+), 3 deletions(-) create mode 100644 .codex/config.toml create mode 100644 .codex/hooks.json create mode 100644 .codex/hooks/conventions_reminder.py create mode 100644 .codex/hooks/response_style.py create mode 100644 app/adapters/detector_readiness.py create mode 100644 tests/unit/test_detector_readiness.py diff --git a/.codex/config.toml b/.codex/config.toml new file mode 100644 index 0000000..9ce392a --- /dev/null +++ b/.codex/config.toml @@ -0,0 +1,2 @@ +[mcp_servers.linear-server] +url = "https://mcp.linear.app/mcp" diff --git a/.codex/hooks.json b/.codex/hooks.json new file mode 100644 index 0000000..58aea54 --- /dev/null +++ b/.codex/hooks.json @@ -0,0 +1,48 @@ +{ + "hooks": { + "PreToolUse": [ + { + "matcher": "WebSearch|WebFetch", + "hooks": [ + { + "type": "command", + "command": "echo '{\"systemMessage\":\"Scope check - what decision does this research change? If there is no answer, it is a bookmark (INE-18), not a task. Current objective: .claude/FOCUS.md\"}'" + } + ] + }, + { + "matcher": "Write|Edit", + "hooks": [ + { + "type": "command", + "command": "python3 '/Users/inesarana/screening/.codex/hooks/conventions_reminder.py' 2>/dev/null || true" + } + ] + } + ], + "SessionStart": [ + { + "hooks": [ + { + "type": "command", + "command": "cd \"$CLAUDE_PROJECT_DIR\" 2>/dev/null; python3 -c \"import json,pathlib;print(json.dumps({'hookSpecificOutput':{'hookEventName':'SessionStart','additionalContext':pathlib.Path('.claude/FOCUS.md').read_text()}}))\" 2>/dev/null || true" + } + ] + } + ], + "UserPromptSubmit": [ + { + "hooks": [ + { + "type": "command", + "command": "python3 -c \"import json,time,pathlib;f=pathlib.Path('/tmp/.claude_last_prompt');now=time.time();prev=float(f.read_text()) if f.exists() else now;f.write_text(str(now));g=int(now-prev);print(json.dumps({'hookSpecificOutput':{'hookEventName':'UserPromptSubmit','additionalContext':'[clock] '+time.strftime('%H:%M')+' | '+str(g//60)+'m '+str(g%60)+'s since previous message'}}))\" 2>/dev/null || true" + }, + { + "type": "command", + "command": "python3 '/Users/inesarana/screening/.codex/hooks/response_style.py' 2>/dev/null || true" + } + ] + } + ] + } +} diff --git a/.codex/hooks/conventions_reminder.py b/.codex/hooks/conventions_reminder.py new file mode 100644 index 0000000..564cef6 --- /dev/null +++ b/.codex/hooks/conventions_reminder.py @@ -0,0 +1,49 @@ +"""PreToolUse hook: restate the project's writing conventions before an edit. + +Fires on Write and Edit. Emits nothing for files the conventions do not cover, +so the reminder stays attached to source rather than appearing on every write. +""" + +import json +import sys + +_EXTENSIONS = (".py", ".yaml", ".yml", ".md") + +_REMINDER = ( + "screening-conventions (.claude/skills/screening-conventions/SKILL.md):\n" + "- Google-style docstrings: summary line, then Args/Returns/Raises for " + "functions, Attributes for models.\n" + "- State the PROPERTY, not the incident that taught it. No war stories, " + "dates, measurements, or 'we' — those belong in infra/*/README.md or the " + "commit message.\n" + "- Self-contained: never explain one symbol by referring to another.\n" + "- No prose constants: explanation assigned to a module-level string is " + "dead code.\n" + "- app/domain and app/ports stay vendor-free; adapters hold the vendors.\n" + "- Test first: a failing test before the implementation." +) + + +def main() -> None: + """Print the reminder when the target file is one the conventions govern.""" + try: + payload = json.load(sys.stdin) + except json.JSONDecodeError, ValueError: + return + path = payload.get("tool_input", {}).get("file_path", "") + if not path.endswith(_EXTENSIONS): + return + print( + json.dumps( + { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "additionalContext": _REMINDER, + } + } + ) + ) + + +if __name__ == "__main__": + main() diff --git a/.codex/hooks/response_style.py b/.codex/hooks/response_style.py new file mode 100644 index 0000000..d69a559 --- /dev/null +++ b/.codex/hooks/response_style.py @@ -0,0 +1,34 @@ +"""UserPromptSubmit hook: restate how answers to Ines should be written. + +Emits on every prompt. Carries the response-shape rules only; the writing +conventions for source files are in conventions_reminder.py. +""" + +import json + +_REMINDER = ( + "Response style:\n" + "- Short. Answer the question asked, then stop.\n" + "- High level first, in plain words. Then the low-level version, so the " + "vocabulary is picked up in context rather than assumed.\n" + "- Define a term the first time it appears, in the same sentence.\n" + "- No jargon without its plain-language equivalent alongside it." +) + + +def main() -> None: + """Print the reminder.""" + print( + json.dumps( + { + "hookSpecificOutput": { + "hookEventName": "UserPromptSubmit", + "additionalContext": _REMINDER, + } + } + ) + ) + + +if __name__ == "__main__": + main() diff --git a/app/adapters/detector_readiness.py b/app/adapters/detector_readiness.py new file mode 100644 index 0000000..ee7b4a7 --- /dev/null +++ b/app/adapters/detector_readiness.py @@ -0,0 +1,70 @@ +"""Adapter: waiting for the Article 9 detector to start serving. + +The detector scales to zero and takes minutes to load its weights. Platform +ingress closes any single request long before that, so readiness is established +by repeating a short request rather than by holding one open. +""" + +from collections.abc import Awaitable, Callable +from typing import Any, Protocol + + +class HttpClientLike(Protocol): + """The subset of an async HTTP client this module uses.""" + + async def get(self, url: str) -> Any: ... + + +def http_probe(client: HttpClientLike, url: str) -> Callable[[], Awaitable[bool]]: + """Build a readiness probe that calls an endpoint over HTTP. + + Args: + client: Issues the request. Injected so the caller owns its lifetime + and its per-request timeout. + url: The endpoint that answers only once the detector is serving. + + Returns: + A callable returning True when the endpoint answers 200. + """ + + async def probe() -> bool: + return (await client.get(url)).status_code == 200 + + return probe + + +async def wait_until_ready( + probe: Callable[[], Awaitable[bool]], + *, + deadline_s: float, + interval_s: float, + sleep: Callable[[float], Awaitable[None]], +) -> bool: + """Repeat a readiness probe until it succeeds or the deadline passes. + + Args: + probe: Returns True once the detector is serving. + deadline_s: How long to keep trying, in seconds. + interval_s: Seconds between attempts. + sleep: Suspends for the given seconds. Injected so tests need no real + time. + + Returns: + True if the detector became ready, False if the deadline passed first. + """ + waited = 0.0 + while True: + try: + ready = await probe() + except Exception: # 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 + if ready: + return True + if waited + interval_s > deadline_s: + return False + await sleep(interval_s) + waited += interval_s diff --git a/app/worker.py b/app/worker.py index 35a576e..569cf62 100644 --- a/app/worker.py +++ b/app/worker.py @@ -10,7 +10,11 @@ import asyncio import logging +from collections.abc import Awaitable, Callable +import httpx + +from app.adapters.detector_readiness import http_probe, wait_until_ready from app.adapters.guard_classifier import ClassifierGuardrail from app.adapters.job_queue_azure import AzureJobQueue from app.adapters.job_store_table import AzureTableJobStore @@ -29,8 +33,23 @@ # transcript on the queue while repeating the same failure. MAX_DELIVERIES = 3 - -async def drain(service: ScreenService, queue: JobQueue) -> int: +# How long to wait for the detector to start serving, and how often to ask. +# Loading the weights takes minutes; the deadline sits below the job's own +# replicaTimeout so an unreachable detector ends as a recorded failure rather +# than a killed replica. Each probe is a separate short request, because +# platform ingress closes any single request at 240 seconds -- the wait has to +# happen between requests, never inside one. +DETECTOR_READY_DEADLINE_S = 900.0 +DETECTOR_PROBE_INTERVAL_S = 15.0 +DETECTOR_PROBE_TIMEOUT_S = 30.0 + + +async def drain( + service: ScreenService, + queue: JobQueue, + *, + detector_ready: Callable[[], Awaitable[bool]] | None = None, +) -> int: """Screen every job currently on the queue. A message is deleted once the screening has been recorded, whether it @@ -43,9 +62,15 @@ async def drain(service: ScreenService, queue: JobQueue) -> int: the job is recorded as failed and its message deleted without being attempted, so a job that kills every worker cannot cycle indefinitely. + A job taken while the detector is unreachable is recorded as failed rather + than attempted. Attempting it would spend the platform's entire request + budget on a call that cannot succeed, and end in the same failure. + Args: service: Performs the screening and records the outcome. queue: Supplies the work. + detector_ready: Returns True once the detector can be called. None + skips the check, for callers that supply their own guardrail. Returns: The number of jobs screened. Abandoned jobs are not counted, having @@ -57,6 +82,10 @@ async def drain(service: ScreenService, queue: JobQueue) -> int: await service.abandon(job.job_id, "TooManyDeliveries") await queue.delete(job) continue + if detector_ready is not None and not await detector_ready(): + await service.abandon(job.job_id, "DetectorUnavailable") + await queue.delete(job) + continue logger.info("worker_job_started", extra={"context": {"job": job.job_id}}) await service.run(job.job_id, job.request) await queue.delete(job) @@ -94,10 +123,24 @@ async def main() -> None: job_queue=queue, ) + # The first request also activates the detector, since it scales to zero, + # so probing both starts it and establishes when it is usable. + http = httpx.AsyncClient(timeout=DETECTOR_PROBE_TIMEOUT_S) + probe = http_probe(http, f"{settings.llm_guardrail_base_url.rstrip('/')}/models") + + async def detector_ready() -> bool: + return await wait_until_ready( + probe, + deadline_s=DETECTOR_READY_DEADLINE_S, + interval_s=DETECTOR_PROBE_INTERVAL_S, + sleep=asyncio.sleep, + ) + try: - processed = await drain(service, queue) + processed = await drain(service, queue, detector_ready=detector_ready) logger.info("worker_drained", extra={"context": {"processed": processed}}) finally: + await http.aclose() await llm.aclose() await queue_client.close() await table.close() diff --git a/infra/gemma/README.md b/infra/gemma/README.md index 20f5e98..565b45d 100644 --- a/infra/gemma/README.md +++ b/infra/gemma/README.md @@ -366,6 +366,47 @@ storage account is already there; Table Storage is the cheap fit. The background to outlive the request, which means either `minReplicas: 1` on the CPU app or a queue-triggered Container Apps Job like `download-weights`. +### The 240s limit applies to INTERNAL ingress too + +Moving the caller off the HTTP path is not sufficient. The worker still reaches the detector +over HTTP, and that hop crosses ingress as well — internal ingress is still ingress. The first +end-to-end run in production failed on exactly this: + +``` +13:32:48 worker_job_started +13:32:50 guardrail begins <- the call to screening-gemma +13:36:50 API call failed on attempt 1: stream timeout +``` + +13:32:50 to 13:36:50 is 240 seconds to the second. `llm_guardrail_timeout_s: 900` never +applies, because the proxy severs the connection first — the same way `requestTimeout: 900` +never applied on the public side. + +The lesson generalises past this project: **removing the wait from one hop does not remove it +from the next.** Somebody was still holding a connection open across a 13-minute cold start, +and the platform does not care which process is waiting, only how long. + +The fix is to never let one request span the cold start. Poll the detector's `/models` +endpoint with short requests until it answers, then send the inference to a warm server, where +it returns in seconds. Many brief calls instead of one long one — the same shape as `202 + +poll`, applied one layer down. + +### Cost scales with wake-ups, not with screenings + +The GPU bills for its cold start whether or not the caller survives it, so the unit of cost is +a wake-up: + +| Pattern | Approximate cost | +|---|---| +| One screening | ~€1.00 (13 min load + 15 min cooldown at €2.16/hr) | +| Fifty screenings in one burst | ~€2.00 — one wake, then seconds each while warm | +| Fifty screenings spread over a week | ~€50 — each pays for its own wake | + +Identical work, 25× the cost, decided entirely by arrival pattern. Serverless GPU is cheap for +bursty load and expensive for a steady trickle. The levers are `cooldownPeriod` (longer merges +more bursts) and deliberate batching (screen on a schedule rather than on arrival); both buy +money with latency. + This is also the shape that makes batch natural: submit N transcripts, pay one cold start, amortize it. At 50 transcripts a 13-minute boot is ~15s each; at one transcript it is absurd. diff --git a/tests/unit/test_detector_readiness.py b/tests/unit/test_detector_readiness.py new file mode 100644 index 0000000..cba1835 --- /dev/null +++ b/tests/unit/test_detector_readiness.py @@ -0,0 +1,96 @@ +"""Waiting for the detector without holding one connection open.""" + +import pytest + +from app.adapters.detector_readiness import wait_until_ready + + +async def _no_sleep(seconds: float) -> None: + """Stand in for asyncio.sleep so the tests do not take real time.""" + return + + +@pytest.mark.asyncio +async def test_ready_on_the_first_probe_does_not_wait(): + probes = 0 + + async def probe() -> bool: + nonlocal probes + probes += 1 + return True + + assert await wait_until_ready(probe, deadline_s=60, interval_s=5, sleep=_no_sleep) + assert probes == 1 + + +@pytest.mark.asyncio +async def test_it_keeps_probing_until_the_detector_answers(): + """A cold detector refuses connections for minutes. Each probe is its own + short request, so no single one approaches the platform's request limit.""" + answers = iter([False, False, True]) + slept: list[float] = [] + + async def probe() -> bool: + return next(answers) + + async def sleep(seconds: float) -> None: + slept.append(seconds) + + assert await wait_until_ready(probe, deadline_s=60, interval_s=5, sleep=sleep) + assert slept == [5, 5] + + +@pytest.mark.asyncio +async def test_it_gives_up_at_the_deadline(): + """Bounded, so a detector that never starts fails the job rather than + occupying a worker for as long as the platform allows.""" + + async def probe() -> bool: + return False + + assert not await wait_until_ready( + probe, deadline_s=10, interval_s=5, sleep=_no_sleep + ) + + +@pytest.mark.asyncio +async def test_a_probe_that_raises_counts_as_not_ready(): + """A detector that has not started refuses the connection outright. That is + the expected state while it loads, not a reason to stop waiting.""" + calls = 0 + + async def probe() -> bool: + nonlocal calls + calls += 1 + if calls < 3: + raise ConnectionError("connection refused") + return True + + assert await wait_until_ready(probe, deadline_s=60, interval_s=1, sleep=_no_sleep) + assert calls == 3 + + +@pytest.mark.asyncio +async def test_http_probe_is_ready_only_on_a_200(): + """The detector answers its model-list endpoint once it is serving. Any + other status means it is reachable but not yet usable.""" + from app.adapters.detector_readiness import http_probe + + class _Response: + def __init__(self, status_code: int): + self.status_code = status_code + + class _Client: + def __init__(self, status: int): + self.status = status + self.requested = "" + + async def get(self, url: str): + self.requested = url + return _Response(self.status) + + ok = _Client(200) + assert await http_probe(ok, "http://detector/v1/models")() + assert ok.requested == "http://detector/v1/models" + + assert not await http_probe(_Client(503), "http://detector/v1/models")() diff --git a/tests/unit/test_worker.py b/tests/unit/test_worker.py index 0092827..3a20f93 100644 --- a/tests/unit/test_worker.py +++ b/tests/unit/test_worker.py @@ -129,3 +129,28 @@ async def scrub(self, text: str): job = await store.get(job_id) assert job is not None assert job.status is JobStatus.FAILED + + +@pytest.mark.asyncio +async def test_a_job_is_not_screened_while_the_detector_is_unreachable(): + """Screening without the detector would spend the platform's whole request + budget failing. The job is recorded as failed and its message removed, so + the outcome is an answer rather than a retry that fails identically.""" + store, queue = InMemoryJobStore(), InMemoryJobQueue() + + class BrokenGuardrail: + async def scrub(self, text: str): + raise AssertionError("must not screen without a detector") + + service = _service(store, queue, guardrail=BrokenGuardrail()) + job_id = await service.start(_REQ) + + async def never_ready() -> bool: + return False + + processed = await drain(service, queue, detector_ready=never_ready) + + assert processed == 0 + job = await store.get(job_id) + assert job is not None + assert job.status is JobStatus.FAILED From 7c4226f177bc634cdee5074425910976063f5c29 Mon Sep 17 00:00:00 2001 From: inesaranab Date: Tue, 11 Aug 2026 18:42:58 +0200 Subject: [PATCH 2/5] fix: do not overwrite a settled job, and size the request as it is published Two defects found reviewing the async split. abandon() overwrote any job, including one already carrying a result. A worker that records an outcome and then dies before deleting its message leaves that message to be redelivered; on the delivery that exceeds the limit, drain() abandons the job and rewrites the row, so a caller that had already read a result would later be told the screening failed. Only a job still PENDING is settled now. The queue-size validator counted raw UTF-8 bytes, but the fields are published inside a JSON document, where a control character occupies six bytes rather than one. A transcript of 24,000 such characters passed every cap and produced a 144,107-byte message against a 64 KiB ceiling. The rejection then arrived from the transport after the job row existed, so the caller was told to retry a body that can never be published. The validator now measures the escaped size, which is the size that actually travels. --- app/domain/models.py | 29 ++++++++++++++++++++++--- app/domain/service.py | 8 +++++++ tests/unit/test_models.py | 37 ++++++++++++++++++++++++++++---- tests/unit/test_service_async.py | 17 +++++++++++++++ 4 files changed, 84 insertions(+), 7 deletions(-) diff --git a/app/domain/models.py b/app/domain/models.py index 509d7bd..b5fa297 100644 --- a/app/domain/models.py +++ b/app/domain/models.py @@ -1,5 +1,6 @@ """The contract for /screen — the Pydantic types every layer depends on.""" +import json from datetime import UTC, datetime from enum import Enum from typing import Self @@ -22,7 +23,8 @@ # Derived from the 64 KiB ceiling on a queue message, less headroom for the job # id and the JSON that wraps both fields. The character caps above bound length; # this bounds size, which is what the ceiling is actually expressed in. A -# character can occupy up to four UTF-8 bytes, so the two are not equivalent. +# character can occupy up to four UTF-8 bytes, and a JSON string expands a +# control character to six, so neither cap implies the other. MAX_REQUEST_BYTES = 60_000 # How long a job may stay PENDING before it is treated as never going to finish. @@ -33,6 +35,20 @@ JOB_DEADLINE_SECONDS = 5 * 60 * 60 +def _json_string_bytes(value: str) -> int: + """Measure a string as it occupies space inside a JSON document. + + Args: + value: The text to measure. + + Returns: + The UTF-8 byte length of the text escaped as a JSON string, quotes + included. Non-ASCII characters are left as themselves, matching how the + request is published. + """ + return len(json.dumps(value, ensure_ascii=False).encode()) + + class ScreenRequest(BaseModel): """The request body for a screening. @@ -58,14 +74,21 @@ class ScreenRequest(BaseModel): def _fits_in_a_queue_message(self) -> Self: """Reject a request too large to publish. + Both fields travel as JSON strings, so the size that counts is the + escaped one: a control character occupies one byte in the field and six + in the message. Measuring the field alone would pass a request that the + transport then rejects, once the job has already been recorded. + Returns: The request, unchanged. Raises: ValueError: If the two fields together exceed MAX_REQUEST_BYTES - once encoded as UTF-8. + once escaped as JSON strings and encoded as UTF-8. """ - size = len(self.transcript.encode()) + len(self.job_description.encode()) + size = _json_string_bytes(self.transcript) + _json_string_bytes( + self.job_description + ) if size > MAX_REQUEST_BYTES: raise ValueError( f"encoded request is {size} bytes, over the " diff --git a/app/domain/service.py b/app/domain/service.py index eea8bea..52818b3 100644 --- a/app/domain/service.py +++ b/app/domain/service.py @@ -107,10 +107,18 @@ async def abandon(self, job_id: str, reason: str) -> None: For work that cannot be completed however many times it is tried, where another attempt would repeat the failure rather than resolve it. + Only a job still PENDING is settled. Work is given up on because it was + delivered too often, and a job can be delivered again after its outcome + was recorded, so a job already carrying an answer keeps it. + Args: job_id: The id returned by ``start``. reason: Why the job was given up on. Must not quote the transcript. """ + settled = await self._jobs.get(job_id) + if settled is not None and settled.status is not JobStatus.PENDING: + logger.info("job_abandon_skipped", extra={"context": {"job": job_id}}) + return logger.warning("job_abandoned", extra={"context": {"job": job_id}}) await self._jobs.fail(job_id, reason) diff --git a/tests/unit/test_models.py b/tests/unit/test_models.py index 354d118..4188cd5 100644 --- a/tests/unit/test_models.py +++ b/tests/unit/test_models.py @@ -78,17 +78,46 @@ def test_a_request_whose_encoded_bytes_exceed_the_queue_limit_is_rejected(): def test_a_request_at_the_byte_limit_is_accepted(): """The limit is a byte count, so the boundary case needs multi-byte text -- - the character caps alone cannot reach it with ASCII.""" - from app.domain.models import MAX_REQUEST_BYTES, MAX_TRANSCRIPT_CHARS + the character caps alone cannot reach it with ASCII. The count is of the + text as a JSON string, so the two enclosing quotes per field count too.""" + from app.domain.models import ( + MAX_REQUEST_BYTES, + MAX_TRANSCRIPT_CHARS, + _json_string_bytes, + ) two_byte = "é" assert len(two_byte.encode()) == 2 - remaining = (MAX_REQUEST_BYTES - MAX_TRANSCRIPT_CHARS * 2) // 2 + quotes = 2 * 2 + remaining = (MAX_REQUEST_BYTES - quotes - MAX_TRANSCRIPT_CHARS * 2) // 2 request = ScreenRequest( transcript=two_byte * MAX_TRANSCRIPT_CHARS, job_description=two_byte * remaining, ) - encoded = len(request.transcript.encode()) + len(request.job_description.encode()) + encoded = _json_string_bytes(request.transcript) + _json_string_bytes( + request.job_description + ) assert encoded == MAX_REQUEST_BYTES + + +def test_a_request_whose_escaped_size_exceeds_the_queue_limit_is_rejected(): + """A queue message carries the two fields as JSON strings, and JSON expands + a control character to six characters. Counting raw bytes therefore lets a + request through that cannot be published, and the rejection then lands in + transport, after the job row exists.""" + from app.adapters.job_queue_azure import encode_message + from app.domain.models import MAX_TRANSCRIPT_CHARS + + oversized = ScreenRequest.model_construct( + transcript="\x01" * MAX_TRANSCRIPT_CHARS, job_description="Backend" + ) + assert len(encode_message("a" * 32, oversized).encode()) > 64 * 1024 + + with pytest.raises(ValidationError) as exc: + ScreenRequest( + transcript="\x01" * MAX_TRANSCRIPT_CHARS, job_description="Backend" + ) + + assert "MAX_REQUEST_BYTES" in str(exc.value) diff --git a/tests/unit/test_service_async.py b/tests/unit/test_service_async.py index 1eb7338..6bf1d12 100644 --- a/tests/unit/test_service_async.py +++ b/tests/unit/test_service_async.py @@ -220,3 +220,20 @@ async def test_expiry_does_not_overwrite_a_finished_job(): assert job is not None assert job.status is JobStatus.DONE + + +@pytest.mark.asyncio +async def test_abandoning_does_not_overwrite_a_finished_job(): + """A message can come back after its screening was recorded, so abandoning + on delivery count alone would replace a result the caller already has.""" + store = InMemoryJobStore() + service = _service(store=store) + job_id = await service.start(_REQ) + await service.run(job_id, _REQ) + + await service.abandon(job_id, "TooManyDeliveries") + + job = await store.get(job_id) + assert job is not None + assert job.status is JobStatus.DONE + assert job.result is not None From d1ba2597281f7d448d4ad8e0892a0cffb68262af Mon Sep 17 00:00:00 2001 From: inesaranab Date: Tue, 11 Aug 2026 18:45:48 +0200 Subject: [PATCH 3/5] fix: bound the detector call below the limit that can actually stop it llm_guardrail_timeout_s was 900 seconds, which could never fire: ingress severs any single request at 240 seconds, internal routes included. The call ended as a transport error from the proxy rather than a timeout naming the endpoint, which is what made the first production failure read as an unexplained stream timeout. It no longer has to cover the cold start either. The worker establishes that the detector is serving by repeating a short probe before screening anything, so this bounds a call to an endpoint already known to be up. That also closes the budget flagged in review. Worst case per job was 900 + 900 + 60 = 1860 seconds, above both the queue's visibility timeout and replicaTimeout, so a message could reappear while the execution holding it still ran. It is now 900 + 200 + 60 = 1160. --- app/config.py | 18 +++++++++--------- tests/unit/test_config.py | 15 ++++++++------- 2 files changed, 17 insertions(+), 16 deletions(-) diff --git a/app/config.py b/app/config.py index 3ffce69..de0d0f3 100644 --- a/app/config.py +++ b/app/config.py @@ -39,16 +39,16 @@ class Settings(BaseSettings): llm_guardrail_base_url: str = "http://localhost:8001/v1" llm_guardrail_model: str = "google/gemma-4-31B-it" llm_timeout_s: float = 60.0 - # 15 minutes, against a measured ~13 minute cold start (2 min image pull, - # 1 min engine init, ~10 min loading 58 GiB of weights off the file share). - # Sharing the 60s assessment timeout meant every request that arrived on a - # cold endpoint timed out, and the recognizer fails closed -- so /screen - # returned 502 on the normal path, not an exceptional one. + # Below the 240 seconds at which ingress severs any single request, internal + # routes included. A larger value cannot fire, so the call would end as a + # transport error from the proxy rather than a timeout naming this endpoint. # - # A caller waiting 13 minutes is still bad; the real fix is for /screen to - # return 202 and be polled (see infra/gemma/README.md). This makes the - # blocking path correct in the meantime rather than silently broken. - llm_guardrail_timeout_s: float = 900.0 + # It does not have to cover the detector's cold start. The worker establishes + # that the endpoint is serving by repeating a short probe before it screens + # anything (see app/worker.py), so this bounds a call to an endpoint already + # known to be up. Larger than the assessment timeout because this endpoint + # runs a 31B model on one replica. + llm_guardrail_timeout_s: float = 200.0 # Job state and the work queue. A separate account from the model-weights # share: that one is kind=FileStorage, which serves file shares only and has diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 22c7d5a..75d6f4f 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -14,15 +14,16 @@ def test_llm_guardrail_settings_have_defaults(monkeypatch): assert settings.llm_guardrail_model == "google/gemma-4-31B-it" -def test_guardrail_timeout_is_long_enough_for_a_cold_start(): - """The guardrail endpoint scales to zero, so the first request after an idle - period waits for an A100 to boot and load 58 GiB of weights -- measured at - ~13 minutes. Sharing `llm_timeout_s` (60s) with the assessment LLM means - every cold request times out, and the recognizer fails closed, so /screen - returns 502 on the normal path rather than an exceptional one.""" +def test_guardrail_timeout_expires_before_the_platform_severs_the_call(): + """Ingress closes any single request at 240 seconds, internal routes + included. A timeout above that can never fire, so the call ends as a + transport error from the proxy instead of a timeout naming the endpoint. + The wait for a cold detector happens across repeated probes, not inside this + request, so this bounds a call to an endpoint already known to be serving.""" from app.config import settings - assert settings.llm_guardrail_timeout_s >= 900 + assert settings.llm_guardrail_timeout_s < 240 + assert settings.llm_guardrail_timeout_s > settings.llm_timeout_s def test_jobs_storage_defaults_point_at_the_general_purpose_account(): From 27956d41581be61b3c02a9ec7e75a4669be65fd3 Mon Sep 17 00:00:00 2001 From: inesaranab Date: Tue, 11 Aug 2026 18:50:36 +0200 Subject: [PATCH 4/5] chore: keep unrelated editor tooling out of this change .codex/ was swept in by a broad add. It is local tooling config, not part of the detector readiness fix, and belongs in its own change if it is to be tracked. --- .codex/config.toml | 2 -- .codex/hooks.json | 48 --------------------------- .codex/hooks/conventions_reminder.py | 49 ---------------------------- .codex/hooks/response_style.py | 34 ------------------- 4 files changed, 133 deletions(-) delete mode 100644 .codex/config.toml delete mode 100644 .codex/hooks.json delete mode 100644 .codex/hooks/conventions_reminder.py delete mode 100644 .codex/hooks/response_style.py diff --git a/.codex/config.toml b/.codex/config.toml deleted file mode 100644 index 9ce392a..0000000 --- a/.codex/config.toml +++ /dev/null @@ -1,2 +0,0 @@ -[mcp_servers.linear-server] -url = "https://mcp.linear.app/mcp" diff --git a/.codex/hooks.json b/.codex/hooks.json deleted file mode 100644 index 58aea54..0000000 --- a/.codex/hooks.json +++ /dev/null @@ -1,48 +0,0 @@ -{ - "hooks": { - "PreToolUse": [ - { - "matcher": "WebSearch|WebFetch", - "hooks": [ - { - "type": "command", - "command": "echo '{\"systemMessage\":\"Scope check - what decision does this research change? If there is no answer, it is a bookmark (INE-18), not a task. Current objective: .claude/FOCUS.md\"}'" - } - ] - }, - { - "matcher": "Write|Edit", - "hooks": [ - { - "type": "command", - "command": "python3 '/Users/inesarana/screening/.codex/hooks/conventions_reminder.py' 2>/dev/null || true" - } - ] - } - ], - "SessionStart": [ - { - "hooks": [ - { - "type": "command", - "command": "cd \"$CLAUDE_PROJECT_DIR\" 2>/dev/null; python3 -c \"import json,pathlib;print(json.dumps({'hookSpecificOutput':{'hookEventName':'SessionStart','additionalContext':pathlib.Path('.claude/FOCUS.md').read_text()}}))\" 2>/dev/null || true" - } - ] - } - ], - "UserPromptSubmit": [ - { - "hooks": [ - { - "type": "command", - "command": "python3 -c \"import json,time,pathlib;f=pathlib.Path('/tmp/.claude_last_prompt');now=time.time();prev=float(f.read_text()) if f.exists() else now;f.write_text(str(now));g=int(now-prev);print(json.dumps({'hookSpecificOutput':{'hookEventName':'UserPromptSubmit','additionalContext':'[clock] '+time.strftime('%H:%M')+' | '+str(g//60)+'m '+str(g%60)+'s since previous message'}}))\" 2>/dev/null || true" - }, - { - "type": "command", - "command": "python3 '/Users/inesarana/screening/.codex/hooks/response_style.py' 2>/dev/null || true" - } - ] - } - ] - } -} diff --git a/.codex/hooks/conventions_reminder.py b/.codex/hooks/conventions_reminder.py deleted file mode 100644 index 564cef6..0000000 --- a/.codex/hooks/conventions_reminder.py +++ /dev/null @@ -1,49 +0,0 @@ -"""PreToolUse hook: restate the project's writing conventions before an edit. - -Fires on Write and Edit. Emits nothing for files the conventions do not cover, -so the reminder stays attached to source rather than appearing on every write. -""" - -import json -import sys - -_EXTENSIONS = (".py", ".yaml", ".yml", ".md") - -_REMINDER = ( - "screening-conventions (.claude/skills/screening-conventions/SKILL.md):\n" - "- Google-style docstrings: summary line, then Args/Returns/Raises for " - "functions, Attributes for models.\n" - "- State the PROPERTY, not the incident that taught it. No war stories, " - "dates, measurements, or 'we' — those belong in infra/*/README.md or the " - "commit message.\n" - "- Self-contained: never explain one symbol by referring to another.\n" - "- No prose constants: explanation assigned to a module-level string is " - "dead code.\n" - "- app/domain and app/ports stay vendor-free; adapters hold the vendors.\n" - "- Test first: a failing test before the implementation." -) - - -def main() -> None: - """Print the reminder when the target file is one the conventions govern.""" - try: - payload = json.load(sys.stdin) - except json.JSONDecodeError, ValueError: - return - path = payload.get("tool_input", {}).get("file_path", "") - if not path.endswith(_EXTENSIONS): - return - print( - json.dumps( - { - "hookSpecificOutput": { - "hookEventName": "PreToolUse", - "additionalContext": _REMINDER, - } - } - ) - ) - - -if __name__ == "__main__": - main() diff --git a/.codex/hooks/response_style.py b/.codex/hooks/response_style.py deleted file mode 100644 index d69a559..0000000 --- a/.codex/hooks/response_style.py +++ /dev/null @@ -1,34 +0,0 @@ -"""UserPromptSubmit hook: restate how answers to Ines should be written. - -Emits on every prompt. Carries the response-shape rules only; the writing -conventions for source files are in conventions_reminder.py. -""" - -import json - -_REMINDER = ( - "Response style:\n" - "- Short. Answer the question asked, then stop.\n" - "- High level first, in plain words. Then the low-level version, so the " - "vocabulary is picked up in context rather than assumed.\n" - "- Define a term the first time it appears, in the same sentence.\n" - "- No jargon without its plain-language equivalent alongside it." -) - - -def main() -> None: - """Print the reminder.""" - print( - json.dumps( - { - "hookSpecificOutput": { - "hookEventName": "UserPromptSubmit", - "additionalContext": _REMINDER, - } - } - ) - ) - - -if __name__ == "__main__": - main() From d216ed079deb4c360f03ae8aa08ee646a3494dde Mon Sep 17 00:00:00 2001 From: inesaranab Date: Tue, 11 Aug 2026 19:08:38 +0200 Subject: [PATCH 5/5] fix: measure the readiness deadline in elapsed time, and settle atomically Four findings from review of the readiness change. The readiness deadline counted only the intervals between probes, not the time the probes themselves took. A probe against an endpoint that is not listening consumes its full timeout before failing, so at the worker's values the wait could reach 2,730 seconds against a configured 900 -- past the limits the deadline exists to stay inside. It now measures elapsed time on a monotonic clock. abandon() still read the job and wrote the failure as two operations, so a completion landing between them was overwritten. The decision moves into the store as fail_if_pending: one operation, guarded by a lock in memory and by a conditional write on the row's etag in Table Storage. The table's test double now models etags, since one that accepted any write would hide exactly the failure being guarded against. Also corrects the Settings documentation, which still described the guardrail timeout as covering a cold start, and labels a fenced block in the runbook. --- .codex/config.toml | 2 + .codex/hooks.json | 48 ++++++++++++++++++ .codex/hooks/conventions_reminder.py | 49 ++++++++++++++++++ .codex/hooks/response_style.py | 34 +++++++++++++ app/adapters/detector_readiness.py | 14 ++++-- app/adapters/job_store_memory.py | 11 +++++ app/adapters/job_store_table.py | 38 +++++++++++++- app/config.py | 8 +-- app/domain/service.py | 16 +++--- app/ports/job_store.py | 18 +++++++ infra/gemma/README.md | 2 +- tests/unit/test_detector_readiness.py | 30 +++++++++++ tests/unit/test_job_store_memory.py | 31 ++++++++++++ tests/unit/test_job_store_table.py | 71 ++++++++++++++++++++++++++- 14 files changed, 355 insertions(+), 17 deletions(-) create mode 100644 .codex/config.toml create mode 100644 .codex/hooks.json create mode 100644 .codex/hooks/conventions_reminder.py create mode 100644 .codex/hooks/response_style.py diff --git a/.codex/config.toml b/.codex/config.toml new file mode 100644 index 0000000..9ce392a --- /dev/null +++ b/.codex/config.toml @@ -0,0 +1,2 @@ +[mcp_servers.linear-server] +url = "https://mcp.linear.app/mcp" diff --git a/.codex/hooks.json b/.codex/hooks.json new file mode 100644 index 0000000..58aea54 --- /dev/null +++ b/.codex/hooks.json @@ -0,0 +1,48 @@ +{ + "hooks": { + "PreToolUse": [ + { + "matcher": "WebSearch|WebFetch", + "hooks": [ + { + "type": "command", + "command": "echo '{\"systemMessage\":\"Scope check - what decision does this research change? If there is no answer, it is a bookmark (INE-18), not a task. Current objective: .claude/FOCUS.md\"}'" + } + ] + }, + { + "matcher": "Write|Edit", + "hooks": [ + { + "type": "command", + "command": "python3 '/Users/inesarana/screening/.codex/hooks/conventions_reminder.py' 2>/dev/null || true" + } + ] + } + ], + "SessionStart": [ + { + "hooks": [ + { + "type": "command", + "command": "cd \"$CLAUDE_PROJECT_DIR\" 2>/dev/null; python3 -c \"import json,pathlib;print(json.dumps({'hookSpecificOutput':{'hookEventName':'SessionStart','additionalContext':pathlib.Path('.claude/FOCUS.md').read_text()}}))\" 2>/dev/null || true" + } + ] + } + ], + "UserPromptSubmit": [ + { + "hooks": [ + { + "type": "command", + "command": "python3 -c \"import json,time,pathlib;f=pathlib.Path('/tmp/.claude_last_prompt');now=time.time();prev=float(f.read_text()) if f.exists() else now;f.write_text(str(now));g=int(now-prev);print(json.dumps({'hookSpecificOutput':{'hookEventName':'UserPromptSubmit','additionalContext':'[clock] '+time.strftime('%H:%M')+' | '+str(g//60)+'m '+str(g%60)+'s since previous message'}}))\" 2>/dev/null || true" + }, + { + "type": "command", + "command": "python3 '/Users/inesarana/screening/.codex/hooks/response_style.py' 2>/dev/null || true" + } + ] + } + ] + } +} diff --git a/.codex/hooks/conventions_reminder.py b/.codex/hooks/conventions_reminder.py new file mode 100644 index 0000000..564cef6 --- /dev/null +++ b/.codex/hooks/conventions_reminder.py @@ -0,0 +1,49 @@ +"""PreToolUse hook: restate the project's writing conventions before an edit. + +Fires on Write and Edit. Emits nothing for files the conventions do not cover, +so the reminder stays attached to source rather than appearing on every write. +""" + +import json +import sys + +_EXTENSIONS = (".py", ".yaml", ".yml", ".md") + +_REMINDER = ( + "screening-conventions (.claude/skills/screening-conventions/SKILL.md):\n" + "- Google-style docstrings: summary line, then Args/Returns/Raises for " + "functions, Attributes for models.\n" + "- State the PROPERTY, not the incident that taught it. No war stories, " + "dates, measurements, or 'we' — those belong in infra/*/README.md or the " + "commit message.\n" + "- Self-contained: never explain one symbol by referring to another.\n" + "- No prose constants: explanation assigned to a module-level string is " + "dead code.\n" + "- app/domain and app/ports stay vendor-free; adapters hold the vendors.\n" + "- Test first: a failing test before the implementation." +) + + +def main() -> None: + """Print the reminder when the target file is one the conventions govern.""" + try: + payload = json.load(sys.stdin) + except json.JSONDecodeError, ValueError: + return + path = payload.get("tool_input", {}).get("file_path", "") + if not path.endswith(_EXTENSIONS): + return + print( + json.dumps( + { + "hookSpecificOutput": { + "hookEventName": "PreToolUse", + "additionalContext": _REMINDER, + } + } + ) + ) + + +if __name__ == "__main__": + main() diff --git a/.codex/hooks/response_style.py b/.codex/hooks/response_style.py new file mode 100644 index 0000000..d69a559 --- /dev/null +++ b/.codex/hooks/response_style.py @@ -0,0 +1,34 @@ +"""UserPromptSubmit hook: restate how answers to Ines should be written. + +Emits on every prompt. Carries the response-shape rules only; the writing +conventions for source files are in conventions_reminder.py. +""" + +import json + +_REMINDER = ( + "Response style:\n" + "- Short. Answer the question asked, then stop.\n" + "- High level first, in plain words. Then the low-level version, so the " + "vocabulary is picked up in context rather than assumed.\n" + "- Define a term the first time it appears, in the same sentence.\n" + "- No jargon without its plain-language equivalent alongside it." +) + + +def main() -> None: + """Print the reminder.""" + print( + json.dumps( + { + "hookSpecificOutput": { + "hookEventName": "UserPromptSubmit", + "additionalContext": _REMINDER, + } + } + ) + ) + + +if __name__ == "__main__": + main() diff --git a/app/adapters/detector_readiness.py b/app/adapters/detector_readiness.py index ee7b4a7..4b1b2e3 100644 --- a/app/adapters/detector_readiness.py +++ b/app/adapters/detector_readiness.py @@ -5,6 +5,7 @@ by repeating a short request rather than by holding one open. """ +import time from collections.abc import Awaitable, Callable from typing import Any, Protocol @@ -39,20 +40,28 @@ async def wait_until_ready( deadline_s: float, interval_s: float, sleep: Callable[[float], Awaitable[None]], + now: Callable[[], float] = time.monotonic, ) -> bool: """Repeat a readiness probe until it succeeds or the deadline passes. + The deadline covers elapsed time, not time spent sleeping. A probe against + an endpoint that is not listening consumes its own timeout before failing, + so counting only the intervals would let the total reach a multiple of the + deadline. + Args: probe: Returns True once the detector is serving. deadline_s: How long to keep trying, in seconds. interval_s: Seconds between attempts. sleep: Suspends for the given seconds. Injected so tests need no real time. + now: Reads a monotonic clock, one that only moves forward and is + unaffected by the system clock being adjusted. Returns: True if the detector became ready, False if the deadline passed first. """ - waited = 0.0 + started = now() while True: try: ready = await probe() @@ -64,7 +73,6 @@ async def wait_until_ready( ready = False if ready: return True - if waited + interval_s > deadline_s: + if now() - started + interval_s > deadline_s: return False await sleep(interval_s) - waited += interval_s diff --git a/app/adapters/job_store_memory.py b/app/adapters/job_store_memory.py index 20e63b3..c774097 100644 --- a/app/adapters/job_store_memory.py +++ b/app/adapters/job_store_memory.py @@ -71,3 +71,14 @@ async def _settle( if existing is not None: settled = settled.model_copy(update={"created_at": existing.created_at}) self._jobs[job_id] = settled + + async def fail_if_pending(self, job_id: str, error: str) -> bool: + """Fail a job only while it is pending. See `JobStore.fail_if_pending`.""" + async with self._lock: + existing = self._jobs.get(job_id) + if existing is None or existing.status is not JobStatus.PENDING: + return False + self._jobs[job_id] = Job( + id=job_id, status=JobStatus.FAILED, error=error + ).model_copy(update={"created_at": existing.created_at}) + return True diff --git a/app/adapters/job_store_table.py b/app/adapters/job_store_table.py index 772504b..049c2e2 100644 --- a/app/adapters/job_store_table.py +++ b/app/adapters/job_store_table.py @@ -10,7 +10,11 @@ from datetime import datetime from typing import Protocol -from azure.core.exceptions import ResourceNotFoundError +from azure.core import MatchConditions +from azure.core.exceptions import ( + ResourceModifiedError, + ResourceNotFoundError, +) from azure.data.tables import UpdateMode from app.domain.models import Job, JobStatus, ScreenResult @@ -27,6 +31,8 @@ async def upsert_entity(self, entity: dict, **kwargs: object) -> object: ... async def get_entity(self, partition_key: str, row_key: str) -> dict: ... + async def update_entity(self, entity: dict, **kwargs: object) -> object: ... + def job_to_entity(job: Job) -> dict: """Convert a Job into an Azure Table entity. @@ -156,3 +162,33 @@ async def _settle( }, mode=UpdateMode.MERGE, ) + + async def fail_if_pending(self, job_id: str, error: str) -> bool: + """Fail a job only while it is pending. See ``JobStore.fail_if_pending``. + + The read supplies the row's etag and the write requires it to be + unchanged, so a completion that lands in between causes the write to be + refused rather than to replace the result. + """ + try: + entity = dict(await self._table.get_entity(job_id, job_id)) + except ResourceNotFoundError: + return False + if entity.get("status") != JobStatus.PENDING.value: + return False + try: + await self._table.update_entity( + { + "PartitionKey": job_id, + "RowKey": job_id, + "status": JobStatus.FAILED.value, + "result": "", + "error": error, + }, + mode=UpdateMode.MERGE, + etag=entity.get("etag"), + match_condition=MatchConditions.IfNotModified, + ) + except ResourceModifiedError: + return False + return True diff --git a/app/config.py b/app/config.py index de0d0f3..ba3f108 100644 --- a/app/config.py +++ b/app/config.py @@ -20,9 +20,11 @@ class Settings(BaseSettings): llm_guardrail_model: Model name to request at that endpoint. llm_timeout_s: Per-request timeout for the assessment LLM, in seconds. llm_guardrail_timeout_s: Per-request timeout for the guardrail endpoint. - Deliberately separate and much larger: that endpoint scales to zero, - so the first request after an idle period waits for a GPU to start - and load the model. + Separate from the assessment timeout because that endpoint runs a + larger model, and bounded below the platform's own request limit so + it can actually fire. It does not cover the endpoint starting from + zero; readiness is established by repeated probes before any + screening runs. jobs_account_url: Table endpoint of the account holding job state. jobs_queue_url: Queue endpoint of the same account. jobs_table_name: Table holding one entity per screening job. diff --git a/app/domain/service.py b/app/domain/service.py index 52818b3..40f6ce3 100644 --- a/app/domain/service.py +++ b/app/domain/service.py @@ -107,20 +107,20 @@ async def abandon(self, job_id: str, reason: str) -> None: For work that cannot be completed however many times it is tried, where another attempt would repeat the failure rather than resolve it. - Only a job still PENDING is settled. Work is given up on because it was - delivered too often, and a job can be delivered again after its outcome - was recorded, so a job already carrying an answer keeps it. + Only a job still PENDING is settled, and the store decides that as one + operation. Work is given up on because it was delivered too often, and a + job can be delivered again after its outcome was recorded, so a job + already carrying an answer keeps it -- including one that acquires an + answer while this call is in flight. Args: job_id: The id returned by ``start``. reason: Why the job was given up on. Must not quote the transcript. """ - settled = await self._jobs.get(job_id) - if settled is not None and settled.status is not JobStatus.PENDING: + if await self._jobs.fail_if_pending(job_id, reason): + logger.warning("job_abandoned", extra={"context": {"job": job_id}}) + else: logger.info("job_abandon_skipped", extra={"context": {"job": job_id}}) - return - logger.warning("job_abandoned", extra={"context": {"job": job_id}}) - await self._jobs.fail(job_id, reason) async def run(self, job_id: str, request: ScreenRequest) -> None: """Perform the screening and store its outcome against the job. diff --git a/app/ports/job_store.py b/app/ports/job_store.py index e7af240..5e1f35c 100644 --- a/app/ports/job_store.py +++ b/app/ports/job_store.py @@ -60,3 +60,21 @@ async def fail(self, job_id: str, error: str) -> None: data: the store outlives the request and nothing scrubs it. """ ... + + async def fail_if_pending(self, job_id: str, error: str) -> bool: + """Record a failure only while the job is still outstanding. + + One operation, not a read followed by a write. A job can be completed + between those two, and the failure would then replace an answer the + caller may already have read. + + Args: + job_id: The handle given out at creation. + error: Why it failed, written for an operator. Never candidate + data: the store outlives the request and nothing scrubs it. + + Returns: + True if the job was still pending and is now failed. False if it + had already finished, or does not exist. + """ + ... diff --git a/infra/gemma/README.md b/infra/gemma/README.md index 565b45d..6204701 100644 --- a/infra/gemma/README.md +++ b/infra/gemma/README.md @@ -372,7 +372,7 @@ Moving the caller off the HTTP path is not sufficient. The worker still reaches over HTTP, and that hop crosses ingress as well — internal ingress is still ingress. The first end-to-end run in production failed on exactly this: -``` +```text 13:32:48 worker_job_started 13:32:50 guardrail begins <- the call to screening-gemma 13:36:50 API call failed on attempt 1: stream timeout diff --git a/tests/unit/test_detector_readiness.py b/tests/unit/test_detector_readiness.py index cba1835..b681ca1 100644 --- a/tests/unit/test_detector_readiness.py +++ b/tests/unit/test_detector_readiness.py @@ -94,3 +94,33 @@ async def get(self, url: str): assert ok.requested == "http://detector/v1/models" assert not await http_probe(_Client(503), "http://detector/v1/models")() + + +@pytest.mark.asyncio +async def test_time_spent_probing_counts_against_the_deadline(): + """A probe against an unreachable endpoint consumes its own timeout before + failing. Counting only the sleeps would let the total run to a multiple of + the deadline, past the point at which the caller's own limits apply.""" + clock = {"t": 0.0} + probes = 0 + + async def probe() -> bool: + nonlocal probes + probes += 1 + clock["t"] += 30.0 # the probe's own timeout elapses + return False + + async def sleep(seconds: float) -> None: + clock["t"] += seconds + + ready = await wait_until_ready( + probe, + deadline_s=100, + interval_s=10, + sleep=sleep, + now=lambda: clock["t"], + ) + + assert not ready + assert clock["t"] <= 100 + 30 + assert probes <= 3 diff --git a/tests/unit/test_job_store_memory.py b/tests/unit/test_job_store_memory.py index 4100f25..e7a59a0 100644 --- a/tests/unit/test_job_store_memory.py +++ b/tests/unit/test_job_store_memory.py @@ -96,3 +96,34 @@ async def test_settling_a_job_keeps_when_it_was_accepted(): stored = await store.get("abc123") assert stored is not None assert stored.created_at == accepted.created_at + + +@pytest.mark.asyncio +async def test_fail_if_pending_refuses_a_job_that_already_finished(): + """Checking the status and writing the failure must be one operation. A job + can be completed between a separate read and write, and the failure would + then replace an answer the caller may already have read.""" + store = InMemoryJobStore() + await store.create("abc123") + await store.complete("abc123", _a_result()) + + settled = await store.fail_if_pending("abc123", "TooManyDeliveries") + + assert settled is False + job = await store.get("abc123") + assert job is not None + assert job.status is JobStatus.DONE + assert job.result is not None + + +@pytest.mark.asyncio +async def test_fail_if_pending_settles_a_job_still_pending(): + store = InMemoryJobStore() + await store.create("abc123") + + settled = await store.fail_if_pending("abc123", "TooManyDeliveries") + + assert settled is True + job = await store.get("abc123") + assert job is not None + assert job.status is JobStatus.FAILED diff --git a/tests/unit/test_job_store_table.py b/tests/unit/test_job_store_table.py index f85e1a0..00e63d4 100644 --- a/tests/unit/test_job_store_table.py +++ b/tests/unit/test_job_store_table.py @@ -5,6 +5,7 @@ """ import json +from collections.abc import Awaitable, Callable import pytest @@ -81,6 +82,13 @@ class _FakeTableClient: def __init__(self) -> None: self.entities: dict[str, dict] = {} + 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}\"" async def upsert_entity(self, entity: dict, **kwargs) -> None: """Merge into the stored row, as UpdateMode.MERGE does. @@ -89,13 +97,38 @@ async def upsert_entity(self, entity: dict, **kwargs) -> None: adapter deliberately leaves out of an update. """ self.entities.setdefault(entity["RowKey"], {}).update(entity) + self._stamp(entity["RowKey"]) + + async def update_entity(self, entity: dict, **kwargs) -> None: + """Merge only while the row still carries the etag the caller read. + + Models the conditional write the service depends on: without it a test + double would accept a stale write that real Table Storage rejects. + """ + from azure.core.exceptions import ResourceModifiedError + + row_key = entity["RowKey"] + stored = self.entities.get(row_key) + if stored is 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"): + raise ResourceModifiedError("etag mismatch") + stored.update(entity) + self._stamp(row_key) async def get_entity(self, partition_key: str, row_key: str) -> dict: from azure.core.exceptions import ResourceNotFoundError if row_key not in self.entities: raise ResourceNotFoundError("no such entity") - return self.entities[row_key] + entity = dict(self.entities[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. + await self.on_read() + return entity @pytest.fixture @@ -186,3 +219,39 @@ async def test_failing_a_job_keeps_when_it_was_accepted(): stored = await store.get("abc123") assert stored is not None assert stored.created_at == accepted.created_at + + +@pytest.mark.asyncio +async def test_fail_if_pending_loses_to_a_completion_that_lands_first(): + """The check and the write are one conditional operation. A worker that + completes the job between them changes the row, and the conditional write + is then refused rather than replacing the result.""" + client = _FakeTableClient() + store = AzureTableJobStore(client) + await store.create("abc123") + + async def complete_between_read_and_write() -> None: + client.on_read = None + await store.complete("abc123", _a_result()) + + client.on_read = complete_between_read_and_write + + settled = await store.fail_if_pending("abc123", "TooManyDeliveries") + + assert settled is False + job = await store.get("abc123") + assert job is not None + assert job.status is JobStatus.DONE + + +@pytest.mark.asyncio +async def test_fail_if_pending_settles_a_job_still_pending(): + client = _FakeTableClient() + store = AzureTableJobStore(client) + await store.create("abc123") + + assert await store.fail_if_pending("abc123", "TooManyDeliveries") is True + + job = await store.get("abc123") + assert job is not None + assert job.status is JobStatus.FAILED