Hands-on lab — Day 98: Section Project: A Complete Data Pipeline

"""Stage 5, part one: configuration resolved by precedence, with provenance.

Day 97's rule, and the Twelve-Factor App's third factor before it: configuration
lives outside the code, and the process reads it from the environment. This
module implements that with four layers, lowest first:

    1. defaults        the values baked into this file
    2. file            a TOML file named by --config-file
    3. environment     PIPELINE_<KEY>, uppercased
    4. command line    an explicit flag

The part people skip is the fourth column. Knowing that ``timeout_seconds`` is
5.0 is half an answer at 3 a.m.; knowing it is 5.0 *because nobody set it* is
the whole answer. So every resolved setting carries where it came from, and
``provenance_table`` prints it.

Secrets are marked, and a marked secret is never printed and never logged.
"""

from __future__ import annotations

import os
import tomllib
from dataclasses import dataclass
from pathlib import Path

REDACTED = "***redacted***"

#: key -> (default value, parser, is_secret)
SPEC: dict[str, tuple[object, str, bool]] = {
    "base_url": ("http://127.0.0.1:8080", "str", False),
    "sources": ("alpha,bravo,charlie", "str", False),
    "database_url": ("sqlite:///pipeline.db", "str", False),
    "api_token": ("", "str", True),
    "timeout_seconds": (5.0, "float", False),
    "retry_attempts": (3, "int", False),
    "retry_backoff_seconds": (0.05, "float", False),
    "window_hours": (12, "int", False),
    "report_at": ("", "str", False),
    "log_level": ("info", "str", False),
}


@dataclass(frozen=True)
class Setting:
    """One resolved setting and the layer that won it."""

    key: str
    value: object
    source: str
    secret: bool

    @property
    def display(self) -> str:
        if self.secret and self.value:
            return REDACTED
        if self.value == "":
            # An empty string is a real answer ("nobody set it"), and printing
            # nothing at all is how a provenance table stops being useful.
            return "<unset>"
        return str(self.value)


def _coerce(raw: object, kind: str) -> object:
    if kind == "int":
        return int(raw)
    if kind == "float":
        return float(raw)
    return str(raw)


@dataclass(frozen=True)
class Config:
    settings: dict[str, Setting]

    def __getitem__(self, key: str) -> object:
        return self.settings[key].value

    def source_of(self, key: str) -> str:
        return self.settings[key].source

    @property
    def source_names(self) -> list[str]:
        raw = str(self["sources"])
        return [name.strip() for name in raw.split(",") if name.strip()]

    @property
    def secrets(self) -> tuple[str, ...]:
        return tuple(
            str(setting.value)
            for setting in self.settings.values()
            if setting.secret and setting.value
        )

    def provenance_table(self) -> str:
        """The printable answer to 'why is it set to that?'."""
        width_key = max(len(k) for k in self.settings)
        width_value = max(len(s.display) for s in self.settings.values())
        width_value = max(width_value, len("value"))
        lines = [
            f"{'setting'.ljust(width_key)}  {'value'.ljust(width_value)}  source",
            f"{'-' * width_key}  {'-' * width_value}  {'-' * 12}",
        ]
        for key in sorted(self.settings):
            setting = self.settings[key]
            lines.append(
                f"{key.ljust(width_key)}  {setting.display.ljust(width_value)}  {setting.source}"
            )
        return "\n".join(lines)


def load_config(
    *,
    config_file: str | Path | None = None,
    environ: dict[str, str] | None = None,
    overrides: dict[str, object] | None = None,
) -> Config:
    """Resolve every setting through the four layers and record which one won.

    ``overrides`` is the command-line layer: pass only the flags the user
    actually gave, because a flag left at its argparse default is not a
    decision and must not outrank the environment.
    """
    environ = os.environ if environ is None else environ
    overrides = overrides or {}

    from_file: dict[str, object] = {}
    if config_file:
        path = Path(config_file)
        if not path.is_file():
            raise FileNotFoundError(f"config file not found: {path}")
        from_file = tomllib.loads(path.read_text(encoding="utf-8"))

    resolved: dict[str, Setting] = {}
    for key, (default, kind, secret) in SPEC.items():
        value: object = default
        source = "default"
        if key in from_file:
            value, source = _coerce(from_file[key], kind), "file"
        env_key = f"PIPELINE_{key.upper()}"
        if environ.get(env_key):
            value, source = _coerce(environ[env_key], kind), "environment"
        if key in overrides and overrides[key] is not None:
            value, source = _coerce(overrides[key], kind), "command line"
        resolved[key] = Setting(key=key, value=value, source=source, secret=secret)

    return Config(settings=resolved)

#!/usr/bin/env python3
"""Run the whole pipeline twice against the local fixture server and show it.

Everything here is offline. The fixture server runs on a thread inside this
process, bound to 127.0.0.1 on a port the kernel picks, and the port is masked
in the output as ``<port>`` so the capture is byte-stable across machines.

Two runs, one database. The point of the second run is the whole day: it fetches
the same nine records, validates the same seven, and stores **none** of them,
because the idempotence key already holds them. The report is identical. The
exit code is identical. Nothing had to be cleaned up first.

    .venv/bin/python examples/demo_run.py
"""

from __future__ import annotations

import io
import json
import os
import sys
import tempfile
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

import fixture_server  # noqa: E402
from config import load_config  # noqa: E402
from ingest import fetch_source  # noqa: E402
from logs import RunLogger, fixed_clock, redact  # noqa: E402
from pipeline import run_pipeline  # noqa: E402

TOKEN = "demo-token-value"
REPORT_AT = "2026-08-16T12:00:00Z"


def rule(title: str) -> None:
    print()
    print(title)
    print("-" * len(title))


def mask(text: str, port: int) -> str:
    return text.replace(f"127.0.0.1:{port}", "127.0.0.1:<port>")


def main() -> int:
    server, port = fixture_server.start_background_server(token=TOKEN)
    base_url = f"http://127.0.0.1:{port}"
    workdir = Path(tempfile.mkdtemp(prefix="day098-"))
    # Work inside the temporary directory so database_url can stay at its
    # default, relative value. Nothing this script prints is then a path that
    # exists only on the machine that ran it.
    origin = Path.cwd()
    os.chdir(workdir)

    environ = {
        "PIPELINE_BASE_URL": base_url,
        "PIPELINE_API_TOKEN": TOKEN,
        "PIPELINE_RETRY_BACKOFF_SECONDS": "0.01",
    }
    overrides = {"report_at": REPORT_AT, "window_hours": 12, "sources": "alpha,bravo,charlie"}
    config = load_config(environ=environ, overrides=overrides)
    # The same resolution with the port replaced by a fixed marker, purely so the
    # printed table is identical on every machine. Only base_url differs.
    display_config = load_config(
        environ={**environ, "PIPELINE_BASE_URL": "http://127.0.0.1:<port>"},
        overrides=overrides,
    )

    print("Day 098 — the whole pipeline, twice")
    print("=" * 36)
    print("fixture server : http://127.0.0.1:<port> (the kernel chose the port)")
    print("database       : a fresh file in a temporary directory")
    print(f"report instant : {REPORT_AT} (a parameter, not the clock)")

    rule("1. Configuration, and where every value came from")
    print(display_config.provenance_table())
    print()
    print("api_token is set and is never printed. That is the point of marking it.")

    outcomes = []
    for number, run_id in enumerate(("run-000000000001", "run-000000000002"), start=1):
        rule(f"{number + 1}. Run {number} — run id {run_id}")
        log_stream = io.StringIO()
        out_stream = io.StringIO()
        logger = RunLogger(
            run_id,
            stream=log_stream,
            clock=fixed_clock(),
            level=str(config["log_level"]),
            secrets=config.secrets,
        )
        outcome = run_pipeline(
            config,
            run_id=run_id,
            logger=logger,
            out=out_stream,
            sleep=lambda _seconds: None,
            started_at=REPORT_AT,
        )
        outcomes.append(outcome)

        print("structured log (stderr), one JSON object per line:")
        for line in log_stream.getvalue().splitlines():
            print(("  " + mask(line, port)).rstrip())
        print()
        print("report (stdout):")
        for line in out_stream.getvalue().splitlines():
            print(("  " + line).rstrip())
        print()
        print(f"exit code: {outcome.exit_code}  ({outcome.status})")

    rule("4. What the second run proves")
    first, second = outcomes
    print(f"  records fetched      run 1: {first.fetched:>2}    run 2: {second.fetched:>2}")
    print(f"  records accepted     run 1: {first.accepted:>2}    run 2: {second.accepted:>2}")
    print(f"  records rejected     run 1: {first.rejected:>2}    run 2: {second.rejected:>2}")
    print(f"  rows inserted        run 1: {first.inserted:>2}    run 2: {second.inserted:>2}")
    print(f"  duplicates skipped   run 1: {first.duplicates:>2}    run 2: {second.duplicates:>2}")
    print(f"  rows in the store    run 1: {first.total_rows:>2}    run 2: {second.total_rows:>2}")
    print(f"  reports identical    {first.report_text == second.report_text}")
    print(f"  exit codes identical {first.exit_code == second.exit_code}")
    print()
    print("  The pipeline was run twice and the data was stored once. Every failure")
    print("  in this design now has the same remedy: run it again.")

    rule("5. Retry only what is worth retrying")
    fixture_server.reset_flaky_counter()
    for source, note in (("bravo", "500 twice, then 200"), ("delta", "404, and it will stay 404")):
        result = fetch_source(
            base_url,
            source,
            token=TOKEN,
            timeout=2.0,
            attempts=3,
            backoff=0.0,
            sleep=lambda _seconds: None,
        )
        print(
            f"  {source:<8} attempts={result.attempts}  ok={result.ok}  "
            f"status={result.status}  ({note})"
        )
    print()
    print("  bravo was worth three attempts. delta was worth one. The difference is")
    print("  whether the status code describes a moment or a mistake.")

    rule("6. The secret, and the upstream that echoed it back")
    leaky = fetch_source(
        base_url, "charlie", token=TOKEN, timeout=2.0, attempts=1, backoff=0.0
    )
    print(f"  raw error body from charlie : {leaky.error}")
    print(f"  after the log redactor      : {redact(leaky.error, config.secrets)}")
    print()
    print("  Nobody wrote code to log the token. The upstream service put it in an")
    print("  error message, and the error message went to the log. Redaction has to")
    print("  live in the logger, where it cannot be forgotten.")

    rule("7. Every log line carries the run id")
    log_stream = io.StringIO()
    logger = RunLogger("run-000000000003", stream=log_stream, clock=fixed_clock(), secrets=config.secrets)
    run_pipeline(
        config,
        run_id="run-000000000003",
        logger=logger,
        out=io.StringIO(),
        sleep=lambda _seconds: None,
        started_at=REPORT_AT,
    )
    records = [json.loads(line) for line in log_stream.getvalue().splitlines()]
    stages = [record["event"] for record in records if record["event"].startswith("stage.")]
    print(f"  log lines          : {len(records)}")
    print(f"  stage summaries    : {len(stages)} -> {', '.join(stages)}")
    print(f"  distinct run ids   : {sorted({record['run_id'] for record in records})}")

    server.shutdown()
    server.server_close()
    os.chdir(origin)
    for path in sorted(workdir.iterdir()):
        path.unlink()
    workdir.rmdir()
    print()
    print(f"temporary database removed: {not workdir.exists()}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""A local fixture server, so the whole pipeline runs with no internet at all.

It binds 127.0.0.1 on port **0**, which asks the operating system for any free
port, then prints the port it was given on the first line of stdout. Hard-coding
a port is how a test suite collides with whatever the learner already has
running, and the handful of ports tutorials reach for are taken on most
developer machines by lunchtime.

Every route exists to make one pipeline promise testable:

    /stations/alpha/readings    200, five records, two of them deliberately bad
                                and one of them a duplicate of another
    /stations/bravo/readings    500, 500, then 200 — the source that proves a
                                retry actually recovers. The counter is per
                                server process, so the SECOND pipeline run gets
                                a 200 on its first attempt, which is what a
                                transient failure looks like in real life.
    /stations/charlie/readings  500 every single time — the source that must be
                                skipped and reported while the others succeed.
                                Its error body echoes the supplied token back,
                                which is a real and common upstream leak and is
                                why the log redactor in logs.py is not
                                decorative.
    /stations/delta/readings    404 — a source name that is simply wrong.
                                Retrying it would waste three round trips to
                                learn the same thing.
    /health                     200 "ok", used only for the readiness loop.

Every request requires ``Authorization: Bearer <token>`` when a token was given
on the command line, and answers 401 otherwise.

Run it by hand if you like:

    .venv/bin/python examples/fixture_server.py --token demo-token-value

Stop it with Ctrl-C. It serves only 127.0.0.1 and exits when its parent dies.
"""

from __future__ import annotations

import argparse
import json
import sys
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer, ThreadingHTTPServer

#: The two well-behaved payloads. Every value here is invented.
PAYLOADS: dict[str, dict[str, object]] = {
    "alpha": {
        "station_id": "alpha",
        "records": [
            {
                "station_id": "alpha",
                "reading_id": "a-1",
                "observed_at": "2026-08-16T09:00:00Z",
                "temperature_c": 18.4,
                "humidity_pct": 61,
            },
            {
                "station_id": "alpha",
                "reading_id": "a-2",
                "observed_at": "2026-08-16T10:00:00Z",
                "temperature_c": 19.0,
                "humidity_pct": 58,
            },
            {
                # Malformed: the temperature arrived as prose.
                "station_id": "alpha",
                "reading_id": "a-3",
                "observed_at": "2026-08-16T11:00:00Z",
                "temperature_c": "warm",
                "humidity_pct": 57,
            },
            {
                # Byte-for-byte duplicate of a-2. Valid, so the validation gate
                # passes it; the store's idempotence key is what drops it.
                "station_id": "alpha",
                "reading_id": "a-2",
                "observed_at": "2026-08-16T10:00:00Z",
                "temperature_c": 19.0,
                "humidity_pct": 58,
            },
            {
                # Out of range: humidity is a percentage.
                "station_id": "alpha",
                "reading_id": "a-5",
                "observed_at": "2026-08-16T12:00:00Z",
                "temperature_c": 21.3,
                "humidity_pct": 155,
            },
        ],
    },
    "bravo": {
        "station_id": "bravo",
        "records": [
            {
                "station_id": "bravo",
                "reading_id": "b-1",
                "observed_at": "2026-08-15T23:30:00Z",
                "temperature_c": 12.2,
                "humidity_pct": 80,
            },
            {
                "station_id": "bravo",
                "reading_id": "b-2",
                "observed_at": "2026-08-16T08:15:00Z",
                "temperature_c": 13.6,
                "humidity_pct": 77,
            },
            {
                "station_id": "bravo",
                "reading_id": "b-3",
                "observed_at": "2026-08-16T11:45:00Z",
                "temperature_c": 15.0,
                "humidity_pct": 74,
            },
            {
                # Valid but wrong: 41.3 C is inside every range the gate
                # checks, and it is a 26.3 C jump in five minutes. No
                # field-level rule can catch this one.
                "station_id": "bravo",
                "reading_id": "b-4",
                "observed_at": "2026-08-16T11:50:00Z",
                "temperature_c": 41.3,
                "humidity_pct": 74,
            },
        ],
    },
}

_bravo_hits = 0
_lock = threading.Lock()


def reset_flaky_counter() -> None:
    """Put bravo back to 'fails the next two attempts'. Used by the tests."""
    global _bravo_hits
    with _lock:
        _bravo_hits = 0


class FixtureHandler(BaseHTTPRequestHandler):
    server_version = "pipeline-fixture/1.0"
    token = ""

    def log_message(self, fmt: str, *args: object) -> None:
        """Silence the default per-request line; the pipeline has its own log."""

    def _send(self, status: int, body: bytes, content_type: str = "application/json") -> None:
        self.send_response(status)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _json(self, status: int, payload: object) -> None:
        self._send(status, json.dumps(payload).encode("utf-8"))

    def do_GET(self) -> None:  # noqa: N802 - the name is fixed by http.server
        global _bravo_hits

        if self.path == "/health":
            self._send(200, b"ok", "text/plain")
            return

        if self.token:
            supplied = self.headers.get("Authorization", "")
            if supplied != f"Bearer {self.token}":
                self._json(401, {"error": "missing or wrong Authorization header"})
                return

        if not self.path.startswith("/stations/") or not self.path.endswith("/readings"):
            self._json(404, {"error": "not found"})
            return

        name = self.path[len("/stations/") : -len("/readings")]

        if name == "bravo":
            with _lock:
                _bravo_hits += 1
                hits = _bravo_hits
            if hits <= 2:
                self._json(500, {"error": f"station bravo is warming up (attempt {hits})"})
                return
            self._json(200, PAYLOADS["bravo"])
            return

        if name == "charlie":
            # A real upstream leak pattern: the error body echoes the secret.
            self._json(
                500,
                {"error": f"upstream credentials rejected for token {self.token or 'none'}"},
            )
            return

        if name in PAYLOADS:
            self._json(200, PAYLOADS[name])
            return

        self._json(404, {"error": f"no such station: {name}"})


def start_background_server(token: str = "") -> tuple[ThreadingHTTPServer, int]:
    """Start the fixture server on a thread and return it with its port.

    Used by demo_run.py and by the test harness so no second process, and no
    fixed port, is ever needed.
    """
    FixtureHandler.token = token
    server = ThreadingHTTPServer(("127.0.0.1", 0), FixtureHandler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server, server.server_address[1]


def main() -> int:
    parser = argparse.ArgumentParser(description="Local fixture server for the Day 98 lab.")
    parser.add_argument("--token", default="", help="require this bearer token on every request")
    args = parser.parse_args()

    FixtureHandler.token = args.token
    server = HTTPServer(("127.0.0.1", 0), FixtureHandler)
    print(server.server_address[1], flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()
    return 0


if __name__ == "__main__":
    sys.exit(main())

"""Stage 1 — Ingest.

**The promise:** every fetch has a deadline, only failures worth retrying are
retried, and a source that never answers does not take the run down with it.

Three decisions, all from Day 78 and Day 84.

**A timeout on every call.** A request with no timeout has no upper bound on how
long your 3 a.m. run takes. ``urllib.request.urlopen`` accepts ``timeout``; the
default is whatever the socket module's default is, which is usually None,
which is forever.

**Retry only what is worth retrying.** A 500 or a 503 means the server had a bad
moment and might not next time. A 404 means the URL is wrong, and it will be
just as wrong in two seconds — retrying it costs three round trips to learn the
same thing and delays every other source. 429 is retryable because it is the
server asking you to slow down, which is exactly what backoff does.

**Be honest about partial success.** ``fetch_all`` never raises. It returns one
``FetchResult`` per source, some with records and some with an error, and the
caller decides what that means. A pipeline that dies because one of five sources
is down has thrown away four sources' worth of good data.

Why ``urllib.request`` and not ``requests`` (Day 78): this pipeline makes one
GET with a timeout and a header. The standard library does that. Every
dependency is a version to pin, a CVE to track and a thing that can break your
scheduled job at 3 a.m., so the bar for adding one is "it earns its place".
``requests`` earns it the moment you need sessions, connection pooling, or
retries with the sophistication of ``urllib3``'s ``Retry``.
"""

from __future__ import annotations

import json
import time
import urllib.error
import urllib.request
from collections.abc import Callable
from dataclasses import dataclass, field

#: Status codes worth a second attempt. Everything else is a decision, not luck.
RETRYABLE_STATUS = frozenset({429, 500, 502, 503, 504})


@dataclass(frozen=True)
class FetchResult:
    """What one source gave us, and what it cost to find out."""

    source: str
    ok: bool
    records: list[dict] = field(default_factory=list)
    attempts: int = 0
    status: int | None = None
    error: str = ""
    retried: bool = False


def _get_json(url: str, *, token: str, timeout: float) -> tuple[int, object]:
    request = urllib.request.Request(url, method="GET")
    request.add_header("Accept", "application/json")
    if token:
        request.add_header("Authorization", f"Bearer {token}")
    with urllib.request.urlopen(request, timeout=timeout) as response:  # noqa: S310
        return response.status, json.loads(response.read().decode("utf-8"))


def fetch_source(
    base_url: str,
    source: str,
    *,
    token: str = "",
    timeout: float = 5.0,
    attempts: int = 3,
    backoff: float = 0.05,
    sleep: Callable[[float], None] = time.sleep,
) -> FetchResult:
    """Fetch one source, retrying only what is worth retrying.

    ``sleep`` is injected so tests do not have to wait for real backoff.
    """
    url = f"{base_url.rstrip('/')}/stations/{source}/readings"
    tried = 0
    last_status: int | None = None
    last_error = ""

    while tried < attempts:
        tried += 1
        try:
            status, payload = _get_json(url, token=token, timeout=timeout)
        except urllib.error.HTTPError as exc:
            # HTTPError is a *file object*. Not closing it leaks a socket, which
            # in a process that runs once an hour forever is a slow resource
            # exhaustion bug that nothing warns you about in production.
            with exc:
                last_status = exc.code
                body = exc.read().decode("utf-8", errors="replace")
            try:
                last_error = str(json.loads(body).get("error", body))
            except (json.JSONDecodeError, AttributeError):
                last_error = body
            if exc.code not in RETRYABLE_STATUS:
                break
        except (urllib.error.URLError, TimeoutError, OSError) as exc:
            last_status = None
            last_error = f"{type(exc).__name__}: {exc}"
        else:
            records = payload.get("records", []) if isinstance(payload, dict) else []
            return FetchResult(
                source=source,
                ok=True,
                records=list(records),
                attempts=tried,
                status=status,
                retried=tried > 1,
            )
        if tried < attempts:
            sleep(backoff * (2 ** (tried - 1)))

    return FetchResult(
        source=source,
        ok=False,
        attempts=tried,
        status=last_status,
        error=last_error,
        retried=tried > 1,
    )


def fetch_all(
    base_url: str,
    sources: list[str],
    *,
    token: str = "",
    timeout: float = 5.0,
    attempts: int = 3,
    backoff: float = 0.05,
    sleep: Callable[[float], None] = time.sleep,
) -> list[FetchResult]:
    """Fetch every source in order. Never raises; a failure is a result."""
    return [
        fetch_source(
            base_url,
            source,
            token=token,
            timeout=timeout,
            attempts=attempts,
            backoff=backoff,
            sleep=sleep,
        )
        for source in sources
    ]

"""Stage 5, part two: one structured log line per stage, with a run id.

Two decisions carry this module, and both come from Day 97.

**Structured, not prose.** ``logging`` in the standard library formats records
through a Formatter; this one emits a single JSON object per line. The reason is
not fashion. A prose line has to be parsed with a regular expression that breaks
the first time somebody adds a word; a JSON line is queryable on the day you
need it, which is always the day something is on fire.

**Redaction is a filter, not a discipline.** Never logging a secret by hand
works right up until an upstream service echoes it back inside an error body —
which the fixture server does on purpose, because real ones do. So the logger
scans every string it is about to emit for every known secret and replaces it.
The check belongs in one place that cannot be forgotten.

The clock is injected. The pipeline passes the real one; the demo and the tests
pass a fixed one, which is what makes the captured logs comparable byte for
byte. A module that reads the clock itself cannot be tested on its output.
"""

from __future__ import annotations

import json
import sys
from collections.abc import Callable
from datetime import datetime, timezone
from typing import Any, TextIO

LEVELS = {"debug": 10, "info": 20, "warning": 30, "error": 40}
REDACTED = "***redacted***"


def utc_clock() -> str:
    """The real clock, as ISO 8601 in UTC with a Z suffix."""
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def fixed_clock(start: str = "2026-08-16T12:00:00Z", step_seconds: int = 1) -> Callable[[], str]:
    """A clock that starts at ``start`` and advances one step per call.

    Deterministic, so a run's log can be compared against a stored capture.
    """
    moment = datetime.strptime(start, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc)
    state = {"n": 0}

    def tick() -> str:
        from datetime import timedelta

        value = moment + timedelta(seconds=step_seconds * state["n"])
        state["n"] += 1
        return value.strftime("%Y-%m-%dT%H:%M:%SZ")

    return tick


def redact(value: Any, secrets: tuple[str, ...]) -> Any:
    """Replace every occurrence of every secret, at any depth."""
    if not secrets:
        return value
    if isinstance(value, str):
        for secret in secrets:
            if secret and secret in value:
                value = value.replace(secret, REDACTED)
        return value
    if isinstance(value, dict):
        return {key: redact(item, secrets) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [redact(item, secrets) for item in value]
    return value


class RunLogger:
    """Emits one JSON object per line, every line carrying the same run id."""

    def __init__(
        self,
        run_id: str,
        *,
        stream: TextIO | None = None,
        clock: Callable[[], str] = utc_clock,
        level: str = "info",
        secrets: tuple[str, ...] = (),
    ) -> None:
        self.run_id = run_id
        self.stream = stream if stream is not None else sys.stderr
        self.clock = clock
        self.threshold = LEVELS[level]
        self.secrets = tuple(s for s in secrets if s)
        self.emitted: list[dict[str, Any]] = []

    def event(self, event: str, level: str = "info", **fields: Any) -> dict[str, Any] | None:
        if LEVELS[level] < self.threshold:
            return None
        record: dict[str, Any] = {
            "ts": self.clock(),
            "level": level,
            "run_id": self.run_id,
            "event": event,
        }
        record.update(redact(fields, self.secrets))
        self.emitted.append(record)
        self.stream.write(json.dumps(record) + "\n")
        self.stream.flush()
        return record

    def events_named(self, event: str) -> list[dict[str, Any]]:
        return [record for record in self.emitted if record["event"] == event]

"""The pipeline: five stages, one run id, and an exit code a scheduler can act on.

This is the only module that knows about all five stages, and it is deliberately
thin. Each stage has already made its own promise; this file's job is to thread
one run id through them, decide what the combination of outcomes *means*, and
say so in a way a scheduler can act on without reading English.

**The exit code is the interface to the scheduler** (Day 81). ``cron`` and
``launchd`` and every CI runner ever written know exactly one thing about your
program: the number it returned. So:

    0   success        every source answered, every record was accepted
    3   partial        the run completed and stored what it could, but at least
                       one source failed permanently or at least one record was
                       rejected. Somebody should look; nothing is on fire.
    1   failure        the run could not do its job at all — no source answered,
                       or the store refused. Page someone.

Collapsing 3 into 0 is the mistake that lets a source go dark for a month.
Collapsing 3 into 1 is the mistake that trains everyone to ignore the alert.

**Where concurrency belongs** (Day 96). Ingest is waiting work: three sources
fetched one after another spend almost all their wall-clock time blocked on a
socket, and running them concurrently is close to free. Validation is not — at
this size it is microseconds of CPU, and a thread pool would cost more in
coordination than it saves. The store is a single SQLite writer by design; two
concurrent writers buy contention, not speed. This module keeps the fetch
sequential because the lab has three sources and determinism is worth more than
milliseconds here, and says so plainly rather than pretending the choice was
forced.
"""

from __future__ import annotations

import argparse
import sys
import uuid
from collections.abc import Callable
from dataclasses import dataclass
from typing import TextIO

from sqlalchemy.orm import Session

import report as report_module
from config import Config, load_config
from ingest import fetch_all
from logs import RunLogger, fixed_clock, utc_clock
from store import build_engine, record_run, store_readings, utc_text
from validate import validate_all

EXIT_SUCCESS = 0
EXIT_FAILURE = 1
EXIT_PARTIAL = 3


@dataclass
class RunOutcome:
    run_id: str
    status: str
    exit_code: int
    fetched: int
    accepted: int
    rejected: int
    inserted: int
    duplicates: int
    total_rows: int
    failed_sources: tuple[str, ...]
    report_text: str


def run_pipeline(
    config: Config,
    *,
    run_id: str,
    logger: RunLogger,
    out: TextIO,
    sleep: Callable[[float], None] | None = None,
    started_at: str | None = None,
) -> RunOutcome:
    """Run all five stages once. Returns the outcome; never raises for data."""
    sources = config.source_names
    started_at = started_at or utc_text()
    logger.event(
        "run.start",
        sources=sources,
        window_hours=config["window_hours"],
        report_at=str(config["report_at"]) or "<now>",
        database_url=str(config["database_url"]),
    )

    # ---- Stage 1: ingest -------------------------------------------------
    kwargs = {} if sleep is None else {"sleep": sleep}
    results = fetch_all(
        str(config["base_url"]),
        sources,
        token=str(config["api_token"]),
        timeout=float(config["timeout_seconds"]),  # type: ignore[arg-type]
        attempts=int(config["retry_attempts"]),  # type: ignore[arg-type]
        backoff=float(config["retry_backoff_seconds"]),  # type: ignore[arg-type]
        **kwargs,
    )
    for result in results:
        if not result.ok:
            logger.event(
                "ingest.source_failed",
                level="warning",
                source=result.source,
                attempts=result.attempts,
                status=result.status,
                error=result.error,
            )
        elif result.retried:
            logger.event(
                "ingest.source_recovered",
                source=result.source,
                attempts=result.attempts,
            )

    fetched = {result.source: result.records for result in results if result.ok}
    failed = tuple(result.source for result in results if not result.ok)
    records_fetched = sum(len(records) for records in fetched.values())
    logger.event(
        "stage.ingest",
        sources_ok=len(fetched),
        sources_failed=len(failed),
        failed_sources=list(failed),
        records_fetched=records_fetched,
        attempts_total=sum(result.attempts for result in results),
    )

    if not fetched:
        logger.event("run.end", level="error", status="failure", exit_code=EXIT_FAILURE)
        return RunOutcome(
            run_id=run_id,
            status="failure",
            exit_code=EXIT_FAILURE,
            fetched=0,
            accepted=0,
            rejected=0,
            inserted=0,
            duplicates=0,
            total_rows=0,
            failed_sources=failed,
            report_text="",
        )

    # ---- Stage 2: validate ----------------------------------------------
    outcome = validate_all(fetched)
    for rejection in outcome.rejected:
        logger.event(
            "validate.rejected",
            level="warning",
            source=rejection.source,
            index=rejection.index,
            reading_id=rejection.reading_id,
            problems=list(rejection.problems),
        )
    logger.event(
        "stage.validate",
        records_in=outcome.considered,
        accepted=len(outcome.accepted),
        rejected=len(outcome.rejected),
        reasons=outcome.reasons(),
    )

    # ---- Stage 3: store --------------------------------------------------
    engine = build_engine(str(config["database_url"]))
    with Session(engine) as session:
        stored = store_readings(session, outcome.accepted, run_id=run_id)
        logger.event(
            "stage.store",
            considered=stored.considered,
            inserted=stored.inserted,
            duplicates_skipped=stored.duplicates,
            total_rows=stored.total_rows,
        )

        # ---- Stage 4: report ---------------------------------------------
        report_at = str(config["report_at"]) or utc_text()
        built = report_module.build_report(
            session,
            report_at=report_at,
            window_hours=int(config["window_hours"]),  # type: ignore[arg-type]
            stations=sources,
        )
        report_text = report_module.format_report(built)
        out.write(report_text + "\n")
        logger.event(
            "stage.report",
            report_at=built.report_at,
            window_start=built.window_start,
            readings_in_window=built.readings_in_window,
            stations=len(built.stations),
            suspect_readings=len(built.suspect),
        )

        # ---- Stage 5: observe --------------------------------------------
        if failed or outcome.rejected:
            status, code = "partial_success", EXIT_PARTIAL
        else:
            status, code = "success", EXIT_SUCCESS

        logger.event(
            "stage.observe",
            status=status,
            exit_code=code,
            run_row=run_id,
            log_lines_so_far=len(logger.emitted) + 1,
        )
        record_run(
            session,
            run_id=run_id,
            started_at=started_at,
            status=status,
            fetched=records_fetched,
            accepted=len(outcome.accepted),
            rejected=len(outcome.rejected),
            inserted=stored.inserted,
            duplicates=stored.duplicates,
        )

    engine.dispose()
    logger.event(
        "run.end",
        level="warning" if code == EXIT_PARTIAL else "info",
        status=status,
        exit_code=code,
        stored_total=stored.total_rows,
    )
    return RunOutcome(
        run_id=run_id,
        status=status,
        exit_code=code,
        fetched=records_fetched,
        accepted=len(outcome.accepted),
        rejected=len(outcome.rejected),
        inserted=stored.inserted,
        duplicates=stored.duplicates,
        total_rows=stored.total_rows,
        failed_sources=failed,
        report_text=report_text,
    )


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="pipeline",
        description="Fetch station readings, validate them, store them once, and report.",
    )
    parser.add_argument("--config-file", default=None, help="TOML file, layer 2 of 4")
    parser.add_argument("--base-url", default=None, help="where the station API lives")
    parser.add_argument("--sources", default=None, help="comma-separated station names")
    parser.add_argument("--database-url", default=None, help="SQLAlchemy URL")
    parser.add_argument("--window-hours", default=None, help="report window, in hours")
    parser.add_argument("--report-at", default=None, help="report instant, ISO 8601 with offset")
    parser.add_argument("--timeout-seconds", default=None, help="per-request deadline")
    parser.add_argument("--retry-attempts", default=None, help="attempts per source")
    parser.add_argument(
        "--log-level", default=None, choices=["debug", "info", "warning", "error"]
    )
    parser.add_argument("--run-id", default=None, help="use this run id instead of a fresh one")
    parser.add_argument(
        "--explain-config",
        action="store_true",
        help="print every setting, its value and which layer set it, then exit 0",
    )
    parser.add_argument(
        "--fixed-clock",
        action="store_true",
        help="log timestamps from a fixed clock, so output is byte-comparable",
    )
    return parser


def main(argv: list[str] | None = None, *, out: TextIO | None = None, err: TextIO | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    out = out if out is not None else sys.stdout
    err = err if err is not None else sys.stderr

    overrides = {
        "base_url": args.base_url,
        "sources": args.sources,
        "database_url": args.database_url,
        "window_hours": args.window_hours,
        "report_at": args.report_at,
        "timeout_seconds": args.timeout_seconds,
        "retry_attempts": args.retry_attempts,
        "log_level": args.log_level,
    }
    try:
        config = load_config(config_file=args.config_file, overrides=overrides)
    except FileNotFoundError as exc:
        err.write(f"configuration error: {exc}\n")
        return EXIT_FAILURE

    if args.explain_config:
        out.write(config.provenance_table() + "\n")
        return EXIT_SUCCESS

    run_id = args.run_id or uuid.uuid4().hex[:12]
    logger = RunLogger(
        run_id,
        stream=err,
        clock=fixed_clock() if args.fixed_clock else utc_clock,
        level=str(config["log_level"]),
        secrets=config.secrets,
    )
    outcome = run_pipeline(config, run_id=run_id, logger=logger, out=out)
    return outcome.exit_code


if __name__ == "__main__":
    sys.exit(main())

# Layer 2 of 4: the settings that belong to this *deployment* rather than to
# this machine or this invocation. Checked into version control, contains no
# secret, and outranked by the environment and by an explicit flag.
#
# Read by: --config-file examples/pipeline.toml

sources = "alpha,bravo,charlie"
window_hours = 12
timeout_seconds = 3.0
retry_attempts = 3
log_level = "info"

"""Stage 4 — Report.

**The promise:** the report answers the question the pipeline exists for, and it
answers it the same way tomorrow as it does today for the same instant.

That second half is the whole reason ``report_at`` is a **parameter** and not a
call to ``datetime.now``. A function that reads the clock cannot be asserted on;
its output changes every time you run it, so "is this number right?" has no
answer you can put in a test. Passing the instant in costs one argument and buys
three things: a testable report, a backfill (ask for last Tuesday and get last
Tuesday's answer), and an incident timeline (ask for 03:00 and see what the
3 a.m. run saw). Day 91 made this choice for the same reason; Day 95 explains
why the instant must be timezone-aware.

The report also carries the one check the validation gate structurally cannot
make. A field validator sees one record. It cannot know that 41.3 Celsius five
minutes after 15.0 Celsius is a broken sensor, because both values are legal.
``suspect_jumps`` compares consecutive readings per station, and it **flags**
rather than deletes: a value that is valid but wrong is a fact about the world
that somebody has to look at, and silently dropping it would replace a visible
anomaly with an invisible gap.
"""

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

from sqlalchemy import select
from sqlalchemy.orm import Session

from store import StoredReading

#: Flag a change larger than this many deci-Celsius inside one hour.
JUMP_THRESHOLD_DC = 100
JUMP_WINDOW_MINUTES = 60


@dataclass(frozen=True)
class StationSummary:
    station_id: str
    readings: int
    min_dc: int | None
    max_dc: int | None
    mean_dc: int | None

    @staticmethod
    def _c(value: int | None) -> str:
        return "-" if value is None else f"{value / 10:.1f}"

    def line(self) -> str:
        return (
            f"  {self.station_id:<12} {self.readings:>8} {self._c(self.min_dc):>9}"
            f" {self._c(self.max_dc):>9} {self._c(self.mean_dc):>9}"
        )


@dataclass(frozen=True)
class SuspectJump:
    station_id: str
    previous_at: str
    observed_at: str
    change_dc: int
    minutes: int

    def line(self) -> str:
        return (
            f"  {self.station_id}: {self.change_dc / 10:+.1f} C in {self.minutes} minutes "
            f"({self.previous_at} -> {self.observed_at})"
        )


@dataclass(frozen=True)
class Report:
    report_at: str
    window_start: str
    window_hours: int
    readings_in_window: int
    total_rows: int
    stations: tuple[StationSummary, ...]
    suspect: tuple[SuspectJump, ...]


def _parse_instant(text: str) -> datetime:
    moment = datetime.fromisoformat(text.replace("Z", "+00:00"))
    if moment.tzinfo is None:
        raise ValueError(f"report_at must carry a UTC offset, got {text!r}")
    return moment.astimezone(timezone.utc)


def build_report(
    session: Session,
    *,
    report_at: str,
    window_hours: int,
    stations: list[str],
) -> Report:
    """Summarise the window ending at ``report_at``, per station."""
    end = _parse_instant(report_at)
    start = end - timedelta(hours=window_hours)
    end_text = end.strftime("%Y-%m-%dT%H:%M:%SZ")
    start_text = start.strftime("%Y-%m-%dT%H:%M:%SZ")

    rows = session.execute(
        select(
            StoredReading.station_id,
            StoredReading.observed_at,
            StoredReading.temperature_dc,
        )
        .where(StoredReading.observed_at >= start_text)
        .where(StoredReading.observed_at <= end_text)
        .order_by(StoredReading.station_id, StoredReading.observed_at)
    ).all()

    by_station: dict[str, list[tuple[str, int]]] = {name: [] for name in stations}
    for station_id, observed_at, temperature_dc in rows:
        by_station.setdefault(station_id, []).append((observed_at, temperature_dc))

    summaries = []
    for station_id in sorted(by_station):
        values = [temp for _, temp in by_station[station_id]]
        if values:
            summaries.append(
                StationSummary(
                    station_id=station_id,
                    readings=len(values),
                    min_dc=min(values),
                    max_dc=max(values),
                    mean_dc=round(sum(values) / len(values)),
                )
            )
        else:
            summaries.append(StationSummary(station_id, 0, None, None, None))

    total = session.execute(select(StoredReading.id)).all()

    return Report(
        report_at=end_text,
        window_start=start_text,
        window_hours=window_hours,
        readings_in_window=len(rows),
        total_rows=len(total),
        stations=tuple(summaries),
        suspect=tuple(suspect_jumps(by_station)),
    )


def suspect_jumps(by_station: dict[str, list[tuple[str, int]]]) -> list[SuspectJump]:
    """Consecutive readings that changed too far, too fast, to be believable."""
    found: list[SuspectJump] = []
    for station_id in sorted(by_station):
        series = sorted(by_station[station_id])
        for (previous_at, previous_dc), (observed_at, observed_dc) in zip(series, series[1:]):
            gap = _parse_instant(observed_at) - _parse_instant(previous_at)
            minutes = int(gap.total_seconds() // 60)
            change = observed_dc - previous_dc
            if minutes <= JUMP_WINDOW_MINUTES and abs(change) > JUMP_THRESHOLD_DC:
                found.append(
                    SuspectJump(
                        station_id=station_id,
                        previous_at=previous_at,
                        observed_at=observed_at,
                        change_dc=change,
                        minutes=minutes,
                    )
                )
    return found


def format_report(report: Report) -> str:
    lines = [
        "Station readings report",
        f"  as of        {report.report_at}",
        f"  window       {report.window_hours}h, from {report.window_start}",
        f"  in window    {report.readings_in_window} of {report.total_rows} stored readings",
        "",
        f"  {'station':<12} {'readings':>8} {'min C':>9} {'max C':>9} {'mean C':>9}",
        f"  {'-' * 12} {'-' * 8} {'-' * 9} {'-' * 9} {'-' * 9}",
    ]
    lines.extend(summary.line() for summary in report.stations)
    lines.append("")
    if report.suspect:
        lines.append(f"  suspect readings ({len(report.suspect)}) — stored and flagged, not dropped:")
        lines.extend(jump.line() for jump in report.suspect)
    else:
        lines.append("  suspect readings (0)")
    return "\n".join(lines)

"""The answer key for ``starter/stages.py`` — all nine exercises completed.

Read this *after* you have tried the exercises, not instead of them. It is here
so the test harness can prove the exercises are achievable rather than merely
asserted, and so the instructor solution has something to point at.

Run the starter suite against it:

    DAY098_SOLUTION=1 .venv/bin/pytest starter -q       # 10 passed
"""

from __future__ import annotations

import io
import json
import sys
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Any, Callable, TextIO

from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator
from sqlalchemy import Integer, String, UniqueConstraint, create_engine, select
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

EXIT_SUCCESS = 0
EXIT_FAILURE = 1
EXIT_PARTIAL = 3
REDACTED = "***redacted***"


@dataclass(frozen=True)
class FetchResult:
    source: str
    ok: bool
    records: list[dict] = field(default_factory=list)
    attempts: int = 0
    status: int | None = None
    error: str = ""


# EXERCISE 2 — a moment, not a mistake.
RETRYABLE_STATUS: frozenset[int] = frozenset({429, 500, 502, 503, 504})


def fetch_source(
    base_url: str,
    source: str,
    *,
    token: str = "",
    timeout: float = 5.0,
    attempts: int = 3,
    backoff: float = 0.05,
    sleep: Callable[[float], None] = time.sleep,
) -> FetchResult:
    """EXERCISE 1 and 2 — a deadline, bounded retries, and only what is worth it."""
    url = f"{base_url.rstrip('/')}/stations/{source}/readings"
    tried = 0
    last_status: int | None = None
    last_error = ""
    while tried < attempts:
        tried += 1
        request = urllib.request.Request(url, method="GET")
        request.add_header("Accept", "application/json")
        if token:
            request.add_header("Authorization", f"Bearer {token}")
        try:
            with urllib.request.urlopen(request, timeout=timeout) as response:  # noqa: S310
                payload = json.loads(response.read().decode("utf-8"))
            return FetchResult(
                source, True, list(payload.get("records", [])), tried, 200
            )
        except urllib.error.HTTPError as exc:
            with exc:
                last_status = exc.code
                body = exc.read().decode("utf-8", errors="replace")
            try:
                last_error = str(json.loads(body).get("error", body))
            except json.JSONDecodeError:
                last_error = body
            if exc.code not in RETRYABLE_STATUS:
                break
        except (urllib.error.URLError, TimeoutError, OSError) as exc:
            last_status = None
            last_error = f"{type(exc).__name__}: {exc}"
        if tried < attempts:
            sleep(backoff * (2 ** (tried - 1)))
    return FetchResult(source, False, [], tried, last_status, last_error)


class Reading(BaseModel):
    """EXERCISE 4 — the gate refuses what the store would have to argue with."""

    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    station_id: str = Field(min_length=1, max_length=32)
    reading_id: str = Field(min_length=1, max_length=64)
    observed_at: datetime
    temperature_c: float = Field(ge=-90.0, le=60.0)
    humidity_pct: int = Field(ge=0, le=100)

    @field_validator("observed_at")
    @classmethod
    def must_be_absolute(cls, value: datetime) -> datetime:
        if value.tzinfo is None:
            raise ValueError("observed_at must carry a UTC offset")
        return value.astimezone(timezone.utc)

    @property
    def temperature_dc(self) -> int:
        return round(self.temperature_c * 10)

    @property
    def observed_at_text(self) -> str:
        return self.observed_at.strftime("%Y-%m-%dT%H:%M:%SZ")


@dataclass(frozen=True)
class Rejection:
    source: str
    index: int
    reading_id: str
    problems: tuple[str, ...]


def validate_all(fetched: dict[str, list[dict]]) -> tuple[list[Reading], list[Rejection]]:
    """EXERCISE 3 — collect every failure; never abort the run for one record."""
    accepted: list[Reading] = []
    rejected: list[Rejection] = []
    for source, records in fetched.items():
        for index, raw in enumerate(records):
            try:
                accepted.append(Reading.model_validate(raw))
            except ValidationError as error:
                problems = tuple(
                    f"{'.'.join(str(part) for part in detail['loc']) or '<record>'}: {detail['msg']}"
                    for detail in error.errors()
                )
                rejected.append(
                    Rejection(
                        source=source,
                        index=index,
                        reading_id=str(raw.get("reading_id", "<no reading_id>")),
                        problems=problems,
                    )
                )
    return accepted, rejected


class Base(DeclarativeBase):
    pass


class StoredReading(Base):
    """EXERCISE 5a — the idempotence key, declared where the database enforces it."""

    __tablename__ = "readings"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    station_id: Mapped[str] = mapped_column(String(32), nullable=False)
    reading_id: Mapped[str] = mapped_column(String(64), nullable=False)
    observed_at: Mapped[str] = mapped_column(String(20), nullable=False)
    temperature_dc: Mapped[int] = mapped_column(Integer, nullable=False)
    humidity_pct: Mapped[int] = mapped_column(Integer, nullable=False)
    ingested_by_run: Mapped[str] = mapped_column(String(32), nullable=False)

    __table_args__ = (
        UniqueConstraint("station_id", "reading_id", name="uq_readings_idempotence"),
    )


@dataclass(frozen=True)
class StoreResult:
    considered: int
    inserted: int
    duplicates: int
    total_rows: int


def store_readings(session: Session, readings: list[Reading], *, run_id: str) -> StoreResult:
    """EXERCISE 5b — insert only what is new, and report it honestly."""
    keys = [(r.station_id, r.reading_id) for r in readings]
    held = set()
    if keys:
        rows = session.execute(
            select(StoredReading.station_id, StoredReading.reading_id)
        ).all()
        held = {(row[0], row[1]) for row in rows}

    seen: set[tuple[str, str]] = set()
    payload: list[dict] = []
    for reading in readings:
        key = (reading.station_id, reading.reading_id)
        if key in held or key in seen:
            continue
        seen.add(key)
        payload.append(
            {
                "station_id": reading.station_id,
                "reading_id": reading.reading_id,
                "observed_at": reading.observed_at_text,
                "temperature_dc": reading.temperature_dc,
                "humidity_pct": reading.humidity_pct,
                "ingested_by_run": run_id,
            }
        )
    if payload:
        session.execute(
            sqlite_insert(StoredReading).on_conflict_do_nothing(
                index_elements=["station_id", "reading_id"]
            ),
            payload,
        )
        session.commit()
    total = len(session.execute(select(StoredReading.id)).all())
    return StoreResult(len(readings), len(payload), len(readings) - len(payload), total)


def build_report(
    session: Session,
    *,
    report_at: str,
    window_hours: int,
    stations: list[str],
) -> dict[str, Any]:
    """EXERCISE 6 — the instant is a parameter, so the answer is reproducible."""
    end = datetime.fromisoformat(report_at.replace("Z", "+00:00"))
    if end.tzinfo is None:
        raise ValueError("report_at must carry a UTC offset")
    end = end.astimezone(timezone.utc)
    start = end - timedelta(hours=window_hours)
    end_text = end.strftime("%Y-%m-%dT%H:%M:%SZ")
    start_text = start.strftime("%Y-%m-%dT%H:%M:%SZ")
    rows = session.execute(
        select(StoredReading.station_id, StoredReading.observed_at, StoredReading.temperature_dc)
        .where(StoredReading.observed_at >= start_text)
        .where(StoredReading.observed_at <= end_text)
    ).all()
    per_station: dict[str, list[int]] = {name: [] for name in stations}
    for station_id, _observed_at, temperature_dc in rows:
        per_station.setdefault(station_id, []).append(temperature_dc)
    return {
        "report_at": end_text,
        "window_start": start_text,
        "readings_in_window": len(rows),
        "stations": {name: len(values) for name, values in sorted(per_station.items())},
    }


def _redact(value: Any, secrets: tuple[str, ...]) -> Any:
    if isinstance(value, str):
        for secret in secrets:
            if secret and secret in value:
                value = value.replace(secret, REDACTED)
        return value
    if isinstance(value, dict):
        return {key: _redact(item, secrets) for key, item in value.items()}
    if isinstance(value, (list, tuple)):
        return [_redact(item, secrets) for item in value]
    return value


class RunLogger:
    """EXERCISE 7 and 9 — a run id on every line, and redaction in one place."""

    def __init__(self, run_id: str, *, stream: TextIO, secrets: tuple[str, ...] = ()) -> None:
        self.run_id = run_id
        self.stream = stream
        self.secrets = tuple(s for s in secrets if s)
        self.emitted: list[dict[str, Any]] = []

    def event(self, event: str, level: str = "info", **fields: Any) -> None:
        record: dict[str, Any] = {"level": level, "run_id": self.run_id, "event": event}
        record.update(_redact(fields, self.secrets))
        self.emitted.append(record)
        self.stream.write(json.dumps(record) + "\n")


def run(
    base_url: str,
    *,
    sources: list[str],
    token: str = "",
    database_url: str = "sqlite://",
    window_hours: int = 12,
    report_at: str = "2026-08-16T12:00:00Z",
    run_id: str = "starter-run",
    log_stream: TextIO | None = None,
    out: TextIO | None = None,
) -> int:
    """EXERCISE 8 — three outcomes, three exit codes."""
    log_stream = log_stream if log_stream is not None else io.StringIO()
    out = out if out is not None else sys.stdout
    logger = RunLogger(run_id, stream=log_stream, secrets=(token,) if token else ())

    results = [
        fetch_source(base_url, source, token=token, backoff=0.0, sleep=lambda _s: None)
        for source in sources
    ]
    fetched = {r.source: r.records for r in results if r.ok}
    failed = [r.source for r in results if not r.ok]
    for result in results:
        if not result.ok:
            logger.event(
                "ingest.source_failed",
                level="warning",
                source=result.source,
                status=result.status,
                error=result.error,
            )
    logger.event(
        "stage.ingest",
        sources_ok=len(fetched),
        sources_failed=len(failed),
        records_fetched=sum(len(records) for records in fetched.values()),
    )
    if not fetched:
        logger.event("run.end", level="error", status="failure")
        out.write("no source answered\n")
        return EXIT_FAILURE

    accepted, rejected = validate_all(fetched)
    logger.event("stage.validate", accepted=len(accepted), rejected=len(rejected))

    engine = create_engine(database_url, future=True)
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        stored = store_readings(session, accepted, run_id=run_id)
        logger.event(
            "stage.store",
            inserted=stored.inserted,
            duplicates_skipped=stored.duplicates,
            total_rows=stored.total_rows,
        )
        summary = build_report(
            session, report_at=report_at, window_hours=window_hours, stations=sources
        )
        logger.event("stage.report", **summary)
        out.write(json.dumps(summary, sort_keys=True) + "\n")

    if failed or rejected:
        status, code = "partial_success", EXIT_PARTIAL
    else:
        status, code = "success", EXIT_SUCCESS
    logger.event("stage.observe", status=status, exit_code=code)
    logger.event("run.end", level="warning" if code else "info", status=status, exit_code=code)
    engine.dispose()
    return code

"""Stage 3 — Store.

**The promise:** running the pipeline twice stores the data once.

That promise is the hinge of the whole design, and it is bought with one thing:
an **idempotence key**, a column or set of columns that identifies a record by
what it *is* rather than by when it arrived. Here it is
``(station_id, reading_id)`` — the pair the source itself assigns, declared
``UNIQUE``.

Two layers enforce it, deliberately:

1. ``store_readings`` asks the database which keys it already holds and inserts
   only the rest. This is what makes the *count* honest: the run can report "4
   new, 3 already held" rather than "7 written".
2. The ``UNIQUE`` constraint plus ``ON CONFLICT DO NOTHING`` catches anything
   layer 1 missed — a duplicate inside the same batch, or a second copy of the
   pipeline that started while this one was between the SELECT and the INSERT.

Layer 1 without layer 2 is a promise the application makes and the database
cannot keep. Layer 2 without layer 1 works and reports nothing useful. The
schema is where correctness lives; the query is where the reporting lives.

Everything else here is Days 88, 91 and 93 applied without ceremony: integer
minor units instead of floats (deci-Celsius — 18.4 C is stored as 184, and the
division by ten happens at the display edge and nowhere else), timestamps as ISO
8601 text in UTC so lexicographic order is chronological order, CHECK
constraints that make the illegal states unrepresentable rather than merely
discouraged, and an index on the column the report actually filters by.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timezone

from sqlalchemy import (
    CheckConstraint,
    Index,
    Integer,
    String,
    UniqueConstraint,
    create_engine,
    func,
    select,
)
from sqlalchemy.dialects.sqlite import insert as sqlite_insert
from sqlalchemy.engine import Engine
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

from validate import Reading


class Base(DeclarativeBase):
    pass


class StoredReading(Base):
    """One reading, as it lives on disk."""

    __tablename__ = "readings"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    station_id: Mapped[str] = mapped_column(String(32), nullable=False)
    reading_id: Mapped[str] = mapped_column(String(64), nullable=False)
    #: ISO 8601 in UTC. Fixed width, most significant field first, so ordering
    #: the text orders the instants (Day 91, Day 95).
    observed_at: Mapped[str] = mapped_column(String(20), nullable=False)
    #: Deci-Celsius as an integer. 18.4 C is 184. No float ever reaches disk.
    temperature_dc: Mapped[int] = mapped_column(Integer, nullable=False)
    humidity_pct: Mapped[int] = mapped_column(Integer, nullable=False)
    #: Which run put it here. This is what makes a bad backfill undoable.
    ingested_by_run: Mapped[str] = mapped_column(String(32), nullable=False)

    __table_args__ = (
        UniqueConstraint("station_id", "reading_id", name="uq_readings_idempotence"),
        CheckConstraint("humidity_pct BETWEEN 0 AND 100", name="ck_readings_humidity"),
        CheckConstraint("temperature_dc BETWEEN -900 AND 600", name="ck_readings_temperature"),
        CheckConstraint("length(observed_at) = 20", name="ck_readings_observed_at_iso"),
        Index("ix_readings_observed_at", "observed_at"),
    )


class RunRow(Base):
    """One row per pipeline run. A run id you cannot look up is a run id you
    cannot act on, and 'delete everything run abc123 wrote' is the cheapest
    recovery a pipeline can offer."""

    __tablename__ = "runs"

    run_id: Mapped[str] = mapped_column(String(32), primary_key=True)
    started_at: Mapped[str] = mapped_column(String(20), nullable=False)
    status: Mapped[str] = mapped_column(String(20), nullable=False)
    records_fetched: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    records_accepted: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    records_rejected: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    records_inserted: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
    records_duplicate: Mapped[int] = mapped_column(Integer, nullable=False, default=0)


@dataclass(frozen=True)
class StoreResult:
    considered: int
    inserted: int
    duplicates: int
    total_rows: int
    inserted_keys: tuple[tuple[str, str], ...] = field(default=())


def build_engine(database_url: str = "sqlite://") -> Engine:
    """Create the engine and the schema. ``sqlite://`` alone means in memory."""
    engine = create_engine(database_url, future=True)
    Base.metadata.create_all(engine)
    return engine


def existing_keys(session: Session, keys: list[tuple[str, str]]) -> set[tuple[str, str]]:
    """Which of these idempotence keys the store already holds."""
    if not keys:
        return set()
    stations = {station for station, _ in keys}
    rows = session.execute(
        select(StoredReading.station_id, StoredReading.reading_id).where(
            StoredReading.station_id.in_(stations)
        )
    ).all()
    held = {(row[0], row[1]) for row in rows}
    return held & set(keys)


def store_readings(
    session: Session,
    readings: list[Reading],
    *,
    run_id: str,
) -> StoreResult:
    """Insert only what is new. Report exactly what happened."""
    keys = [(reading.station_id, reading.reading_id) for reading in readings]
    already = existing_keys(session, keys)

    seen_in_batch: set[tuple[str, str]] = set()
    rows: list[dict] = []
    for reading in readings:
        key = (reading.station_id, reading.reading_id)
        if key in already or key in seen_in_batch:
            continue
        seen_in_batch.add(key)
        rows.append(
            {
                "station_id": reading.station_id,
                "reading_id": reading.reading_id,
                "observed_at": reading.observed_at_text,
                "temperature_dc": reading.temperature_dc,
                "humidity_pct": reading.humidity_pct,
                "ingested_by_run": run_id,
            }
        )

    if rows:
        # Layer 2: the database's own guarantee, in case layer 1 raced.
        statement = sqlite_insert(StoredReading).on_conflict_do_nothing(
            index_elements=["station_id", "reading_id"]
        )
        session.execute(statement, rows)
        session.commit()

    total = session.execute(select(func.count()).select_from(StoredReading)).scalar_one()
    return StoreResult(
        considered=len(readings),
        inserted=len(rows),
        duplicates=len(readings) - len(rows),
        total_rows=int(total),
        inserted_keys=tuple(sorted((row["station_id"], row["reading_id"]) for row in rows)),
    )


def record_run(
    session: Session,
    *,
    run_id: str,
    started_at: str,
    status: str,
    fetched: int,
    accepted: int,
    rejected: int,
    inserted: int,
    duplicates: int,
) -> None:
    """Write (or overwrite) this run's row. Re-running a run id is not an error;
    it is a rerun, and the counts it reports are the counts of the rerun."""
    statement = (
        sqlite_insert(RunRow)
        .values(
            run_id=run_id,
            started_at=started_at,
            status=status,
            records_fetched=fetched,
            records_accepted=accepted,
            records_rejected=rejected,
            records_inserted=inserted,
            records_duplicate=duplicates,
        )
        .on_conflict_do_update(
            index_elements=["run_id"],
            set_={
                "started_at": started_at,
                "status": status,
                "records_fetched": fetched,
                "records_accepted": accepted,
                "records_rejected": rejected,
                "records_inserted": inserted,
                "records_duplicate": duplicates,
            },
        )
    )
    session.execute(statement)
    session.commit()


def utc_text(moment: datetime | None = None) -> str:
    moment = moment or datetime.now(timezone.utc)
    return moment.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

"""Stage 2 — Validate.

**The promise:** nothing enters the store that the store's own constraints would
have to argue with, every bad record is reported well enough to fix the source,
and one bad record does not end the run.

This is Day 94's gate at a boundary, with one addition that matters more than
the model itself: ``collect``. The obvious implementation raises on the first
failure, and it is wrong for a pipeline. If a source starts sending 400 bad
records out of 10,000, "the first one failed" is a bug report you cannot act on.
"3,942 records rejected, 3,940 of them because humidity_pct exceeded 100, first
offender b-4471 at 2026-08-16T04:15:00Z" is a message you can forward to
whoever owns the sensor.

``Reading`` is deliberately strict:

* ``extra="forbid"`` — an unexpected field is a signal that the source changed
  shape, and silently ignoring it is how a schema drift goes unnoticed for a
  quarter.
* ``observed_at`` must be timezone-aware (Day 95). A naive timestamp is not an
  instant; it is an instant plus an assumption, and the assumption is usually
  the assumption of whoever wrote the code, not whoever ran it.
* The ranges are physical, not arbitrary: humidity is a percentage, and the
  recorded temperature extremes on Earth sit inside -90 to 60 Celsius.

Note what the gate can*not* catch, because the lab has a record for it: bravo's
b-4 reports 41.3 Celsius five minutes after 15.0 Celsius. Every field is legal.
Only a rule about the *sequence* sees it, and that rule lives in the report.
"""

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timezone

from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator


class Reading(BaseModel):
    """One sensor reading, as it is allowed to exist inside this pipeline."""

    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    station_id: str = Field(min_length=1, max_length=32)
    reading_id: str = Field(min_length=1, max_length=64)
    observed_at: datetime
    temperature_c: float = Field(ge=-90.0, le=60.0)
    humidity_pct: int = Field(ge=0, le=100)

    @field_validator("observed_at")
    @classmethod
    def must_be_absolute(cls, value: datetime) -> datetime:
        if value.tzinfo is None:
            raise ValueError("observed_at must carry a UTC offset (it is an instant, not a date)")
        return value.astimezone(timezone.utc)

    @property
    def temperature_dc(self) -> int:
        """Deci-Celsius. The store keeps an integer; see store.py for why."""
        return round(self.temperature_c * 10)

    @property
    def observed_at_text(self) -> str:
        return self.observed_at.strftime("%Y-%m-%dT%H:%M:%SZ")


@dataclass(frozen=True)
class Rejection:
    """One record that did not get in, and enough detail to fix its source."""

    source: str
    index: int
    reading_id: str
    problems: tuple[str, ...]

    def __str__(self) -> str:
        return f"{self.source}[{self.index}] {self.reading_id}: " + "; ".join(self.problems)


@dataclass(frozen=True)
class ValidationOutcome:
    accepted: list[Reading]
    rejected: list[Rejection]

    @property
    def considered(self) -> int:
        return len(self.accepted) + len(self.rejected)

    def reasons(self) -> dict[str, int]:
        """How many rejections per distinct problem, worst first."""
        counts: dict[str, int] = {}
        for rejection in self.rejected:
            for problem in rejection.problems:
                field_name = problem.split(":", 1)[0]
                counts[field_name] = counts.get(field_name, 0) + 1
        return dict(sorted(counts.items(), key=lambda item: (-item[1], item[0])))


def _problems(error: ValidationError) -> tuple[str, ...]:
    out = []
    for detail in error.errors():
        location = ".".join(str(part) for part in detail["loc"]) or "<record>"
        out.append(f"{location}: {detail['msg']}")
    return tuple(out)


def validate_records(source: str, raw_records: list[dict]) -> ValidationOutcome:
    """Validate every record. Collect the failures; never raise."""
    accepted: list[Reading] = []
    rejected: list[Rejection] = []
    for index, raw in enumerate(raw_records):
        try:
            accepted.append(Reading.model_validate(raw))
        except ValidationError as error:
            identifier = str(raw.get("reading_id", "<no reading_id>")) if isinstance(raw, dict) else "<not an object>"
            rejected.append(
                Rejection(
                    source=source,
                    index=index,
                    reading_id=identifier,
                    problems=_problems(error),
                )
            )
    return ValidationOutcome(accepted=accepted, rejected=rejected)


def validate_all(fetched: dict[str, list[dict]]) -> ValidationOutcome:
    """Run the gate across every source's records, in source order."""
    accepted: list[Reading] = []
    rejected: list[Rejection] = []
    for source, records in fetched.items():
        outcome = validate_records(source, records)
        accepted.extend(outcome.accepted)
        rejected.extend(outcome.rejected)
    return ValidationOutcome(accepted=accepted, rejected=rejected)

lesson_id: D098
day: 98
kind: project
languages: [python, sql, bash]
setup_commands:
  - cd labs/sections/programming-with-python/day-098-section-project-a-complete-data-pipeline
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
  - '.venv/bin/python -c "import sqlalchemy, pydantic; print(sqlalchemy.__version__, pydantic.VERSION)"'
  - export PYTHONPATH=examples
run_commands:
  - bash tests/run_tests.sh
  - .venv/bin/python examples/demo_run.py
  - 'PIPELINE_LOG_LEVEL=warning PIPELINE_API_TOKEN=demo-token-value .venv/bin/python examples/pipeline.py --config-file examples/pipeline.toml --window-hours 24 --explain-config'
  - .venv/bin/python examples/fixture_server.py --token demo-token-value
  - 'PIPELINE_API_TOKEN=demo-token-value .venv/bin/python examples/pipeline.py --base-url http://127.0.0.1:PORT --sources alpha,bravo,charlie --report-at 2026-08-16T12:00:00Z --window-hours 12 --run-id run-cli000001 --fixed-clock'
  - .venv/bin/pytest starter -q
  - DAY098_SOLUTION=1 .venv/bin/pytest starter -q
test_commands:
  - bash tests/run_tests.sh
cleanup_commands:
  - rm -f pipeline.db
  - find . -type d -name __pycache__ -prune -exec rm -rf -- {} +
  - rm -rf starter/.pytest_cache .pytest_cache
  - '.venv removal is optional: rm -rf .venv'
  - 'git checkout -- starter/  # optional: reset your work'
requires_network: true
requires_api_key: false
estimated_minutes: 45
last_executed: '2026-08-16'
executed_on: 'macOS 26.5.2 (Apple Silicon, arm64), Python 3.14.0, SQLAlchemy 2.0.51, pydantic 2.13.4, SQLite 3.53.3, pytest 9.1.1, bash 3.2.57 — bash tests/run_tests.sh -> 84 checks, 0 failure(s), exit 0'
network_note: 'Network is needed once, for pip install of the three pinned packages. At run time every request goes to a fixture server bound to 127.0.0.1 on a kernel-chosen port; section 10 of the harness fails if any URL in the lab points anywhere else.'

# Why each pin exists, and what is deliberately absent

```
SQLAlchemy==2.0.51
pydantic==2.13.4
pytest==9.1.1
```

Three dependencies for a pipeline that fetches over HTTP, validates, stores in a
real schema, reports and logs. That is not an accident; it is the point of the
last section of this course. Every dependency is a version to pin, a
vulnerability feed to watch, and one more thing that can break the job that runs
while you are asleep. The bar for adding one is that it earns its place.

| Pin | What it does here | What it replaces |
| --- | --- | --- |
| `SQLAlchemy==2.0.51` | The declarative models, the schema with its constraints and index, and the `INSERT ... ON CONFLICT DO NOTHING` that makes the store idempotent | Hand-written SQL strings and hand-written parameter binding (Days 88-91 did that on purpose, so you know what is being replaced) |
| `pydantic==2.13.4` | The validation gate: types, ranges, the timezone requirement, `extra="forbid"`, and error objects with a field path and a message | A wall of `if not isinstance(...)` that never produces a usable error report |
| `pytest==9.1.1` | The starter exercises | Nothing; it is the runner from Week 11 |

## What is deliberately NOT here

**`requests`.** Day 78 taught it and it is an excellent library. This pipeline
makes one GET with a timeout and one header, and `urllib.request` in the
standard library does exactly that. `requests` earns its place the moment you
need connection pooling across many calls, a `Session` with shared headers, or
`urllib3`'s `Retry` with its `Retry-After` handling — none of which this
pipeline needs. `examples/ingest.py` says so in its module docstring.

**Airflow, Dagster, Prefect, dbt.** The lesson describes all four from their
documentation, states plainly that no output from any of them is reproduced,
and section 1 of `tests/run_tests.sh` asserts that none of them is importable
here — so the claim cannot go quietly stale if somebody installs one later.

**Alembic.** Schema migration is Day 88's subject and SQLAlchemy's migration
tool is worth knowing about; this lab creates its schema once with
`Base.metadata.create_all` and never changes it, so a migration tool would be
scaffolding around nothing.

## Installing

```bash
cd labs/sections/programming-with-python/day-098-section-project-a-complete-data-pipeline
python3 -m venv .venv
.venv/bin/pip install -r requirements/requirements.txt
```

That install is the only moment this lab needs the network. Everything after it
talks to a fixture server on 127.0.0.1, and section 10 of the harness asserts
that no URL anywhere in the lab points at anything else.

SQLAlchemy==2.0.51
pydantic==2.13.4
pytest==9.1.1

# The brief: make a finished-looking pipeline actually keep its promises

You have inherited `stages.py`. It runs. It fetches from three weather stations,
validates what comes back, stores it, prints a summary and exits. On a good day,
against healthy sources, you would never know anything was wrong with it.

Your job is the bad day.

Nine exercises, marked `EXERCISE n` in `stages.py`. Each one turns a stage that
*works* into a stage that keeps a **promise** — a statement about what happens
when something goes wrong at three in the morning with nobody watching.

| # | Stage | The promise you are adding |
| --- | --- | --- |
| 1 | Ingest | Every fetch has a deadline, and a source that blinks gets another chance |
| 2 | Ingest | Only failures worth retrying are retried |
| 3 | Validate | One bad record does not end the run; every bad record is collected |
| 4 | Validate | The gate refuses what the store would have to argue with |
| 5 | Store | Running twice stores once |
| 6 | Report | The report instant is a parameter, so the answer is reproducible |
| 7 | Observe | Every log line carries the run id |
| 8 | Observe | Partial success has its own exit code |
| 9 | Observe | A secret cannot reach the log, even when an upstream echoes it back |

## How to work

```bash
cd labs/sections/programming-with-python/day-098-section-project-a-complete-data-pipeline
.venv/bin/pytest starter -q          # 1 passed, 9 skipped — the starting line
```

The one passing test proves the skeleton runs end to end. Read what it asserts:
the skeleton **reports its own failure**, because it aborts at the first
malformed record. That is exercise 3, and you can see it before you have written
a line.

Then, for each exercise in order:

1. Read the `EXERCISE n` block in `stages.py`. It names the exact change.
2. Make the change.
3. Delete that exercise's `@exercise(...)` decorator in `test_stages.py`.
4. `.venv/bin/pytest starter -q` — the test tells you whether the promise holds.

Each exercise's docstring also names a `-k` filter, so you can run one at a time:

```bash
.venv/bin/pytest starter -q -k idempotent
```

## What the fixture server does to you

Nothing in this lab reaches the internet. `examples/fixture_server.py` binds
127.0.0.1 on a port the kernel picks, and it is hostile on purpose:

- **alpha** answers immediately with five records: two good, one with a
  temperature of `"warm"`, one that repeats an earlier record's id exactly, and
  one with a humidity of 155 per cent.
- **bravo** returns 500 twice and then 200. Its counter is per server process,
  so the *second* pipeline run finds it healthy — which is what a transient
  failure actually looks like.
- **charlie** returns 500 every time, and its error body politely quotes your
  API token back at you.
- **delta** returns 404. Retrying it is a waste of three round trips.

One of bravo's records is the interesting one: `b-4` reports 41.3 Celsius five
minutes after 15.0 Celsius. Every field is legal. Exercise 4 will not catch it
and is not supposed to. Think about where that check belongs before you look at
`examples/report.py`.

## When you are done

```bash
.venv/bin/pytest starter -q          # 10 passed
bash tests/run_tests.sh              # 84 checks, 0 failure(s)
```

The reference implementation of all nine is in `examples/stages_solved.py`, and
the production version — the same promises, written out at full size across five
modules — is the rest of `examples/`. Read the reference *after* you have tried,
not instead.

"""Test wiring: import paths and one fixture server for the whole session."""

from __future__ import annotations

import importlib
import os
import sys
from pathlib import Path

import pytest

LAB = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(LAB / "starter"))
sys.path.insert(0, str(LAB / "examples"))

import fixture_server  # noqa: E402

TOKEN = "demo-token-value"

# The test harness runs this same suite twice: once against the skeleton the
# learner receives, and once against examples/stages_solved.py, so the nine
# exercises are proved achievable rather than merely asserted to be.
if os.environ.get("DAY098_SOLUTION"):
    sys.modules["stages"] = importlib.import_module("stages_solved")


@pytest.fixture(scope="session")
def base_url() -> str:
    """A fixture server on 127.0.0.1, on a port the kernel picks.

    Session-scoped, so bravo's "fail twice then succeed" counter is shared by
    every test in the run. Tests that care about it call
    ``fixture_server.reset_flaky_counter()`` first.
    """
    server, port = fixture_server.start_background_server(token=TOKEN)
    yield f"http://127.0.0.1:{port}"
    server.shutdown()
    server.server_close()


@pytest.fixture
def token() -> str:
    return TOKEN

[pytest]
# Warnings are errors here. A DeprecationWarning from SQLAlchemy or pydantic is
# the library telling you that the thing you just wrote will stop working, and
# the only reliable way to read that message is to make it impossible to ignore.
filterwarnings =
    error
testpaths = .

"""Your work. A pipeline that runs, and breaks every promise it should keep.

Nothing here is broken in the sense of raising a mysterious error. It is worse
than that: it is a pipeline that looks finished. It fetches, it validates, it
stores, it reports, it exits. Point it at a healthy source on a good day and you
would never know. The nine exercises below turn each of its five stages into a
stage that keeps a promise.

Run the skeleton first, before you change anything:

    .venv/bin/pytest starter -q          # 1 passed, 9 skipped

The one passing test proves the skeleton runs end to end and reports failure —
because it aborts at the first malformed record. That is exercise 3.

Work in order. After each exercise, delete that exercise's ``@exercise(...)``
decorator in ``test_stages.py`` and run pytest again.
"""

from __future__ import annotations

import io
import json
import sys
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Any, Callable, TextIO

from pydantic import BaseModel, ValidationError
from sqlalchemy import Integer, String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column

EXIT_SUCCESS = 0
EXIT_FAILURE = 1
EXIT_PARTIAL = 3


# ---------------------------------------------------------------------------
# Stage 1 — Ingest
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class FetchResult:
    source: str
    ok: bool
    records: list[dict] = field(default_factory=list)
    attempts: int = 0
    status: int | None = None
    error: str = ""


#: Statuses worth a second attempt.
#:
#:     EXERCISE 2 — Retry only what is worth retrying.
#:     A 500 means the server had a bad moment. A 404 means the URL is wrong,
#:     and it will be just as wrong in two seconds. Fill this set with the
#:     codes that describe a moment rather than a mistake, then use it in
#:     fetch_source() to break out of the loop instead of retrying.
#:     Check with: .venv/bin/pytest starter -q -k retryable
RETRYABLE_STATUS: frozenset[int] = frozenset()


def fetch_source(
    base_url: str,
    source: str,
    *,
    token: str = "",
    timeout: float = 5.0,
    attempts: int = 3,
    backoff: float = 0.05,
    sleep: Callable[[float], None] = time.sleep,
) -> FetchResult:
    """Fetch one source.

    EXERCISE 1 — Give this a deadline and a second chance.
    Right now it makes exactly one attempt and passes no ``timeout`` to
    ``urlopen``, so a hung server hangs your 3 a.m. run forever and a source
    that blinks is lost for the day. Wrap the request in a loop that tries up
    to ``attempts`` times, pass ``timeout=timeout`` to ``urlopen``, and call
    ``sleep(backoff * 2 ** (tried - 1))`` between attempts.
    Check with: .venv/bin/pytest starter -q -k retries
    """
    url = f"{base_url.rstrip('/')}/stations/{source}/readings"
    request = urllib.request.Request(url, method="GET")
    request.add_header("Accept", "application/json")
    if token:
        request.add_header("Authorization", f"Bearer {token}")
    try:
        with urllib.request.urlopen(request) as response:  # noqa: S310
            payload = json.loads(response.read().decode("utf-8"))
            return FetchResult(source, True, list(payload.get("records", [])), 1, response.status)
    except urllib.error.HTTPError as exc:
        # HTTPError is a file object; not closing it leaks a socket.
        with exc:
            body = exc.read().decode("utf-8", errors="replace")
        try:
            message = str(json.loads(body).get("error", body))
        except json.JSONDecodeError:
            message = body
        return FetchResult(source, False, [], 1, exc.code, message)
    except (urllib.error.URLError, TimeoutError, OSError) as exc:
        return FetchResult(source, False, [], 1, None, f"{type(exc).__name__}: {exc}")


# ---------------------------------------------------------------------------
# Stage 2 — Validate
# ---------------------------------------------------------------------------
class Reading(BaseModel):
    """One reading.

    EXERCISE 4 — Make this model refuse what the store would have to argue with.
    As written it accepts humidity of 155 and a naive timestamp, and it silently
    ignores any extra field the source starts sending. Add
    ``model_config = ConfigDict(extra="forbid")``, constrain ``humidity_pct`` to
    0-100 and ``temperature_c`` to -90.0-60.0 with ``Field(ge=..., le=...)``,
    and add a ``field_validator`` on ``observed_at`` that rejects a value whose
    ``tzinfo`` is None — an instant without an offset is an assumption, not a
    time (Day 95).
    Check with: .venv/bin/pytest starter -q -k out_of_range
    """

    station_id: str
    reading_id: str
    observed_at: datetime
    temperature_c: float
    humidity_pct: int

    @property
    def temperature_dc(self) -> int:
        return round(self.temperature_c * 10)

    @property
    def observed_at_text(self) -> str:
        return self.observed_at.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


@dataclass(frozen=True)
class Rejection:
    source: str
    index: int
    reading_id: str
    problems: tuple[str, ...]


def validate_all(fetched: dict[str, list[dict]]) -> tuple[list[Reading], list[Rejection]]:
    """Validate every fetched record.

    EXERCISE 3 — Collect the failures instead of dying on the first.
    This raises the moment one record is malformed, which throws away every good
    record that came with it and tells whoever owns the source exactly one thing
    that is wrong. Catch ``ValidationError`` per record, build a ``Rejection``
    from ``error.errors()`` (each entry has ``loc`` and ``msg``), and return
    both lists.
    Check with: .venv/bin/pytest starter -q -k collects
    """
    accepted: list[Reading] = []
    rejected: list[Rejection] = []
    for source, records in fetched.items():
        for index, raw in enumerate(records):
            accepted.append(Reading.model_validate(raw))  # raises on the first bad one
    return accepted, rejected


# ---------------------------------------------------------------------------
# Stage 3 — Store
# ---------------------------------------------------------------------------
class Base(DeclarativeBase):
    pass


class StoredReading(Base):
    """One reading on disk.

    EXERCISE 5a — Declare the idempotence key.
    Add to ``__table_args__`` a
    ``UniqueConstraint("station_id", "reading_id", name="uq_readings_idempotence")``.
    This is the promise the database keeps even when your code is wrong.
    """

    __tablename__ = "readings"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    station_id: Mapped[str] = mapped_column(String(32), nullable=False)
    reading_id: Mapped[str] = mapped_column(String(64), nullable=False)
    observed_at: Mapped[str] = mapped_column(String(20), nullable=False)
    temperature_dc: Mapped[int] = mapped_column(Integer, nullable=False)
    humidity_pct: Mapped[int] = mapped_column(Integer, nullable=False)
    ingested_by_run: Mapped[str] = mapped_column(String(32), nullable=False)

    __table_args__ = ()


@dataclass(frozen=True)
class StoreResult:
    considered: int
    inserted: int
    duplicates: int
    total_rows: int


def store_readings(session: Session, readings: list[Reading], *, run_id: str) -> StoreResult:
    """Write the accepted readings.

    EXERCISE 5b — Store what is new, and only what is new.
    This inserts everything it is given, so the duplicate inside one payload
    lands twice and a second run of the pipeline doubles the table. Select the
    ``(station_id, reading_id)`` pairs the table already holds, skip those and
    any repeat inside this batch, and report ``inserted`` and ``duplicates``
    honestly. Then use ``sqlalchemy.dialects.sqlite.insert(...)
    .on_conflict_do_nothing(index_elements=["station_id", "reading_id"])`` so
    the database backs you up.
    Check with: .venv/bin/pytest starter -q -k idempotent
    """
    for reading in readings:
        session.add(
            StoredReading(
                station_id=reading.station_id,
                reading_id=reading.reading_id,
                observed_at=reading.observed_at_text,
                temperature_dc=reading.temperature_dc,
                humidity_pct=reading.humidity_pct,
                ingested_by_run=run_id,
            )
        )
    session.commit()
    total = len(session.execute(select(StoredReading.id)).all())
    return StoreResult(len(readings), len(readings), 0, total)


# ---------------------------------------------------------------------------
# Stage 4 — Report
# ---------------------------------------------------------------------------
def build_report(session: Session, *, window_hours: int, stations: list[str]) -> dict[str, Any]:
    """Summarise the last ``window_hours``.

    EXERCISE 6 — Make the instant a parameter.
    This reads the clock, so its answer changes every second and no test can
    assert on it. Add a keyword-only ``report_at: str`` argument, parse it with
    ``datetime.fromisoformat(report_at.replace("Z", "+00:00"))``, require an
    offset, and derive the window from it. Then the report is testable, a
    backfill is possible, and "what did the 3 a.m. run see?" has an answer.
    Check with: .venv/bin/pytest starter -q -k fixed_instant
    """
    end = datetime.now(timezone.utc)
    start = end - timedelta(hours=window_hours)
    end_text = end.strftime("%Y-%m-%dT%H:%M:%SZ")
    start_text = start.strftime("%Y-%m-%dT%H:%M:%SZ")
    rows = session.execute(
        select(StoredReading.station_id, StoredReading.observed_at, StoredReading.temperature_dc)
        .where(StoredReading.observed_at >= start_text)
        .where(StoredReading.observed_at <= end_text)
    ).all()
    per_station: dict[str, list[int]] = {name: [] for name in stations}
    for station_id, _observed_at, temperature_dc in rows:
        per_station.setdefault(station_id, []).append(temperature_dc)
    return {
        "report_at": end_text,
        "window_start": start_text,
        "readings_in_window": len(rows),
        "stations": {name: len(values) for name, values in sorted(per_station.items())},
    }


# ---------------------------------------------------------------------------
# Stage 5 — Observe
# ---------------------------------------------------------------------------
class RunLogger:
    """One JSON object per line.

    EXERCISE 7 — Carry the run id on every line.
    A log line with no run id cannot be joined to the run that wrote it, and at
    3 a.m. that is the difference between "this run failed" and "something
    failed". Put ``run_id`` into every record.

    EXERCISE 9 — Redact secrets in the logger, not by remembering.
    charlie's error body echoes the API token back, so the token reaches the log
    without anybody writing code to log it. Store ``secrets`` on the logger and
    replace every occurrence in every string value with ``"***redacted***"``
    before writing. Strings nested inside lists and dicts count.
    Check with: .venv/bin/pytest starter -q -k "run_id or secret"
    """

    def __init__(self, run_id: str, *, stream: TextIO, secrets: tuple[str, ...] = ()) -> None:
        self.run_id = run_id
        self.stream = stream
        self.secrets = secrets
        self.emitted: list[dict[str, Any]] = []

    def event(self, event: str, level: str = "info", **fields: Any) -> None:
        record = {"level": level, "event": event}
        record.update(fields)
        self.emitted.append(record)
        self.stream.write(json.dumps(record) + "\n")


def run(
    base_url: str,
    *,
    sources: list[str],
    token: str = "",
    database_url: str = "sqlite://",
    window_hours: int = 12,
    run_id: str = "starter-run",
    log_stream: TextIO | None = None,
    out: TextIO | None = None,
) -> int:
    """Run all five stages once and return an exit code.

    EXERCISE 8 — Say partial success out loud.
    This returns 0 whenever it did not crash, so a scheduler cannot tell a clean
    run from one where a source has been dark for a week. Return
    ``EXIT_PARTIAL`` (3) when any source failed permanently or any record was
    rejected, ``EXIT_FAILURE`` (1) when no source answered at all, and
    ``EXIT_SUCCESS`` (0) only when neither happened.
    Check with: .venv/bin/pytest starter -q -k exit_code
    """
    log_stream = log_stream if log_stream is not None else io.StringIO()
    out = out if out is not None else sys.stdout
    logger = RunLogger(run_id, stream=log_stream, secrets=(token,) if token else ())

    results = [fetch_source(base_url, source, token=token) for source in sources]
    fetched = {result.source: result.records for result in results if result.ok}
    failed = [result.source for result in results if not result.ok]
    for result in results:
        if not result.ok:
            logger.event(
                "ingest.source_failed",
                level="warning",
                source=result.source,
                status=result.status,
                error=result.error,
            )
    logger.event(
        "stage.ingest",
        sources_ok=len(fetched),
        sources_failed=len(failed),
        records_fetched=sum(len(records) for records in fetched.values()),
    )
    if not fetched:
        logger.event("run.end", level="error", status="failure")
        out.write("no source answered\n")
        return EXIT_FAILURE

    try:
        accepted, rejected = validate_all(fetched)
    except ValidationError as error:
        logger.event("run.end", level="error", status="failure", error=str(error).splitlines()[0])
        out.write("aborted at the first bad record\n")
        return EXIT_FAILURE
    logger.event("stage.validate", accepted=len(accepted), rejected=len(rejected))

    engine = create_engine(database_url, future=True)
    Base.metadata.create_all(engine)
    with Session(engine) as session:
        stored = store_readings(session, accepted, run_id=run_id)
        logger.event(
            "stage.store",
            inserted=stored.inserted,
            duplicates_skipped=stored.duplicates,
            total_rows=stored.total_rows,
        )
        summary = build_report(session, window_hours=window_hours, stations=sources)
        logger.event("stage.report", **summary)
        out.write(json.dumps(summary, sort_keys=True) + "\n")
    logger.event("stage.observe", status="success")
    logger.event("run.end", status="success")
    engine.dispose()
    return EXIT_SUCCESS

"""Nine promises, nine tests. One baseline test that passes before you start.

Each skipped test names the exercise in ``stages.py`` that unblocks it. Do the
exercise, delete that test's ``@exercise(...)`` decorator, and run:

    .venv/bin/pytest starter -q
"""

from __future__ import annotations

import io
import json
import os

import pytest

import fixture_server
import stages

#: When the harness runs this suite against the completed reference, the skip
#: marks must not apply — the whole point is to see all ten pass.
SOLVED = bool(os.environ.get("DAY098_SOLUTION"))


def exercise(number: int, what: str):
    """Skip until the learner has done the exercise; never skip for the key."""
    return pytest.mark.skipif(not SOLVED, reason=f"Exercise {number} — {what}")


SOURCES = ["alpha", "bravo", "charlie"]
REPORT_AT = "2026-08-16T12:00:00Z"


def _run(base_url: str, token: str, **kwargs):
    log = io.StringIO()
    out = io.StringIO()
    code = stages.run(
        base_url,
        sources=SOURCES,
        token=token,
        log_stream=log,
        out=out,
        **kwargs,
    )
    return code, log.getvalue(), out.getvalue()


# ---------------------------------------------------------------------------
# Baseline — passes on the untouched skeleton.
# ---------------------------------------------------------------------------
def test_the_skeleton_runs_and_reports_its_own_failure(base_url, token):
    """The starter is not broken. It is finished-looking and wrong.

    It aborts at the first malformed record, which is exercise 3 — and it says
    so rather than crashing, so you can see the shape of the thing before you
    start improving it.
    """
    fixture_server.reset_flaky_counter()
    code, log_text, out_text = _run(base_url, token)
    assert code in (stages.EXIT_FAILURE, stages.EXIT_SUCCESS, stages.EXIT_PARTIAL)
    assert log_text.strip(), "the skeleton must emit at least one log line"
    for line in log_text.splitlines():
        json.loads(line)  # every line is a complete JSON object


# ---------------------------------------------------------------------------
# Stage 1 — Ingest
# ---------------------------------------------------------------------------
@exercise(1, "give fetch_source a timeout and retries")
def test_a_flaky_source_recovers_after_retries(base_url, token):
    fixture_server.reset_flaky_counter()
    result = stages.fetch_source(
        base_url, "bravo", token=token, attempts=3, backoff=0.0, sleep=lambda _s: None
    )
    assert result.ok, "bravo answers on the third attempt; one attempt is not enough"
    assert result.attempts == 3
    assert len(result.records) == 4


@exercise(2, "retry only retryable statuses")
def test_a_wrong_url_is_not_retried(base_url, token):
    result = stages.fetch_source(
        base_url, "delta", token=token, attempts=3, backoff=0.0, sleep=lambda _s: None
    )
    assert not result.ok
    assert result.status == 404
    assert result.attempts == 1, "a 404 will be a 404 next time too"


# ---------------------------------------------------------------------------
# Stage 2 — Validate
# ---------------------------------------------------------------------------
@exercise(3, "collect every bad record instead of raising")
def test_validation_collects_rather_than_aborts(base_url, token):
    fixture_server.reset_flaky_counter()
    fetched = {
        source: stages.fetch_source(
            base_url, source, token=token, attempts=3, backoff=0.0, sleep=lambda _s: None
        ).records
        for source in ("alpha", "bravo")
    }
    accepted, rejected = stages.validate_all(fetched)
    assert len(accepted) + len(rejected) == 9
    assert len(rejected) >= 1
    assert {r.reading_id for r in rejected} >= {"a-3"}
    assert all(r.problems for r in rejected), "a rejection with no reason cannot be fixed"


@exercise(4, "constrain the model so out-of-range values are rejected")
def test_out_of_range_values_are_rejected(base_url, token):
    fixture_server.reset_flaky_counter()
    fetched = {
        source: stages.fetch_source(
            base_url, source, token=token, attempts=3, backoff=0.0, sleep=lambda _s: None
        ).records
        for source in ("alpha", "bravo")
    }
    accepted, rejected = stages.validate_all(fetched)
    assert len(accepted) == 7
    assert {r.reading_id for r in rejected} == {"a-3", "a-5"}
    problems = " ".join(p for r in rejected for p in r.problems)
    assert "humidity_pct" in problems and "temperature_c" in problems


# ---------------------------------------------------------------------------
# Stage 3 — Store
# ---------------------------------------------------------------------------
@exercise(5, "idempotence key, then insert only what is new")
def test_running_twice_stores_once(base_url, token, tmp_path):
    fixture_server.reset_flaky_counter()
    url = f"sqlite:///{tmp_path / 'pipeline.db'}"
    first_code, _, _ = _run(base_url, token, database_url=url, run_id="run-one")
    second_code, second_log, _ = _run(base_url, token, database_url=url, run_id="run-two")
    events = [json.loads(line) for line in second_log.splitlines()]
    store = next(e for e in events if e["event"] == "stage.store")
    assert store["inserted"] == 0, "the second run must store nothing new"
    assert store["duplicates_skipped"] == 7
    assert store["total_rows"] == 6, "six distinct readings, however many times you run"


# ---------------------------------------------------------------------------
# Stage 4 — Report
# ---------------------------------------------------------------------------
@exercise(6, "make the report instant a parameter")
def test_the_report_is_built_at_a_fixed_instant(base_url, token, tmp_path):
    fixture_server.reset_flaky_counter()
    url = f"sqlite:///{tmp_path / 'pipeline.db'}"
    _run(base_url, token, database_url=url)
    from sqlalchemy import create_engine
    from sqlalchemy.orm import Session

    engine = create_engine(url, future=True)
    with Session(engine) as session:
        summary = stages.build_report(
            session, report_at=REPORT_AT, window_hours=12, stations=SOURCES
        )
    engine.dispose()
    assert summary["report_at"] == REPORT_AT
    assert summary["window_start"] == "2026-08-16T00:00:00Z"
    assert summary["readings_in_window"] == 5
    assert summary["stations"] == {"alpha": 2, "bravo": 3, "charlie": 0}


# ---------------------------------------------------------------------------
# Stage 5 — Observe
# ---------------------------------------------------------------------------
@exercise(7, "carry the run id on every log line")
def test_every_log_line_carries_the_run_id(base_url, token, tmp_path):
    fixture_server.reset_flaky_counter()
    url = f"sqlite:///{tmp_path / 'pipeline.db'}"
    _, log_text, _ = _run(base_url, token, database_url=url, run_id="run-abc123")
    events = [json.loads(line) for line in log_text.splitlines()]
    assert events, "a run with no log is a run you cannot investigate"
    assert {e.get("run_id") for e in events} == {"run-abc123"}
    stage_events = [e["event"] for e in events if e["event"].startswith("stage.")]
    assert stage_events == [
        "stage.ingest",
        "stage.validate",
        "stage.store",
        "stage.report",
        "stage.observe",
    ]


@exercise(8, "return 3 for partial success")
def test_partial_success_gets_its_own_exit_code(base_url, token, tmp_path):
    fixture_server.reset_flaky_counter()
    url = f"sqlite:///{tmp_path / 'pipeline.db'}"
    code, _, _ = _run(base_url, token, database_url=url)
    assert code == stages.EXIT_PARTIAL, (
        "charlie failed permanently and two records were rejected: that is not success"
    )


@exercise(9, "redact secrets inside the logger")
def test_no_secret_reaches_the_log(base_url, token, tmp_path):
    fixture_server.reset_flaky_counter()
    url = f"sqlite:///{tmp_path / 'pipeline.db'}"
    _, log_text, out_text = _run(base_url, token, database_url=url)
    assert token not in log_text, "charlie's error body echoed the token straight into the log"
    assert token not in out_text
    assert "***redacted***" in log_text, "redaction must leave a visible mark, not a silent gap"

#!/usr/bin/env bash
# Tests for the Day 098 lab. Run from the lab directory:
#   bash tests/run_tests.sh
#
# This harness checks the five promises the pipeline makes, and it checks each
# one by observing behaviour rather than by reading the source:
#
#   * INGEST     a source that fails twice recovers on the third attempt; a 404
#                is attempted exactly once; a permanently failing source does
#                not take the run down with it.
#   * VALIDATE   two deliberately bad records are rejected, counted and
#                explained, and the seven good ones still get through — and the
#                record that is valid but wrong passes the gate, because no
#                field-level rule can see it.
#   * STORE      the pipeline is run twice against one database and the second
#                run inserts nothing. The UNIQUE constraint is then attacked
#                directly, to prove the guarantee is the database's and not the
#                application's good intentions.
#   * REPORT     built at a fixed instant, so its numbers are assertable, and
#                the implausible jump is flagged rather than dropped.
#   * OBSERVE    exactly one structured log line per stage, every line carrying
#                the same run id, configuration provenance printed correctly,
#                the API token absent from every byte of the log, and an exit
#                code of 3 for partial success.
#
# Plus: the starter skeleton runs, the nine exercises are proved achievable by
# running the same suite against the completed reference, the captures still
# match a live run, and the lab leaves nothing behind.
#
# Everything runs offline against a fixture server bound to 127.0.0.1 on a port
# the kernel chooses. Deterministic, non-interactive, exits 0 only if every
# check passes.
set -u

export PYTHONDONTWRITEBYTECODE=1

lab_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"

# Bytecode left by an EARLIER command is not this run's litter. The README
# documents `pytest starter -q`, and running it writes .pyc files that would
# then fail the cleanliness check at the end of this script -- failing the
# reader for following the instructions. Clearing them here makes that final
# check measure what it claims to: what THIS run left behind. `.venv` is
# untouched, because the packages' own bytecode is theirs, not ours.
find "${lab_dir}" -name '.venv' -prune -o -type d -name '__pycache__' -exec rm -rf {} + 2>/dev/null || true
find "${lab_dir}" -name '.venv' -prune -o -type d -name '.pytest_cache' -exec rm -rf {} + 2>/dev/null || true

failures=0
checks=0

check() {
  local label="$1" ok="$2"
  checks=$((checks + 1))
  if [ "${ok}" = "yes" ]; then
    echo "  ok: ${label}"
  else
    echo "  FAIL: ${label}"
    failures=$((failures + 1))
  fi
}

check_eq() {
  local label="$1" want="$2" got="$3"
  checks=$((checks + 1))
  if [ "${want}" = "${got}" ]; then
    echo "  ok: ${label}"
  else
    echo "  FAIL: ${label}"
    echo "        expected: ${want}"
    echo "        actual  : ${got}"
    failures=$((failures + 1))
  fi
}

check_grep() {
  local label="$1" file="$2" pattern="$3"
  checks=$((checks + 1))
  if grep -qE "${pattern}" "${file}"; then
    echo "  ok: ${label}"
  else
    echo "  FAIL: ${label}"
    echo "        no line in $(basename "${file}") matched: ${pattern}"
    failures=$((failures + 1))
  fi
}

check_absent() {
  local label="$1" file="$2" pattern="$3"
  checks=$((checks + 1))
  if grep -qF "${pattern}" "${file}"; then
    echo "  FAIL: ${label}"
    echo "        found in $(basename "${file}"): ${pattern}"
    failures=$((failures + 1))
  else
    echo "  ok: ${label}"
  fi
}

# Resolve a tool: an explicit override, then this lab's .venv, then PATH.
# Fails loudly with install instructions rather than skipping quietly.
resolve_tool() {
  local tool="$1" override="$2"
  if [ -n "${override}" ] && [ -x "${override}" ]; then echo "${override}"; return 0; fi
  if [ -x "${lab_dir}/.venv/bin/${tool}" ]; then echo "${lab_dir}/.venv/bin/${tool}"; return 0; fi
  if command -v "${tool}" >/dev/null 2>&1; then command -v "${tool}"; return 0; fi
  return 1
}

install_hint() {
  echo "  Install the pinned dependencies with:" >&2
  echo "    cd ${lab_dir}" >&2
  echo "    python3 -m venv .venv" >&2
  echo "    .venv/bin/pip install -r requirements/requirements.txt" >&2
  echo "  Or point this suite at an existing interpreter:" >&2
  echo "    PYTHON=/path/to/python3 PYTEST=/path/to/pytest bash tests/run_tests.sh" >&2
}

python_bin="$(resolve_tool python3 "${PYTHON:-}")" || {
  echo "FAIL: python3 not found." >&2
  install_hint
  exit 1
}

pytest_bin="$(resolve_tool pytest "${PYTEST:-}")" || {
  echo "FAIL: pytest not found." >&2
  install_hint
  exit 1
}

for module in sqlalchemy pydantic; do
  if ! "${python_bin}" -c "import ${module}" >/dev/null 2>&1; then
    echo "FAIL: ${module} is not importable from ${python_bin}." >&2
    echo "  This lab is a pipeline built on it, so there is nothing to fall back to." >&2
    install_hint
    exit 1
  fi
done

work="$(mktemp -d)"
server_pid=""
cleanup() {
  if [ -n "${server_pid}" ]; then
    kill "${server_pid}" >/dev/null 2>&1
    # Reap it quietly: without the wait, bash prints its own "Terminated" line
    # to stderr after this script has already reported its result.
    wait "${server_pid}" >/dev/null 2>&1
  fi
  rm -rf "${work}"
}
trap cleanup EXIT

export PYTHONPATH="${lab_dir}/examples"
TOKEN="demo-token-value"

echo "Day 098 — Section Project: A Complete Data Pipeline"
echo

# ---------------------------------------------------------------------------
echo "1. Environment — the versions actually in use"
# ---------------------------------------------------------------------------
"${python_bin}" - > "${work}/versions.txt" <<'PY'
import platform
import sqlite3

import pydantic
import sqlalchemy

print(f"python {platform.python_version()}")
print(f"sqlalchemy {sqlalchemy.__version__}")
print(f"pydantic {pydantic.VERSION}")
print(f"sqlite {sqlite3.sqlite_version}")
PY
sed 's/^/    /' "${work}/versions.txt"

pinned_sqla="$(grep -iE '^SQLAlchemy==' "${lab_dir}/requirements/requirements.txt" | cut -d= -f3)"
pinned_pyd="$(grep -iE '^pydantic==' "${lab_dir}/requirements/requirements.txt" | cut -d= -f3)"
check_eq "the installed SQLAlchemy is the version requirements.txt pins" \
  "${pinned_sqla}" "$(awk '/^sqlalchemy /{print $2}' "${work}/versions.txt")"
check_eq "the installed pydantic is the version requirements.txt pins" \
  "${pinned_pyd}" "$(awk '/^pydantic /{print $2}' "${work}/versions.txt")"

# The lesson states plainly that no orchestrator is exercised here. Prove the
# claim is still true rather than letting the text go quietly stale.
for orchestrator in airflow dagster prefect dbt; do
  check "the lesson's claim that ${orchestrator} is not installed here holds" \
    "$("${python_bin}" -c "import ${orchestrator}" >/dev/null 2>&1 && echo no || echo yes)"
done

# ---------------------------------------------------------------------------
echo
echo "2. The whole pipeline, twice — the demo run"
# ---------------------------------------------------------------------------
"${python_bin}" "${lab_dir}/examples/demo_run.py" > "${work}/demo.txt" 2>&1
demo_status=$?
check_eq "demo_run.py exits 0" "0" "${demo_status}"

check_grep "run 1 fetched nine records from the two sources that answered" \
  "${work}/demo.txt" '"event": "stage.ingest", "sources_ok": 2, "sources_failed": 1, "failed_sources": \["charlie"\], "records_fetched": 9, "attempts_total": 7'
check_grep "bravo recovered on its third attempt" \
  "${work}/demo.txt" '"event": "ingest.source_recovered", "source": "bravo", "attempts": 3'
check_grep "charlie failed permanently after three attempts" \
  "${work}/demo.txt" '"event": "ingest.source_failed", "source": "charlie", "attempts": 3, "status": 500'
check_grep "the validation gate accepted 7 and rejected 2, and said why" \
  "${work}/demo.txt" '"event": "stage.validate", "records_in": 9, "accepted": 7, "rejected": 2, "reasons": \{"humidity_pct": 1, "temperature_c": 1\}'
check_grep "run 1 stored 6 rows and skipped the in-payload duplicate" \
  "${work}/demo.txt" '"event": "stage.store", "considered": 7, "inserted": 6, "duplicates_skipped": 1, "total_rows": 6'
check_grep "run 2 stored NOTHING — the idempotence key held" \
  "${work}/demo.txt" '"event": "stage.store", "considered": 7, "inserted": 0, "duplicates_skipped": 7, "total_rows": 6'
check_grep "both runs exited 3, not 0 — one source is dark" \
  "${work}/demo.txt" '"event": "run.end", "status": "partial_success", "exit_code": 3, "stored_total": 6'
check_grep "the two runs produced identical reports" \
  "${work}/demo.txt" '^  reports identical    True$'
check_grep "and identical exit codes" \
  "${work}/demo.txt" '^  exit codes identical True$'
check_grep "bravo was worth three attempts" \
  "${work}/demo.txt" '^  bravo    attempts=3  ok=True  status=200'
check_grep "delta was worth exactly one — a 404 will be a 404 next time" \
  "${work}/demo.txt" '^  delta    attempts=1  ok=False  status=404'
check_grep "the fixture server really did echo the token back in an error body" \
  "${work}/demo.txt" 'raw error body from charlie : upstream credentials rejected for token demo-token-value'
check_grep "and the redactor caught it" \
  "${work}/demo.txt" 'after the log redactor      : upstream credentials rejected for token \*\*\*redacted\*\*\*'
check_grep "exactly five stage summaries, in pipeline order" \
  "${work}/demo.txt" '^  stage summaries    : 5 -> stage\.ingest, stage\.validate, stage\.store, stage\.report, stage\.observe$'
check_grep "the demo removed its temporary database" \
  "${work}/demo.txt" '^temporary database removed: True$'
check_absent "no absolute home path leaked into the demo output" \
  "${work}/demo.txt" "/Users/"

# ---------------------------------------------------------------------------
echo
echo "3. The report — fixed instant, exact values, and the record that is"
echo "   valid but wrong"
# ---------------------------------------------------------------------------
check_grep "alpha: 2 readings, 18.4 to 19.0, mean 18.7" \
  "${work}/demo.txt" '^    alpha               2      18\.4      19\.0      18\.7$'
check_grep "bravo: 3 readings, 13.6 to 41.3, mean 23.3" \
  "${work}/demo.txt" '^    bravo               3      13\.6      41\.3      23\.3$'
check_grep "charlie: 0 readings, and it is REPORTED rather than omitted" \
  "${work}/demo.txt" '^    charlie             0         -         -         -$'
check_grep "5 of the 6 stored readings fall inside the 12-hour window" \
  "${work}/demo.txt" '^    in window    5 of 6 stored readings$'
check_grep "the implausible jump is flagged, not deleted" \
  "${work}/demo.txt" '^    bravo: \+26\.3 C in 5 minutes \(2026-08-16T11:45:00Z -> 2026-08-16T11:50:00Z\)$'

# The window boundary is a decision, not an accident: b-1 at 23:30 the previous
# day is stored and deliberately outside a 12-hour window ending at noon.
"${python_bin}" - > "${work}/window.txt" <<'PY'
from datetime import datetime, timedelta, timezone

end = datetime(2026, 8, 16, 12, 0, tzinfo=timezone.utc)
for hours in (12, 24):
    start = end - timedelta(hours=hours)
    stored = {
        "a-1": "2026-08-16T09:00:00Z",
        "a-2": "2026-08-16T10:00:00Z",
        "b-1": "2026-08-15T23:30:00Z",
        "b-2": "2026-08-16T08:15:00Z",
        "b-3": "2026-08-16T11:45:00Z",
        "b-4": "2026-08-16T11:50:00Z",
    }
    inside = [
        key
        for key, text in stored.items()
        if start <= datetime.fromisoformat(text.replace("Z", "+00:00")) <= end
    ]
    print(f"WINDOW {hours} {len(inside)} {' '.join(sorted(inside))}")
PY
check_grep "a 12-hour window holds 5 readings" "${work}/window.txt" '^WINDOW 12 5 a-1 a-2 b-2 b-3 b-4$'
check_grep "a 24-hour window holds all 6 — the window is the parameter" \
  "${work}/window.txt" '^WINDOW 24 6 a-1 a-2 b-1 b-2 b-3 b-4$'

# ---------------------------------------------------------------------------
echo
echo "4. Idempotence, attacked directly"
# ---------------------------------------------------------------------------
"${python_bin}" - > "${work}/idempotence.txt" 2>&1 <<'PY'
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from store import StoredReading, build_engine, store_readings
from validate import Reading

raw = {
    "station_id": "alpha",
    "reading_id": "a-1",
    "observed_at": "2026-08-16T09:00:00Z",
    "temperature_c": 18.4,
    "humidity_pct": 61,
}
reading = Reading.model_validate(raw)
engine = build_engine("sqlite://")
with Session(engine) as session:
    first = store_readings(session, [reading], run_id="r1")
    second = store_readings(session, [reading], run_id="r2")
    third = store_readings(session, [reading, reading], run_id="r3")
    print(f"FIRST inserted={first.inserted} total={first.total_rows}")
    print(f"SECOND inserted={second.inserted} duplicates={second.duplicates} total={second.total_rows}")
    print(f"BATCH inserted={third.inserted} duplicates={third.duplicates} total={third.total_rows}")

    # Now go around the application entirely and ask the database to break it.
    try:
        session.add(
            StoredReading(
                station_id="alpha",
                reading_id="a-1",
                observed_at="2026-08-16T09:00:00Z",
                temperature_dc=184,
                humidity_pct=61,
                ingested_by_run="rogue",
            )
        )
        session.commit()
        print("CONSTRAINT enforced=False")
    except IntegrityError as exc:
        session.rollback()
        first = str(exc).splitlines()[0]
        print(f"CONSTRAINT enforced=True says={first.split(') ', 1)[1]}")

    # And prove the CHECK constraints are real too.
    try:
        session.execute(
            StoredReading.__table__.insert().values(
                station_id="alpha",
                reading_id="rogue-1",
                observed_at="2026-08-16T09:00:00Z",
                temperature_dc=184,
                humidity_pct=155,
                ingested_by_run="rogue",
            )
        )
        session.commit()
        print("CHECK humidity_enforced=False")
    except IntegrityError:
        session.rollback()
        print("CHECK humidity_enforced=True")
engine.dispose()
PY
check_grep "the first store inserts the row" "${work}/idempotence.txt" '^FIRST inserted=1 total=1$'
check_grep "the second store inserts nothing and says so" \
  "${work}/idempotence.txt" '^SECOND inserted=0 duplicates=1 total=1$'
check_grep "a duplicate INSIDE one batch is caught too" \
  "${work}/idempotence.txt" '^BATCH inserted=0 duplicates=2 total=1$'
check_grep "the UNIQUE constraint refuses a write that bypasses the application" \
  "${work}/idempotence.txt" '^CONSTRAINT enforced=True says=UNIQUE constraint failed: readings\.station_id, readings\.reading_id$'
check_grep "and the humidity CHECK constraint refuses an impossible percentage" \
  "${work}/idempotence.txt" '^CHECK humidity_enforced=True$'

# ---------------------------------------------------------------------------
echo
echo "5. Validation — collected, counted, explained; and what it cannot see"
# ---------------------------------------------------------------------------
"${python_bin}" - > "${work}/validation.txt" <<'PY'
from validate import validate_all

records = [
    {"station_id": "x", "reading_id": "ok", "observed_at": "2026-08-16T09:00:00Z",
     "temperature_c": 18.4, "humidity_pct": 61},
    {"station_id": "x", "reading_id": "bad-temp", "observed_at": "2026-08-16T09:00:00Z",
     "temperature_c": "warm", "humidity_pct": 61},
    {"station_id": "x", "reading_id": "bad-hum", "observed_at": "2026-08-16T09:00:00Z",
     "temperature_c": 18.4, "humidity_pct": 155},
    {"station_id": "x", "reading_id": "naive", "observed_at": "2026-08-16T09:00:00",
     "temperature_c": 18.4, "humidity_pct": 61},
    {"station_id": "x", "reading_id": "extra", "observed_at": "2026-08-16T09:00:00Z",
     "temperature_c": 18.4, "humidity_pct": 61, "battery_pct": 90},
    {"station_id": "x", "reading_id": "valid-but-wrong", "observed_at": "2026-08-16T09:05:00Z",
     "temperature_c": 41.3, "humidity_pct": 61},
]
outcome = validate_all({"x": records})
print(f"CONSIDERED {outcome.considered} ACCEPTED {len(outcome.accepted)} REJECTED {len(outcome.rejected)}")
for rejection in outcome.rejected:
    print(f"REJECT {rejection}")
print(f"REASONS {outcome.reasons()}")
print(f"DECI {[r.temperature_dc for r in outcome.accepted]}")
PY
check_grep "six records in, two accepted, four rejected — and the run continued" \
  "${work}/validation.txt" '^CONSIDERED 6 ACCEPTED 2 REJECTED 4$'
check_grep "a non-numeric temperature is named with its field and its reason" \
  "${work}/validation.txt" '^REJECT x\[1\] bad-temp: temperature_c: Input should be a valid number'
check_grep "an out-of-range humidity is named too" \
  "${work}/validation.txt" '^REJECT x\[2\] bad-hum: humidity_pct: Input should be less than or equal to 100$'
check_grep "a timestamp with no offset is refused — it is not an instant" \
  "${work}/validation.txt" '^REJECT x\[3\] naive: observed_at: Value error, observed_at must carry a UTC offset'
check_grep "an unexpected field is refused rather than silently ignored" \
  "${work}/validation.txt" '^REJECT x\[4\] extra: battery_pct: Extra inputs are not permitted$'
check_grep "the reasons are counted per field, so a source owner can be told" \
  "${work}/validation.txt" "^REASONS \{'battery_pct': 1, 'humidity_pct': 1, 'observed_at': 1, 'temperature_c': 1\}$"
check_grep "the record that is valid but WRONG passes the gate — 41.3 C is legal" \
  "${work}/validation.txt" '^DECI \[184, 413\]$'

# ---------------------------------------------------------------------------
echo
echo "6. Configuration — four layers, resolved and printed"
# ---------------------------------------------------------------------------
(cd "${lab_dir}" && PIPELINE_LOG_LEVEL=warning PIPELINE_API_TOKEN="${TOKEN}" \
  "${python_bin}" examples/pipeline.py \
  --config-file examples/pipeline.toml --window-hours 24 --explain-config \
  > "${work}/provenance.txt" 2>&1)
provenance_status=$?
check_eq "--explain-config exits 0" "0" "${provenance_status}"
check_grep "a value nobody set is reported as a default" \
  "${work}/provenance.txt" '^retry_backoff_seconds  0\.05                   default$'
check_grep "a value from the TOML file is attributed to the file" \
  "${work}/provenance.txt" '^timeout_seconds        3\.0                    file$'
check_grep "the environment outranks the file" \
  "${work}/provenance.txt" '^log_level              warning                environment$'
check_grep "and an explicit flag outranks the environment" \
  "${work}/provenance.txt" '^window_hours           24                     command line$'
check_grep "the secret's SOURCE is reported and its VALUE is not" \
  "${work}/provenance.txt" '^api_token              \*\*\*redacted\*\*\*         environment$'
check_absent "the token never appears in the provenance table" \
  "${work}/provenance.txt" "${TOKEN}"

# ---------------------------------------------------------------------------
echo
echo "7. The command-line pipeline against a real server, and its exit code"
# ---------------------------------------------------------------------------
"${python_bin}" "${lab_dir}/examples/fixture_server.py" --token "${TOKEN}" \
  > "${work}/port.txt" 2>"${work}/server.err" &
server_pid=$!
port=""
for _ in 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20; do
  port="$(head -1 "${work}/port.txt" 2>/dev/null)"
  [ -n "${port}" ] && break
  "${python_bin}" -c "import time; time.sleep(0.1)"
done
check "the fixture server started and announced a port" \
  "$([ -n "${port}" ] && echo yes || echo no)"

if [ -z "${port}" ]; then
  echo "  (skipping the live-server section: no port)" >&2
else
  run_dir="${work}/run"
  mkdir -p "${run_dir}"
  (cd "${run_dir}" && PIPELINE_API_TOKEN="${TOKEN}" "${python_bin}" \
    "${lab_dir}/examples/pipeline.py" \
    --base-url "http://127.0.0.1:${port}" \
    --sources alpha,bravo,charlie \
    --report-at 2026-08-16T12:00:00Z \
    --window-hours 12 \
    --run-id run-cli000001 \
    --fixed-clock \
    > "${work}/cli-stdout.txt" 2> "${work}/cli-stderr.txt")
  cli_status=$?
  check_eq "a partially successful run exits 3, not 0 and not 1" "3" "${cli_status}"

  check_grep "the report went to stdout" \
    "${work}/cli-stdout.txt" '^Station readings report$'
  check_grep "with the same numbers the demo produced" \
    "${work}/cli-stdout.txt" '^  in window    5 of 6 stored readings$'
  check_absent "nothing but the report went to stdout" \
    "${work}/cli-stdout.txt" '"event":'

  stage_lines="$(grep -c '"event": "stage\.' "${work}/cli-stderr.txt")"
  check_eq "exactly five stage lines, one per stage" "5" "${stage_lines}"
  run_ids="$("${python_bin}" - "${work}/cli-stderr.txt" <<'PY'
import json
import sys

with open(sys.argv[1], encoding="utf-8") as handle:
    ids = {json.loads(line)["run_id"] for line in handle if line.strip()}
print(len(ids), " ".join(sorted(ids)))
PY
)"
  check_eq "every line carries the one run id" "1 run-cli000001" "${run_ids}"
  check_grep "every line is valid JSON with a timestamp and a level" \
    "${work}/cli-stderr.txt" '^\{"ts": "2026-08-16T12:00:0[0-9]Z", "level": "(info|warning)", "run_id": "run-cli000001"'
  check_absent "the API token appears nowhere in the log" \
    "${work}/cli-stderr.txt" "${TOKEN}"
  check_grep "and the place it would have leaked shows the redaction instead" \
    "${work}/cli-stderr.txt" 'upstream credentials rejected for token \*\*\*redacted\*\*\*'

  # A second run of the same command must change nothing at all.
  (cd "${run_dir}" && PIPELINE_API_TOKEN="${TOKEN}" "${python_bin}" \
    "${lab_dir}/examples/pipeline.py" \
    --base-url "http://127.0.0.1:${port}" \
    --sources alpha,bravo,charlie \
    --report-at 2026-08-16T12:00:00Z \
    --window-hours 12 \
    --run-id run-cli000002 \
    --fixed-clock \
    > "${work}/cli-stdout-2.txt" 2> "${work}/cli-stderr-2.txt")
  second_status=$?
  check_eq "the second run also exits 3" "3" "${second_status}"
  checks=$((checks + 1))
  if diff -q "${work}/cli-stdout.txt" "${work}/cli-stdout-2.txt" >/dev/null 2>&1; then
    echo "  ok: the second run's report is byte-for-byte identical"
  else
    echo "  FAIL: the second run's report differs"
    diff "${work}/cli-stdout.txt" "${work}/cli-stdout-2.txt" | head -8 | sed 's/^/        /'
    failures=$((failures + 1))
  fi
  check_grep "and it inserted nothing, because everything was already held" \
    "${work}/cli-stderr-2.txt" '"event": "stage.store", "considered": 7, "inserted": 0, "duplicates_skipped": 7, "total_rows": 6'

  # The runs table is the record that makes a bad backfill undoable.
  runs="$("${python_bin}" - "${run_dir}/pipeline.db" <<'PY'
import sqlite3
import sys

connection = sqlite3.connect(sys.argv[1])
rows = connection.execute(
    "SELECT run_id, status, records_inserted, records_duplicate FROM runs ORDER BY run_id"
).fetchall()
for row in rows:
    print("RUN", *row)
owners = connection.execute(
    "SELECT ingested_by_run, count(*) FROM readings GROUP BY ingested_by_run ORDER BY 1"
).fetchall()
for row in owners:
    print("OWNER", *row)
connection.close()
PY
)"
  printf '%s\n' "${runs}" > "${work}/runs.txt"
  check_grep "the first run's row records six inserts" \
    "${work}/runs.txt" '^RUN run-cli000001 partial_success 6 1$'
  check_grep "the second run's row records none" \
    "${work}/runs.txt" '^RUN run-cli000002 partial_success 0 7$'
  check_grep "and every stored row still names the run that wrote it" \
    "${work}/runs.txt" '^OWNER run-cli000001 6$'

  # Point it at nothing at all: no source answers, and that is a failure, not
  # a partial success.
  (cd "${run_dir}" && PIPELINE_API_TOKEN="${TOKEN}" "${python_bin}" \
    "${lab_dir}/examples/pipeline.py" \
    --base-url "http://127.0.0.1:${port}" \
    --sources delta \
    --report-at 2026-08-16T12:00:00Z \
    --run-id run-cli000003 \
    --fixed-clock \
    > "${work}/cli-stdout-3.txt" 2> "${work}/cli-stderr-3.txt")
  third_status=$?
  check_eq "a run where no source answers exits 1, not 3" "1" "${third_status}"
  check_grep "and says so at error level" \
    "${work}/cli-stderr-3.txt" '"level": "error", "run_id": "run-cli000003", "event": "run.end", "status": "failure"'
  check_grep "a 404 source is attempted once, not three times" \
    "${work}/cli-stderr-3.txt" '"event": "stage.ingest", "sources_ok": 0, "sources_failed": 1, "failed_sources": \["delta"\], "records_fetched": 0, "attempts_total": 1'
fi

# ---------------------------------------------------------------------------
echo
echo "8. The starter — a skeleton that runs, and nine reachable exercises"
# ---------------------------------------------------------------------------
(cd "${lab_dir}" && "${pytest_bin}" starter -q > "${work}/starter.txt" 2>&1)
starter_status=$?
check_eq "pytest starter exits 0 on the unmodified skeleton" "0" "${starter_status}"
check_grep "one baseline test passes and nine exercises wait" \
  "${work}/starter.txt" '1 passed, 9 skipped'

exercise_markers="$(grep -c 'EXERCISE [0-9]' "${lab_dir}/starter/stages.py")"
check_eq "stages.py carries ten exercise markers (stage 3 has two parts)" "10" "${exercise_markers}"
named_tests="$(grep -c '^@exercise(' "${lab_dir}/starter/test_stages.py")"
check_eq "and each has a test that names it" "9" "${named_tests}"

(cd "${lab_dir}" && DAY098_SOLUTION=1 "${pytest_bin}" starter -q > "${work}/solved.txt" 2>&1)
solved_status=$?
check_eq "the completed reference passes the same suite" "0" "${solved_status}"
check_grep "all ten tests pass against examples/stages_solved.py" \
  "${work}/solved.txt" '10 passed'

# ---------------------------------------------------------------------------
echo
echo "9. Captured output still matches a live run"
# ---------------------------------------------------------------------------
# Compare a capture against a live run. The optional third argument is a sed
# expression applied to BOTH sides first, for the one capture that legitimately
# contains a duration: pytest prints "in 0.64s", and asserting on a stopwatch is
# how a suite becomes flaky on somebody else's machine.
compare() {
  local name="$1" live="$2" normalise="${3:-}"
  local stored="${lab_dir}/expected-output/${name}"
  checks=$((checks + 1))
  if [ ! -f "${stored}" ]; then
    echo "  FAIL: expected-output/${name} is missing"
    failures=$((failures + 1))
  else
    local a="${work}/cmp-stored-${name}" b="${work}/cmp-live-${name}"
    if [ -n "${normalise}" ]; then
      sed "${normalise}" "${stored}" > "${a}"
      sed "${normalise}" "${live}" > "${b}"
    else
      cp "${stored}" "${a}"
      cp "${live}" "${b}"
    fi
    if diff -q "${a}" "${b}" >/dev/null 2>&1; then
      echo "  ok: expected-output/${name} matches this run"
    else
      echo "  FAIL: expected-output/${name} differs from this run"
      diff "${a}" "${b}" | head -12 | sed 's/^/        /'
      failures=$((failures + 1))
    fi
  fi
}
compare "demo.txt" "${work}/demo.txt"
compare "config-provenance.txt" "${work}/provenance.txt"
compare "starter-progress.txt" "${work}/starter.txt" 's/ in [0-9.]*s$/ in <duration>/'
if [ -n "${port}" ]; then
  cat "${work}/cli-stdout.txt" > "${work}/cli-run.txt"
  echo "--- structured log (stderr) ---" >> "${work}/cli-run.txt"
  cat "${work}/cli-stderr.txt" >> "${work}/cli-run.txt"
  compare "cli-run.txt" "${work}/cli-run.txt"
fi

# ---------------------------------------------------------------------------
echo
echo "10. Hygiene — offline, self-contained, leaving nothing behind"
# ---------------------------------------------------------------------------
"${python_bin}" - "${lab_dir}" > "${work}/hygiene.txt" <<'PY'
import re
import sys
from pathlib import Path

root = Path(sys.argv[1])
skip = {".venv", "__pycache__", ".pytest_cache"}
urls, sudo_lines = set(), []
comment = re.compile(r"^\s*(#|--)")
for path in sorted(root.rglob("*")):
    if not path.is_file() or path.suffix not in {".py", ".sh", ".ini", ".toml"}:
        continue
    if skip & set(path.parts):
        continue
    for number, line in enumerate(
        path.read_text(encoding="utf-8", errors="ignore").splitlines(), 1
    ):
        for url in re.findall(r"https?://[^\s\"')]+", line):
            urls.add(url)
        if re.search(r"(^|[;|&(]\s*)sudo\s", line) and not comment.match(line):
            sudo_lines.append(f"{path.name}:{number}")
offsite = sorted(u for u in urls if not u.startswith("http://127.0.0.1"))
print("OFFSITE " + " ".join(offsite))
print("SUDO " + " ".join(sudo_lines))
PY
check_eq "every URL in this lab points at 127.0.0.1 and nowhere else" "OFFSITE" \
  "$(grep '^OFFSITE ' "${work}/hygiene.txt" | sed 's/ *$//')"
check_eq "no line in this lab would actually invoke sudo" "SUDO" \
  "$(grep '^SUDO ' "${work}/hygiene.txt" | sed 's/ *$//')"

check "no captured output leaks an absolute home path" \
  "$(grep -rl '/Users/\|/home/' "${lab_dir}/expected-output" >/dev/null 2>&1 && echo no || echo yes)"
check "this suite created no database file inside the lab directory" \
  "$(find "${lab_dir}" -name '*.db' -not -path '*/.venv/*' | grep -q . && echo no || echo yes)"
check "and left no __pycache__ behind" \
  "$(find "${lab_dir}" -type d -name '__pycache__' -not -path '*/.venv/*' | grep -q . && echo no || echo yes)"
check "the fixture server binds 127.0.0.1 on port 0, never a fixed port" \
  "$(grep -q '("127.0.0.1", 0)' "${lab_dir}/examples/fixture_server.py" && echo yes || echo no)"

echo
echo "${checks} checks, ${failures} failure(s)."
[ "${failures}" -eq 0 ]