Drain the worker on deploy and make interrupted or malformed runs fail cleanly

infra/install_switchboard.sh

@@ -199,6 +199,9 @@ ExecStartPre=/usr/bin/find /var/lib/tin-lite-proxy-grants -type f -delete
 ExecStart=/usr/local/bin/uv run --frozen --no-dev tin-lite serve
 Restart=always
 RestartSec=5
+# SIGTERM drains the Temporal worker (TIN_LITE_WORKER_GRACEFUL_SHUTDOWN_SECONDS,
+# default 300s) while HTTP keeps serving, then HTTP gets 20s. Stay above both.
+TimeoutStopSec=420
 NoNewPrivileges=true
 PrivateTmp=true
 ProtectSystem=strict

src/tin_lite/activities.py

@@ -117,7 +117,11 @@
     validate_site_health_model_route,
 )
 from tin_lite.system_wiki import read_system_wiki_document
-from tin_lite.usage_capture import external_usage_scope, observation_key
+from tin_lite.usage_capture import (
+    ObservationAlreadyRecorded,
+    external_usage_scope,
+    observation_key,
+)
 from tin_lite.visibility import (
     ResponseCheckpoint,
     ResponseRequest,
@@ -1129,6 +1133,9 @@ async def garden_project_memory(self, run_id_text: str) -> None:
                         )
                         if current_index is not None:
                             owned_section = extract_owned_section(current_index.decode("utf-8"))
+                    await _refuse_repeated_model_request(
+                        self._db, conn, run_id=run_id, step="memory", label="project memory"
+                    )
                     with external_usage_scope(self._db, conn, run_id, "memory"):
                         memory_index = await self._await_with_heartbeats(
                             reporter.garden(
@@ -1163,12 +1170,10 @@ async def garden_project_memory(self, run_id_text: str) -> None:
                         "source_run_ids": [str(source.run_id) for source in sources],
                     },
                 )
-            except Exception as exc:
-                await self._db.fail_effect(
-                    conn,
-                    execution_key=execution_key,
-                    error_message=_safe_failure(exc),
-                )
+            except BaseException as exc:
+                await _record_effect_failure(self._db, conn, execution_key=execution_key, exc=exc)
+                if isinstance(exc, ObservationAlreadyRecorded):
+                    raise _interrupted_model_request("project memory") from exc
                 raise
 
     @activity.defn(name="project_memory_result")
@@ -1311,6 +1316,9 @@ async def generate_scan_report(self, run_id_text: str) -> None:
                             content=await self._integration_evidence(project),
                         )
                     )
+                    await _refuse_repeated_model_request(
+                        self._db, conn, run_id=run_id, step="scan", label="project scan"
+                    )
                     with external_usage_scope(self._db, conn, run_id, "scan"):
                         report = await self._await_with_heartbeats(
                             reporter.report(
@@ -1344,12 +1352,10 @@ async def generate_scan_report(self, run_id_text: str) -> None:
                         "source_refs": [source.artifact_ref for source in sources],
                     },
                 )
-            except Exception as exc:
-                await self._db.fail_effect(
-                    conn,
-                    execution_key=execution_key,
-                    error_message=_safe_failure(exc),
-                )
+            except BaseException as exc:
+                await _record_effect_failure(self._db, conn, execution_key=execution_key, exc=exc)
+                if isinstance(exc, ObservationAlreadyRecorded):
+                    raise _interrupted_model_request("project scan") from exc
                 raise
 
     @activity.defn(name="project_scan_result")
@@ -3714,7 +3720,15 @@ async def _procedure_companions(self, run, project, procedure, revision):
                 primary = await self._storage.read_procedure_checkpoint(
                     repo_id=project.state_repo_id, revision=revision, path=procedure.output_path
                 )
-                await validate_pair(self._storage, project, run, primary, raw)
+                try:
+                    await validate_pair(self._storage, project, run, primary, raw)
+                except ValueError as exc:
+                    # The saved pair cannot change on retry; say so once instead of thrice.
+                    raise ApplicationError(
+                        f"Brand capture output is invalid: {exc}",
+                        type="BrandCaptureInvalid",
+                        non_retryable=True,
+                    ) from exc
         elif procedure.output_validator == article_review.VALIDATOR:
             article_review.validate_notes(
                 raw, revision=procedure.review_revision_context is not None
@@ -4467,14 +4481,17 @@ async def _weekly_brief_effect(
                 return existing.result
             await self._db.start_effect(conn, execution_key=execution_key, operation=operation)
             try:
+                await _refuse_repeated_model_request(
+                    self._db, conn, run_id=run_id, step="weekly_brief", label="weekly brief"
+                )
                 with external_usage_scope(self._db, conn, run_id, "weekly_brief"):
                     result = await execute()
                 await self._db.complete_effect(conn, execution_key=execution_key, result=result)
                 return result
-            except Exception as exc:
-                await self._db.fail_effect(
-                    conn, execution_key=execution_key, error_message=_safe_failure(exc)
-                )
+            except BaseException as exc:
+                await _record_effect_failure(self._db, conn, execution_key=execution_key, exc=exc)
+                if isinstance(exc, ObservationAlreadyRecorded):
+                    raise _interrupted_model_request("weekly brief") from exc
                 raise
 
     async def _pinned_native_reporter(self, run, reporter, key):
@@ -4642,6 +4659,9 @@ async def _answer_page_effect(
                 return existing.result
             await self._db.start_effect(conn, execution_key=execution_key, operation=operation)
             try:
+                await _refuse_repeated_model_request(
+                    self._db, conn, run_id=run_id, step="answer_page", label="answer page"
+                )
                 with external_usage_scope(self._db, conn, run_id, "answer_page"):
                     result = await execute()
                 await self._db.complete_effect(
@@ -4650,12 +4670,10 @@ async def _answer_page_effect(
                     result=result,
                 )
                 return result
-            except Exception as exc:
-                await self._db.fail_effect(
-                    conn,
-                    execution_key=execution_key,
-                    error_message=_safe_failure(exc),
-                )
+            except BaseException as exc:
+                await _record_effect_failure(self._db, conn, execution_key=execution_key, exc=exc)
+                if isinstance(exc, ObservationAlreadyRecorded):
+                    raise _interrupted_model_request("answer page") from exc
                 raise
 
     async def _answer_page_sources(self, *, run_id: UUID, project) -> list[AnswerPageSource]:
@@ -5141,3 +5159,45 @@ def _safe_failure(exc: BaseException) -> str:
     """Name the exception with its text, so a failed receipt says what actually went wrong."""
     text = scrub_secrets(" ".join(str(exc).split()))
     return f"{type(exc).__name__}: {text or 'operation failed'}"[:FAILURE_MESSAGE_LIMIT]
+
+
+INTERRUPTED_MODEL_REQUEST = (
+    "the model request was interrupted and was not repeated; start the run again"
+)
+
+
+def _interrupted_model_request(label: str) -> ApplicationError:
+    """A metered request whose outcome is unknown is never bought again on retry."""
+    return ApplicationError(
+        f"{label}: {INTERRUPTED_MODEL_REQUEST}",
+        type="ModelRequestInterrupted",
+        non_retryable=True,
+    )
+
+
+async def _refuse_repeated_model_request(db, conn, *, run_id: UUID, step: str, label: str) -> None:
+    if await db.get_effect(observation_key(run_id, step, "responses"), conn=conn) is not None:
+        raise _interrupted_model_request(label)
+
+
+async def _record_effect_failure(db, conn, *, execution_key: str, exc: BaseException) -> None:
+    """Mark the owning receipt failed even while the activity is being cancelled.
+
+    A deploy cancels in-flight activities; a receipt left 'started' would hide the
+    interruption from the next attempt. The write finishes before the effect lock
+    (and its connection) is released, then the cancellation continues.
+    """
+    write = asyncio.ensure_future(
+        db.fail_effect(conn, execution_key=execution_key, error_message=_safe_failure(exc))
+    )
+    interrupted = False
+    while True:
+        try:
+            await asyncio.shield(write)
+            break
+        except asyncio.CancelledError:
+            if write.done():
+                raise
+            interrupted = True
+    if interrupted and not isinstance(exc, asyncio.CancelledError):
+        raise asyncio.CancelledError

src/tin_lite/brand_capture.py

@@ -1,7 +1,9 @@
 """First capture preparation and preservation, over ordinary project files and receipts."""
 
 import hashlib
+import logging
 import re
+from functools import partial
 from urllib.parse import urlsplit
 
 from tin_lite import brand_contract as contract
@@ -11,6 +13,7 @@
 
 PACKET_MAX = 32_000
 PREPARATION = "brand_capture_preparation"
+logger = logging.getLogger(__name__)
 
 
 def public_url(value):
@@ -82,7 +85,7 @@ async def resolve_brand(storage, project, revision):
         result["diagnostics"].append(str(exc))
         return result
     try:
-        result["assessment"] = contract.assessment(result["guide"])
+        result["assessment"] = contract.assessment(result["guide"], notes=result["diagnostics"])
     except ValueError:
         result["diagnostics"].append(
             "Optional brand assessment is unavailable; the guide is valid."
@@ -225,8 +228,14 @@ async def prepare(self, run, procedure):
 async def validate_pair(storage, project, run, brand, design):
     """New files use the contract; any existing destination must be carried byte-for-byte."""
     generated_sources = []
+    notes: list[str] = []
     for path, raw, limit, validate_new in (
-        (contract.BRAND_PATH, brand, contract.BRAND_MAX, contract.validate_new_brand),
+        (
+            contract.BRAND_PATH,
+            brand,
+            contract.BRAND_MAX,
+            partial(contract.validate_new_brand, notes=notes),
+        ),
         (contract.DESIGN_PATH, design, contract.DESIGN_MAX, contract.validate_new_design),
     ):
         validate_document(raw, limit)
@@ -250,3 +259,6 @@ async def validate_pair(storage, project, run, brand, design):
             for k in brand_sources.keys() & design_sources.keys()
         ):
             raise ValueError("The same source ID cannot identify different sources across the pair")
+    for note in notes:
+        logger.info("brand capture run %s: %s", run.id, note)
+    return notes

src/tin_lite/brand_contract.py

@@ -19,6 +19,7 @@
 )
 CLASSES = {"distinctive_strong", "distinctive_inconsistent", "competent_generic", "weak", "unknown"}
 EVIDENCE = {"adequate", "partial", "insufficient", "conflicting"}
+METHOD = {"kind": "visual_capture", "rubric_version": "marketing-brand.v1"}
 _HEX = re.compile(r"#[0-9a-fA-F]{6}\Z")
 _FENCES = re.compile(r"^```json\s*\n(.*?)\n```[ \t]*$", re.M | re.S)
 _REFERENCE = re.compile(r'^\[([a-z][a-z0-9_-]{0,63})\]:\s+(\S+)(?:\s+"[^"\n]*")?\s*$', re.M)
@@ -132,11 +133,27 @@ def sources(text):
     return found
 
 
-def assessment(text, *, required=False):
+def assessment(text, *, required=False, notes=None):
+    """The validated assessment; its method is always the canonical constant.
+
+    Code owns the method. A capture that omits it or describes it differently (for
+    example "packet_only") is normalized, with a note appended to ``notes``; coverage
+    limits belong in ``evidence_status`` and the findings, not in the method.
+    """
     item = block(text, ASSESSMENT_SCHEMA, 8_000, required=required)
     if item is None:
         return None
-    value = item[0]
+    value = dict(item[0])
+    if "method" in value and not isinstance(value["method"], dict):
+        raise ValueError("Brand assessment method must be an object")
+    if value.get("method") != METHOD:
+        if notes is not None:
+            notes.append(
+                "Brand assessment method was "
+                + ("missing" if "method" not in value else "non-standard")
+                + f"; recorded as {METHOD['kind']} ({METHOD['rubric_version']})."
+            )
+        value["method"] = dict(METHOD)
     if (
         set(value)
         != {
@@ -169,8 +186,6 @@ def assessment(text, *, required=False):
         raise ValueError("Invalid capture timestamp") from exc
     if stamp.tzinfo is None:
         raise ValueError("Capture timestamp must include its timezone")
-    if value["method"] != {"kind": "visual_capture", "rubric_version": "marketing-brand.v1"}:
-        raise ValueError("Unsupported brand assessment method")
     findings = value["findings"]
     if not isinstance(findings, list) or not 1 <= len(findings) <= 5:
         raise ValueError("Brand assessment requires one to five grounded findings")
@@ -203,10 +218,10 @@ def sections(text, headings):
     return result
 
 
-def validate_new_brand(text):
+def validate_new_brand(text, *, notes=None):
     parts = sections(text, BRAND_SECTIONS)
     palette = tokens(text)
-    assessment(parts["Assessment and sources"], required=True)
+    assessment(parts["Assessment and sources"], required=True, notes=notes)
     if block(text, TOKEN_SCHEMA, 2_000)[2] != len(text.rstrip()):
         raise ValueError("Brand tokens must be the final block")
     visual = parts["Visual style"].lower()

src/tin_lite/cli.py

@@ -196,6 +196,38 @@ async def _validate_community(
         raise SystemExit(1)
 
 
+def _serve() -> None:
+    from tin_lite.main import app  # imported here: building the app reads settings
+    from tin_lite.serving import DrainingServer
+
+    settings = get_settings()
+
+    async def drain_worker() -> None:
+        runtime = getattr(app.state, "runtime", None)
+        if runtime is not None:
+            await runtime.worker.shutdown()
+
+    # OAuth providers return short-lived authorization codes in callback query
+    # strings. Uvicorn's generic access logger includes the full query string;
+    # disable it and retain only deliberate application-level event logging.
+    config = uvicorn.Config(
+        app,
+        host="0.0.0.0",  # noqa: S104
+        port=8000,
+        access_log=False,
+        # Persistent MCP/SSE connections must not hold shutdown until systemd
+        # kills the process. This applies after the worker drain below.
+        timeout_graceful_shutdown=20,
+    )
+    DrainingServer(
+        config,
+        drain=drain_worker,
+        # The worker cancels activities at its graceful timeout; the margin covers
+        # reporting their results and stopping the pollers.
+        drain_timeout=settings.worker_graceful_shutdown_seconds + 60,
+    ).run()
+
+
 def main() -> None:
     parser = argparse.ArgumentParser(prog="tin-lite")
     commands = parser.add_subparsers(dest="command", required=True)
@@ -246,18 +278,7 @@ def main() -> None:
     elif args.command == "validate-community":
         asyncio.run(_validate_community(args.root, args.private, args.package))
     elif args.command == "serve":
-        # OAuth providers return short-lived authorization codes in callback query
-        # strings. Uvicorn's generic access logger includes the full query string;
-        # disable it and retain only deliberate application-level event logging.
-        uvicorn.run(
-            "tin_lite.main:app",
-            host="0.0.0.0",  # noqa: S104
-            port=8000,
-            access_log=False,
-            # Persistent MCP/SSE connections must not hold shutdown until systemd
-            # kills the process. Leave time for the lifespan/worker cleanup too.
-            timeout_graceful_shutdown=20,
-        )
+        _serve()
     elif args.command == "create-project":
         asyncio.run(_create_project(args.name, args.repo_id))
     elif args.command == "grant-project-access":