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..4b1b2e3 --- /dev/null +++ b/app/adapters/detector_readiness.py @@ -0,0 +1,78 @@ +"""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. +""" + +import time +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]], + 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. + """ + started = now() + 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 now() - started + interval_s > deadline_s: + return False + await sleep(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 3ffce69..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. @@ -39,16 +41,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/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..40f6ce3 100644 --- a/app/domain/service.py +++ b/app/domain/service.py @@ -107,12 +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, 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. """ - logger.warning("job_abandoned", extra={"context": {"job": job_id}}) - await self._jobs.fail(job_id, reason) + 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}}) 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/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..6204701 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: + +```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 +``` + +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_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(): diff --git a/tests/unit/test_detector_readiness.py b/tests/unit/test_detector_readiness.py new file mode 100644 index 0000000..b681ca1 --- /dev/null +++ b/tests/unit/test_detector_readiness.py @@ -0,0 +1,126 @@ +"""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")() + + +@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 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 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