Hands-on lab — Day 81: Scheduling and Background Jobs

"""The clock as an injected boundary.

Day 74 named six boundaries a unit test must not cross, and the clock was the
first of them. Today the clock is not incidental — it is the subject. Every
question worth asking about a scheduled job ("has today's run already
happened?", "is this run overdue?", "what instant does 02:30 mean on the day
the clocks change?") is a question about time, and none of it is testable if
the answer comes from `datetime.now()` buried three calls deep.

So nothing in this lab calls `datetime.now()` except `system_clock`, which is
the one adapter at the edge. Everything else takes a `Clock` parameter.
"""

from __future__ import annotations

import datetime as dt
from collections.abc import Callable
from zoneinfo import ZoneInfo

#: A clock is any zero-argument callable that returns an aware datetime.
#: That is the whole interface. It needs no class and no library.
Clock = Callable[[], dt.datetime]

UTC = dt.timezone.utc


def system_clock(tz: str = "UTC") -> Clock:
    """The real clock: the only function in this lab that reads the system time.

    Returns an *aware* datetime. A naive datetime (one with no ``tzinfo``) is
    the root of most scheduling bugs, because it silently means "whatever the
    machine happens to think local time is" — and a cron job's machine often
    disagrees with your laptop.
    """
    zone = UTC if tz.upper() == "UTC" else ZoneInfo(tz)
    return lambda: dt.datetime.now(tz=zone)


def frozen_clock(moment: dt.datetime) -> Clock:
    """A clock stuck at one instant. Five words of code; replaces a library."""
    if moment.tzinfo is None:
        raise ValueError("frozen_clock needs an aware datetime, not a naive one")
    return lambda: moment


def ticking_clock(start: dt.datetime, step: dt.timedelta) -> Clock:
    """A clock that advances by ``step`` every time it is read.

    Useful for measuring a duration in a test without any duration passing:
    the runner reads the clock once before the work and once after, so a
    one-second step makes every job take exactly one second, deterministically.
    """
    if start.tzinfo is None:
        raise ValueError("ticking_clock needs an aware datetime, not a naive one")
    state = {"now": start}

    def read() -> dt.datetime:
        current = state["now"]
        state["now"] = current + step
        return current

    return read


class FakeTime:
    """A stand-in for ``time.monotonic`` and ``time.sleep`` together.

    ``sched.scheduler`` takes its time source and its delay function as
    constructor arguments precisely so that they can be replaced. Handing it a
    ``FakeTime`` makes a scheduler that "waits" six hours in microseconds,
    which is how you test a schedule instead of watching one.
    """

    def __init__(self, start: float = 0.0) -> None:
        self.now = start
        self.slept: list[float] = []

    def time(self) -> float:
        return self.now

    def sleep(self, seconds: float) -> None:
        self.slept.append(seconds)
        self.now += seconds

    @property
    def total_slept(self) -> float:
        return sum(self.slept)

"""Make this directory importable so the tests can `import runner` directly.

Without this, `pytest examples` collects the test files but cannot import the
modules beside them, because the lab has no package layout — deliberately, so
that every file here can also be run with `python3 examples/<name>.py`.
"""

from __future__ import annotations

import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent
if str(HERE) not in sys.path:
    sys.path.insert(0, str(HERE))

"""A small, honest cron-expression parser.

Five fields, separated by whitespace:

    minute  hour  day-of-month  month  day-of-week
    0-59    0-23  1-31          1-12   0-6 (0 = Sunday, 7 also accepted)

Each field is ``*``, a number, a comma-separated list, an ``a-b`` range, or a
step written ``*/n`` or ``a-b/n``. This module implements exactly that, and
nothing it does not implement is silently accepted: an unparseable field
raises ``CronError`` rather than matching everything, because a schedule that
quietly means "every minute" is a bad way to find out you made a typo.

The one rule everybody gets wrong is in :func:`matches`. When BOTH the
day-of-month and the day-of-week fields are restricted, cron ORs them: the job
runs on days matching either. Everywhere else the fields are ANDed. This
module reproduces that rule and the test suite pins it.

Real cron implementations differ in extensions (``@daily``, ``L``, ``W``,
names like ``MON``, seconds fields in some libraries). This parser handles the
classic five numeric fields only, and says so rather than pretending.
"""

from __future__ import annotations

import datetime as dt
from dataclasses import dataclass


class CronError(ValueError):
    """Raised when an expression cannot be parsed."""


FIELD_RANGES: tuple[tuple[str, int, int], ...] = (
    ("minute", 0, 59),
    ("hour", 0, 23),
    ("day-of-month", 1, 31),
    ("month", 1, 12),
    ("day-of-week", 0, 7),
)


def _parse_field(spec: str, name: str, low: int, high: int) -> tuple[frozenset[int], bool]:
    """Return the set of matching values and whether the field is restricted."""
    restricted = spec.strip() != "*"
    values: set[int] = set()
    for part in spec.split(","):
        part = part.strip()
        if not part:
            raise CronError(f"{name}: empty item in {spec!r}")
        step = 1
        if "/" in part:
            part, _, step_text = part.partition("/")
            try:
                step = int(step_text)
            except ValueError:
                raise CronError(f"{name}: step {step_text!r} is not a number") from None
            if step < 1:
                raise CronError(f"{name}: step must be 1 or more, got {step}")
        if part == "*":
            start, end = low, high
        elif "-" in part.lstrip("-"):
            start_text, _, end_text = part.partition("-")
            start, end = _int(start_text, name), _int(end_text, name)
        else:
            start = end = _int(part, name)
        if start > end:
            raise CronError(f"{name}: range {part!r} runs backwards")
        if start < low or end > high:
            raise CronError(f"{name}: {part!r} is outside {low}-{high}")
        values.update(range(start, end + 1, step))
    if name == "day-of-week" and 7 in values:
        # Both 0 and 7 mean Sunday. Normalise so matching has one answer.
        values.discard(7)
        values.add(0)
    return frozenset(values), restricted


def _int(text: str, name: str) -> int:
    try:
        return int(text)
    except ValueError:
        raise CronError(f"{name}: {text!r} is not a number") from None


@dataclass(frozen=True)
class CronSchedule:
    """A parsed five-field cron expression."""

    expression: str
    minutes: frozenset[int]
    hours: frozenset[int]
    days_of_month: frozenset[int]
    months: frozenset[int]
    days_of_week: frozenset[int]
    dom_restricted: bool
    dow_restricted: bool

    def matches(self, moment: dt.datetime) -> bool:
        """Would cron fire at this minute?

        Seconds are ignored: cron's resolution is one minute.
        """
        if moment.minute not in self.minutes:
            return False
        if moment.hour not in self.hours:
            return False
        if moment.month not in self.months:
            return False
        # Python: Monday is 0 and Sunday is 6. Cron: Sunday is 0.
        dow = (moment.weekday() + 1) % 7
        dom_hit = moment.day in self.days_of_month
        dow_hit = dow in self.days_of_week
        if self.dom_restricted and self.dow_restricted:
            return dom_hit or dow_hit  # the OR-not-AND rule
        return dom_hit and dow_hit

    def next_run_after(self, moment: dt.datetime, *, horizon_days: int = 400) -> dt.datetime:
        """The first matching minute strictly after ``moment``.

        Walks forward a minute at a time. That is not clever, but for a
        schedule that fires at least once a year it is fast enough (a daily
        job is found within 1440 steps) and it is obviously correct, which
        matters more here than speed.
        """
        candidate = moment.replace(second=0, microsecond=0) + dt.timedelta(minutes=1)
        limit = candidate + dt.timedelta(days=horizon_days)
        while candidate < limit:
            if self.matches(candidate):
                return candidate
            candidate += dt.timedelta(minutes=1)
        raise CronError(
            f"{self.expression!r} has no run within {horizon_days} days of {moment.isoformat()}"
        )

    def describe(self) -> str:
        """A plain-English summary, so a generated line can be read back."""
        parts = [
            f"minute {_summarise(self.minutes)}",
            f"hour {_summarise(self.hours)}",
            f"day-of-month {'any' if not self.dom_restricted else _summarise(self.days_of_month)}",
            f"month {'any' if self.months == frozenset(range(1, 13)) else _summarise(self.months)}",
            f"day-of-week {'any' if not self.dow_restricted else _summarise(self.days_of_week)}",
        ]
        note = ""
        if self.dom_restricted and self.dow_restricted:
            note = "  (day-of-month OR day-of-week — cron ORs these two when both are set)"
        return " · ".join(parts) + note


def _summarise(values: frozenset[int]) -> str:
    ordered = sorted(values)
    if len(ordered) > 6:
        return f"{ordered[0]}..{ordered[-1]} ({len(ordered)} values)"
    return ",".join(str(v) for v in ordered)


def parse(expression: str) -> CronSchedule:
    """Parse a five-field cron expression, or raise :class:`CronError`."""
    fields = expression.split()
    if len(fields) != 5:
        raise CronError(
            f"expected 5 fields (minute hour day-of-month month day-of-week), "
            f"got {len(fields)} in {expression!r}"
        )
    parsed = [
        _parse_field(field, name, low, high)
        for field, (name, low, high) in zip(fields, FIELD_RANGES, strict=True)
    ]
    return CronSchedule(
        expression=expression,
        minutes=parsed[0][0],
        hours=parsed[1][0],
        days_of_month=parsed[2][0],
        months=parsed[3][0],
        days_of_week=parsed[4][0],
        dom_restricted=parsed[2][1],
        dow_restricted=parsed[4][1],
    )

date,station,celsius
2026-07-17,ALPHA,17.4
2026-07-17,ALPHA,19.1
2026-07-17,BRAVO,21.8
2026-07-17,BRAVO,22.6
2026-07-18,ALPHA,18.2
2026-07-18,ALPHA,20.9
2026-07-18,ALPHA,23.5
2026-07-18,BRAVO,24.1
2026-07-18,BRAVO,25.0
2026-07-19,ALPHA,16.8
2026-07-19,ALPHA,18.0
2026-07-19,ALPHA,19.6
2026-07-19,BRAVO,20.2
2026-07-19,BRAVO,26.4
2026-07-19,CHARLIE,31.7
2026-07-20,ALPHA,15.5
2026-07-20,BRAVO,27.3

"""The whole day in one run, in eight parts and under a second.

    python3 examples/demo.py

Nothing here waits, nothing here schedules anything, and nothing here is left
running afterwards. Every duration you see was computed against an injected
clock.
"""

from __future__ import annotations

import datetime as dt
import json
import tempfile
from pathlib import Path

from clock import frozen_clock, ticking_clock
from cronexpr import parse
from gen_schedules import JobSchedule
from inprocess import deadline_corrected_loop, naive_sleep_loop, run_sched_schedule
from joblock import AlreadyRunning, job_lock
from reportjob import generate_daily_report
from runner import EXIT_MEANINGS, jsonl_logger, run_job
from timezones import classify_wall_time, daily_instants_local, gaps_between
from watchdog import check_heartbeat

UTC = dt.timezone.utc
HERE = Path(__file__).resolve().parent
NOW = dt.datetime(2026, 7, 20, 2, 30, tzinfo=UTC)
DAY = dt.date(2026, 7, 19)


def heading(number: int, text: str) -> None:
    print()
    print(f"{number}. {text}")
    print("=" * (len(text) + 3))


def main() -> int:
    heading(1, "A sleep loop drifts, and the drift never stops growing")
    naive = naive_sleep_loop(runs=100, interval=60, work_seconds=5)
    fixed = deadline_corrected_loop(runs=100, interval=60, work_seconds=5)
    print("  work(); sleep(60) with 5s of work")
    print(f"      gap between runs : {naive.starts[1] - naive.starts[0]:.0f}s (you wrote 60)")
    print(f"      run 100 is late by: {naive.final_drift:.0f}s")
    print("  sleep until the next deadline instead")
    print(f"      gap between runs : {fixed.starts[1] - fixed.starts[0]:.0f}s")
    print(f"      run 100 is late by: {fixed.final_drift:.0f}s")

    heading(2, "sched with an injected clock: six hours of schedule, no waiting")
    fired, fake = run_sched_schedule(delays=[3600, 7200, 21600])
    for when, index in fired:
        print(f"      event {index} fired at t+{when:.0f}s")
    print(f"      total time 'waited': {fake.total_slept:.0f}s of fake time, 0s of real time")

    with tempfile.TemporaryDirectory(prefix="day081-demo.") as workdir:
        out = Path(workdir)

        heading(3, "Idempotence: run it twice, get one report")
        for attempt in (1, 2):
            status, path = generate_daily_report(
                source=HERE / "data" / "readings.csv",
                output_dir=out,
                report_date=DAY,
                generated_at=NOW,
            )
            print(f"      run {attempt}: {status:8s} -> {path.name}")
        files = sorted(p.name for p in out.glob("report-*.json"))
        print(f"      files on disk: {files}")
        payload = json.loads((out / f"report-{DAY.isoformat()}.json").read_text())
        print(f"      readings summarised: {payload['reading_count']}")

        heading(4, "The lock: a second run refuses rather than doubling the work")
        lock = out / "daily-report.lock"
        did_work: list[str] = []
        with job_lock(lock):
            run = run_job(
                name="daily-report",
                work=lambda: did_work.append("worked") or {},
                clock=frozen_clock(NOW),
                lock_path=lock,
                log=lambda event: None,
            )
        print(f"      status   : {run.status}")
        print(f"      exit code: {run.exit_code}  ({EXIT_MEANINGS[run.exit_code]})")
        print(f"      work done: {bool(did_work)}")
        try:
            with job_lock(lock):
                print("      lock is free again once the holder finished: yes")
        except AlreadyRunning:
            print("      lock is free again once the holder finished: no")

        heading(5, "A hung job hits its timeout instead of blocking every later run")
        import time

        started = time.monotonic()
        hung = run_job(
            name="daily-report",
            work=lambda: time.sleep(30),
            clock=ticking_clock(NOW, dt.timedelta(seconds=30)),
            lock_path=out / "hang.lock",
            log=jsonl_logger(out / "job.log"),
            timeout_seconds=0.2,
        )
        print("      the work asked for 30s, the budget was 0.2s")
        print(f"      status   : {hung.status}")
        print(f"      exit code: {hung.exit_code}  ({EXIT_MEANINGS[hung.exit_code]})")
        print(f"      real time spent: {time.monotonic() - started:.2f}s")
        print("      log line:")
        print("        " + (out / "job.log").read_text().strip())

        heading(6, "The dead man's switch: alerting on silence, not on failure")
        heartbeat = out / "heartbeat.json"
        heartbeat.write_text(
            json.dumps({"job": "daily-report", "last_success": (NOW - dt.timedelta(hours=6)).isoformat()})
        )
        for label, moment in (
            ("six hours ago", NOW),
            ("two days later, nothing has run", NOW + dt.timedelta(days=2)),
        ):
            verdict = check_heartbeat(
                heartbeat_path=heartbeat,
                clock=frozen_clock(moment),
                max_age=dt.timedelta(hours=26),
            )
            print(f"      {label}:")
            print(f"        {verdict.state.upper()}: {verdict.message}")

    heading(7, "Time zones: the two mornings a local schedule is wrong")
    for naive_time in (dt.datetime(2026, 3, 8, 2, 30), dt.datetime(2026, 11, 1, 1, 30)):
        verdict = classify_wall_time(naive_time, "America/New_York")
        print(f"      {verdict.wall_time} America/New_York -> {verdict.kind}")
        print(f"        {verdict.note}")
    spring = gaps_between(
        daily_instants_local(
            start_date=dt.date(2026, 3, 6), days=5, hour=12, minute=0,
            zone_name="America/New_York",
        )
    )
    utc_gaps = gaps_between(
        daily_instants_local(
            start_date=dt.date(2026, 3, 6), days=5, hour=12, minute=0, zone_name="UTC"
        )
    )
    print(f"      hours between daily 12:00 runs, local: {spring}")
    print(f"      hours between daily 12:00 runs, UTC  : {utc_gaps}")

    heading(8, "One schedule, three dialects — generated, not installed")
    schedule = JobSchedule(label="com.example.dailyreport", minute=30, hour=2)
    print("      cron    : " + schedule.cron_expression)
    print("      launchd : StartCalendarInterval Hour=2 Minute=30")
    print("      systemd : OnCalendar=" + schedule.on_calendar)
    print("      reads as: " + parse(schedule.cron_expression).describe())
    print()
    print("      Nothing was scheduled. No crontab, no launchd job, no systemd timer.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""Generate schedule files for cron, launchd and systemd — and install nothing.

This script writes text files into a directory you name. That is all it does.
It never runs ``crontab``, never runs ``launchctl``, never runs ``systemctl``,
and never touches anything under your home directory unless you point
``--out`` there yourself. The install commands are *printed* so you can read
them; running them is your decision, made deliberately, on a machine you
intend to schedule something on.

One schedule definition produces all three formats, which is the point: the
cron five-field expression, the launchd ``StartCalendarInterval`` dictionary
and the systemd ``OnCalendar`` string all say the same thing in three
dialects, and seeing them side by side is the fastest way to learn any of them.

    python3 examples/gen_schedules.py --out examples/schedules \\
        --hour 2 --minute 30 --project-dir /opt/reports
"""

from __future__ import annotations

import argparse
import datetime as dt
from dataclasses import dataclass
from pathlib import Path
from xml.sax.saxutils import escape

from cronexpr import parse

WEEKDAY_NAMES = ("Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat")
SYSTEMD_DAYS = ("Sun", "Mon", "Tue", "Wed", "Thu", "Fri", "Sat")


@dataclass(frozen=True)
class JobSchedule:
    """One schedule, expressible in three dialects."""

    label: str
    minute: int
    hour: int
    days_of_week: tuple[int, ...] = ()  # empty means every day; 0 = Sunday
    project_dir: str = "/opt/reports"
    python: str = "/usr/bin/python3"
    script: str = "job.py"
    log_dir: str = "/var/log/reports"
    timezone: str = "UTC"

    # ---- the command every dialect runs -------------------------------

    @property
    def command(self) -> str:
        return f"{self.python} {self.project_dir}/{self.script} --output-dir {self.project_dir}/out"

    @property
    def argv(self) -> tuple[str, ...]:
        return (
            self.python,
            f"{self.project_dir}/{self.script}",
            "--output-dir",
            f"{self.project_dir}/out",
        )

    # ---- cron ---------------------------------------------------------

    @property
    def cron_expression(self) -> str:
        dow = ",".join(str(d) for d in self.days_of_week) if self.days_of_week else "*"
        return f"{self.minute} {self.hour} * * {dow}"

    def cron_line(self) -> str:
        """The crontab entry, with the environment a cron job does NOT inherit."""
        stdout = f"{self.log_dir}/{self.label}.log"
        return "\n".join(
            [
                "# Generated schedule. Read it, then decide whether to install it.",
                "# Install with:  crontab -l > my.cron ; cat this-file >> my.cron ; crontab my.cron",
                "# List with:     crontab -l          Edit with:  crontab -e",
                "#",
                "# cron gives a job almost no environment: a short PATH, no shell",
                "# profile, HOME set but nothing sourced, and the home directory as",
                "# the working directory. Everything the job needs is therefore set",
                "# here, explicitly, rather than assumed.",
                "SHELL=/bin/sh",
                "PATH=/usr/local/bin:/usr/bin:/bin",
                f"TZ={self.timezone}",
                "MAILTO=",
                "",
                "# min hour day-of-month month day-of-week  command",
                f"{self.cron_expression} cd {self.project_dir} && {self.command} "
                f">> {stdout} 2>&1",
                "",
            ]
        )

    # ---- launchd (macOS) ----------------------------------------------

    def launchd_plist(self) -> str:
        args = "\n".join(f"      <string>{escape(a)}</string>" for a in self.argv)
        weekday_entries = ""
        if self.days_of_week:
            weekday_entries = "\n".join(
                "    <dict>\n"
                f"      <key>Weekday</key><integer>{day}</integer>\n"
                f"      <key>Hour</key><integer>{self.hour}</integer>\n"
                f"      <key>Minute</key><integer>{self.minute}</integer>\n"
                "    </dict>"
                for day in self.days_of_week
            )
            calendar = f"  <key>StartCalendarInterval</key>\n  <array>\n{weekday_entries}\n  </array>"
        else:
            calendar = (
                "  <key>StartCalendarInterval</key>\n"
                "  <dict>\n"
                f"    <key>Hour</key><integer>{self.hour}</integer>\n"
                f"    <key>Minute</key><integer>{self.minute}</integer>\n"
                "  </dict>"
            )
        return f"""<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
  "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
  <!-- Label: the unique name launchd knows this job by. Reverse-DNS by
       convention, and the file should be named after it. -->
  <key>Label</key>
  <string>{escape(self.label)}</string>

  <!-- ProgramArguments: argv, one element per array entry. NOT a shell
       command line: there is no shell here, so no globbing, no pipes, and
       no quoting rules to get wrong. -->
  <key>ProgramArguments</key>
  <array>
{args}
  </array>

{calendar}

  <!-- RunAtLoad false: loading the job should not immediately run it.
       Setting this true is a common surprise during installation. -->
  <key>RunAtLoad</key>
  <false/>

  <!-- launchd starts the job with a minimal environment, exactly like cron.
       Anything the job needs must be stated. -->
  <key>WorkingDirectory</key>
  <string>{escape(self.project_dir)}</string>
  <key>EnvironmentVariables</key>
  <dict>
    <key>PATH</key><string>/usr/local/bin:/usr/bin:/bin</string>
    <key>TZ</key><string>{escape(self.timezone)}</string>
  </dict>

  <!-- Where output goes. Without these, stdout and stderr are discarded and
       a failing job leaves no trace at all. -->
  <key>StandardOutPath</key>
  <string>{escape(self.log_dir)}/{escape(self.label)}.out.log</string>
  <key>StandardErrorPath</key>
  <string>{escape(self.log_dir)}/{escape(self.label)}.err.log</string>
</dict>
</plist>
"""

    # ---- systemd (Linux) ----------------------------------------------

    @property
    def on_calendar(self) -> str:
        if self.days_of_week:
            days = ",".join(SYSTEMD_DAYS[d] for d in self.days_of_week)
            return f"{days} *-*-* {self.hour:02d}:{self.minute:02d}:00"
        return f"*-*-* {self.hour:02d}:{self.minute:02d}:00"

    def systemd_service(self) -> str:
        exec_start = " ".join(self.argv)
        return f"""# {self.label}.service — WHAT to run. It has no schedule of its own.
# Check the file with:   systemd-analyze verify {self.label}.service
# Run it once by hand:   systemctl --user start {self.label}.service
[Unit]
Description=Daily station report
# The timer will not start the job before the network is up.
After=network-online.target

[Service]
# oneshot: this is a task that finishes, not a daemon that stays up.
# systemd counts the unit as active until the process exits.
Type=oneshot
WorkingDirectory={self.project_dir}
Environment=PATH=/usr/local/bin:/usr/bin:/bin
Environment=TZ={self.timezone}
ExecStart={exec_start}
# A hard ceiling, enforced by systemd rather than by the job itself.
TimeoutStartSec=600
# Everything the job prints goes to the journal, tagged with this identifier:
#   journalctl --user -u {self.label}.service -n 50
StandardOutput=journal
StandardError=journal
SyslogIdentifier={self.label}
"""

    def systemd_timer(self) -> str:
        return f"""# {self.label}.timer — WHEN to run it. Pairs with {self.label}.service.
# Install (user scope, no root):
#   cp {self.label}.service {self.label}.timer ~/.config/systemd/user/
#   systemctl --user daemon-reload
#   systemctl --user enable --now {self.label}.timer
# Inspect:  systemctl --user list-timers
[Unit]
Description=Run the daily station report

[Timer]
OnCalendar={self.on_calendar}
# Persistent: if the machine was off at the scheduled moment, run once as
# soon as it comes back. This is the catch-up behaviour cron does not have.
Persistent=true
# Spread load: start somewhere in the first minute rather than exactly on
# the second, so a fleet of machines does not stampede one server.
RandomizedDelaySec=60
AccuracySec=1s
Unit={self.label}.service

[Install]
WantedBy=timers.target
"""

    # ---- the human-readable summary ------------------------------------

    def summary(self) -> str:
        schedule = parse(self.cron_expression)
        base = dt.datetime(2026, 7, 19, 0, 0, tzinfo=dt.timezone.utc)
        upcoming = []
        cursor = base
        for _ in range(3):
            cursor = schedule.next_run_after(cursor)
            upcoming.append(cursor.strftime("%Y-%m-%d %H:%M"))
        return "\n".join(
            [
                f"label          : {self.label}",
                f"cron           : {self.cron_expression}",
                f"launchd        : StartCalendarInterval Hour={self.hour} Minute={self.minute}",
                f"systemd        : OnCalendar={self.on_calendar}",
                f"reads as       : {schedule.describe()}",
                f"next three (from {base:%Y-%m-%d %H:%M} UTC): " + ", ".join(upcoming),
            ]
        )


def write_all(schedule: JobSchedule, out_dir: Path) -> list[Path]:
    """Write the four schedule files. Returns the paths written."""
    out_dir.mkdir(parents=True, exist_ok=True)
    written = []
    for name, text in (
        (f"{schedule.label}.cron", schedule.cron_line()),
        (f"{schedule.label}.plist", schedule.launchd_plist()),
        (f"{schedule.label}.service", schedule.systemd_service()),
        (f"{schedule.label}.timer", schedule.systemd_timer()),
    ):
        path = out_dir / name
        path.write_text(text, encoding="utf-8")
        written.append(path)
    return written


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="gen_schedules.py",
        description=(
            "Write cron, launchd and systemd schedule files for one job. "
            "Installs nothing; prints the install commands so you can decide."
        ),
    )
    parser.add_argument("--out", type=Path, required=True, help="directory to write into")
    parser.add_argument("--label", default="com.example.dailyreport")
    parser.add_argument("--hour", type=int, default=2)
    parser.add_argument("--minute", type=int, default=30)
    parser.add_argument(
        "--weekday",
        type=int,
        action="append",
        default=None,
        help="0=Sunday .. 6=Saturday; repeat for several. Omit for every day.",
    )
    parser.add_argument("--project-dir", default="/opt/reports")
    parser.add_argument("--python", default="/usr/bin/python3")
    parser.add_argument("--log-dir", default="/var/log/reports")
    parser.add_argument("--timezone", default="UTC")
    args = parser.parse_args(argv)

    schedule = JobSchedule(
        label=args.label,
        minute=args.minute,
        hour=args.hour,
        days_of_week=tuple(args.weekday or ()),
        project_dir=args.project_dir,
        python=args.python,
        log_dir=args.log_dir,
        timezone=args.timezone,
    )
    written = write_all(schedule, args.out)
    print(schedule.summary())
    print()
    print("written:")
    for path in written:
        print(f"  {path}")
    print()
    print("NOTHING was installed. To install, you would run — deliberately, yourself:")
    print("  cron    : crontab -l > my.cron && cat "
          f"{args.out}/{schedule.label}.cron >> my.cron && crontab my.cron")
    print("  launchd : cp "
          f"{args.out}/{schedule.label}.plist ~/Library/LaunchAgents/ && "
          f"launchctl load ~/Library/LaunchAgents/{schedule.label}.plist")
    print("  systemd : cp "
          f"{args.out}/{schedule.label}.{{service,timer}} ~/.config/systemd/user/ && "
          f"systemctl --user daemon-reload && systemctl --user enable --now {schedule.label}.timer")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""Hold the job lock for a fixed number of seconds, then let go and exit.

This exists so the test suite can prove the overlap protection without racing:
start this, wait for it to say READY, run the real job, and assert the job
exits 75 having done nothing.

It always exits on its own. The test suite also waits for it explicitly, so
nothing is left running after the lab — that rule is not negotiable here.

    python3 examples/hold_lock.py /tmp/reports/daily-report.lock 2
"""

from __future__ import annotations

import sys
import time

from joblock import AlreadyRunning, job_lock


def main(argv: list[str]) -> int:
    if len(argv) != 3:
        print("usage: hold_lock.py <lock-path> <seconds>", file=sys.stderr)
        return 2
    path, seconds = argv[1], float(argv[2])
    try:
        with job_lock(path):
            print("READY", flush=True)
            time.sleep(seconds)
    except AlreadyRunning as exc:
        print(f"could not take the lock: {exc}", file=sys.stderr)
        return 75
    print("RELEASED", flush=True)
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))

"""The three in-process ways to run something later, and what each costs.

Working from the inside out, before we leave the process at all:

1. **A sleep loop.** ``while True: work(); time.sleep(interval)``. Everyone
   writes this first. It drifts, because the interval is measured from when
   the work *finished*, not from when it was supposed to start — so every run
   is late by the accumulated duration of every run before it.
2. **``sched.scheduler``.** The standard library's event scheduler. It takes
   its time source and its delay function as arguments, which is a small
   design decision with a large consequence: you can hand it a fake clock and
   test six hours of schedule in microseconds.
3. **``threading.Timer``.** One callback, once, after a delay, on its own
   thread. Fine for a single delayed action inside a running program; not a
   scheduler.

All three share the fatal property: they die with the process. Close the
terminal, deploy a new version, reboot the laptop — the schedule is gone, and
nothing tells you. That is why real recurring work belongs to the operating
system's scheduler, and why the rest of this lab is about that.

Every function here takes its time source as a parameter, so nothing in this
file ever actually waits.
"""

from __future__ import annotations

import sched
from collections.abc import Callable
from dataclasses import dataclass

from clock import FakeTime


@dataclass(frozen=True)
class LoopTrace:
    """When each run started, and how late it was against the ideal schedule."""

    starts: tuple[float, ...]
    interval: float

    @property
    def lateness(self) -> tuple[float, ...]:
        first = self.starts[0]
        return tuple(
            round(start - (first + index * self.interval), 6)
            for index, start in enumerate(self.starts)
        )

    @property
    def final_drift(self) -> float:
        return self.lateness[-1]


def naive_sleep_loop(
    *, runs: int, interval: float, work_seconds: float, fake: FakeTime | None = None
) -> LoopTrace:
    """``work(); sleep(interval)`` — the version that drifts.

    With a 60-second interval and 5 seconds of work, run 100 starts 495
    seconds late. The schedule is not 60 seconds; it is 65, and nobody wrote
    65 anywhere.
    """
    fake = fake or FakeTime()
    starts = []
    for _ in range(runs):
        starts.append(fake.time())
        fake.sleep(work_seconds)  # the work
        fake.sleep(interval)  # the wait
    return LoopTrace(tuple(starts), interval)


def deadline_corrected_loop(
    *, runs: int, interval: float, work_seconds: float, fake: FakeTime | None = None
) -> LoopTrace:
    """Sleep until the *next deadline*, not for a fixed span. No drift.

    The fix is three lines: track the next scheduled instant, and sleep for
    whatever is left of it. If a run overruns the interval entirely the sleep
    is zero and the next run starts immediately — which is a decision you have
    now made on purpose rather than by accident.
    """
    fake = fake or FakeTime()
    starts = []
    next_due = fake.time()
    for _ in range(runs):
        starts.append(fake.time())
        fake.sleep(work_seconds)
        next_due += interval
        remaining = next_due - fake.time()
        fake.sleep(max(0.0, remaining))
    return LoopTrace(tuple(starts), interval)


def run_sched_schedule(
    *, delays: list[float], fake: FakeTime | None = None
) -> tuple[list[tuple[float, int]], FakeTime]:
    """Drive ``sched.scheduler`` with a fake clock: hours of schedule, no waiting.

    ``sched.scheduler(timefunc, delayfunc)`` is stdlib dependency injection
    from 1990s Python, and it is the reason a schedule built on ``sched`` is
    testable while one built on ``time.sleep`` is not.
    """
    fake = fake or FakeTime()
    fired: list[tuple[float, int]] = []
    scheduler = sched.scheduler(timefunc=fake.time, delayfunc=fake.sleep)

    def record(index: int) -> None:
        fired.append((fake.time(), index))

    for index, delay in enumerate(delays):
        scheduler.enter(delay, priority=1, action=record, argument=(index,))
    scheduler.run()
    return fired, fake


def periodic_with_sched(
    *, interval: float, runs: int, work: Callable[[], None] | None = None,
    fake: FakeTime | None = None,
) -> list[float]:
    """``sched`` has no repeat: a recurring job re-enters itself. Still no drift.

    The re-entry is scheduled against ``next_due`` rather than "now", so the
    same deadline correction applies.
    """
    fake = fake or FakeTime()
    scheduler = sched.scheduler(timefunc=fake.time, delayfunc=fake.sleep)
    starts: list[float] = []
    state = {"remaining": runs, "next_due": fake.time()}

    def tick() -> None:
        starts.append(fake.time())
        if work is not None:
            work()
        state["remaining"] -= 1
        if state["remaining"] > 0:
            state["next_due"] += interval
            scheduler.enterabs(state["next_due"], priority=1, action=tick)

    scheduler.enterabs(state["next_due"], priority=1, action=tick)
    scheduler.run()
    return starts

"""The command a scheduler would run — and the command you run by hand today.

Two subcommands, in the argparse style of Day 80:

    python3 examples/job.py run   --output-dir /tmp/reports
    python3 examples/job.py watch --heartbeat-file /tmp/reports/heartbeat.json

``run`` is the job. ``watch`` is the dead man's switch that notices when
``run`` has quietly stopped happening. They are separate programs on purpose:
a watchdog that lives inside the job it watches cannot report that the job did
not start.

``--now`` injects the clock. It exists so that every time-dependent behaviour
in this lab — "which day's report is this?", "is the heartbeat stale?", "what
happens at 02:30 on the day the clocks change?" — can be exercised in a
fraction of a second instead of waited for. That is Day 74's boundary lesson
applied to the one boundary that scheduling is entirely made of.

Exit codes:
    0    the work is done, or was already done
    1    the work raised
    75   another copy holds the lock; nothing was done  (sysexits EX_TEMPFAIL)
    124  the work exceeded its timeout and was interrupted  (as GNU timeout does)
"""

from __future__ import annotations

import argparse
import datetime as dt
import sys
import time
from pathlib import Path

from clock import Clock, frozen_clock, system_clock
from reportjob import generate_daily_report
from runner import (
    EXIT_MEANINGS,
    EXIT_OK,
    combined_logger,
    jsonl_logger,
    run_job,
    stream_logger,
)
from watchdog import check_heartbeat

HERE = Path(__file__).resolve().parent
DEFAULT_DATA = HERE / "data" / "readings.csv"


def build_clock(now: str | None, tz: str) -> Clock:
    if now is None:
        return system_clock(tz)
    moment = dt.datetime.fromisoformat(now)
    if moment.tzinfo is None:
        raise SystemExit(
            "--now needs a timezone offset, for example 2026-07-20T02:30:00+00:00. "
            "A naive timestamp means 'whatever this machine thinks local time is', "
            "which is the bug this lab is about."
        )
    return frozen_clock(moment)


def command_run(args: argparse.Namespace) -> int:
    clock = build_clock(args.now, args.timezone)
    output_dir = Path(args.output_dir)
    report_date = (
        dt.date.fromisoformat(args.date)
        if args.date
        else (clock().date() - dt.timedelta(days=1))
    )

    def work() -> dict[str, object]:
        if args.simulate_hang:
            time.sleep(args.simulate_hang)
        if args.simulate_failure:
            raise RuntimeError("the upstream feed returned nothing (simulated)")
        status, path = generate_daily_report(
            source=Path(args.data),
            output_dir=output_dir,
            report_date=report_date,
            generated_at=clock(),
        )
        return {
            "status": "ok" if status == "written" else "skipped",
            "report_date": report_date.isoformat(),
            "output": str(path),
            "action": status,
        }

    loggers = [stream_logger()]
    if args.log_file:
        loggers.append(jsonl_logger(args.log_file))

    run = run_job(
        name=args.name,
        work=work,
        clock=clock,
        lock_path=args.lock_file or (output_dir / f"{args.name}.lock"),
        log=combined_logger(*loggers),
        timeout_seconds=args.timeout,
        heartbeat_path=args.heartbeat_file
        or (output_dir / f"{args.name}.heartbeat.json"),
    )
    if run.exit_code != EXIT_OK:
        print(
            f"{args.name}: {run.status} -> exit {run.exit_code} "
            f"({EXIT_MEANINGS[run.exit_code]})",
            file=sys.stderr,
        )
    return run.exit_code


def command_watch(args: argparse.Namespace) -> int:
    clock = build_clock(args.now, args.timezone)
    verdict = check_heartbeat(
        heartbeat_path=args.heartbeat_file,
        clock=clock,
        max_age=dt.timedelta(minutes=args.max_age_minutes),
    )
    print(f"{verdict.state.upper()}: {verdict.message}")
    return 1 if verdict.alerting else 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="job.py",
        description="A scheduled daily report, and the watchdog that notices when it stops.",
        epilog="This program never installs itself into any scheduler. See gen_schedules.py.",
    )
    parser.add_argument("--now", help="freeze the clock, e.g. 2026-07-20T02:30:00+00:00")
    parser.add_argument("--timezone", default="UTC", help="zone for the real clock (default UTC)")
    sub = parser.add_subparsers(dest="command", required=True)

    run_parser = sub.add_parser("run", help="generate one daily report, idempotently")
    run_parser.add_argument("--output-dir", required=True, type=Path)
    run_parser.add_argument("--data", default=str(DEFAULT_DATA), help="the readings CSV")
    run_parser.add_argument("--date", help="report date (default: the day before --now)")
    run_parser.add_argument("--name", default="daily-report", help="job name used in logs and locks")
    run_parser.add_argument("--lock-file", type=Path)
    run_parser.add_argument("--log-file", type=Path, help="append JSON lines here as well")
    run_parser.add_argument("--heartbeat-file", type=Path)
    run_parser.add_argument("--timeout", type=float, default=60.0, help="seconds (0 disables)")
    run_parser.add_argument(
        "--simulate-hang", type=float, default=0.0, help="sleep this long inside the work"
    )
    run_parser.add_argument(
        "--simulate-failure", action="store_true", help="raise inside the work"
    )
    run_parser.set_defaults(func=command_run)

    watch_parser = sub.add_parser("watch", help="alert if the last success is too old")
    watch_parser.add_argument("--heartbeat-file", required=True, type=Path)
    watch_parser.add_argument("--max-age-minutes", type=float, default=2880.0)
    watch_parser.set_defaults(func=command_watch)
    return parser


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    if getattr(args, "timeout", None) == 0:
        args.timeout = None
    return int(args.func(args))


if __name__ == "__main__":
    raise SystemExit(main())

"""A lock that stops two copies of one job running at once.

The failure this prevents is specific. A job scheduled every five minutes that
usually takes forty seconds is fine until the day it takes six minutes —
because then the scheduler starts the next copy while the first is still
working, and now two processes are writing the same file, counting the same
rows, or calling the same paid API. Cron will not stop that happening. Nothing
will, unless the job stops itself.

The mechanism here is ``fcntl.flock`` with ``LOCK_NB``: an advisory lock held
by the operating system on an open file descriptor. Two properties make it the
right tool:

* it is atomic — there is no window between "check" and "take" for a second
  process to slip through, which is exactly the bug in the obvious
  "if the lockfile exists, exit" implementation;
* the kernel releases it when the process exits, however it exits. A job
  killed with SIGKILL, or a machine that loses power, does not leave a stale
  lock that blocks every future run — which is the other failure of the
  naive version, and the more annoying one at three in the morning.

``fcntl`` is POSIX: macOS and Linux have it, Windows does not. The Windows
equivalent is ``msvcrt.locking`` or a named mutex; this lab runs on macOS and
Linux, and the README says so plainly.
"""

from __future__ import annotations

import contextlib
import fcntl
import os
from collections.abc import Iterator
from pathlib import Path


class AlreadyRunning(RuntimeError):
    """Raised when another process already holds the job's lock."""

    def __init__(self, path: Path, holder_pid: str) -> None:
        super().__init__(f"another run already holds {path} (pid {holder_pid})")
        self.path = path
        self.holder_pid = holder_pid


@contextlib.contextmanager
def job_lock(path: str | os.PathLike[str]) -> Iterator[Path]:
    """Hold an exclusive, non-blocking lock for the duration of the block.

    Raises :class:`AlreadyRunning` immediately if the lock is held elsewhere.
    Failing fast is deliberate: a scheduled job that *waits* for the lock just
    queues up copies of itself, and a queue of overdue jobs is how a slow
    Monday becomes an outage.

    The process id is written into the file purely so a human reading it later
    knows who to look for. It is not used for locking — the file descriptor is.
    """
    lock_path = Path(path)
    lock_path.parent.mkdir(parents=True, exist_ok=True)
    handle = open(lock_path, "a+")  # noqa: SIM115 - closed in the finally below
    try:
        try:
            fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError:
            handle.seek(0)
            holder = handle.read().strip() or "unknown"
            raise AlreadyRunning(lock_path, holder) from None
        handle.seek(0)
        handle.truncate()
        handle.write(f"{os.getpid()}\n")
        handle.flush()
        try:
            yield lock_path
        finally:
            fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
    finally:
        handle.close()

"""The work itself: a daily report, written idempotently.

The job is deliberately dull — read yesterday's readings out of a CSV, total
them per station, write a JSON file. The interesting part is not what it
computes but the two properties it has:

**Idempotence.** Running it twice for the same day produces one report, not
two, and not a doubled one. This is the single most important property a
scheduled job can have, because retries, catch-up runs and a nervous operator
typing the command again all mean "run it twice", and every one of those
happens eventually.

**Atomic output.** The file is written to a temporary name in the same
directory and then moved into place with ``os.replace``, which on POSIX is an
atomic rename. Without that, a job killed halfway through leaves a truncated
JSON file that *looks* finished to the next run — so the next run skips it,
and the truncated file lives forever. This is one of those bugs that is
invisible for months and then ruins a morning.
"""

from __future__ import annotations

import csv
import datetime as dt
import json
import os
import statistics
import tempfile
from dataclasses import dataclass, field
from pathlib import Path


@dataclass(frozen=True)
class StationSummary:
    """One station's numbers for one day."""

    station: str
    count: int
    minimum: float
    maximum: float
    mean: float

    def as_dict(self) -> dict[str, float | int | str]:
        return {
            "station": self.station,
            "count": self.count,
            "min_celsius": round(self.minimum, 2),
            "max_celsius": round(self.maximum, 2),
            "mean_celsius": round(self.mean, 2),
        }


@dataclass(frozen=True)
class DailyReport:
    """The whole report for one day. A value — it writes nothing itself."""

    report_date: dt.date
    generated_at: dt.datetime
    stations: tuple[StationSummary, ...] = field(default_factory=tuple)

    @property
    def reading_count(self) -> int:
        return sum(s.count for s in self.stations)

    def as_dict(self) -> dict[str, object]:
        return {
            "report_date": self.report_date.isoformat(),
            "generated_at": self.generated_at.isoformat(),
            "reading_count": self.reading_count,
            "stations": [s.as_dict() for s in self.stations],
        }


def load_readings(source: Path, report_date: dt.date) -> dict[str, list[float]]:
    """Read the CSV and group one day's Celsius values by station."""
    grouped: dict[str, list[float]] = {}
    wanted = report_date.isoformat()
    with open(source, newline="", encoding="utf-8") as handle:
        for row in csv.DictReader(handle):
            if row["date"] != wanted:
                continue
            grouped.setdefault(row["station"], []).append(float(row["celsius"]))
    return grouped


def build_report(
    readings: dict[str, list[float]],
    *,
    report_date: dt.date,
    generated_at: dt.datetime,
) -> DailyReport:
    """Pure: values in, value out. No clock, no filesystem, no exceptions to catch."""
    summaries = tuple(
        StationSummary(
            station=station,
            count=len(values),
            minimum=min(values),
            maximum=max(values),
            mean=statistics.fmean(values),
        )
        for station, values in sorted(readings.items())
    )
    return DailyReport(
        report_date=report_date, generated_at=generated_at, stations=summaries
    )


def report_path(output_dir: Path, report_date: dt.date) -> Path:
    """The output name IS the idempotence key. One day, one file, one name."""
    return Path(output_dir) / f"report-{report_date.isoformat()}.json"


def already_written(output_dir: Path, report_date: dt.date) -> bool:
    """Has a *complete* report for this day already been written?

    "Complete" means the file parses as JSON and carries the marker key. A
    partially written file fails both tests, so a crashed run is retried rather
    than mistaken for a success.
    """
    path = report_path(output_dir, report_date)
    if not path.exists():
        return False
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, UnicodeDecodeError):
        return False
    return payload.get("report_date") == report_date.isoformat()


def write_report_atomically(report: DailyReport, output_dir: Path) -> Path:
    """Write to a temporary file in the same directory, then rename into place.

    Same directory matters: ``os.replace`` is only atomic within one
    filesystem, and ``/tmp`` is frequently a different one.
    """
    destination = report_path(output_dir, report.report_date)
    destination.parent.mkdir(parents=True, exist_ok=True)
    handle = tempfile.NamedTemporaryFile(
        "w",
        encoding="utf-8",
        dir=destination.parent,
        prefix=destination.name + ".",
        suffix=".partial",
        delete=False,
    )
    try:
        with handle:
            json.dump(report.as_dict(), handle, indent=2, sort_keys=True)
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(handle.name, destination)
    except BaseException:
        Path(handle.name).unlink(missing_ok=True)
        raise
    return destination


def generate_daily_report(
    *,
    source: Path,
    output_dir: Path,
    report_date: dt.date,
    generated_at: dt.datetime,
) -> tuple[str, Path]:
    """The whole unit of work. Returns ``("written" | "skipped", path)``.

    The skip branch is what makes the job idempotent, and it is checked before
    any work is done rather than after, so a repeat run is also cheap.
    """
    if already_written(output_dir, report_date):
        return "skipped", report_path(output_dir, report_date)
    readings = load_readings(source, report_date)
    report = build_report(
        readings, report_date=report_date, generated_at=generated_at
    )
    return "written", write_report_atomically(report, output_dir)

"""The part that makes a job survivable: lock, timeout, log, exit code, heartbeat.

Scheduling something is one line in a crontab. Everything that makes the
scheduled thing trustworthy lives here, and it is the same five concerns every
time:

1. **One at a time.** Take a lock; if someone else has it, exit immediately
   with a distinct code rather than doing the work twice.
2. **Bounded.** A job with no timeout can hang forever holding the lock, which
   means every later run is also skipped, which means the job silently stops.
   A hang is worse than a crash precisely because nothing reports it.
3. **Logged with context.** One structured line per run, with a run id, the
   status, the duration and the exit code. "It failed last Tuesday" is only
   answerable if the line exists.
4. **Honest exit codes.** The scheduler and any wrapper only see the number.
   0 means the work is done; anything else means look.
5. **A heartbeat.** Every success writes down when it succeeded. A separate
   watchdog reads that file. Without it, a job that stops being scheduled at
   all produces no failure and no output — and no alert.

Exit codes follow conventions that already exist rather than inventing new
numbers: 75 is ``EX_TEMPFAIL`` from the BSD ``sysexits.h`` list ("temporary
failure, try again later"), and 124 is the code GNU ``timeout`` uses when it
kills a command.
"""

from __future__ import annotations

import datetime as dt
import json
import signal
import sys
import traceback
from collections.abc import Callable, Mapping
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, TextIO

from clock import Clock
from joblock import AlreadyRunning, job_lock

EXIT_OK = 0
EXIT_FAILED = 1
EXIT_ALREADY_RUNNING = 75  # sysexits.h EX_TEMPFAIL
EXIT_TIMEOUT = 124  # the code GNU timeout(1) uses

EXIT_MEANINGS: dict[int, str] = {
    EXIT_OK: "the work is done (or was already done)",
    EXIT_FAILED: "the work raised an exception",
    EXIT_ALREADY_RUNNING: "another copy holds the lock; nothing was done",
    EXIT_TIMEOUT: "the work exceeded its timeout and was interrupted",
}


class JobTimeout(TimeoutError):
    """Raised inside the job when its time budget expires."""


@dataclass(frozen=True)
class JobRun:
    """The record of one invocation. Everything the log line is built from."""

    name: str
    run_id: str
    status: str  # "ok" | "skipped" | "failed" | "timeout" | "already-running"
    exit_code: int
    started_at: dt.datetime
    finished_at: dt.datetime
    duration_seconds: float
    detail: Mapping[str, Any] = field(default_factory=dict)

    def as_event(self) -> dict[str, Any]:
        return {
            "job": self.name,
            "run_id": self.run_id,
            "status": self.status,
            "exit_code": self.exit_code,
            "started_at": self.started_at.isoformat(),
            "finished_at": self.finished_at.isoformat(),
            "duration_seconds": round(self.duration_seconds, 3),
            **dict(self.detail),
        }


Logger = Callable[[Mapping[str, Any]], None]


def jsonl_logger(path: str | Path) -> Logger:
    """Append one JSON object per line. Machine-readable, greppable, append-only.

    One line per event, never a multi-line traceback in the middle of the
    stream: a log you cannot process with ``grep`` and ``json.loads`` is a log
    nobody processes.
    """
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)

    def write(event: Mapping[str, Any]) -> None:
        with open(target, "a", encoding="utf-8") as handle:
            handle.write(json.dumps(dict(event), sort_keys=True) + "\n")

    return write


def stream_logger(stream: TextIO | None = None) -> Logger:
    """The same events, to standard output — which is where cron mails them from."""
    target = stream if stream is not None else sys.stdout

    def write(event: Mapping[str, Any]) -> None:
        print(json.dumps(dict(event), sort_keys=True), file=target)

    return write


def combined_logger(*loggers: Logger) -> Logger:
    def write(event: Mapping[str, Any]) -> None:
        for logger in loggers:
            logger(event)

    return write


def make_run_id(name: str, started: dt.datetime) -> str:
    """Deterministic under a frozen clock, which is what makes runs assertable."""
    return f"{name}-{started.strftime('%Y%m%dT%H%M%S%z')}"


def write_heartbeat(path: str | Path, run: JobRun) -> None:
    """Record the last success. This file is the dead man's switch."""
    target = Path(path)
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(
        json.dumps(
            {
                "job": run.name,
                "run_id": run.run_id,
                "last_success": run.finished_at.isoformat(),
            },
            sort_keys=True,
        )
        + "\n",
        encoding="utf-8",
    )


def run_job(
    *,
    name: str,
    work: Callable[[], Mapping[str, Any]],
    clock: Clock,
    lock_path: str | Path,
    log: Logger,
    timeout_seconds: float | None = None,
    heartbeat_path: str | Path | None = None,
) -> JobRun:
    """Run ``work`` once, under a lock, under a time budget, and write it down.

    ``work`` is a zero-argument callable returning a mapping that is merged
    into the log event. Injecting it — instead of hard-coding the report
    generator here — is what lets the test suite check the lock, the timeout
    and the logging with a two-line fake job.
    """
    started = clock()
    run_id = make_run_id(name, started)

    def finish(
        status: str, exit_code: int, detail: Mapping[str, Any]
    ) -> JobRun:
        finished = clock()
        run = JobRun(
            name=name,
            run_id=run_id,
            status=status,
            exit_code=exit_code,
            started_at=started,
            finished_at=finished,
            duration_seconds=(finished - started).total_seconds(),
            detail=detail,
        )
        log(run.as_event())
        return run

    try:
        with job_lock(lock_path):
            with _time_budget(timeout_seconds):
                try:
                    detail = dict(work())
                except JobTimeout:
                    return finish(
                        "timeout",
                        EXIT_TIMEOUT,
                        {
                            "error": "JobTimeout",
                            "timeout_seconds": timeout_seconds,
                            "message": f"work exceeded {timeout_seconds}s and was interrupted",
                        },
                    )
                except Exception as exc:  # noqa: BLE001 - deliberate: log and exit non-zero
                    return finish(
                        "failed",
                        EXIT_FAILED,
                        {
                            "error": type(exc).__name__,
                            "message": str(exc),
                            "traceback": traceback.format_exc(limit=3).strip().splitlines()[-1],
                        },
                    )
    except AlreadyRunning as exc:
        return finish(
            "already-running",
            EXIT_ALREADY_RUNNING,
            {"lock_path": str(exc.path), "holder_pid": exc.holder_pid},
        )

    status = str(detail.pop("status", "ok"))
    run = finish(status, EXIT_OK, detail)
    if heartbeat_path is not None:
        write_heartbeat(heartbeat_path, run)
    return run


class _time_budget:
    """A wall-clock guard built on ``signal.setitimer`` and ``SIGALRM``.

    Honest limits, because they matter: this is POSIX-only, it works only in
    the main thread, and it interrupts Python at the next opportunity — a call
    blocked deep inside a C library may not notice. It is fine for the common
    cases (a sleep, a socket read, a loop) and it is not a substitute for
    supervising the job as a child process. ``supervise.py`` in this directory
    does the stronger version.
    """

    def __init__(self, seconds: float | None) -> None:
        self.seconds = seconds
        self._previous: Any = None

    def __enter__(self) -> _time_budget:
        if self.seconds is None:
            return self

        def on_alarm(signum: int, frame: Any) -> None:
            raise JobTimeout(f"exceeded {self.seconds}s")

        self._previous = signal.signal(signal.SIGALRM, on_alarm)
        signal.setitimer(signal.ITIMER_REAL, self.seconds)
        return self

    def __exit__(self, *exc_info: Any) -> None:
        if self.seconds is None:
            return
        signal.setitimer(signal.ITIMER_REAL, 0)
        if self._previous is not None:
            signal.signal(signal.SIGALRM, self._previous)

# Generated schedule. Read it, then decide whether to install it.
# Install with:  crontab -l > my.cron ; cat this-file >> my.cron ; crontab my.cron
# List with:     crontab -l          Edit with:  crontab -e
#
# cron gives a job almost no environment: a short PATH, no shell
# profile, HOME set but nothing sourced, and the home directory as
# the working directory. Everything the job needs is therefore set
# here, explicitly, rather than assumed.
SHELL=/bin/sh
PATH=/usr/local/bin:/usr/bin:/bin
TZ=UTC
MAILTO=

# min hour day-of-month month day-of-week  command
30 2 * * * cd /opt/reports && /usr/bin/python3 /opt/reports/job.py --output-dir /opt/reports/out >> /var/log/reports/com.example.dailyreport.log 2>&1

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
  "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
  <!-- Label: the unique name launchd knows this job by. Reverse-DNS by
       convention, and the file should be named after it. -->
  <key>Label</key>
  <string>com.example.dailyreport</string>

  <!-- ProgramArguments: argv, one element per array entry. NOT a shell
       command line: there is no shell here, so no globbing, no pipes, and
       no quoting rules to get wrong. -->
  <key>ProgramArguments</key>
  <array>
      <string>/usr/bin/python3</string>
      <string>/opt/reports/job.py</string>
      <string>--output-dir</string>
      <string>/opt/reports/out</string>
  </array>

  <key>StartCalendarInterval</key>
  <dict>
    <key>Hour</key><integer>2</integer>
    <key>Minute</key><integer>30</integer>
  </dict>

  <!-- RunAtLoad false: loading the job should not immediately run it.
       Setting this true is a common surprise during installation. -->
  <key>RunAtLoad</key>
  <false/>

  <!-- launchd starts the job with a minimal environment, exactly like cron.
       Anything the job needs must be stated. -->
  <key>WorkingDirectory</key>
  <string>/opt/reports</string>
  <key>EnvironmentVariables</key>
  <dict>
    <key>PATH</key><string>/usr/local/bin:/usr/bin:/bin</string>
    <key>TZ</key><string>UTC</string>
  </dict>

  <!-- Where output goes. Without these, stdout and stderr are discarded and
       a failing job leaves no trace at all. -->
  <key>StandardOutPath</key>
  <string>/var/log/reports/com.example.dailyreport.out.log</string>
  <key>StandardErrorPath</key>
  <string>/var/log/reports/com.example.dailyreport.err.log</string>
</dict>
</plist>

# com.example.dailyreport.service — WHAT to run. It has no schedule of its own.
# Check the file with:   systemd-analyze verify com.example.dailyreport.service
# Run it once by hand:   systemctl --user start com.example.dailyreport.service
[Unit]
Description=Daily station report
# The timer will not start the job before the network is up.
After=network-online.target

[Service]
# oneshot: this is a task that finishes, not a daemon that stays up.
# systemd counts the unit as active until the process exits.
Type=oneshot
WorkingDirectory=/opt/reports
Environment=PATH=/usr/local/bin:/usr/bin:/bin
Environment=TZ=UTC
ExecStart=/usr/bin/python3 /opt/reports/job.py --output-dir /opt/reports/out
# A hard ceiling, enforced by systemd rather than by the job itself.
TimeoutStartSec=600
# Everything the job prints goes to the journal, tagged with this identifier:
#   journalctl --user -u com.example.dailyreport.service -n 50
StandardOutput=journal
StandardError=journal
SyslogIdentifier=com.example.dailyreport

# com.example.dailyreport.timer — WHEN to run it. Pairs with com.example.dailyreport.service.
# Install (user scope, no root):
#   cp com.example.dailyreport.service com.example.dailyreport.timer ~/.config/systemd/user/
#   systemctl --user daemon-reload
#   systemctl --user enable --now com.example.dailyreport.timer
# Inspect:  systemctl --user list-timers
[Unit]
Description=Run the daily station report

[Timer]
OnCalendar=*-*-* 02:30:00
# Persistent: if the machine was off at the scheduled moment, run once as
# soon as it comes back. This is the catch-up behaviour cron does not have.
Persistent=true
# Spread load: start somewhere in the first minute rather than exactly on
# the second, so a fleet of machines does not stampede one server.
RandomizedDelaySec=60
AccuracySec=1s
Unit=com.example.dailyreport.service

[Install]
WantedBy=timers.target

# Generated schedule. Read it, then decide whether to install it.
# Install with:  crontab -l > my.cron ; cat this-file >> my.cron ; crontab my.cron
# List with:     crontab -l          Edit with:  crontab -e
#
# cron gives a job almost no environment: a short PATH, no shell
# profile, HOME set but nothing sourced, and the home directory as
# the working directory. Everything the job needs is therefore set
# here, explicitly, rather than assumed.
SHELL=/bin/sh
PATH=/usr/local/bin:/usr/bin:/bin
TZ=UTC
MAILTO=

# min hour day-of-month month day-of-week  command
15 6 * * 1,2,3,4,5 cd /opt/reports && /usr/bin/python3 /opt/reports/job.py --output-dir /opt/reports/out >> /var/log/reports/com.example.weekdayreport.log 2>&1

<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
  "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
  <!-- Label: the unique name launchd knows this job by. Reverse-DNS by
       convention, and the file should be named after it. -->
  <key>Label</key>
  <string>com.example.weekdayreport</string>

  <!-- ProgramArguments: argv, one element per array entry. NOT a shell
       command line: there is no shell here, so no globbing, no pipes, and
       no quoting rules to get wrong. -->
  <key>ProgramArguments</key>
  <array>
      <string>/usr/bin/python3</string>
      <string>/opt/reports/job.py</string>
      <string>--output-dir</string>
      <string>/opt/reports/out</string>
  </array>

  <key>StartCalendarInterval</key>
  <array>
    <dict>
      <key>Weekday</key><integer>1</integer>
      <key>Hour</key><integer>6</integer>
      <key>Minute</key><integer>15</integer>
    </dict>
    <dict>
      <key>Weekday</key><integer>2</integer>
      <key>Hour</key><integer>6</integer>
      <key>Minute</key><integer>15</integer>
    </dict>
    <dict>
      <key>Weekday</key><integer>3</integer>
      <key>Hour</key><integer>6</integer>
      <key>Minute</key><integer>15</integer>
    </dict>
    <dict>
      <key>Weekday</key><integer>4</integer>
      <key>Hour</key><integer>6</integer>
      <key>Minute</key><integer>15</integer>
    </dict>
    <dict>
      <key>Weekday</key><integer>5</integer>
      <key>Hour</key><integer>6</integer>
      <key>Minute</key><integer>15</integer>
    </dict>
  </array>

  <!-- RunAtLoad false: loading the job should not immediately run it.
       Setting this true is a common surprise during installation. -->
  <key>RunAtLoad</key>
  <false/>

  <!-- launchd starts the job with a minimal environment, exactly like cron.
       Anything the job needs must be stated. -->
  <key>WorkingDirectory</key>
  <string>/opt/reports</string>
  <key>EnvironmentVariables</key>
  <dict>
    <key>PATH</key><string>/usr/local/bin:/usr/bin:/bin</string>
    <key>TZ</key><string>UTC</string>
  </dict>

  <!-- Where output goes. Without these, stdout and stderr are discarded and
       a failing job leaves no trace at all. -->
  <key>StandardOutPath</key>
  <string>/var/log/reports/com.example.weekdayreport.out.log</string>
  <key>StandardErrorPath</key>
  <string>/var/log/reports/com.example.weekdayreport.err.log</string>
</dict>
</plist>

# com.example.weekdayreport.service — WHAT to run. It has no schedule of its own.
# Check the file with:   systemd-analyze verify com.example.weekdayreport.service
# Run it once by hand:   systemctl --user start com.example.weekdayreport.service
[Unit]
Description=Daily station report
# The timer will not start the job before the network is up.
After=network-online.target

[Service]
# oneshot: this is a task that finishes, not a daemon that stays up.
# systemd counts the unit as active until the process exits.
Type=oneshot
WorkingDirectory=/opt/reports
Environment=PATH=/usr/local/bin:/usr/bin:/bin
Environment=TZ=UTC
ExecStart=/usr/bin/python3 /opt/reports/job.py --output-dir /opt/reports/out
# A hard ceiling, enforced by systemd rather than by the job itself.
TimeoutStartSec=600
# Everything the job prints goes to the journal, tagged with this identifier:
#   journalctl --user -u com.example.weekdayreport.service -n 50
StandardOutput=journal
StandardError=journal
SyslogIdentifier=com.example.weekdayreport

# com.example.weekdayreport.timer — WHEN to run it. Pairs with com.example.weekdayreport.service.
# Install (user scope, no root):
#   cp com.example.weekdayreport.service com.example.weekdayreport.timer ~/.config/systemd/user/
#   systemctl --user daemon-reload
#   systemctl --user enable --now com.example.weekdayreport.timer
# Inspect:  systemctl --user list-timers
[Unit]
Description=Run the daily station report

[Timer]
OnCalendar=Mon,Tue,Wed,Thu,Fri *-*-* 06:15:00
# Persistent: if the machine was off at the scheduled moment, run once as
# soon as it comes back. This is the catch-up behaviour cron does not have.
Persistent=true
# Spread load: start somewhere in the first minute rather than exactly on
# the second, so a fleet of machines does not stampede one server.
RandomizedDelaySec=60
AccuracySec=1s
Unit=com.example.weekdayreport.service

[Install]
WantedBy=timers.target

"""The stronger timeout: run the job as a child process and kill the group.

``runner.py`` guards the work with ``SIGALRM``, which is simple and enough for
most jobs. It has two holes worth knowing about: it cannot interrupt a call
that is blocked inside a C library, and it cannot do anything at all about a
grandchild process the job started.

Supervising the job as a child closes both. ``subprocess.run(..., timeout=n)``
raises ``TimeoutExpired`` and kills the child — but only the child. A job that
launched its own helpers leaves them orphaned and still running, which is
exactly the "no background process left behind" rule this lab cares about. The
fix is ``start_new_session=True``, which puts the child in its own process
group, and ``os.killpg`` to signal the whole group.

The escalation is the usual one: ask politely with SIGTERM, wait a moment, and
only then use SIGKILL, which cannot be caught or ignored.

Run it directly to watch a hung command die on schedule:

    python3 examples/supervise.py --timeout 1 -- sleep 30
"""

from __future__ import annotations

import argparse
import os
import signal
import subprocess
import sys
import time
from dataclasses import dataclass


@dataclass(frozen=True)
class Supervised:
    exit_code: int
    timed_out: bool
    stdout: str
    stderr: str


def run_supervised(
    argv: list[str],
    *,
    timeout: float,
    grace: float = 0.5,
) -> Supervised:
    """Run ``argv``, killing the whole process group if it overruns."""
    process = subprocess.Popen(
        argv,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        start_new_session=True,  # its own process group, so killpg reaches helpers
    )
    try:
        stdout, stderr = process.communicate(timeout=timeout)
        return Supervised(process.returncode, False, stdout, stderr)
    except subprocess.TimeoutExpired:
        _terminate_group(process, grace)
        stdout, stderr = process.communicate()
        return Supervised(124, True, stdout, stderr)


def _terminate_group(process: subprocess.Popen[str], grace: float) -> None:
    """SIGTERM the group, wait ``grace`` seconds, then SIGKILL what is left."""
    try:
        group = os.getpgid(process.pid)
    except ProcessLookupError:
        return
    for sig, wait in ((signal.SIGTERM, grace), (signal.SIGKILL, 0.0)):
        try:
            os.killpg(group, sig)
        except ProcessLookupError:
            return
        deadline = time.monotonic() + wait
        while time.monotonic() < deadline:
            if process.poll() is not None:
                return
            time.sleep(0.02)
        if process.poll() is not None:
            return


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        prog="supervise.py",
        description="Run a command with a hard wall-clock timeout on its whole process group.",
    )
    parser.add_argument("--timeout", type=float, required=True, help="seconds before the kill")
    parser.add_argument("--grace", type=float, default=0.5, help="seconds between TERM and KILL")
    parser.add_argument("command", nargs=argparse.REMAINDER, help="the command, after --")
    args = parser.parse_args(argv)

    command = [c for c in args.command if c != "--"]
    if not command:
        parser.error("no command given; put it after --")

    result = run_supervised(command, timeout=args.timeout, grace=args.grace)
    if result.stdout:
        sys.stdout.write(result.stdout)
    if result.stderr:
        sys.stderr.write(result.stderr)
    if result.timed_out:
        print(
            f"supervise: killed after {args.timeout}s -> exit 124",
            file=sys.stderr,
        )
    return result.exit_code


if __name__ == "__main__":
    raise SystemExit(main())

"""The cron expression tests, including the field everybody gets wrong."""

from __future__ import annotations

import datetime as dt

import pytest
from cronexpr import CronError, parse

UTC = dt.timezone.utc


def at(year, month, day, hour=0, minute=0):
    return dt.datetime(year, month, day, hour, minute, tzinfo=UTC)


def test_five_fields_are_required():
    with pytest.raises(CronError, match="expected 5 fields"):
        parse("30 2 * *")


def test_a_field_outside_its_range_is_rejected_not_ignored():
    with pytest.raises(CronError, match="outside 0-23"):
        parse("0 24 * * *")


def test_a_typo_raises_rather_than_silently_meaning_every_minute():
    with pytest.raises(CronError):
        parse("*/ 2 * * *")


@pytest.mark.parametrize(
    ("expression", "moment", "expected"),
    [
        ("30 2 * * *", at(2026, 7, 19, 2, 30), True),
        ("30 2 * * *", at(2026, 7, 19, 2, 31), False),
        ("*/15 * * * *", at(2026, 7, 19, 9, 45), True),
        ("*/15 * * * *", at(2026, 7, 19, 9, 46), False),
        ("0 9-17 * * 1-5", at(2026, 7, 20, 9, 0), True),  # a Monday
        ("0 9-17 * * 1-5", at(2026, 7, 19, 9, 0), False),  # a Sunday: only one day field is set, so AND
        ("0 0 1 1 *", at(2027, 1, 1, 0, 0), True),
    ],
)
def test_matching(expression, moment, expected):
    assert parse(expression).matches(moment) is expected


def test_sunday_is_both_zero_and_seven():
    assert parse("0 3 * * 0").days_of_week == parse("0 3 * * 7").days_of_week


def test_day_of_month_and_day_of_week_are_ORed_not_ANDed():
    """The famous gotcha: `0 0 13 * 5` is the 13th OR any Friday, not Friday the 13th.

    2026-11-13 is a Friday, so it matches on both counts. 2026-07-13 is a
    Monday and matches only the day-of-month. 2026-07-17 is a Friday and
    matches only the day-of-week. Under an AND reading the last two would not
    fire at all — and a job written that way runs about eight times more often
    than its author expected.
    """
    schedule = parse("0 0 13 * 5")
    assert schedule.dom_restricted and schedule.dow_restricted
    assert schedule.matches(at(2026, 11, 13))  # Friday the 13th: both
    assert schedule.matches(at(2026, 7, 13))  # a Monday: day-of-month only
    assert schedule.matches(at(2026, 7, 17))  # a Friday: day-of-week only
    assert not schedule.matches(at(2026, 7, 14))  # neither


def test_only_one_of_the_two_day_fields_restricted_means_AND():
    schedule = parse("0 0 * * 5")  # every Friday
    assert schedule.matches(at(2026, 7, 17))
    assert not schedule.matches(at(2026, 7, 13))


def test_next_run_after_a_daily_schedule():
    schedule = parse("30 2 * * *")
    assert schedule.next_run_after(at(2026, 7, 19, 1, 0)) == at(2026, 7, 19, 2, 30)
    assert schedule.next_run_after(at(2026, 7, 19, 2, 30)) == at(2026, 7, 20, 2, 30)


def test_next_run_skips_to_the_right_weekday():
    schedule = parse("0 6 * * 1")  # Mondays at 06:00
    # 2026-07-19 is a Sunday; the next Monday is the 20th.
    assert schedule.next_run_after(at(2026, 7, 19, 12, 0)) == at(2026, 7, 20, 6, 0)


def test_describe_names_the_or_rule():
    assert "OR" in parse("0 0 13 * 5").describe()
    assert "OR" not in parse("30 2 * * *").describe()

"""Drift, measured — and a `sched` schedule run through six hours in no time."""

from __future__ import annotations

from clock import FakeTime
from inprocess import (
    deadline_corrected_loop,
    naive_sleep_loop,
    periodic_with_sched,
    run_sched_schedule,
)


def test_the_naive_sleep_loop_drifts_by_the_work_duration_every_run():
    """`work(); sleep(60)` with 5 seconds of work is a 65-second schedule.

    Nobody wrote 65 anywhere. After 100 runs the job is 495 seconds — more
    than eight minutes — behind where its author believes it is, and the
    error grows without limit.
    """
    trace = naive_sleep_loop(runs=100, interval=60, work_seconds=5)
    assert trace.lateness[0] == 0
    assert trace.lateness[1] == 5
    assert trace.final_drift == 495
    assert trace.starts[1] - trace.starts[0] == 65


def test_the_deadline_corrected_loop_does_not_drift():
    trace = deadline_corrected_loop(runs=100, interval=60, work_seconds=5)
    assert set(trace.lateness) == {0}
    assert trace.starts[1] - trace.starts[0] == 60


def test_a_run_longer_than_the_interval_makes_the_next_one_immediate():
    """Correction cannot invent time: overrun means the next run starts at once."""
    trace = deadline_corrected_loop(runs=5, interval=10, work_seconds=25)
    assert trace.starts[1] - trace.starts[0] == 25
    assert trace.final_drift > 0  # honest: this schedule cannot be met


def test_sched_runs_a_six_hour_schedule_instantly():
    fired, fake = run_sched_schedule(delays=[3600, 7200, 21600])
    assert [index for _, index in fired] == [0, 1, 2]
    assert [when for when, _ in fired] == [3600, 7200, 21600]
    assert fake.total_slept == 21600  # six hours "waited", zero seconds spent


def test_sched_runs_events_in_time_order_not_insertion_order():
    fired, _ = run_sched_schedule(delays=[300, 60, 120])
    assert [index for _, index in fired] == [1, 2, 0]


def test_a_recurring_sched_job_re_enters_itself_without_drifting():
    fake = FakeTime()
    starts = periodic_with_sched(
        interval=900, runs=8, work=lambda: fake.sleep(11), fake=fake
    )
    assert starts == [0, 900, 1800, 2700, 3600, 4500, 5400, 6300]

"""The operational properties: idempotence, locking, timeout, logging, heartbeat.

Every test here injects the clock, so a run that "takes a second" or a
heartbeat that is "two days old" costs nothing at all.
"""

from __future__ import annotations

import datetime as dt
import json
from pathlib import Path

import pytest
from clock import frozen_clock, ticking_clock
from joblock import AlreadyRunning, job_lock
from reportjob import (
    already_written,
    build_report,
    generate_daily_report,
    load_readings,
    report_path,
)
from runner import (
    EXIT_ALREADY_RUNNING,
    EXIT_FAILED,
    EXIT_OK,
    EXIT_TIMEOUT,
    JobTimeout,
    jsonl_logger,
    run_job,
)

UTC = dt.timezone.utc
NOW = dt.datetime(2026, 7, 20, 2, 30, tzinfo=UTC)
DAY = dt.date(2026, 7, 19)
DATA = Path(__file__).resolve().parent / "data" / "readings.csv"


@pytest.fixture
def collected():
    events: list[dict] = []
    return events, events.append


# --------------------------------------------------------------------------
# The work itself
# --------------------------------------------------------------------------


def test_the_report_summarises_only_the_requested_day():
    readings = load_readings(DATA, DAY)
    assert sorted(readings) == ["ALPHA", "BRAVO", "CHARLIE"]
    report = build_report(readings, report_date=DAY, generated_at=NOW)
    assert report.reading_count == 6
    alpha = next(s for s in report.stations if s.station == "ALPHA")
    assert alpha.count == 3
    assert alpha.minimum == 16.8
    assert alpha.maximum == 19.6
    assert round(alpha.mean, 4) == round((16.8 + 18.0 + 19.6) / 3, 4)


def test_running_the_job_twice_leaves_exactly_one_report(tmp_path):
    """Idempotence: the property that makes retries and catch-up runs safe."""
    first = generate_daily_report(
        source=DATA, output_dir=tmp_path, report_date=DAY, generated_at=NOW
    )
    second = generate_daily_report(
        source=DATA,
        output_dir=tmp_path,
        report_date=DAY,
        generated_at=NOW + dt.timedelta(hours=1),
    )
    assert first[0] == "written"
    assert second[0] == "skipped"
    assert first[1] == second[1]
    assert sorted(p.name for p in tmp_path.iterdir()) == [f"report-{DAY.isoformat()}.json"]
    payload = json.loads(first[1].read_text())
    assert payload["reading_count"] == 6
    # The second run did not rewrite the file: the timestamp is the first run's.
    assert payload["generated_at"] == NOW.isoformat()


def test_a_truncated_output_file_is_not_mistaken_for_a_finished_run(tmp_path):
    """A crash mid-write must produce a retry, not a permanent silent skip."""
    partial = report_path(tmp_path, DAY)
    partial.write_text('{"report_date": "2026-07-19", "stat')
    assert already_written(tmp_path, DAY) is False
    status, path = generate_daily_report(
        source=DATA, output_dir=tmp_path, report_date=DAY, generated_at=NOW
    )
    assert status == "written"
    assert json.loads(path.read_text())["reading_count"] == 6


def test_no_partial_files_are_left_behind(tmp_path):
    generate_daily_report(
        source=DATA, output_dir=tmp_path, report_date=DAY, generated_at=NOW
    )
    assert [p.name for p in tmp_path.iterdir() if p.name.endswith(".partial")] == []


# --------------------------------------------------------------------------
# The lock
# --------------------------------------------------------------------------


def test_the_lock_is_exclusive_within_one_process(tmp_path):
    path = tmp_path / "job.lock"
    with job_lock(path):
        with pytest.raises(AlreadyRunning):
            with job_lock(path):
                pytest.fail("the second acquisition should have been refused")


def test_the_lock_is_released_even_when_the_work_raises(tmp_path):
    path = tmp_path / "job.lock"
    with pytest.raises(ValueError):
        with job_lock(path):
            raise ValueError("boom")
    with job_lock(path):  # must not raise
        pass


def test_a_second_run_under_a_held_lock_exits_75_and_does_no_work(tmp_path, collected):
    events, log = collected
    path = tmp_path / "job.lock"
    calls = []

    with job_lock(path):
        run = run_job(
            name="daily-report",
            work=lambda: calls.append("worked") or {},
            clock=frozen_clock(NOW),
            lock_path=path,
            log=log,
        )
    assert run.exit_code == EXIT_ALREADY_RUNNING
    assert run.status == "already-running"
    assert calls == []  # the work never ran — this is the assertion that matters
    assert events[-1]["status"] == "already-running"


# --------------------------------------------------------------------------
# The runner: statuses, exit codes, timeout, logs, heartbeat
# --------------------------------------------------------------------------


def test_a_successful_run_exits_zero_and_logs_one_line(tmp_path, collected):
    events, log = collected
    run = run_job(
        name="daily-report",
        work=lambda: {"rows": 6},
        clock=ticking_clock(NOW, dt.timedelta(seconds=2)),
        lock_path=tmp_path / "job.lock",
        log=log,
    )
    assert run.exit_code == EXIT_OK
    assert run.status == "ok"
    assert run.duration_seconds == 2.0
    assert len(events) == 1
    assert events[0]["run_id"] == "daily-report-20260720T023000+0000"
    assert events[0]["rows"] == 6


def test_a_failing_run_exits_one_and_names_the_exception(tmp_path, collected):
    events, log = collected

    def work():
        raise RuntimeError("the upstream feed returned nothing")

    run = run_job(
        name="daily-report",
        work=work,
        clock=frozen_clock(NOW),
        lock_path=tmp_path / "job.lock",
        log=log,
    )
    assert run.exit_code == EXIT_FAILED
    assert events[0]["error"] == "RuntimeError"
    assert "upstream feed" in events[0]["message"]


def test_a_hung_job_is_interrupted_by_the_timeout(tmp_path, collected):
    """A real hang, killed by a real SIGALRM — the only test here that waits.

    It waits for 0.2 seconds, not for the 30 the job asks for, which is the
    whole point of having a timeout at all.
    """
    import time

    events, log = collected
    started = time.monotonic()
    run = run_job(
        name="daily-report",
        work=lambda: time.sleep(30),
        clock=frozen_clock(NOW),
        lock_path=tmp_path / "job.lock",
        log=log,
        timeout_seconds=0.2,
    )
    elapsed = time.monotonic() - started
    assert run.exit_code == EXIT_TIMEOUT
    assert run.status == "timeout"
    assert elapsed < 5, f"the timeout did not fire; the test waited {elapsed:.1f}s"
    assert events[0]["error"] == "JobTimeout"


def test_the_timeout_releases_the_lock_so_the_next_run_can_start(tmp_path, collected):
    import time

    events, log = collected
    lock = tmp_path / "job.lock"
    run_job(
        name="daily-report",
        work=lambda: time.sleep(30),
        clock=frozen_clock(NOW),
        lock_path=lock,
        log=log,
        timeout_seconds=0.2,
    )
    with job_lock(lock):  # must not raise: a hang must not block every later run
        pass


def test_a_timeout_of_none_disables_the_alarm(tmp_path, collected):
    _, log = collected
    run = run_job(
        name="daily-report",
        work=lambda: {},
        clock=frozen_clock(NOW),
        lock_path=tmp_path / "job.lock",
        log=log,
        timeout_seconds=None,
    )
    assert run.exit_code == EXIT_OK


def test_jobtimeout_is_a_timeouterror():
    assert issubclass(JobTimeout, TimeoutError)


def test_the_log_is_one_json_object_per_line(tmp_path):
    log_path = tmp_path / "job.log"
    log = jsonl_logger(log_path)
    for index in range(3):
        run_job(
            name="daily-report",
            work=lambda i=index: {"index": i},
            clock=frozen_clock(NOW + dt.timedelta(minutes=index)),
            lock_path=tmp_path / "job.lock",
            log=log,
        )
    lines = log_path.read_text().strip().splitlines()
    assert len(lines) == 3
    events = [json.loads(line) for line in lines]
    assert [e["index"] for e in events] == [0, 1, 2]
    assert {e["run_id"] for e in events} == {
        "daily-report-20260720T023000+0000",
        "daily-report-20260720T023100+0000",
        "daily-report-20260720T023200+0000",
    }


def test_only_a_success_writes_a_heartbeat(tmp_path, collected):
    _, log = collected
    heartbeat = tmp_path / "heartbeat.json"

    def failing():
        raise RuntimeError("no")

    run_job(
        name="daily-report",
        work=failing,
        clock=frozen_clock(NOW),
        lock_path=tmp_path / "job.lock",
        log=log,
        heartbeat_path=heartbeat,
    )
    assert not heartbeat.exists()

    run_job(
        name="daily-report",
        work=lambda: {},
        clock=frozen_clock(NOW),
        lock_path=tmp_path / "job.lock",
        log=log,
        heartbeat_path=heartbeat,
    )
    assert json.loads(heartbeat.read_text())["last_success"] == NOW.isoformat()


def test_a_skipped_run_still_counts_as_success(tmp_path, collected):
    """An idempotent no-op must exit 0. Alerting on it would train people to ignore alerts."""
    events, log = collected
    run = run_job(
        name="daily-report",
        work=lambda: {"status": "skipped", "action": "skipped"},
        clock=frozen_clock(NOW),
        lock_path=tmp_path / "job.lock",
        log=log,
    )
    assert run.exit_code == EXIT_OK
    assert run.status == "skipped"
    assert events[0]["status"] == "skipped"

"""The generated schedule files say what they claim — and install nothing.

The last two tests are the ones that matter for safety: they read every file
in this lab and assert that no code path anywhere executes `crontab`,
`launchctl` or `systemctl`. The install commands exist only as text to be
read.
"""

from __future__ import annotations

import datetime as dt
import re
from pathlib import Path

from cronexpr import parse
from gen_schedules import JobSchedule, write_all

UTC = dt.timezone.utc
HERE = Path(__file__).resolve().parent
LAB = HERE.parent

DAILY = JobSchedule(label="com.example.dailyreport", minute=30, hour=2)
WEEKDAYS = JobSchedule(
    label="com.example.weekdayreport", minute=15, hour=6, days_of_week=(1, 2, 3, 4, 5)
)


def test_the_generated_cron_line_parses_to_the_intended_schedule():
    schedule = parse(DAILY.cron_expression)
    assert DAILY.cron_expression == "30 2 * * *"
    base = dt.datetime(2026, 7, 19, 0, 0, tzinfo=UTC)
    assert schedule.next_run_after(base) == dt.datetime(2026, 7, 19, 2, 30, tzinfo=UTC)
    assert schedule.next_run_after(
        dt.datetime(2026, 7, 19, 2, 30, tzinfo=UTC)
    ) == dt.datetime(2026, 7, 20, 2, 30, tzinfo=UTC)


def test_the_weekday_schedule_skips_the_weekend():
    schedule = parse(WEEKDAYS.cron_expression)
    assert WEEKDAYS.cron_expression == "15 6 * * 1,2,3,4,5"
    # 2026-07-17 is a Friday; the next run is Monday the 20th.
    assert schedule.next_run_after(
        dt.datetime(2026, 7, 17, 12, 0, tzinfo=UTC)
    ) == dt.datetime(2026, 7, 20, 6, 15, tzinfo=UTC)


def test_the_cron_file_sets_the_environment_cron_does_not_give_you():
    text = DAILY.cron_line()
    assert "PATH=/usr/local/bin:/usr/bin:/bin" in text
    assert "SHELL=/bin/sh" in text
    assert "TZ=UTC" in text
    assert f"cd {DAILY.project_dir}" in text  # cron starts in $HOME, not your project
    assert ">>" in text and "2>&1" in text  # output goes somewhere on purpose


def test_all_three_dialects_agree_on_the_same_moment():
    """One schedule definition, three syntaxes, 02:30 in every one of them."""
    plist = DAILY.launchd_plist()
    assert DAILY.cron_expression == "30 2 * * *"
    assert "<key>Hour</key><integer>2</integer>" in plist
    assert "<key>Minute</key><integer>30</integer>" in plist
    assert DAILY.on_calendar == "*-*-* 02:30:00"


def test_the_launchd_plist_declares_every_key_that_matters():
    plist = DAILY.launchd_plist()
    for required in (
        "Label",  # what launchd calls the job
        "ProgramArguments",  # argv, not a shell command line
        "StartCalendarInterval",  # when
        "RunAtLoad",  # loading must not mean running
        "StandardOutPath",  # otherwise output is discarded
        "StandardErrorPath",
        "EnvironmentVariables",  # launchd gives a minimal environment, like cron
    ):
        assert f"<key>{required}</key>" in plist, f"{required} missing from the plist"
    assert "<false/>" in plist  # RunAtLoad is false
    for argument in DAILY.argv:
        assert f"<string>{argument}</string>" in plist


def test_the_weekday_plist_has_one_calendar_entry_per_day():
    plist = WEEKDAYS.launchd_plist()
    assert plist.count("<key>Weekday</key>") == 5
    for day in (1, 2, 3, 4, 5):
        assert f"<key>Weekday</key><integer>{day}</integer>" in plist


def test_the_systemd_pair_splits_what_from_when():
    service = DAILY.systemd_service()
    timer = DAILY.systemd_timer()
    assert "Type=oneshot" in service
    assert "ExecStart=" in service
    assert "OnCalendar" not in service  # the service has no schedule of its own
    assert "OnCalendar=*-*-* 02:30:00" in timer
    assert "Persistent=true" in timer  # catch-up after downtime, which cron lacks
    assert "WantedBy=timers.target" in timer


def test_writing_the_files_produces_four_readable_artefacts(tmp_path):
    written = write_all(DAILY, tmp_path)
    assert sorted(p.suffix for p in written) == [".cron", ".plist", ".service", ".timer"]
    for path in written:
        assert path.read_text(encoding="utf-8").strip()


def test_the_committed_example_files_match_what_the_generator_produces():
    """The files in examples/schedules are generated, not hand-edited."""
    committed = LAB / "examples" / "schedules"
    assert (committed / "com.example.dailyreport.cron").read_text() == DAILY.cron_line()
    assert (committed / "com.example.dailyreport.plist").read_text() == DAILY.launchd_plist()
    assert (
        committed / "com.example.dailyreport.service"
    ).read_text() == DAILY.systemd_service()
    assert (committed / "com.example.dailyreport.timer").read_text() == DAILY.systemd_timer()


# --------------------------------------------------------------------------
# Safety: this lab installs nothing, anywhere, ever.
# --------------------------------------------------------------------------

INSTALLERS = re.compile(r"\b(crontab|launchctl|systemctl|launchd|at|batch)\b")
EXECUTION = re.compile(
    r"(subprocess\.(run|call|check_call|check_output|Popen)|os\.(system|exec|spawn)|"
    r"^\s*(crontab|launchctl|systemctl)\s)",
    re.MULTILINE,
)


def _source_files() -> list[Path]:
    files = []
    for pattern in ("*.py", "*.sh"):
        files.extend(sorted((LAB / "examples").glob(pattern)))
        files.extend(sorted((LAB / "starter").glob(pattern)))
        files.extend(sorted((LAB / "tests").glob(pattern)))
    return files


def test_no_file_in_this_lab_executes_a_scheduler_command():
    """The rule, enforced: install commands are printed, never run."""
    offenders = []
    for path in _source_files():
        for number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
            if EXECUTION.search(line) and INSTALLERS.search(line):
                offenders.append(f"{path.name}:{number}: {line.strip()}")
    assert offenders == [], "a file appears to execute a scheduler command:\n" + "\n".join(
        offenders
    )


def test_the_generator_writes_only_where_it_is_told(tmp_path):
    """No default output path, no home directory, no surprises."""
    import gen_schedules

    before = sorted(tmp_path.iterdir())
    assert before == []
    gen_schedules.main(
        ["--out", str(tmp_path), "--hour", "3", "--minute", "5", "--label", "test.job"]
    )
    names = sorted(p.name for p in tmp_path.iterdir())
    assert names == [
        "test.job.cron",
        "test.job.plist",
        "test.job.service",
        "test.job.timer",
    ]

"""Daylight saving, tested in milliseconds because the clock is a parameter.

Waiting for 2 November to find out what a job does on 1 November is not a
plan. These tests ask the operating system's own time zone database what the
transitions are, and assert on the answer.
"""

from __future__ import annotations

import datetime as dt

from timezones import classify_wall_time, daily_instants_local, gaps_between

UTC = dt.timezone.utc
ZONE = "America/New_York"


def test_an_ordinary_wall_time_happens_exactly_once():
    verdict = classify_wall_time(dt.datetime(2026, 7, 19, 2, 30), ZONE)
    assert verdict.kind == "normal"
    assert len(verdict.instants) == 1


def test_0230_does_not_exist_on_the_spring_forward_morning():
    """2026-03-08: US clocks jump 02:00 to 03:00, so 02:30 never happens.

    A daily job at 02:30 local time therefore has no instant to run at, and
    Python resolves the impossible wall time to 03:30 local — an hour later
    than anyone intended.
    """
    verdict = classify_wall_time(dt.datetime(2026, 3, 8, 2, 30), ZONE)
    assert verdict.kind == "skipped"
    assert not verdict.exists
    assert verdict.instants[0] == dt.datetime(2026, 3, 8, 7, 30, tzinfo=UTC)


def test_0130_happens_twice_on_the_fall_back_morning():
    """2026-11-01: 01:30 occurs once as EDT and again an hour later as EST."""
    verdict = classify_wall_time(dt.datetime(2026, 11, 1, 1, 30), ZONE)
    assert verdict.kind == "repeated"
    assert verdict.instants == (
        dt.datetime(2026, 11, 1, 5, 30, tzinfo=UTC),
        dt.datetime(2026, 11, 1, 6, 30, tzinfo=UTC),
    )
    assert (verdict.instants[1] - verdict.instants[0]) == dt.timedelta(hours=1)


def test_a_local_daily_schedule_has_a_23_hour_and_a_25_hour_day():
    spring = gaps_between(
        daily_instants_local(
            start_date=dt.date(2026, 3, 6), days=5, hour=12, minute=0, zone_name=ZONE
        )
    )
    autumn = gaps_between(
        daily_instants_local(
            start_date=dt.date(2026, 10, 30), days=5, hour=12, minute=0, zone_name=ZONE
        )
    )
    assert 23.0 in spring, spring
    assert 25.0 in autumn, autumn


def test_a_utc_schedule_is_always_exactly_24_hours():
    """The whole argument for UTC, in one assertion."""
    for start in (dt.date(2026, 3, 6), dt.date(2026, 10, 30)):
        gaps = gaps_between(
            daily_instants_local(
                start_date=start, days=5, hour=12, minute=0, zone_name="UTC"
            )
        )
        assert set(gaps) == {24.0}

"""The dead man's switch: alerting on silence rather than on failure.

Each of these tests describes a real outage shape. None of them takes longer
than a millisecond, because the only thing that had to pass was time, and time
is a parameter.
"""

from __future__ import annotations

import datetime as dt
import json

from clock import frozen_clock
from watchdog import MISSING, OK, STALE, UNREADABLE, check_heartbeat

UTC = dt.timezone.utc
NOW = dt.datetime(2026, 7, 20, 9, 0, tzinfo=UTC)
BUDGET = dt.timedelta(hours=26)  # one daily interval plus a couple of hours


def write_heartbeat(path, moment):
    path.write_text(json.dumps({"job": "daily-report", "last_success": moment.isoformat()}))


def test_a_recent_success_is_quiet(tmp_path):
    path = tmp_path / "heartbeat.json"
    write_heartbeat(path, NOW - dt.timedelta(hours=7))
    verdict = check_heartbeat(heartbeat_path=path, clock=frozen_clock(NOW), max_age=BUDGET)
    assert verdict.state == OK
    assert not verdict.alerting


def test_one_missed_run_is_tolerated(tmp_path):
    """A single late run should not page anyone. Two intervals is the usual budget."""
    path = tmp_path / "heartbeat.json"
    write_heartbeat(path, NOW - dt.timedelta(hours=25))
    assert check_heartbeat(
        heartbeat_path=path, clock=frozen_clock(NOW), max_age=BUDGET
    ).state == OK


def test_a_job_that_stopped_two_days_ago_alerts(tmp_path):
    """The outage nothing else catches: the job was removed and never ran again.

    No failure, no traceback, no non-zero exit — because there was no run.
    Only the absence of a success says anything at all.
    """
    path = tmp_path / "heartbeat.json"
    write_heartbeat(path, NOW - dt.timedelta(days=2))
    verdict = check_heartbeat(heartbeat_path=path, clock=frozen_clock(NOW), max_age=BUDGET)
    assert verdict.state == STALE
    assert verdict.alerting
    assert "stopped running" in verdict.message
    assert verdict.age_seconds == 2 * 86400


def test_a_job_that_never_ran_at_all_alerts(tmp_path):
    verdict = check_heartbeat(
        heartbeat_path=tmp_path / "nothing.json", clock=frozen_clock(NOW), max_age=BUDGET
    )
    assert verdict.state == MISSING
    assert verdict.alerting


def test_a_corrupt_heartbeat_alerts_rather_than_passing(tmp_path):
    path = tmp_path / "heartbeat.json"
    path.write_text("{ this is not json")
    assert check_heartbeat(
        heartbeat_path=path, clock=frozen_clock(NOW), max_age=BUDGET
    ).state == UNREADABLE


def test_a_naive_timestamp_is_refused_rather_than_guessed(tmp_path):
    """Comparing an aware 'now' with a naive record would raise; guessing would lie."""
    path = tmp_path / "heartbeat.json"
    path.write_text(json.dumps({"last_success": "2026-07-20T08:00:00"}))
    verdict = check_heartbeat(
        heartbeat_path=path, clock=frozen_clock(NOW), max_age=BUDGET
    )
    assert verdict.state == UNREADABLE
    assert "naive" in verdict.message


def test_the_boundary_is_exact(tmp_path):
    path = tmp_path / "heartbeat.json"
    write_heartbeat(path, NOW - BUDGET)
    assert check_heartbeat(
        heartbeat_path=path, clock=frozen_clock(NOW), max_age=BUDGET
    ).state == OK
    write_heartbeat(path, NOW - BUDGET - dt.timedelta(seconds=1))
    assert check_heartbeat(
        heartbeat_path=path, clock=frozen_clock(NOW), max_age=BUDGET
    ).state == STALE

"""What "02:30 every day" means on the two days a year it does not.

A schedule written in local time is a schedule with two broken days per year.
On the spring-forward morning some wall-clock times do not happen; on the
autumn morning some happen twice. A daily job at 02:30 local therefore misses
a run in spring, and a daily job at 01:30 local runs twice in autumn — once as
daylight time and once as standard time, an hour apart in real elapsed time.

If the job is idempotent, the doubled run is harmless and the missing one is a
gap. If it is not idempotent, the doubled run is a doubled invoice.

Everything below is computed from Python's ``zoneinfo``, which reads the IANA
time zone database that ships with the operating system. The rules encoded
there are the real ones, so these functions tell you what your machine
actually believes rather than what a lesson asserts.

The answer, every time, is: **schedule in UTC and convert for display.** UTC
has no transitions, so 24 hours after an instant is always the same clock time
and always exactly 24 hours.
"""

from __future__ import annotations

import datetime as dt
from dataclasses import dataclass
from zoneinfo import ZoneInfo

UTC = dt.timezone.utc


@dataclass(frozen=True)
class WallClockVerdict:
    """What one local wall-clock time turns out to mean."""

    wall_time: str
    zone: str
    kind: str  # "normal" | "skipped" | "repeated"
    instants: tuple[dt.datetime, ...]
    note: str

    @property
    def exists(self) -> bool:
        return self.kind != "skipped"


def classify_wall_time(naive: dt.datetime, zone_name: str) -> WallClockVerdict:
    """Does this local wall-clock time happen once, twice, or not at all?

    The method is the one PEP 495 made possible: build the same wall time with
    ``fold=0`` and ``fold=1`` and compare their UTC offsets.

    * offsets equal, and the value round-trips through UTC unchanged -> normal;
    * offsets differ, and both round-trip -> the time happens twice (fall back);
    * the value does not round-trip -> the time never happens (spring forward),
      and Python resolves it to the instant one hour away.
    """
    zone = ZoneInfo(zone_name)
    first = naive.replace(tzinfo=zone, fold=0)
    second = naive.replace(tzinfo=zone, fold=1)
    label = naive.strftime("%Y-%m-%d %H:%M")

    first_utc = first.astimezone(UTC)
    round_tripped = first_utc.astimezone(zone).replace(tzinfo=None)

    if round_tripped != naive:
        return WallClockVerdict(
            label,
            zone_name,
            "skipped",
            (first_utc,),
            (
                f"{label} never appears on the wall clock in {zone_name}; "
                f"the clocks jump over it. Python resolves it to "
                f"{first_utc.astimezone(zone):%H:%M %Z}, an hour later than intended."
            ),
        )

    if first.utcoffset() != second.utcoffset():
        return WallClockVerdict(
            label,
            zone_name,
            "repeated",
            (first_utc, second.astimezone(UTC)),
            (
                f"{label} happens twice in {zone_name}: once at "
                f"{first_utc:%H:%M} UTC ({first:%Z}) and again an hour later at "
                f"{second.astimezone(UTC):%H:%M} UTC ({second:%Z}). "
                "A job scheduled then runs twice unless it is idempotent."
            ),
        )

    return WallClockVerdict(
        label,
        zone_name,
        "normal",
        (first_utc,),
        f"{label} in {zone_name} is exactly one instant: {first_utc:%H:%M} UTC.",
    )


def daily_instants_local(
    *,
    start_date: dt.date,
    days: int,
    hour: int,
    minute: int,
    zone_name: str,
) -> list[dt.datetime]:
    """The UTC instants a "hour:minute every day, local time" job would fire at."""
    zone = ZoneInfo(zone_name)
    out = []
    for offset in range(days):
        day = start_date + dt.timedelta(days=offset)
        local = dt.datetime(day.year, day.month, day.day, hour, minute, tzinfo=zone)
        out.append(local.astimezone(UTC))
    return out


def gaps_between(instants: list[dt.datetime]) -> list[float]:
    """Hours between consecutive runs. For a daily job in UTC these are all 24."""
    return [
        (later - earlier).total_seconds() / 3600
        for earlier, later in zip(instants, instants[1:], strict=False)
    ]

"""Alerting on silence: the check that catches a job which simply stopped.

Every alert most people write fires on failure. Failure is the easy case: the
job ran, something went wrong, it exited non-zero, and something noticed.

The case that actually bites is the opposite. Somebody edits the crontab and
drops a line. A machine is rebuilt and the timer is not re-enabled. A lock file
on a network share is never released and every run exits 75 — quietly, because
75 is not a crash. In all three the job produces no error, because it produces
nothing at all, and a monitor watching for errors sees a clean, quiet,
completely broken system. Teams discover this months later, usually by
noticing that a number stopped moving.

The fix is a **dead man's switch**: the job writes down each success, and a
separate check alerts when that record gets too old. It inverts the question
from "did anything fail?" to "did the thing that should have happened, happen?"
— and only the second question has an answer when the job is gone.

The check itself is fifteen lines and takes the clock as a parameter, so
"what does this report the morning after a job stops?" is a test, not a wait.
"""

from __future__ import annotations

import datetime as dt
import json
from dataclasses import dataclass
from pathlib import Path

from clock import Clock

OK = "ok"
STALE = "stale"
MISSING = "missing"
UNREADABLE = "unreadable"


@dataclass(frozen=True)
class WatchdogVerdict:
    """What the watchdog concluded, and why — in words a pager can carry."""

    state: str
    message: str
    last_success: dt.datetime | None = None
    age_seconds: float | None = None

    @property
    def alerting(self) -> bool:
        return self.state != OK


def check_heartbeat(
    *,
    heartbeat_path: str | Path,
    clock: Clock,
    max_age: dt.timedelta,
) -> WatchdogVerdict:
    """Alert if the last recorded success is older than ``max_age``.

    Choose ``max_age`` as roughly two intervals plus the job's normal runtime.
    Too tight and one slow run pages somebody at 4 a.m.; too loose and a job
    can be dead for a day before anyone hears about it. Two intervals is the
    usual compromise: it tolerates exactly one missed run and no more.
    """
    path = Path(heartbeat_path)
    now = clock()
    if not path.exists():
        return WatchdogVerdict(
            MISSING,
            f"no heartbeat at {path.name}: the job has never recorded a success",
        )
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
        last_success = dt.datetime.fromisoformat(payload["last_success"])
    except (json.JSONDecodeError, KeyError, ValueError, UnicodeDecodeError) as exc:
        return WatchdogVerdict(
            UNREADABLE, f"heartbeat at {path.name} could not be read: {type(exc).__name__}"
        )
    if last_success.tzinfo is None:
        return WatchdogVerdict(
            UNREADABLE,
            f"heartbeat at {path.name} holds a naive timestamp; it cannot be compared safely",
        )
    age = (now - last_success).total_seconds()
    if age > max_age.total_seconds():
        return WatchdogVerdict(
            STALE,
            (
                f"last success was {_humanise(age)} ago "
                f"(budget {_humanise(max_age.total_seconds())}) — the job has stopped running"
            ),
            last_success,
            age,
        )
    return WatchdogVerdict(
        OK, f"last success {_humanise(age)} ago, within budget", last_success, age
    )


def _humanise(seconds: float) -> str:
    seconds = float(seconds)
    if seconds < 90:
        return f"{seconds:.0f}s"
    if seconds < 5400:
        return f"{seconds / 60:.0f}m"
    if seconds < 172800:
        return f"{seconds / 3600:.1f}h"
    return f"{seconds / 86400:.1f}d"

lesson_id: D081
day: 81
kind: python-program
languages: [python, bash]
setup_commands:
  - cd labs/sections/programming-with-python/day-081-scheduling-and-background-jobs
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
  - .venv/bin/pytest --version
run_commands:
  - python3 examples/demo.py
  - python3 examples/job.py --now 2026-07-20T02:30:00+00:00 run --output-dir /tmp/day081
  - 'python3 examples/job.py --now 2026-07-20T02:35:00+00:00 run --output-dir /tmp/day081   # idempotent: skipped'
  - 'python3 examples/job.py --now 2026-07-22T02:30:00+00:00 run --output-dir /tmp/day081 --date 2026-07-17 --simulate-hang 30 --timeout 1   # exit 124'
  - 'python3 examples/job.py --now 2026-08-01T09:00:00+00:00 watch --heartbeat-file /tmp/day081/daily-report.heartbeat.json --max-age-minutes 1560   # exit 1'
  - 'python3 examples/supervise.py --timeout 1 -- sleep 30   # exit 124'
  - 'python3 examples/gen_schedules.py --out /tmp/day081/schedules --hour 2 --minute 30   # writes files; installs nothing'
  - .venv/bin/pytest examples -q
  - .venv/bin/pytest starter -q
test_commands:
  - bash tests/run_tests.sh
cleanup_commands:
  - rm -rf /tmp/day081
  - rm -rf .venv
  - 'git checkout -- starter/  # optional: reset your work'
requires_network: true
requires_api_key: false
estimated_minutes: 30
last_executed: '2026-07-19'
executed_on: 'macOS 26.5.1 (Apple Silicon), Python 3.14.0, pytest 9.1.1, bash 3.2.57 — bash tests/run_tests.sh -> 56 checks, 0 failure(s), exit 0; pytest examples -q -> 61 passed in 0.54s; pytest starter -q -> 1 passed, 8 skipped. Nothing was installed into cron, launchd or systemd, and no process was left running (asserted by section 8 of the runner). requires_network is true only for the one-time pytest install; no test opens a socket.'

# Dependencies

One package, and it is only there to run the tests.

## pytest 9.1.1

- **What it is:** the test framework you have used since Day 71.
- **Why it is here:** the reference suite in `examples/` and your exercises in
  `starter/` are pytest tests. The bash runner (`tests/run_tests.sh`) drives
  pytest and then adds the end-to-end checks that need real processes and real
  exit codes.
- **Licence and cost:** MIT. Free and open source, no account, no key.
- **Install:** `python3 -m venv .venv` then
  `.venv/bin/pip install -r requirements/requirements.txt`.

## Everything else is the standard library — deliberately

Scheduling is one of the areas where Python's own batteries genuinely are
enough, and this lab makes that argument by not installing anything to do the
work:

| Module | What it does here |
| --- | --- |
| `sched` | the event scheduler, driven by an injected clock so nothing waits |
| `signal` | `SIGALRM` and `setitimer` for the in-process timeout |
| `fcntl` | `flock` for the "only one copy at a time" lock (POSIX only) |
| `subprocess` | supervising a child process and killing its process group |
| `datetime`, `zoneinfo` | aware timestamps and the real IANA time zone rules |
| `json`, `csv` | the report, and the structured log |
| `argparse` | the command-line interface, as on Day 80 |
| `tempfile`, `os.replace` | the atomic write that makes the job idempotent |
| `statistics` | `fmean` for the per-station averages |

## Libraries the lesson discusses but does NOT install

`schedule`, `APScheduler`, `croniter` and `Celery` are all real, free and open
source, and the lesson's Alternatives section describes each one accurately —
including what it would add and what it would cost you. None of them is
installed here, and no code in this lab imports any of them. Where the lesson
shows their syntax, it says plainly that the snippet was written from the
project's documented interface rather than captured from a run on this
machine.

## Network

Installing pytest needs the network once. **The lab itself never does** — no
test, script or check in this directory opens a socket, and
`tests/run_tests.sh` asserts that no networking module is even imported.

## Platform note

`fcntl` is POSIX. macOS and Linux have it; Windows does not, and the Windows
equivalent is `msvcrt.locking` or a named mutex. Run this lab under WSL on
Windows.

# Day 081 — the only third-party package this lab needs.
# Everything else it uses ships with Python: sched, signal, fcntl, subprocess,
# datetime, zoneinfo, json, csv, argparse, tempfile, statistics.
pytest==9.1.1

"""Make this directory importable, and put examples/ on the path too.

Your starter modules are imported by name (`import joblock`), and a few of the
exercises lean on finished pieces from `examples/` so you can work on one idea
at a time. `starter` comes first on the path, so your version always wins.
"""

from __future__ import annotations

import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent
EXAMPLES = HERE.parent / "examples"
for directory in (EXAMPLES, HERE):
    if str(directory) not in sys.path:
        sys.path.insert(0, str(directory))

"""YOUR WORKING FILE — exercises 1 to 4.

Five functions, each one an operational property a scheduled job needs. They
are ordered the way you would add them to a real job, and each one is worth
the few lines it costs.

Run your work with:

    .venv/bin/pytest starter -q

Every exercise is marked `@pytest.mark.skip` in `starter/test_myjob.py`.
Delete the skip line for the exercise you are attempting, then make it pass.
The finished versions live in `examples/` — read them after you have tried,
not before, because the whole value here is in getting the lock wrong once.
"""

from __future__ import annotations

import contextlib
import datetime as dt
import json
import os
import tempfile
from collections.abc import Callable, Iterator, Mapping
from pathlib import Path
from typing import Any

# ---------------------------------------------------------------------------
# Exercise 1 — idempotence
# ---------------------------------------------------------------------------
# Running a job twice must produce one result. Retries, catch-up runs, and an
# operator typing the command a second time all mean "run it twice", and all
# three happen.
#
# Two functions. `output_written` answers "has a COMPLETE result already been
# produced for this day?" — complete meaning the file parses as JSON and
# carries the right `report_date`, so a half-written file from a crashed run
# is retried rather than mistaken for a success. `write_atomically` writes to
# a temporary name in the SAME directory and then `os.replace`s it into place,
# which is atomic on POSIX. Same directory matters: os.replace is only atomic
# within one filesystem.
#
# Prove it with: pytest starter -q -k "idempot or partial"


def output_path(output_dir: Path, report_date: dt.date) -> Path:
    """The output name IS the idempotence key: one day, one file, one name."""
    return Path(output_dir) / f"report-{report_date.isoformat()}.json"


def output_written(output_dir: Path, report_date: dt.date) -> bool:
    """Return True only if a COMPLETE result already exists for this date."""
    raise NotImplementedError("Exercise 1a: check the file exists, parses, and matches the date")


def write_atomically(payload: Mapping[str, Any], output_dir: Path, report_date: dt.date) -> Path:
    """Write JSON via a temporary file in the same directory, then os.replace."""
    raise NotImplementedError("Exercise 1b: tempfile.NamedTemporaryFile(dir=...) then os.replace")


# ---------------------------------------------------------------------------
# Exercise 2 — the lock
# ---------------------------------------------------------------------------
# A job that takes longer than its interval will eventually be started while
# the previous copy is still running. Two copies writing one file is a
# corrupted file; two copies calling a paid API is a doubled bill.
#
# Use `fcntl.flock` with `LOCK_EX | LOCK_NB`. Do NOT write
# "if the lock file exists: exit" — there is a window between the check and
# the create where a second process slips through, and a stale file from a
# killed process blocks every run afterwards. flock has neither problem: it is
# atomic, and the kernel releases it when the process dies, however it dies.
#
# Raise AlreadyRunning when the lock is held. Release it in a finally, so a
# job that raises does not leave the lock held.
#
# Prove it with: pytest starter -q -k lock


class AlreadyRunning(RuntimeError):
    """Raised when another process already holds the job's lock."""


@contextlib.contextmanager
def my_job_lock(path: str | os.PathLike[str]) -> Iterator[Path]:
    """Hold an exclusive non-blocking lock for the block; raise AlreadyRunning if taken."""
    raise NotImplementedError("Exercise 2: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)")
    yield Path(path)  # noqa: B901 - keeps this a generator function for the contextmanager


# ---------------------------------------------------------------------------
# Exercise 3 — the timeout
# ---------------------------------------------------------------------------
# A job with no time budget can hang forever holding the lock, which silently
# stops every later run. A hang is worse than a crash: a crash is reported.
#
# Use `signal.signal(signal.SIGALRM, handler)` and
# `signal.setitimer(signal.ITIMER_REAL, seconds)`. The handler raises
# JobTimeout. Cancel the timer in a finally — an alarm left armed fires
# somewhere unrelated later, which is a genuinely confusing bug.
#
# `seconds is None` must mean "no limit" rather than "zero".
#
# Prove it with: pytest starter -q -k timeout


class JobTimeout(TimeoutError):
    """Raised inside the job when its time budget expires."""


@contextlib.contextmanager
def time_budget(seconds: float | None) -> Iterator[None]:
    """Raise JobTimeout if the block runs longer than `seconds`."""
    raise NotImplementedError("Exercise 3: SIGALRM + setitimer, cancelled in a finally")
    yield  # noqa: B901 - keeps this a generator function for the contextmanager


# ---------------------------------------------------------------------------
# Exercise 4 — structured logging
# ---------------------------------------------------------------------------
# You will not be watching when this fails. The log line is the entire record
# of what happened, so it needs enough context to answer "which run, when, how
# long, what happened, what did it exit with" without any other evidence.
#
# Write ONE JSON object per line, appended. One line per event means the file
# can be processed with grep and json.loads; a multi-line traceback in the
# middle of the stream means it cannot.
#
# Include at least: job, run_id, status, exit_code, started_at, finished_at,
# duration_seconds. Timestamps come from the injected clock, never from
# datetime.now() — that is what makes the log assertable in a test.
#
# Prove it with: pytest starter -q -k log


def make_event(
    *,
    job: str,
    status: str,
    exit_code: int,
    started_at: dt.datetime,
    finished_at: dt.datetime,
    **extra: Any,
) -> dict[str, Any]:
    """Build the log event. run_id must be derived from the name and start time."""
    raise NotImplementedError("Exercise 4a: build the dict, including run_id and duration_seconds")


def append_jsonl(path: str | Path, event: Mapping[str, Any]) -> None:
    """Append exactly one JSON object, on one line, to `path`."""
    raise NotImplementedError("Exercise 4b: open(path, 'a') and write json.dumps(event) + newline")


# ---------------------------------------------------------------------------
# Provided, so you can concentrate on the four exercises above.
# ---------------------------------------------------------------------------


def frozen_clock(moment: dt.datetime) -> Callable[[], dt.datetime]:
    if moment.tzinfo is None:
        raise ValueError("use an aware datetime")
    return lambda: moment


def sample_payload(report_date: dt.date, generated_at: dt.datetime) -> dict[str, Any]:
    return {
        "report_date": report_date.isoformat(),
        "generated_at": generated_at.isoformat(),
        "reading_count": 6,
    }


__all__ = [
    "AlreadyRunning",
    "JobTimeout",
    "append_jsonl",
    "frozen_clock",
    "make_event",
    "my_job_lock",
    "output_path",
    "output_written",
    "sample_payload",
    "time_budget",
    "write_atomically",
]

# Silence unused-import warnings while the exercises are unfinished; the
# finished versions use every one of these.
_UNUSED = (json, tempfile)

# Your notes — exercises 5 and 6

Write your answers underneath each heading, in sentences. These are the
questions that separate "I scheduled a job" from "I operate a job", and they
are worth more than the code above.

## Exercise 5 — write the schedule files, install nothing

Run the generator against a directory of your own:

```bash
python3 examples/gen_schedules.py --out starter/schedules \
    --hour 2 --minute 30 --project-dir /opt/reports --timezone UTC
```

Open all four files it wrote. Then answer:

### 5a. What does cron give your job, and what does it not?

_(List at least four things about a cron job's environment that differ from
your interactive terminal, and say which line of the generated `.cron` file
compensates for each one.)_

### 5b. What is the install command for each of the three schedulers, and why have you not run it?

_(Quote the three commands from the generator's output. Then say, in one
sentence each, what would change on this machine if you ran them.)_

### 5c. Which of the three would you choose for a job on a laptop, and which for a job on a server that must not miss a run?

_(Name the specific feature that decides it.)_

## Exercise 6 — the operational questions

### 6a. Your job takes 40 seconds and runs every 5 minutes. One day the upstream feed is slow and it takes 6 minutes. Describe exactly what happens, minute by minute, with and without the lock.

_(Your answer here.)_

### 6b. The machine was off from Friday evening until Monday morning. Should the three missed daily runs be made up, one made up, or none? Justify the answer for a report job, and then for a job that sends an email to a customer.

_(Your answer here.)_

### 6c. Your job has run successfully every day for six months. Today somebody edits the crontab and drops the line. How long until anybody notices, and what single mechanism would have caught it the next morning?

_(Your answer here.)_

### 6d. Which of the properties you built today would you add first to a job you inherited, and why that one?

_(Your answer here.)_

"""The tests for your exercises. Delete a `skip` line, then make it pass.

The first test needs no work from you — it runs green immediately, so you can
confirm your setup before you change anything. Everything after it is skipped
until you remove the decorator.

    .venv/bin/pytest starter -q          # 1 passed, 7 skipped, to begin with
    .venv/bin/pytest starter -q -k lock  # just exercise 2
"""

from __future__ import annotations

import datetime as dt
import json
import time

import pytest
from myjob import (
    AlreadyRunning,
    JobTimeout,
    append_jsonl,
    frozen_clock,
    make_event,
    my_job_lock,
    output_path,
    output_written,
    sample_payload,
    time_budget,
    write_atomically,
)

UTC = dt.timezone.utc
NOW = dt.datetime(2026, 7, 20, 2, 30, tzinfo=UTC)
DAY = dt.date(2026, 7, 19)


def test_the_setup_works():
    """No exercise here. If this passes, pytest can import your module."""
    assert output_path("/tmp/x", DAY).name == "report-2026-07-19.json"
    assert frozen_clock(NOW)() == NOW


# --- Exercise 1: idempotence ------------------------------------------------


@pytest.mark.skip(reason="Exercise 1 — delete this line when you attempt it")
def test_running_twice_leaves_exactly_one_output(tmp_path):
    assert output_written(tmp_path, DAY) is False
    first = write_atomically(sample_payload(DAY, NOW), tmp_path, DAY)
    assert output_written(tmp_path, DAY) is True
    # A second run must see the first one's work and do nothing.
    if not output_written(tmp_path, DAY):
        write_atomically(sample_payload(DAY, NOW + dt.timedelta(hours=1)), tmp_path, DAY)
    files = sorted(p.name for p in tmp_path.iterdir())
    assert files == ["report-2026-07-19.json"]
    assert json.loads(first.read_text())["generated_at"] == NOW.isoformat()


@pytest.mark.skip(reason="Exercise 1 — delete this line when you attempt it")
def test_a_partial_file_is_retried_not_skipped(tmp_path):
    """A crash mid-write must not look like a finished run for ever."""
    output_path(tmp_path, DAY).write_text('{"report_date": "2026-07-19", "read')
    assert output_written(tmp_path, DAY) is False
    write_atomically(sample_payload(DAY, NOW), tmp_path, DAY)
    assert output_written(tmp_path, DAY) is True
    assert [p.name for p in tmp_path.iterdir() if p.name.endswith(".partial")] == []


# --- Exercise 2: the lock ---------------------------------------------------


@pytest.mark.skip(reason="Exercise 2 — delete this line when you attempt it")
def test_a_second_lock_is_refused_immediately(tmp_path):
    path = tmp_path / "job.lock"
    started = time.monotonic()
    with my_job_lock(path):
        with pytest.raises(AlreadyRunning):
            with my_job_lock(path):
                pytest.fail("the second acquisition should have been refused")
    # "Refused", not "queued": a scheduled job must never wait for the lock.
    assert time.monotonic() - started < 1.0


@pytest.mark.skip(reason="Exercise 2 — delete this line when you attempt it")
def test_the_lock_is_released_even_when_the_work_raises(tmp_path):
    path = tmp_path / "job.lock"
    with pytest.raises(ValueError):
        with my_job_lock(path):
            raise ValueError("boom")
    with my_job_lock(path):
        pass  # must not raise


# --- Exercise 3: the timeout ------------------------------------------------


@pytest.mark.skip(reason="Exercise 3 — delete this line when you attempt it")
def test_a_hung_block_raises_jobtimeout_quickly():
    started = time.monotonic()
    with pytest.raises(JobTimeout):
        with time_budget(0.2):
            time.sleep(30)
    elapsed = time.monotonic() - started
    assert elapsed < 5, f"the alarm did not fire; the test waited {elapsed:.1f}s"


@pytest.mark.skip(reason="Exercise 3 — delete this line when you attempt it")
def test_a_budget_of_none_means_no_limit_and_leaves_no_alarm_armed():
    with time_budget(None):
        pass
    with time_budget(0.5):
        pass
    # If the timer were not cancelled, this sleep would be interrupted.
    time.sleep(0.7)


# --- Exercise 4: structured logging -----------------------------------------


@pytest.mark.skip(reason="Exercise 4 — delete this line when you attempt it")
def test_the_event_carries_enough_to_debug_a_run_you_did_not_watch():
    event = make_event(
        job="daily-report",
        status="ok",
        exit_code=0,
        started_at=NOW,
        finished_at=NOW + dt.timedelta(seconds=41),
        report_date=DAY.isoformat(),
    )
    for key in (
        "job",
        "run_id",
        "status",
        "exit_code",
        "started_at",
        "finished_at",
        "duration_seconds",
    ):
        assert key in event, f"the log event has no {key}"
    assert event["duration_seconds"] == 41
    assert event["report_date"] == "2026-07-19"
    assert NOW.strftime("%Y%m%d") in event["run_id"]


@pytest.mark.skip(reason="Exercise 4 — delete this line when you attempt it")
def test_the_log_is_one_json_object_per_line(tmp_path):
    path = tmp_path / "job.log"
    for index in range(3):
        append_jsonl(
            path,
            make_event(
                job="daily-report",
                status="ok",
                exit_code=0,
                started_at=NOW + dt.timedelta(minutes=index),
                finished_at=NOW + dt.timedelta(minutes=index, seconds=5),
                index=index,
            ),
        )
    lines = path.read_text().strip().splitlines()
    assert len(lines) == 3
    assert [json.loads(line)["index"] for line in lines] == [0, 1, 2]

#!/usr/bin/env bash
# Tests for the Day 081 lab. Run from the lab directory:
#   bash tests/run_tests.sh
#
# This suite proves the operational properties of a scheduled job WITHOUT
# scheduling anything. In particular, section 8 asserts the safety rule this
# lab is built around: nothing was installed into any real scheduler, and no
# background process was left running.
#
# The two checks worth reading first:
#
#   * "a second run under a held lock exits 75 and does no work" starts a
#     helper that takes the lock, waits for it to report READY, and then runs
#     the real job. The helper is always waited for, so the suite cannot leave
#     a process behind;
#   * "a hung job is killed by the timeout" runs a job that asks to sleep for
#     30 seconds with a 1-second budget, asserts exit 124, and asserts the
#     whole thing finished in under 10 seconds.
#
# No network, non-interactive, deterministic. Exits 0 only if every check
# passes.
set -u

export PYTHONDONTWRITEBYTECODE=1

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

cleanup() {
  # Never leave anything running or lying about, whatever happened above.
  if [ -n "${helper_pid}" ] && kill -0 "${helper_pid}" 2>/dev/null; then
    kill "${helper_pid}" 2>/dev/null
    wait "${helper_pid}" 2>/dev/null
  fi
  [ -n "${work_dir}" ] && [ -d "${work_dir}" ] && rm -rf "${work_dir}"
  return 0
}
trap cleanup EXIT INT TERM

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

# Resolve pytest: an explicit override, then this lab's .venv, then whatever
# is on PATH. Fails loudly with instructions rather than silently skipping.
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
}

pytest_bin="$(resolve_tool pytest "${PYTEST:-}")" || {
  echo "FAIL: pytest not found." >&2
  echo "  Install it with:" >&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 pytest: PYTEST=/path/to/pytest bash tests/run_tests.sh" >&2
  exit 1
}

python_bin="$(command -v python3 || true)"
if [ -z "${python_bin}" ]; then
  echo "FAIL: python3 not found on PATH." >&2
  exit 1
fi

work_dir="$(mktemp -d "${TMPDIR:-/tmp}/day081.XXXXXX")"

echo "Day 081 — A Job That Survives Being Ignored"
echo

# --------------------------------------------------------------------------
echo "1. The tools"
# --------------------------------------------------------------------------

version_line="$("${pytest_bin}" --version 2>&1 | head -1)"
case "${version_line}" in
  pytest*) check "pytest --version reports a pytest ( ${version_line} )" "yes" ;;
  *) check "pytest --version reports a pytest ( ${version_line} )" "no" ;;
esac

if "${python_bin}" -c "import fcntl, sched, signal, zoneinfo" 2>/dev/null; then
  check "fcntl, sched, signal and zoneinfo are all standard library — nothing to install" "yes"
else
  check "fcntl, sched, signal and zoneinfo are all standard library" "no"
fi

# --------------------------------------------------------------------------
echo
echo "2. The reference suite"
# --------------------------------------------------------------------------

examples_out="$(cd "${lab_dir}" && "${pytest_bin}" examples -q -p no:cacheprovider 2>&1)"
examples_exit=$?
if [ "${examples_exit}" -eq 0 ]; then
  check "pytest examples exits 0" "yes"
else
  check "pytest examples exits 0 (got ${examples_exit})" "no"
  echo "${examples_out}" | tail -25
fi
case "${examples_out}" in
  *"61 passed"*) check "pytest examples reports 61 passed" "yes" ;;
  *) check "pytest examples reports 61 passed (got: $(printf '%s' "${examples_out}" | tail -1))" "no" ;;
esac

# The suite must run fast, because everything slow in it was injected away.
suite_seconds="$(printf '%s\n' "${examples_out}" | sed -n 's/.* in \([0-9.]*\)s.*/\1/p' | tail -1)"
if [ -n "${suite_seconds}" ] && "${python_bin}" -c "import sys; sys.exit(0 if float('${suite_seconds}') < 10 else 1)"; then
  check "the whole suite runs in ${suite_seconds}s — nothing in it waits for a schedule" "yes"
else
  check "the whole suite runs in under 10s (got ${suite_seconds:-unknown}s)" "no"
fi

# --------------------------------------------------------------------------
echo
echo "3. Idempotence — running it twice leaves exactly one result"
# --------------------------------------------------------------------------

out_dir="${work_dir}/out"
run_one="$(cd "${lab_dir}" && "${python_bin}" examples/job.py \
  --now 2026-07-20T02:30:00+00:00 run --output-dir "${out_dir}" \
  --log-file "${out_dir}/job.log" 2>&1)"
first_exit=$?
run_two="$(cd "${lab_dir}" && "${python_bin}" examples/job.py \
  --now 2026-07-20T02:35:00+00:00 run --output-dir "${out_dir}" \
  --log-file "${out_dir}/job.log" 2>&1)"
second_exit=$?

[ "${first_exit}" -eq 0 ] \
  && check "the first run exits 0" "yes" \
  || check "the first run exits 0 (got ${first_exit})" "no"
[ "${second_exit}" -eq 0 ] \
  && check "the second run also exits 0 — an idempotent no-op is a success" "yes" \
  || check "the second run also exits 0 (got ${second_exit})" "no"

case "${run_one}" in
  *'"action": "written"'*) check "the first run reports action=written" "yes" ;;
  *) check "the first run reports action=written" "no" ;;
esac
case "${run_two}" in
  *'"action": "skipped"'*) check "the second run reports action=skipped — it did NOT redo the work" "yes" ;;
  *) check "the second run reports action=skipped" "no" ;;
esac

report_count="$(find "${out_dir}" -maxdepth 1 -name 'report-*.json' | wc -l | tr -d ' ')"
if [ "${report_count}" -eq 1 ]; then
  check "two runs produced exactly one report file" "yes"
else
  check "two runs produced exactly one report file (got ${report_count})" "no"
fi

# The second run must not have rewritten the file: its timestamp is the first's.
if "${python_bin}" -c "
import json, sys
payload = json.load(open('${out_dir}/report-2026-07-19.json'))
sys.exit(0 if payload['generated_at'] == '2026-07-20T02:30:00+00:00'
         and payload['reading_count'] == 6 else 1)
"; then
  check "the report still carries the FIRST run's timestamp and 6 readings" "yes"
else
  check "the report still carries the first run's timestamp and 6 readings" "no"
fi

partials="$(find "${out_dir}" -maxdepth 1 -name '*.partial*' | wc -l | tr -d ' ')"
if [ "${partials}" -eq 0 ]; then
  check "no partial files were left behind (the write was atomic)" "yes"
else
  check "no partial files were left behind (found ${partials})" "no"
fi

log_lines="$(wc -l < "${out_dir}/job.log" | tr -d ' ')"
if [ "${log_lines}" -eq 2 ]; then
  check "the log has one JSON line per run (2 lines for 2 runs)" "yes"
else
  check "the log has one JSON line per run (got ${log_lines})" "no"
fi
if "${python_bin}" -c "
import json, sys
lines = open('${out_dir}/job.log').read().strip().splitlines()
events = [json.loads(line) for line in lines]
needed = {'job','run_id','status','exit_code','started_at','finished_at','duration_seconds'}
sys.exit(0 if all(needed <= set(e) for e in events) else 1)
"; then
  check "every log line carries job, run_id, status, exit_code, times and duration" "yes"
else
  check "every log line carries job, run_id, status, exit_code, times and duration" "no"
fi

# --------------------------------------------------------------------------
echo
echo "4. Overlap — a second run under a held lock refuses to start"
# --------------------------------------------------------------------------

lock_path="${out_dir}/daily-report.lock"
helper_fifo="${work_dir}/helper.out"
# Started as a direct child (no subshell) so that ${helper_pid} really is the
# python process and `kill` reaches it rather than a wrapper.
"${python_bin}" "${lab_dir}/examples/hold_lock.py" "${lock_path}" 6 > "${helper_fifo}" 2>&1 &
helper_pid=$!

# Wait for READY rather than sleeping a fixed time.
ready="no"
for _ in $(seq 1 100); do
  if [ -s "${helper_fifo}" ] && grep -q READY "${helper_fifo}" 2>/dev/null; then
    ready="yes"
    break
  fi
  "${python_bin}" -c "import time; time.sleep(0.05)"
done
check "the lock helper took the lock and reported READY" "${ready}"

locked_out="$(cd "${lab_dir}" && "${python_bin}" examples/job.py \
  --now 2026-07-21T02:30:00+00:00 run --output-dir "${out_dir}" \
  --date 2026-07-20 --lock-file "${lock_path}" 2>&1)"
locked_exit=$?

if [ "${locked_exit}" -eq 75 ]; then
  check "a run under a held lock exits 75 (EX_TEMPFAIL), not 0 and not 1" "yes"
else
  check "a run under a held lock exits 75 (got ${locked_exit})" "no"
fi
case "${locked_out}" in
  *'"status": "already-running"'*) check "it logs status=already-running" "yes" ;;
  *) check "it logs status=already-running" "no" ;;
esac
if [ ! -f "${out_dir}/report-2026-07-20.json" ]; then
  check "the refused run did NOT do the work — no report for 2026-07-20" "yes"
else
  check "the refused run did NOT do the work" "no"
fi

# Stop the helper and reap it. Nothing is left running.
kill "${helper_pid}" 2>/dev/null
wait "${helper_pid}" 2>/dev/null
sleep_probe=0
while kill -0 "${helper_pid}" 2>/dev/null && [ "${sleep_probe}" -lt 40 ]; do
  "${python_bin}" -c "import time; time.sleep(0.05)"
  sleep_probe=$((sleep_probe + 1))
done
if kill -0 "${helper_pid}" 2>/dev/null; then
  check "the lock helper is gone once the check finishes" "no"
else
  check "the lock helper is gone once the check finishes" "yes"
fi
helper_pid=""

# And the lock is usable again immediately afterwards.
after_out="$(cd "${lab_dir}" && "${python_bin}" examples/job.py \
  --now 2026-07-21T02:30:00+00:00 run --output-dir "${out_dir}" \
  --date 2026-07-20 --lock-file "${lock_path}" 2>&1)"
after_exit=$?
if [ "${after_exit}" -eq 0 ] && [ -f "${out_dir}/report-2026-07-20.json" ]; then
  check "once the holder exits, the next run takes the lock and does the work" "yes"
else
  check "once the holder exits, the next run takes the lock and does the work (exit ${after_exit})" "no"
fi

# --------------------------------------------------------------------------
echo
echo "5. Timeouts — a hung job is killed, and does not block the next run"
# --------------------------------------------------------------------------

hang_start="$(date +%s)"
hang_out="$(cd "${lab_dir}" && "${python_bin}" examples/job.py \
  --now 2026-07-22T02:30:00+00:00 run --output-dir "${out_dir}" \
  --date 2026-07-17 --lock-file "${out_dir}/hang.lock" \
  --simulate-hang 30 --timeout 1 2>&1)"
hang_exit=$?
hang_elapsed=$(( $(date +%s) - hang_start ))

if [ "${hang_exit}" -eq 124 ]; then
  check "a job that hangs for 30s with a 1s budget exits 124, as GNU timeout does" "yes"
else
  check "a job that hangs for 30s with a 1s budget exits 124 (got ${hang_exit})" "no"
fi
if [ "${hang_elapsed}" -lt 10 ]; then
  check "it was killed after about a second, not after thirty (took ${hang_elapsed}s)" "yes"
else
  check "it was killed after about a second (took ${hang_elapsed}s)" "no"
fi
case "${hang_out}" in
  *'"error": "JobTimeout"'*) check "the timeout is logged as JobTimeout with its budget" "yes" ;;
  *) check "the timeout is logged as JobTimeout with its budget" "no" ;;
esac
if [ ! -f "${out_dir}/report-2026-07-17.json" ]; then
  check "the timed-out run produced no output" "yes"
else
  check "the timed-out run produced no output" "no"
fi

# The stronger form: supervise a child process and kill its whole group.
#
# The child is a uniquely named symlink to sleep rather than plain `sleep 30`.
# The orphan check below is a pattern match over the whole process table, so a
# generic name would match any unrelated process on the machine that happens to
# be sleeping — the check would then fail for reasons having nothing to do with
# this lab. The name carries this harness's own pid, so it can only match the
# child we started.
probe_name="day081-probe-sleep-$$"
probe_path="${work_dir}/${probe_name}"
ln -s "$(command -v sleep)" "${probe_path}"
sup_start="$(date +%s)"
(cd "${lab_dir}" && "${python_bin}" examples/supervise.py --timeout 1 -- "${probe_path}" 30 >/dev/null 2>&1)
sup_exit=$?
sup_elapsed=$(( $(date +%s) - sup_start ))
if [ "${sup_exit}" -eq 124 ] && [ "${sup_elapsed}" -lt 10 ]; then
  check "supervise.py kills an overrunning child process group and exits 124" "yes"
else
  check "supervise.py kills an overrunning child (exit ${sup_exit}, ${sup_elapsed}s)" "no"
fi
if pgrep -f "${probe_name}" >/dev/null 2>&1; then
  check "no supervised child survived the supervisor" "no"
else
  check "no supervised child survived the supervisor" "yes"
fi

# --------------------------------------------------------------------------
echo
echo "6. The watchdog — alerting on silence"
# --------------------------------------------------------------------------

heartbeat="${out_dir}/daily-report.heartbeat.json"
if [ -f "${heartbeat}" ]; then
  check "a successful run wrote a heartbeat" "yes"
else
  check "a successful run wrote a heartbeat" "no"
fi

(cd "${lab_dir}" && "${python_bin}" examples/job.py --now 2026-07-21T09:00:00+00:00 \
  watch --heartbeat-file "${heartbeat}" --max-age-minutes 1560 >/dev/null 2>&1)
fresh_exit=$?
[ "${fresh_exit}" -eq 0 ] \
  && check "the watchdog is quiet while the job is running (exit 0)" "yes" \
  || check "the watchdog is quiet while the job is running (got ${fresh_exit})" "no"

stale_out="$(cd "${lab_dir}" && "${python_bin}" examples/job.py --now 2026-08-01T09:00:00+00:00 \
  watch --heartbeat-file "${heartbeat}" --max-age-minutes 1560 2>&1)"
stale_exit=$?
if [ "${stale_exit}" -eq 1 ]; then
  check "eleven days later, with no run at all, the watchdog alerts (exit 1)" "yes"
else
  check "eleven days later the watchdog alerts (got ${stale_exit})" "no"
fi
case "${stale_out}" in
  *"stopped running"*) check "the alert says the job has stopped running" "yes" ;;
  *) check "the alert says the job has stopped running (got: ${stale_out})" "no" ;;
esac

missing_exit=0
(cd "${lab_dir}" && "${python_bin}" examples/job.py --now 2026-07-21T09:00:00+00:00 \
  watch --heartbeat-file "${work_dir}/never.json" >/dev/null 2>&1) || missing_exit=$?
[ "${missing_exit}" -eq 1 ] \
  && check "a job that has never succeeded also alerts" "yes" \
  || check "a job that has never succeeded also alerts (got ${missing_exit})" "no"

# --------------------------------------------------------------------------
echo
echo "7. The schedule files say what they claim"
# --------------------------------------------------------------------------

gen_out="$(cd "${lab_dir}" && "${python_bin}" examples/gen_schedules.py \
  --out "${work_dir}/schedules" --hour 2 --minute 30 2>&1)"
gen_exit=$?
[ "${gen_exit}" -eq 0 ] \
  && check "gen_schedules.py exits 0" "yes" \
  || check "gen_schedules.py exits 0 (got ${gen_exit})" "no"

for suffix in cron plist service timer; do
  if [ -s "${work_dir}/schedules/com.example.dailyreport.${suffix}" ]; then
    check "it wrote a non-empty .${suffix} file" "yes"
  else
    check "it wrote a non-empty .${suffix} file" "no"
  fi
done

case "${gen_out}" in
  *"NOTHING was installed"*) check "the generator states plainly that it installed nothing" "yes" ;;
  *) check "the generator states plainly that it installed nothing" "no" ;;
esac

if "${python_bin}" -c "
import sys
sys.path.insert(0, '${lab_dir}/examples')
import datetime as dt
from cronexpr import parse
line = [l for l in open('${work_dir}/schedules/com.example.dailyreport.cron')
        if l.strip() and not l.startswith('#') and ' ' in l and l.split()[0].isdigit()]
schedule = parse(' '.join(line[0].split()[:5]))
base = dt.datetime(2026, 7, 19, 0, 0, tzinfo=dt.timezone.utc)
first = schedule.next_run_after(base)
second = schedule.next_run_after(first)
sys.exit(0 if (first == dt.datetime(2026, 7, 19, 2, 30, tzinfo=dt.timezone.utc)
               and second == dt.datetime(2026, 7, 20, 2, 30, tzinfo=dt.timezone.utc)) else 1)
"; then
  check "the generated cron line parses to 02:30 daily, as intended" "yes"
else
  check "the generated cron line parses to 02:30 daily, as intended" "no"
fi

if grep -q 'PATH=/usr/local/bin:/usr/bin:/bin' "${work_dir}/schedules/com.example.dailyreport.cron" \
   && grep -q '^SHELL=' "${work_dir}/schedules/com.example.dailyreport.cron"; then
  check "the cron file sets PATH and SHELL explicitly — cron supplies almost nothing" "yes"
else
  check "the cron file sets PATH and SHELL explicitly" "no"
fi
if grep -q 'Persistent=true' "${work_dir}/schedules/com.example.dailyreport.timer"; then
  check "the systemd timer sets Persistent=true for catch-up after downtime" "yes"
else
  check "the systemd timer sets Persistent=true" "no"
fi
if grep -q '<key>RunAtLoad</key>' "${work_dir}/schedules/com.example.dailyreport.plist"; then
  check "the launchd plist declares RunAtLoad so loading does not mean running" "yes"
else
  check "the launchd plist declares RunAtLoad" "no"
fi

# --------------------------------------------------------------------------
echo
echo "8. SAFETY — nothing was installed, nothing was left running"
# --------------------------------------------------------------------------

# 8a. No lab code executes a scheduler command. This is the structural
#     guarantee: the install commands exist only as text to be read.
#     (This runner reads `crontab -l` in 8b, which is read-only and is the
#     check itself, so tests/ is deliberately outside the scan.)
if grep -rnE '(subprocess\.[A-Za-z_]+|os\.system|os\.exec[a-z]*)[^\n]*(crontab|launchctl|systemctl)' \
     "${lab_dir}/examples" "${lab_dir}/starter" 2>/dev/null >/dev/null; then
  check "no learner-facing file executes crontab, launchctl or systemctl" "no"
else
  check "no learner-facing file executes crontab, launchctl or systemctl" "yes"
fi

# 8b. The user's real crontab does not contain this lab's job.
if command -v crontab >/dev/null 2>&1; then
  real_crontab="$(crontab -l 2>/dev/null || true)"
  case "${real_crontab}" in
    *dailyreport*|*"examples/job.py"*)
      check "the real crontab contains no entry from this lab" "no" ;;
    *) check "the real crontab contains no entry from this lab" "yes" ;;
  esac
else
  check "crontab is not installed here, so nothing could have been added to it" "yes"
fi

# 8c. No launchd agent and no systemd user unit was installed.
agents="${HOME}/Library/LaunchAgents"
if [ -d "${agents}" ] && ls "${agents}" 2>/dev/null | grep -q 'com.example.dailyreport'; then
  check "no launchd agent named com.example.dailyreport exists in the user's LaunchAgents" "no"
else
  check "no launchd agent named com.example.dailyreport exists in the user's LaunchAgents" "yes"
fi

units="${HOME}/.config/systemd/user"
if [ -d "${units}" ] && ls "${units}" 2>/dev/null | grep -q 'com.example.dailyreport'; then
  check "no systemd user unit named com.example.dailyreport was installed" "no"
else
  check "no systemd user unit named com.example.dailyreport was installed" "yes"
fi

# 8d. Nothing from this lab is still running.
if pgrep -f 'hold_lock.py' >/dev/null 2>&1; then
  check "no hold_lock.py process survived the suite" "no"
else
  check "no hold_lock.py process survived the suite" "yes"
fi
if pgrep -f 'examples/job.py' >/dev/null 2>&1; then
  check "no job.py process survived the suite" "no"
else
  check "no job.py process survived the suite" "yes"
fi

# 8e. The lab wrote nothing outside its own directory and the temporary one.
if [ -z "$(find "${lab_dir}" -maxdepth 1 -name 'report-*.json' 2>/dev/null)" ]; then
  check "no report file was left in the lab directory itself" "yes"
else
  check "no report file was left in the lab directory itself" "no"
fi

# --------------------------------------------------------------------------
echo
echo "9. The starter is runnable before you start"
# --------------------------------------------------------------------------

starter_out="$(cd "${lab_dir}" && "${pytest_bin}" starter -q -p no:cacheprovider 2>&1)"
starter_exit=$?
if [ "${starter_exit}" -eq 0 ]; then
  check "pytest starter exits 0 with the exercises unfinished" "yes"
else
  check "pytest starter exits 0 with the exercises unfinished (got ${starter_exit})" "no"
  echo "${starter_out}" | tail -15
fi
case "${starter_out}" in
  *"1 passed, 8 skipped"*) check "the starter has 1 worked test and 8 skipped exercises" "yes" ;;
  *) check "the starter has 1 worked test and 8 skipped exercises" "no" ;;
esac

for heading in "Exercise 5" "Exercise 6" "5a." "6b." "6c."; do
  if grep -q "${heading}" "${lab_dir}/starter/NOTES.md"; then
    check "NOTES.md asks the '${heading}' question" "yes"
  else
    check "NOTES.md asks the '${heading}' question" "no"
  fi
done

# --------------------------------------------------------------------------
echo
echo "10. Nothing here reaches the network"
# --------------------------------------------------------------------------

if grep -rqE '^\s*(import|from)\s+(socket|urllib|http|requests|ftplib|smtplib)\b' \
     "${lab_dir}/examples" "${lab_dir}/starter" 2>/dev/null; then
  check "no networking module is imported anywhere in examples/ or starter/" "no"
else
  check "no networking module is imported anywhere in examples/ or starter/" "yes"
fi

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