From 781df6a6d3c4ea1861a135e6c65c9b61dcaf99ee Mon Sep 17 00:00:00 2001 From: Ege Demir Date: Fri, 25 Sep 2026 10:36:23 +0000 Subject: [PATCH] Drain the worker on deploy and make interrupted or malformed runs fail cleanly - SIGTERM now drains the Temporal worker (graceful_shutdown_timeout, default 300s via TIN_LITE_WORKER_GRACEFUL_SHUTDOWN_SECONDS) while HTTP and the Codex relay keep serving; uvicorn's own 20s shutdown follows. WorkerGroup shutdown is idempotent and no longer hangs on a worker that never started. install_switchboard.sh sets TimeoutStopSec=420. - Weekly brief, answer page, project memory and project scan receipts are marked failed even on cancellation, and a retry whose model request was already observed fails non-retryably instead of "External request already observed". begin_observation raises ObservationAlreadyRecorded. - brand.capture: code owns the assessment method; variants are normalized with a note, non-objects are rejected, and an invalid pair fails once. - visibility.audit: question IDs are normalized deterministically before validation and receipt keys; adjudication references follow. - commit_project_changes maps code.storage RefUpdateError to invalid (no changes), conflict (stale ref) or a recorded storage failure. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01Ua8Ysd7jtjWKHGg9DrUs4n --- infra/install_switchboard.sh | 3 + src/tin_lite/activities.py | 108 +++++++++++--- src/tin_lite/brand_capture.py | 16 +- src/tin_lite/brand_contract.py | 27 +++- src/tin_lite/cli.py | 45 ++++-- src/tin_lite/code_storage.py | 18 ++- src/tin_lite/main.py | 2 + src/tin_lite/project_files.py | 6 + src/tin_lite/runtime.py | 4 + src/tin_lite/serving.py | 71 +++++++++ src/tin_lite/settings.py | 6 + src/tin_lite/usage_capture.py | 12 +- src/tin_lite/visibility.py | 53 ++++++- src/tin_lite/worker_group.py | 37 ++++- tests/test_answer_page.py | 2 +- tests/test_brand_capture.py | 66 +++++++++ tests/test_cli_serve.py | 36 +++-- tests/test_effect_interruption.py | 97 ++++++++++++ tests/test_project_commit_errors.py | 133 +++++++++++++++++ tests/test_project_memory.py | 2 +- tests/test_runtime.py | 5 + tests/test_scan_report.py | 2 +- tests/test_visibility_audit.py | 58 ++++++++ tests/test_worker_drain.py | 138 ++++++++++++++++++ .../skills/brand-capture/CONTRACT.md | 8 +- workflow_packages/brand.capture/workflow.json | 2 +- 26 files changed, 890 insertions(+), 67 deletions(-) create mode 100644 src/tin_lite/serving.py create mode 100644 tests/test_effect_interruption.py create mode 100644 tests/test_project_commit_errors.py create mode 100644 tests/test_worker_drain.py diff --git a/infra/install_switchboard.sh b/infra/install_switchboard.sh index 409b90e4..86e7bb0a 100755 --- a/infra/install_switchboard.sh +++ b/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 diff --git a/src/tin_lite/activities.py b/src/tin_lite/activities.py index 06b12794..6ddd9314 100644 --- a/src/tin_lite/activities.py +++ b/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 diff --git a/src/tin_lite/brand_capture.py b/src/tin_lite/brand_capture.py index b51606ae..0f4a8037 100644 --- a/src/tin_lite/brand_capture.py +++ b/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 diff --git a/src/tin_lite/brand_contract.py b/src/tin_lite/brand_contract.py index d2caf386..4d877be8 100644 --- a/src/tin_lite/brand_contract.py +++ b/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() diff --git a/src/tin_lite/cli.py b/src/tin_lite/cli.py index 37b04199..1a448132 100644 --- a/src/tin_lite/cli.py +++ b/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": diff --git a/src/tin_lite/code_storage.py b/src/tin_lite/code_storage.py index f8fab59d..8d311b56 100644 --- a/src/tin_lite/code_storage.py +++ b/src/tin_lite/code_storage.py @@ -13,7 +13,7 @@ import jwt from cryptography.hazmat.primitives import serialization from pierre_storage import GitStorage -from pierre_storage.errors import ApiError +from pierre_storage.errors import ApiError, RefUpdateError from pierre_storage.types import GitFileMode, Repo from pierre_storage.version import get_user_agent @@ -89,6 +89,18 @@ class SandboxRemotes: ephemeral_auth_header: str +_REF_CONFLICTS = {"conflict", "precondition_failed", "409", "412"} + + +def _project_commit_error(exc: RefUpdateError) -> Exception: + """Name a rejected member commit in the terms the file service and MCP understand.""" + if "no changes" in (exc.message or "").lower(): + return ValueError("no changes: files already match expected_revision") + if {str(exc.reason).lower(), str(exc.status).lower()} & _REF_CONFLICTS: + return RuntimeError("canonical project state changed before file commit") + return RuntimeError(f"code.storage rejected the project file commit ({exc.reason})") + + class CodeStorage: def __init__(self, *, organization: str, private_key: str) -> None: self._organization = organization @@ -940,11 +952,13 @@ async def commit_project_changes( try: result = await builder.send() return result["commit_sha"], tuple(sorted(changed_paths)) - except Exception: + except Exception as exc: recent = await repo.list_commits(branch=branch, limit=1, ttl=300) commits = recent.get("commits", []) if commits and commits[0].get("message") == commit_message: return commits[0]["sha"], tuple(sorted(changed_paths)) + if isinstance(exc, RefUpdateError): + raise _project_commit_error(exc) from exc raise async def revert_latest_project_commit( diff --git a/src/tin_lite/main.py b/src/tin_lite/main.py index 08129105..9bd02296 100644 --- a/src/tin_lite/main.py +++ b/src/tin_lite/main.py @@ -72,6 +72,8 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: if billing_task is not None: billing_task.cancel() await asyncio.gather(billing_task, return_exceptions=True) + # Usually already drained by the serve command's SIGTERM handling; the + # group's shutdown is idempotent and then returns at once. await runtime.worker.shutdown() posthog_client = getattr(mcp_server, "posthog_client", None) if posthog_client is not None: diff --git a/src/tin_lite/project_files.py b/src/tin_lite/project_files.py index 31b5741c..9120b801 100644 --- a/src/tin_lite/project_files.py +++ b/src/tin_lite/project_files.py @@ -238,6 +238,12 @@ async def commit( message=normalized_message, changes=normalized, ) + except ValueError: + # Nothing was committed; do not leave the request looking in flight. + await self._database.fail_project_file_change( + project_id=project.id, request_id=request_id, error_code="invalid" + ) + raise except RuntimeError as exc: await self._database.fail_project_file_change( project_id=project.id, diff --git a/src/tin_lite/runtime.py b/src/tin_lite/runtime.py index d3a30b17..dcdb4d43 100644 --- a/src/tin_lite/runtime.py +++ b/src/tin_lite/runtime.py @@ -1,6 +1,7 @@ from __future__ import annotations from dataclasses import dataclass +from datetime import timedelta from pathlib import Path from temporalio import activity @@ -427,6 +428,7 @@ async def build_runtime(settings: Settings) -> RuntimeServices: by_name = {activity._Definition.must_from_callable(fn).name: fn for fn in activities} if set(by_name) != CODEX_ACTIVITIES | TRUSTED_ACTIVITIES: raise RuntimeError("Every registered activity needs an explicit worker lane") + graceful_shutdown = timedelta(seconds=settings.worker_graceful_shutdown_seconds) worker = WorkerGroup( Worker( temporal, @@ -435,6 +437,7 @@ async def build_runtime(settings: Settings) -> RuntimeServices: # Project child-workflow IDs serialize execution; this is only the # machine's parallel capacity across projects, not an account lock. max_concurrent_activities=4, + graceful_shutdown_timeout=graceful_shutdown, workflows=registered_workflows(), workflow_runner=workflow_runner(), activities=activities, @@ -444,6 +447,7 @@ async def build_runtime(settings: Settings) -> RuntimeServices: temporal, task_queue=trusted_task_queue(settings.task_queue), max_concurrent_activities=4, + graceful_shutdown_timeout=graceful_shutdown, activities=[by_name[name] for name in sorted(TRUSTED_ACTIVITIES)], ), ) diff --git a/src/tin_lite/serving.py b/src/tin_lite/serving.py new file mode 100644 index 00000000..6a356ccc --- /dev/null +++ b/src/tin_lite/serving.py @@ -0,0 +1,71 @@ +"""Serve HTTP while the Temporal worker drains on shutdown. + +Activities running in the worker call back into this same process (the Codex relay +under /internal/codex-api). If uvicorn stopped first, a deploy would cut off every +in-flight model call. On the first SIGTERM/SIGINT this server therefore keeps +accepting requests, asks the worker to stop polling and finish its activities +(bounded by the worker's graceful timeout), and only then lets uvicorn begin its +own HTTP shutdown. +""" + +from __future__ import annotations + +import asyncio +import logging +import signal +from collections.abc import Awaitable, Callable +from types import FrameType + +import uvicorn + +logger = logging.getLogger(__name__) + +Drain = Callable[[], Awaitable[None]] + + +class DrainingServer(uvicorn.Server): + def __init__(self, config: uvicorn.Config, *, drain: Drain, drain_timeout: float) -> None: + super().__init__(config) + self._drain = drain + self._drain_timeout = drain_timeout + self._loop: asyncio.AbstractEventLoop | None = None + self._drain_task: asyncio.Task[None] | None = None + self._drain_requested = False + + async def serve(self, sockets=None) -> None: # type: ignore[no-untyped-def] + self._loop = asyncio.get_running_loop() + await super().serve(sockets) + + def handle_exit(self, sig: int, frame: FrameType | None) -> None: + # Runs as a synchronous signal handler, so it only schedules work on the loop. + if self._drain_requested and not self.should_exit: + self._captured_signals.append(sig) + if sig == signal.SIGINT: + # An operator's second Ctrl-C stops waiting for the drain. + logger.warning("Second interrupt during worker drain; stopping HTTP now") + self.should_exit = True + # Repeated SIGTERMs (systemd and the `uv run` parent both send one) are + # absorbed; systemd's TimeoutStopSec is the hard bound. + return + if self._drain_requested or self._loop is None or self._loop.is_closed(): + super().handle_exit(sig, frame) + return + self._drain_requested = True + self._captured_signals.append(sig) + self._loop.call_soon_threadsafe(self._start_drain) + + def _start_drain(self) -> None: + self._drain_task = asyncio.create_task(self._drain_then_exit(), name="worker-drain") + + async def _drain_then_exit(self) -> None: + logger.info("Draining the Temporal worker before stopping HTTP") + try: + await asyncio.wait_for(self._drain(), timeout=self._drain_timeout) + except TimeoutError: + logger.error("Worker drain exceeded %ss; stopping HTTP anyway", self._drain_timeout) + except Exception: + logger.exception("Worker drain failed; stopping HTTP anyway") + else: + logger.info("Worker drained; stopping HTTP") + finally: + self.should_exit = True diff --git a/src/tin_lite/settings.py b/src/tin_lite/settings.py index a3b599bb..604776c4 100644 --- a/src/tin_lite/settings.py +++ b/src/tin_lite/settings.py @@ -191,6 +191,12 @@ class Settings(BaseSettings): default="tin-lite-codex-studio", alias="TIN_LITE_E2B_STUDIO_TEMPLATE" ) task_queue: str = Field(default="tin-lite-checkpoint-a", alias="TIN_LITE_TASK_QUEUE") + # On SIGTERM the worker stops polling and in-flight activities get this long to + # finish before Temporal cancels them. HTTP (including the Codex relay those + # activities call) keeps serving meanwhile; systemd TimeoutStopSec must exceed it. + worker_graceful_shutdown_seconds: int = Field( + default=300, ge=0, alias="TIN_LITE_WORKER_GRACEFUL_SHUTDOWN_SECONDS" + ) switchboard_public_url: str = Field( default="http://127.0.0.1:8000", alias="TIN_LITE_PUBLIC_URL" ) diff --git a/src/tin_lite/usage_capture.py b/src/tin_lite/usage_capture.py index 59c92822..06fd73c3 100644 --- a/src/tin_lite/usage_capture.py +++ b/src/tin_lite/usage_capture.py @@ -76,6 +76,14 @@ def billable_search_calls(output): return searches +class ObservationAlreadyRecorded(RuntimeError): + """A metered request for this run step was already attempted. + + Its outcome may be unknown, so it is never silently repeated; callers convert + this into a terminal run error. + """ + + def observation_key(run_id, step, endpoint): return f"usage:{run_id}:external:{hashlib.sha256(f'{step}:{endpoint}'.encode()).hexdigest()}" @@ -90,7 +98,9 @@ async def begin_observation(provider, category, endpoint, *, request=None): if existing is not None: # The owning workflow reconciles its output. An observation is not permission # to repeat a request whose outcome is unknown. - raise RuntimeError("External request already observed; recover its owning effect") + raise ObservationAlreadyRecorded( + "External request already observed; recover its owning effect" + ) record = { "version": 1, "run_id": str(run_id), diff --git a/src/tin_lite/visibility.py b/src/tin_lite/visibility.py index 58df4ddc..747d3227 100644 --- a/src/tin_lite/visibility.py +++ b/src/tin_lite/visibility.py @@ -17,6 +17,8 @@ MAX_SOURCES_PER_ANSWER = 20 MAX_VISIBILITY_RESPONSE_BYTES = 250_000 DOMAIN_PATTERN = r"(?:[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,63}" +QUESTION_ID_MAX = 32 +QUESTION_ID_PATTERN = r"^[a-z0-9_-]{1,32}$" ResponseRequest = Callable[[], Awaitable[dict[str, Any]]] ResponseCheckpoint = Callable[[ResponseRequest], Awaitable[dict[str, Any]]] QUESTION_FAMILIES = ( @@ -108,6 +110,8 @@ async def prepare_panel( panel["model"] = self.model if isinstance(panel.get("target"), dict): panel["target"]["domain"] = _canonical_domain(panel["target"].get("domain")) + # Before validation and before the activity derives answer receipts from them. + _normalize_question_ids(panel) _validate_panel(panel, target_request=target_request) if "CANDIDATE_BANK_V1" in self._skill_suite: _validate_candidate_bank(panel) @@ -352,7 +356,7 @@ def _panel_schema(*, candidate_bank: bool = False) -> dict[str, Any]: "type": "object", "additionalProperties": False, "properties": { - "id": {"type": "string"}, + "id": {"type": "string", "pattern": QUESTION_ID_PATTERN}, "family": {"type": "string", "enum": list(QUESTION_FAMILIES)}, "fit": { "type": "string", @@ -486,6 +490,47 @@ def _adjudication_schema() -> dict[str, Any]: } +def _question_id_base(value: str) -> str: + return re.sub(r"[^a-z0-9_-]", "_", value.strip().lower())[:QUESTION_ID_MAX] + + +def _normalize_question_ids(panel: dict[str, Any]) -> dict[str, str]: + """Make model-chosen question IDs safe and unique, deterministically. + + IDs name answer receipts, so the same panel response always yields the same IDs: + lowercase, other characters replaced with "_", at most 32 characters, and a + "-2", "-3"... suffix for a repeat. Returns the original-to-normalized map. + """ + questions = panel.get("questions") + mapping: dict[str, str] = {} + if not isinstance(questions, list): + return mapping + used: set[str] = set() + for index, item in enumerate(questions, start=1): + if not isinstance(item, dict) or not isinstance(item.get("id"), str): + continue + original = item["id"] + base = _question_id_base(original) or f"q{index}" + candidate, attempt = base, 2 + while candidate in used: + suffix = f"-{attempt}" + candidate = base[: QUESTION_ID_MAX - len(suffix)] + suffix + attempt += 1 + used.add(candidate) + item["id"] = candidate + mapping.setdefault(original, candidate) + return mapping + + +def _question_ref(value: Any, question_ids: list[str]) -> Any: + """Map a reference written in the pre-normalization spelling onto a panel ID.""" + if isinstance(value, str) and value not in question_ids: + normalized = _question_id_base(value) + if normalized in question_ids: + return normalized + return value + + def _validate_panel(panel: dict[str, Any], *, target_request: str) -> None: target = panel.get("target") questions = panel.get("questions") @@ -535,7 +580,7 @@ def _validate_panel(panel: dict[str, Any], *, target_request: str) -> None: text = item.get("text") if ( not isinstance(question_id, str) - or not re.fullmatch(r"[a-z0-9_-]{1,32}", question_id) + or not re.fullmatch(QUESTION_ID_PATTERN, question_id) or question_id in ids ): raise VisibilityProtocolError("visibility question ID is invalid") @@ -576,6 +621,7 @@ def _normalize_adjudication( for outcome in outcomes: if not isinstance(outcome, dict): raise VisibilityProtocolError("visibility outcome is invalid") + outcome["question_id"] = _question_ref(outcome.get("question_id"), question_ids) question_id = outcome.get("question_id") if question_id not in question_ids or question_id in by_id: raise VisibilityProtocolError("visibility outcome question is invalid") @@ -632,6 +678,9 @@ def _normalize_adjudication( if not isinstance(recommendation.get("action"), str) or not recommendation["action"]: raise VisibilityProtocolError("visibility recommendation action is invalid") evidence_ids = recommendation.get("evidence_question_ids") + if isinstance(evidence_ids, list): + evidence_ids = [_question_ref(item, question_ids) for item in evidence_ids] + recommendation["evidence_question_ids"] = evidence_ids if not isinstance(evidence_ids, list) or any( item not in question_ids for item in evidence_ids ): diff --git a/src/tin_lite/worker_group.py b/src/tin_lite/worker_group.py index f76f7c86..29ff8576 100644 --- a/src/tin_lite/worker_group.py +++ b/src/tin_lite/worker_group.py @@ -8,16 +8,51 @@ class WorkerGroup: + """Runs the workers together and shuts them down exactly once. + + Shutdown is idempotent: the deploy drain (on SIGTERM) and the application + lifespan both request it, and the second request only waits for the first. + """ + def __init__(self, *workers: Worker) -> None: self.workers = workers + self._tasks: list[asyncio.Task[None]] | None = None + self._shutdown: asyncio.Task[None] | None = None async def run(self) -> None: tasks = [asyncio.create_task(worker.run()) for worker in self.workers] + self._tasks = tasks try: await asyncio.gather(*tasks) finally: await self.shutdown() await asyncio.gather(*tasks, return_exceptions=True) + @property + def is_shutdown(self) -> bool: + return self._shutdown is not None and self._shutdown.done() + async def shutdown(self) -> None: - await asyncio.gather(*(worker.shutdown() for worker in self.workers)) + if self._shutdown is None: + self._shutdown = asyncio.create_task(self._stop(), name="temporal-worker-shutdown") + # A cancelled caller must not cancel the drain that other callers await. + await asyncio.shield(self._shutdown) + + async def _stop(self) -> None: + if self._tasks is None: + return # never started, so there is nothing to drain + await asyncio.gather( + *(self._stop_one(w, t) for w, t in zip(self.workers, self._tasks, strict=True)) + ) + + @staticmethod + async def _stop_one(worker: Worker, run_task: asyncio.Task[None]) -> None: + if run_task.done(): + return # already finished, including a worker that failed to start + # Worker.shutdown() waits for a completion event that a worker failing before + # it starts polling never sets; its run task finishing is enough in that case. + stop = asyncio.create_task(worker.shutdown()) + await asyncio.wait({stop, run_task}, return_when=asyncio.FIRST_COMPLETED) + if not stop.done(): + stop.cancel() + await asyncio.gather(stop, return_exceptions=True) diff --git a/tests/test_answer_page.py b/tests/test_answer_page.py index 41e6f008..e32e4f75 100644 --- a/tests/test_answer_page.py +++ b/tests/test_answer_page.py @@ -220,7 +220,7 @@ async def list_memory_source_runs(self, **values): assert values["exclude_run_id"] == self.run.id return [self.visibility_run] - async def get_effect(self, execution_key): + async def get_effect(self, execution_key, conn=None): return self.receipts.get(execution_key) async def project_success(self, **values): diff --git a/tests/test_brand_capture.py b/tests/test_brand_capture.py index 6dffe783..0fa91457 100644 --- a/tests/test_brand_capture.py +++ b/tests/test_brand_capture.py @@ -421,3 +421,69 @@ async def test_validated_capture_pair_is_reviewed_then_adopted(publication_db): assert f.storage.repo.trees[f.storage.repo.head][brand.DESIGN_PATH][1] == DESIGN await activities.project_codex_procedure_result(str(run.id)) assert (await f.db.get_run(run.id)).status.value == "succeeded" + + +@pytest.mark.parametrize( + "method", + [ + {"kind": "packet_only", "rubric_version": "marketing-brand.v1"}, + {"kind": "visual_capture"}, + {}, + None, # omitted + ], +) +def test_variant_assessment_method_is_normalized_with_a_note(method): + report = deepcopy(ASSESSMENT) + if method is None: + del report["method"] + else: + report["method"] = method + notes = [] + assert brand.validate_new_brand(brand_doc(assessment=report).decode(), notes=notes) == TOKENS + value = brand.assessment(brand_doc(assessment=report).decode()) + assert value["method"] == brand.METHOD + assert len(notes) == 1 and "visual_capture" in notes[0] + canonical = [] + brand.validate_new_brand(brand_doc().decode(), notes=canonical) + assert canonical == [] + + +@pytest.mark.parametrize("method", ["visual_capture", ["visual_capture"], 1]) +def test_assessment_method_that_is_not_an_object_is_rejected(method): + report = {**ASSESSMENT, "method": method} + with pytest.raises(ValueError, match="method must be an object"): + brand.validate_new_brand(brand_doc(assessment=report).decode()) + + +async def test_malformed_capture_pair_fails_once_without_retries(publication_db): + from dataclasses import replace + + from temporalio.exceptions import ApplicationError + + f = await setup(publication_db) + run = await start(f, product_url="https://example.com/") + prepared = await f.sources.prepare(run, f.procedure) + base = prepared["project_revision"] + run = replace(run, expected_head_sha=base) + procedure = f.procedure.resolve_inputs(run.input, run_id=run.id) + activities = TinActivities( + database=f.db, + storage=f.storage, + settings=f.settings, + sandboxes=SimpleNamespace(), + integrations=f.runtime.integrations, + ) + variant = {**ASSESSMENT, "method": {"kind": "packet_only"}} + revision = f.storage.repo.edit( + {procedure.output_path: brand_doc(assessment=variant), procedure.companion_path: DESIGN} + ) + f.storage.repo.head = base + assert await activities._procedure_companions(run, f.project, procedure, revision) + malformed = {**ASSESSMENT, "method": "visual_capture"} + revision = f.storage.repo.edit( + {procedure.output_path: brand_doc(assessment=malformed), procedure.companion_path: DESIGN} + ) + f.storage.repo.head = base + with pytest.raises(ApplicationError, match="method must be an object") as failed: + await activities._procedure_companions(run, f.project, procedure, revision) + assert failed.value.non_retryable and failed.value.type == "BrandCaptureInvalid" diff --git a/tests/test_cli_serve.py b/tests/test_cli_serve.py index 2426d4cd..3a871178 100644 --- a/tests/test_cli_serve.py +++ b/tests/test_cli_serve.py @@ -1,17 +1,31 @@ -from unittest.mock import Mock +import asyncio +import sys +from types import SimpleNamespace +from unittest.mock import AsyncMock -from tin_lite import cli +from tin_lite import cli, serving def test_serve_bounds_connection_drain_without_logging_oauth_urls(monkeypatch): - serve = Mock() - monkeypatch.setattr(cli.uvicorn, "run", serve) + app = SimpleNamespace(state=SimpleNamespace()) + monkeypatch.setitem(sys.modules, "tin_lite.main", SimpleNamespace(app=app)) + monkeypatch.setattr( + cli, "get_settings", lambda: SimpleNamespace(worker_graceful_shutdown_seconds=300) + ) + started = [] + monkeypatch.setattr(serving.DrainingServer, "run", lambda self: started.append(self)) monkeypatch.setattr("sys.argv", ["tin-lite", "serve"]) cli.main() - serve.assert_called_once_with( - "tin_lite.main:app", - host="0.0.0.0", # noqa: S104 - port=8000, - access_log=False, - timeout_graceful_shutdown=20, - ) + + (server,) = started + assert server.config.app is app + assert server.config.host == "0.0.0.0" # noqa: S104 + assert server.config.port == 8000 + assert server.config.access_log is False + assert server.config.timeout_graceful_shutdown == 20 + assert server._drain_timeout == 360 + + asyncio.run(server._drain()) # before startup there is no worker to drain + app.state.runtime = SimpleNamespace(worker=SimpleNamespace(shutdown=AsyncMock())) + asyncio.run(server._drain()) + app.state.runtime.worker.shutdown.assert_awaited_once() diff --git a/tests/test_effect_interruption.py b/tests/test_effect_interruption.py new file mode 100644 index 00000000..1d2f29f1 --- /dev/null +++ b/tests/test_effect_interruption.py @@ -0,0 +1,97 @@ +"""A deploy that cancels a model call must leave a failed receipt and a terminal retry.""" + +import asyncio + +import pytest +from temporalio.exceptions import ApplicationError +from test_procedure_publication import activity_fixture +from test_procedure_publication import publication_db as publication_db + +from tin_lite.activities import INTERRUPTED_MODEL_REQUEST +from tin_lite.usage_capture import ( + ObservationAlreadyRecorded, + begin_observation, + external_usage_scope, +) + +EFFECTS = [ + ("_weekly_brief_effect", "weekly_brief:model"), + ("_answer_page_effect", "answer_page:model"), +] + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("method", "suffix"), EFFECTS) +async def test_cancelled_model_call_fails_receipt_and_is_never_repeated( + publication_db, method, suffix +): + db = publication_db + activities, _, run, _ = await activity_fixture(db) + effect = getattr(activities, method) + dispatched = asyncio.Event() + + async def interrupted_call(): + await begin_observation("openai", "model", "responses") + dispatched.set() + await asyncio.Event().wait() # the deploy cancels the activity here + + task = asyncio.create_task(effect(run_id=run.id, execute=interrupted_call)) + await asyncio.wait_for(dispatched.wait(), 5) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + receipt = await db.get_effect(f"{run.id}:{suffix}") + assert receipt is not None and receipt.status == "failed" + assert "CancelledError" in (receipt.error_message or "") + + async def must_not_run(): + raise AssertionError("an interrupted paid request was repeated") + + with pytest.raises(ApplicationError) as retried: + await effect(run_id=run.id, execute=must_not_run) + assert retried.value.non_retryable + assert retried.value.type == "ModelRequestInterrupted" + assert INTERRUPTED_MODEL_REQUEST in retried.value.message + assert "already observed" not in retried.value.message + receipt = await db.get_effect(f"{run.id}:{suffix}") + assert receipt.status == "failed" + assert INTERRUPTED_MODEL_REQUEST in (receipt.error_message or "") + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("method", "suffix"), EFFECTS) +async def test_observation_conflict_inside_the_call_is_terminal(publication_db, method, suffix): + db = publication_db + activities, _, run, _ = await activity_fixture(db) + effect = getattr(activities, method) + + async def second_request(): + await begin_observation("openai", "model", "other-endpoint") + await begin_observation("openai", "model", "other-endpoint") + + with pytest.raises(ApplicationError) as failed: + await effect(run_id=run.id, execute=second_request) + assert failed.value.non_retryable + assert isinstance(failed.value.__cause__, ObservationAlreadyRecorded) + assert isinstance(failed.value.__cause__, RuntimeError) # older callers still match + receipt = await db.get_effect(f"{run.id}:{suffix}") + assert receipt.status == "failed" + + +@pytest.mark.asyncio +async def test_receipt_wedged_as_started_by_an_older_release_is_recovered(publication_db): + db = publication_db + activities, _, run, _ = await activity_fixture(db) + key = f"{run.id}:weekly_brief:model" + async with db.effect_lock(key, "weekly_brief_model") as (conn, _): + await db.start_effect(conn, execution_key=key, operation="weekly_brief_model") + with external_usage_scope(db, conn, run.id, "weekly_brief"): + await begin_observation("openai", "model", "responses") + + async def must_not_run(): + raise AssertionError("an interrupted paid request was repeated") + + with pytest.raises(ApplicationError, match="was not repeated") as retried: + await activities._weekly_brief_effect(run_id=run.id, execute=must_not_run) + assert retried.value.non_retryable + assert (await db.get_effect(key)).status == "failed" diff --git a/tests/test_project_commit_errors.py b/tests/test_project_commit_errors.py new file mode 100644 index 00000000..2327e4a8 --- /dev/null +++ b/tests/test_project_commit_errors.py @@ -0,0 +1,133 @@ +"""code.storage ref-update rejections reach members and agents as invalid or conflict.""" + +from __future__ import annotations + +from contextlib import asynccontextmanager +from types import SimpleNamespace +from unittest.mock import AsyncMock +from uuid import uuid4 + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec +from pierre_storage.errors import RefUpdateError + +from tin_lite.code_storage import CodeStorage +from tin_lite.mcp_errors import tool_error +from tin_lite.project_files import ProjectFileService, StaleProjectRevisionError + +HEAD = "a" * 40 +CHANGES = [{"operation": "upsert", "path": "notes.md", "content": "same as before"}] + + +class RejectingRepo: + def __init__(self, error: RefUpdateError) -> None: + self.error = error + + async def list_files(self, **values) -> dict: + return {"paths": ["notes.md"]} + + async def list_commits(self, **values) -> dict: + return {"commits": [{"sha": HEAD, "message": "an earlier commit"}]} + + def create_commit(self, **values): + error = self.error + + class Builder: + def add_file_from_string(self, path: str, content: str): + return self + + async def send(self) -> dict: + raise error + + return Builder() + + +def storage_rejecting(monkeypatch: pytest.MonkeyPatch, error: RefUpdateError) -> CodeStorage: + pem = ( + ec.generate_private_key(ec.SECP256R1()) + .private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + .decode() + ) + storage = CodeStorage(organization="tin", private_key=pem) + repo = RejectingRepo(error) + + async def get_repo(repo_id: str): + return repo + + async def head_sha(repo, branch: str): + return HEAD + + monkeypatch.setattr(storage, "get_repo", get_repo) + monkeypatch.setattr(storage, "head_sha", head_sha) + return storage + + +def file_service(storage: CodeStorage): + @asynccontextmanager + async def lock(**values): + yield + + database = SimpleNamespace( + project_file_change_lock=lock, + get_project_file_change=AsyncMock(return_value=None), + start_project_file_change=AsyncMock(), + fail_project_file_change=AsyncMock(), + complete_project_file_change=AsyncMock(), + ) + return ProjectFileService(database=database, storage=storage), database # type: ignore[arg-type] + + +async def commit(service): + project = SimpleNamespace(id=uuid4(), state_repo_id="projects/p", canonical_branch="main") + return await service.commit( + project=project, + actor_clerk_user_id="user_member", + client_id=None, + request_id=uuid4(), + expected_revision=HEAD, + message="tick the plan", + changes=CHANGES, + ) + + +@pytest.mark.asyncio +async def test_no_changes_is_invalid_and_the_request_is_not_left_started(monkeypatch): + storage = storage_rejecting(monkeypatch, RefUpdateError("no changes to commit")) + service, database = file_service(storage) + with pytest.raises(ValueError, match="no changes") as caught: + await commit(service) + assert str(tool_error(caught.value)) == ( + "invalid: no changes: files already match expected_revision" + ) + assert database.fail_project_file_change.await_args.kwargs["error_code"] == "invalid" + database.complete_project_file_change.assert_not_awaited() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error", + [ + RefUpdateError("expected head moved", status="409", reason="conflict"), + RefUpdateError("precondition failed", status="precondition_failed"), + ], +) +async def test_ref_conflict_is_a_stale_revision(monkeypatch, error): + service, database = file_service(storage_rejecting(monkeypatch, error)) + with pytest.raises(StaleProjectRevisionError) as caught: + await commit(service) + assert str(tool_error(caught.value)).startswith("conflict: canonical project state changed") + assert database.fail_project_file_change.await_args.kwargs["error_code"] == "stale_revision" + + +@pytest.mark.asyncio +async def test_other_ref_rejection_is_recorded_as_a_storage_failure(monkeypatch): + error = RefUpdateError("upstream unavailable", status="503", reason="unavailable") + service, database = file_service(storage_rejecting(monkeypatch, error)) + with pytest.raises(RuntimeError, match=r"rejected the project file commit \(unavailable\)"): + await commit(service) + assert database.fail_project_file_change.await_args.kwargs["error_code"] == "storage_failed" diff --git a/tests/test_project_memory.py b/tests/test_project_memory.py index 3a47eeb7..9c2758f9 100644 --- a/tests/test_project_memory.py +++ b/tests/test_project_memory.py @@ -80,7 +80,7 @@ async def list_memory_source_runs(self, **values) -> list[WorkflowRun]: assert values["exclude_run_id"] == self.run.id return [self.source] - async def get_effect(self, execution_key: str): + async def get_effect(self, execution_key: str, conn=None): return self.receipts.get(execution_key) async def project_memory_success(self, **values) -> None: diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 032fdf9c..ce212e9b 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -1,5 +1,6 @@ from __future__ import annotations +from datetime import timedelta from types import SimpleNamespace from unittest.mock import AsyncMock @@ -70,6 +71,7 @@ def register(*args, **kwargs): temporal_namespace="tin-lite-dev", temporal_api_key=secret, task_queue="tin-lite-checkpoint-a", + worker_graceful_shutdown_seconds=300, switchboard_public_url="https://lite.tin.computer", ) @@ -83,6 +85,9 @@ def register(*args, **kwargs): assert original["interceptors"] assert trusted["task_queue"] == settings.task_queue + "-trusted" assert trusted["max_concurrent_activities"] == 4 + # A deploy lets in-flight activities finish instead of cancelling them at once. + assert original["graceful_shutdown_timeout"] == timedelta(minutes=5) + assert trusted["graceful_shutdown_timeout"] == timedelta(minutes=5) assert not trusted.get("workflows") original_names = { activity._Definition.must_from_callable(fn).name for fn in original["activities"] diff --git a/tests/test_scan_report.py b/tests/test_scan_report.py index 84f1bfab..e35280dc 100644 --- a/tests/test_scan_report.py +++ b/tests/test_scan_report.py @@ -72,7 +72,7 @@ async def mark_run_running(self, run_id: UUID) -> None: async def list_memory_source_runs(self, **values) -> list[WorkflowRun]: raise AssertionError("ready project memory should be the only project scan source") - async def get_effect(self, execution_key: str): + async def get_effect(self, execution_key: str, conn=None): return self.receipts.get(execution_key) async def project_success(self, **values) -> None: diff --git a/tests/test_visibility_audit.py b/tests/test_visibility_audit.py index 190bd6f5..62ca0b6f 100644 --- a/tests/test_visibility_audit.py +++ b/tests/test_visibility_audit.py @@ -487,3 +487,61 @@ async def test_visibility_duplicate_execution_reuses_calls_commit_and_projection assert database.run.status == RunStatus.SUCCEEDED assert database.run.artifact_path == "reports/AI_VISIBILITY.md" assert database.events == ["visibility_audit_created", "visibility_audit_ready"] + + +@pytest.mark.asyncio +async def test_visibility_question_ids_are_normalized_and_adjudication_lines_up() -> None: + raw = panel() + long_id = "Buyer Question About Automation Tools" + for item, question_id in zip(raw["questions"], ["Q1", "Q 2", "q3", "Q3", long_id], strict=True): + item["id"] = question_id + normalized = ["q1", "q_2", "q3", "q3-2", "buyer_question_about_automation_"] + scored_raw = adjudication() + for outcome, question_id in zip(scored_raw["outcomes"], normalized, strict=True): + outcome["question_id"] = question_id + scored_raw["outcomes"][0]["question_id"] = "Q1" # the pre-normalization spelling + scored_raw["recommendations"][0]["evidence_question_ids"] = ["Q1", "Q 2", "q3-2"] + responses = FakeResponses( + response_text("resp_panel", json.dumps(raw)), + response_text("resp_score", json.dumps(scored_raw)), + ) + auditor = VisibilityAuditor(responses=responses, skill_suite="# Audit rules") + + prepared = await auditor.prepare_panel( + project_name="virvid.app", target_request="virvid.app", sources=[] + ) + panel_schema = responses.payloads[0]["text"]["format"]["schema"] + id_schema = panel_schema["properties"]["questions"]["items"]["properties"]["id"] + assert id_schema["pattern"] == "^[a-z0-9_-]{1,32}$" + assert [item["id"] for item in prepared["questions"]] == normalized + again = await VisibilityAuditor( + responses=FakeResponses(response_text("resp_panel", json.dumps(raw))), + skill_suite="# Audit rules", + ).prepare_panel(project_name="virvid.app", target_request="virvid.app", sources=[]) + assert again["panel_hash"] == prepared["panel_hash"] # receipts keyed by these IDs + + answer = response_text("resp_web", "Virvid is one product to evaluate.", searched=True) + searched = await VisibilityAuditor( + responses=FakeResponses(answer), skill_suite="# Audit rules" + ).answer(question=prepared["questions"][0]["text"], searched=True) + measurements = [ + {"question_id": item["id"], "searched": searched, "probe": searched} + for item in prepared["questions"] + ] + scored = await auditor.adjudicate(panel=prepared, measurements=measurements) + assert [item["question_id"] for item in scored["outcomes"]] == normalized + assert scored["recommendations"][0]["evidence_question_ids"] == ["q1", "q_2", "q3-2"] + + +@pytest.mark.asyncio +async def test_visibility_question_id_that_cannot_be_normalized_still_fails() -> None: + raw = panel() + raw["questions"][0]["id"] = 7 + auditor = VisibilityAuditor( + responses=FakeResponses(response_text("resp_panel", json.dumps(raw))), + skill_suite="# Audit rules", + ) + with pytest.raises(VisibilityProtocolError, match="question ID is invalid"): + await auditor.prepare_panel( + project_name="virvid.app", target_request="virvid.app", sources=[] + ) diff --git a/tests/test_worker_drain.py b/tests/test_worker_drain.py new file mode 100644 index 00000000..c4c7b326 --- /dev/null +++ b/tests/test_worker_drain.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +import asyncio +import signal + +import httpx +import pytest +import uvicorn + +from tin_lite.serving import DrainingServer +from tin_lite.worker_group import WorkerGroup + + +async def _ok_app(scope, receive, send): # type: ignore[no-untyped-def] + assert scope["type"] == "http" + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"ok"}) + + +def _server(drain, *, drain_timeout: float = 5) -> DrainingServer: # type: ignore[no-untyped-def] + config = uvicorn.Config(_ok_app, host="127.0.0.1", port=0, lifespan="off", log_level="error") + return DrainingServer(config, drain=drain, drain_timeout=drain_timeout) + + +@pytest.mark.asyncio +async def test_exit_handler_waits_for_worker_shutdown_before_http_exit(): + release = asyncio.Event() + seen: list[bool] = [] + server: DrainingServer + + async def drain() -> None: + seen.append(server.should_exit) + await release.wait() + seen.append(server.should_exit) + + server = _server(drain) + server._loop = asyncio.get_running_loop() + server.handle_exit(signal.SIGTERM, None) + assert server.should_exit is False + await asyncio.sleep(0.01) + assert server._drain_task is not None and not server._drain_task.done() + # The second SIGTERM (systemd and `uv run` both deliver one) does not cut the drain. + server.handle_exit(signal.SIGTERM, None) + await asyncio.sleep(0.01) + assert server.should_exit is False + release.set() + await server._drain_task + assert seen == [False, False] + assert server.should_exit is True + assert server.force_exit is False + + +@pytest.mark.asyncio +async def test_second_interrupt_stops_waiting_and_timeout_still_exits(): + server = _server(lambda: asyncio.Event().wait()) + server._loop = asyncio.get_running_loop() + server.handle_exit(signal.SIGINT, None) + await asyncio.sleep(0.01) + assert server.should_exit is False + server.handle_exit(signal.SIGINT, None) + assert server.should_exit is True + server._drain_task.cancel() + + timed = _server(lambda: asyncio.Event().wait(), drain_timeout=0.05) + timed._loop = asyncio.get_running_loop() + timed.handle_exit(signal.SIGTERM, None) + await asyncio.sleep(0.01) + await asyncio.wait_for(timed._drain_task, 2) + assert timed.should_exit is True + + +@pytest.mark.asyncio +async def test_http_keeps_serving_while_worker_drains(): + release = asyncio.Event() + drained = asyncio.Event() + + async def drain() -> None: + drained.set() + await release.wait() + + server = _server(drain) + serving = asyncio.create_task(server.serve()) + for _ in range(500): # uvicorn exposes no startup event + if server.started: + break + await asyncio.sleep(0.01) + port = server.servers[0].sockets[0].getsockname()[1] + server.handle_exit(signal.SIGTERM, None) + # Signals delivered for real are re-raised after serve(); this one was simulated. + server._captured_signals.clear() + await asyncio.wait_for(drained.wait(), 2) + async with httpx.AsyncClient() as client: + response = await client.get(f"http://127.0.0.1:{port}/") + assert response.status_code == 200 and response.text == "ok" + assert not serving.done() + release.set() + await asyncio.wait_for(serving, 5) + + +class _FakeWorker: + def __init__(self, *, fail_on_start: bool = False) -> None: + self.fail_on_start = fail_on_start + self.stop = asyncio.Event() + self.shutdowns = 0 + + async def run(self) -> None: + if self.fail_on_start: + raise RuntimeError("namespace check failed") + await self.stop.wait() + await asyncio.sleep(0.02) # activities finishing + + async def shutdown(self) -> None: + self.shutdowns += 1 + self.stop.set() + if self.fail_on_start: + await asyncio.Event().wait() # a never-started Temporal worker never completes + + +@pytest.mark.asyncio +async def test_worker_group_shutdown_is_idempotent_for_drain_then_lifespan(): + worker = _FakeWorker() + group = WorkerGroup(worker) # type: ignore[arg-type] + running = asyncio.create_task(group.run()) + await asyncio.sleep(0) + await asyncio.gather(group.shutdown(), group.shutdown()) + assert group.is_shutdown + await group.shutdown() # the lifespan's later call returns at once + await asyncio.wait_for(running, 1) + assert worker.shutdowns == 1 + + +@pytest.mark.asyncio +async def test_worker_group_shutdown_does_not_hang_on_unstarted_workers(): + await asyncio.wait_for(WorkerGroup(_FakeWorker()).shutdown(), 1) # type: ignore[arg-type] + healthy, broken = _FakeWorker(), _FakeWorker(fail_on_start=True) + with pytest.raises(RuntimeError, match="namespace check failed"): + await asyncio.wait_for(WorkerGroup(healthy, broken).run(), 1) # type: ignore[arg-type] + assert healthy.shutdowns == 1 diff --git a/workflow_packages/brand.capture/skills/brand-capture/CONTRACT.md b/workflow_packages/brand.capture/skills/brand-capture/CONTRACT.md index 15ea6c36..2a69e19d 100644 --- a/workflow_packages/brand.capture/skills/brand-capture/CONTRACT.md +++ b/workflow_packages/brand.capture/skills/brand-capture/CONTRACT.md @@ -36,8 +36,12 @@ example, not a diagnosis or a default identity: Allowed identity classes: distinctive_strong, distinctive_inconsistent, competent_generic, weak, unknown. Evidence status: adequate, partial, insufficient, conflicting. One to five findings, each with observation, source_ids and generation_implication. The block is at most -8,000 bytes. Do not add probabilities or other fields. A packet-only capture must say it used -supplied observations; method describes the rubric, not proof of independent browser coverage. +8,000 bytes. Do not add probabilities or other fields. `method` is a fixed literal: always +exactly `{"kind": "visual_capture", "rubric_version": "marketing-brand.v1"}`, whatever you +observed. It names the rubric, not your coverage. Express a packet-only or otherwise limited +capture through `evidence_status` (usually `partial` or `insufficient`) and in the findings +and the Assessment and sources prose, for example "supplied observations only; no browser +capture". Declare every source ID with a Markdown reference definition, for example: `[site-home]: https://example.com/ "Homepage, observed at actual capture time"`. diff --git a/workflow_packages/brand.capture/workflow.json b/workflow_packages/brand.capture/workflow.json index 1de2a579..c3d592f0 100644 --- a/workflow_packages/brand.capture/workflow.json +++ b/workflow_packages/brand.capture/workflow.json @@ -4,7 +4,7 @@ "key": "brand.capture", "title": "Capture brand and design", "description": "Inspect your website and optional repository once. Propose a concise brand guide for future marketing and document the product design you have today. Review both before using them; existing documents stay unchanged.", - "version": "1.0.1", + "version": "1.0.2", "executor": "codex.procedure", "system": "creative-studio", "schedule_modes": ["on_demand"],