Hands-on lab — Day 84: Shipping an Automation Toolkit

"""feedkit — a small, installable, scheduled automation toolkit.

The package is deliberately layered so that the boundaries sit at the edges:

    core            pure data in, pure data out — no network, no clock, no disk
    config          the four-layer precedence, resolved by a pure function
    logging_setup   structured JSON logging with secret redaction
    state           the state file, written atomically, plus the run lock
    adapters        the network and the clock — the only impure module
    runner          the order of one unattended run
    cli             argparse subcommands and the two console entry points

Read `core.py` first. It is where the interesting decisions live, and it is
readable without knowing anything about HTTP.
"""

__all__ = ["__version__"]

__version__ = "1.0.0"

"""The edges: the network, the clock, and sleeping.

Everything in this module talks to something outside the process. That is the
whole reason it is a separate module — the core can be tested with plain data
because none of this leaks into it, and this module can be swapped for a fake
in a test because the runner receives it as an argument rather than importing
it. Day 74's rule, applied to the two boundaries that hurt most.

Note the constructor of `HttpFetcher`: it takes a session, a timeout, a retry
count, a backoff base AND a `sleeper`. Injecting the sleeper is what lets the
test suite exercise three retries in microseconds instead of seconds, without
anybody having to patch `time.sleep` globally and hope.
"""

from __future__ import annotations

import time
from datetime import datetime, timezone
from typing import Any, Callable, Protocol

import requests


class FetchError(RuntimeError):
    """A source could not be fetched, after every retry was spent."""


class Clock(Protocol):
    """The clock, as an interface, so a test can hand over a fixed time."""

    def now_iso(self) -> str: ...


class SystemClock:
    """The real clock. UTC, always — a job that runs at 02:30 local time runs
    twice or not at all on the two days a year the offset changes."""

    def now_iso(self) -> str:
        return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


class FixedClock:
    """A clock that never moves. Shipped rather than hidden in a test file,
    because it belongs to the design, not to the tests."""

    def __init__(self, value: str) -> None:
        self.value = value

    def now_iso(self) -> str:
        return self.value


#: Statuses worth trying again: the server said "not now", not "never".
RETRYABLE_STATUS = frozenset({408, 425, 429, 500, 502, 503, 504})


class HttpFetcher:
    """Fetch one source's JSON, with a timeout and bounded retries.

    Three things here are not optional in an unattended job:

    * a **timeout** on every request — the default in `requests` is no timeout
      at all, and a job with no timeout does not fail, it hangs, which is the
      one outcome no supervisor can see;
    * **bounded** retries with exponential backoff — unbounded retries turn a
      failing dependency into a self-inflicted outage;
    * a retry decision based on WHAT went wrong. A 503 is worth another go; a
      404 or a 401 will be a 404 or a 401 forever, and retrying it is just
      noise you will pay for in someone else's server logs.
    """

    def __init__(
        self,
        session: requests.Session,
        base_url: str,
        token: str = "",
        timeout: float = 5.0,
        retries: int = 3,
        backoff_seconds: float = 0.5,
        sleeper: Callable[[float], None] = time.sleep,
        logger: Any = None,
    ) -> None:
        self.session = session
        self.base_url = base_url.rstrip("/")
        self.token = token
        self.timeout = timeout
        self.retries = max(1, retries)
        self.backoff_seconds = backoff_seconds
        self.sleeper = sleeper
        self.logger = logger

    def _headers(self) -> dict[str, str]:
        headers = {
            "Accept": "application/json",
            # Identify the client honestly — Day 79's rule, and the thing that
            # lets an operator find you when your job misbehaves.
            "User-Agent": "feedkit/1.0 (personal automation toolkit)",
        }
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"
        return headers

    def fetch(self, source: str) -> tuple[Any, int]:
        """Return the decoded payload and the attempt number that succeeded."""
        url = f"{self.base_url}/feed/{source}.json"
        last_error = "no attempt was made"

        for attempt in range(1, self.retries + 1):
            try:
                response = self.session.get(url, headers=self._headers(), timeout=self.timeout)
            except requests.RequestException as exc:
                last_error = f"{type(exc).__name__}: {exc}"
                if self.logger:
                    self.logger.warning(
                        "fetch attempt failed",
                        extra={"source": source, "attempt": attempt, "status": "transport"},
                    )
            else:
                if response.status_code == 200:
                    try:
                        return response.json(), attempt
                    except ValueError as exc:
                        # A body that is not JSON will not become JSON on a
                        # retry. Fail now.
                        raise FetchError(f"response was not JSON: {exc}") from exc
                last_error = f"HTTP {response.status_code}"
                if response.status_code not in RETRYABLE_STATUS:
                    raise FetchError(f"{last_error} (not retryable)")
                if self.logger:
                    self.logger.warning(
                        "fetch attempt failed",
                        extra={
                            "source": source,
                            "attempt": attempt,
                            "status": response.status_code,
                        },
                    )

            if attempt < self.retries:
                self.sleeper(self.backoff_seconds * (2 ** (attempt - 1)))

        raise FetchError(f"{last_error} after {self.retries} attempts")


def build_session() -> requests.Session:
    """One Session for the whole run, so the connection is reused across
    sources instead of being renegotiated for each one."""
    return requests.Session()

"""The command line: three subcommands over one shared core, plus the
scheduled entry point.

`feedkit fetch` does the work. `feedkit report` renders what has been
collected. `feedkit status` says when the last successful run was and whether
the toolkit has gone quiet. `feedkit-scheduled` is what the crontab, launchd
job or systemd timer invokes — the same fetch, with the settings a machine
wants rather than the settings a human wants.

Both entry points are declared in `pyproject.toml` under
`[project.scripts]`, which is what turns them into commands on PATH when the
package is installed. That is Day 83's mechanism doing the work Day 80's
argparse designed.
"""

from __future__ import annotations

import argparse
import sys
from pathlib import Path
from typing import Any, Sequence, TextIO

from . import adapters, config as config_module, core, logging_setup, runner
from . import state as state_module

VERSION = "1.0.0"


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="feedkit",
        description=(
            "Collect entries from configured JSON sources, on a schedule, "
            "without processing anything twice."
        ),
        epilog=(
            "Settings resolve in this order, weakest first: defaults, "
            "configuration file, environment, flags. Run "
            "`feedkit status --explain-config` to see where each value came from. "
            "The access token is read from FEEDKIT_TOKEN and is never written to "
            "a file or a log."
        ),
    )
    # Global options are the CONFIGURATION SETTINGS — the fourth and strongest
    # layer of the precedence. Keeping them all on the main parser rather than
    # scattering them across subcommands means `--max-items` means the same
    # thing everywhere, and `status --explain-config` can show the effect of
    # any of them.
    parser.add_argument("--version", action="version", version=f"feedkit {VERSION}")
    parser.add_argument("--config", metavar="PATH", help="configuration file to read")
    parser.add_argument("--base-url", dest="base_url", help="root address of the source server")
    parser.add_argument("--log-level", dest="log_level", choices=sorted(logging_setup.LEVELS))
    parser.add_argument("--state-file", dest="state_file", metavar="PATH")
    parser.add_argument("--sources", help="comma-separated list, overriding the configuration")
    parser.add_argument(
        "--max-items",
        dest="max_items",
        type=int,
        metavar="N",
        help="most previously unseen entries to accept from one source in one run",
    )
    parser.add_argument("--retries", type=int, metavar="N", help="attempts per source")

    subcommands = parser.add_subparsers(dest="command", metavar="COMMAND")

    fetch = subcommands.add_parser("fetch", help="collect new entries from every source")
    fetch.add_argument(
        "--dry-run",
        action="store_true",
        help="do everything except write the state file, and say what would have changed",
    )

    report = subcommands.add_parser("report", help="show what has been collected")
    report.add_argument("--limit", dest="report_limit", type=int, metavar="N")

    status = subcommands.add_parser("status", help="show the last successful run")
    status.add_argument(
        "--max-age-minutes",
        dest="max_age_minutes",
        type=int,
        metavar="N",
        help="watchdog allowance; exit 3 when the last success is older than this",
    )
    status.add_argument(
        "--explain-config",
        action="store_true",
        help="print every setting, its value, and which layer it came from",
    )
    return parser


CONFIG_FLAGS = (
    "config",
    "base_url",
    "log_level",
    "state_file",
    "sources",
    "max_items",
    "retries",
    "report_limit",
    "max_age_minutes",
)


def flags_from(args: argparse.Namespace) -> dict[str, Any]:
    """Only the parsed arguments that are configuration settings, and only the
    ones actually supplied — argparse leaves the rest as None, and None is what
    tells the resolver 'this layer has no opinion'."""
    return {name: getattr(args, name, None) for name in CONFIG_FLAGS}


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

    if not args.command:
        parser.print_help(out)
        return core.EXIT_FATAL

    try:
        settings, config_path = config_module.load(flags_from(args))
    except config_module.ConfigError as exc:
        print(f"feedkit: configuration error: {exc}", file=sys.stderr)
        return core.EXIT_FATAL

    run_id = runner.new_run_id()
    logger = logging_setup.configure(
        settings.log_level,
        run_id=run_id,
        secrets=[settings.token] if settings.token else [],
        stream=out,
    )
    state_path = Path(settings.state_file).expanduser()
    lock_path = state_path.with_suffix(state_path.suffix + ".lock")

    try:
        if args.command == "fetch":
            return _cmd_fetch(args, settings, state_path, lock_path, logger, out, run_id)
        if args.command == "report":
            return _cmd_report(settings, state_path, out)
        if args.command == "status":
            return _cmd_status(args, settings, config_path, state_path, out)
    except state_module.StateError as exc:
        print(f"feedkit: {exc}", file=sys.stderr)
        return core.EXIT_FATAL

    parser.print_help(out)
    return core.EXIT_FATAL


def _cmd_fetch(
    args: argparse.Namespace,
    settings: config_module.Config,
    state_path: Path,
    lock_path: Path,
    logger: Any,
    out: TextIO,
    run_id: str,
) -> int:
    session = adapters.build_session()
    try:
        fetcher = adapters.HttpFetcher(
            session=session,
            base_url=settings.base_url,
            token=settings.token,
            timeout=settings.timeout_seconds,
            retries=settings.retries,
            backoff_seconds=settings.backoff_seconds,
            logger=logger,
        )
        summary, code = runner.run_fetch(
            settings,
            fetcher,
            adapters.SystemClock(),
            state_path,
            lock_path,
            logger,
            dry_run=args.dry_run,
            run_id=run_id,
        )
    finally:
        session.close()

    print(core.format_summary(summary, run_id=run_id, dry_run=args.dry_run), file=out)
    return code


def _cmd_report(settings: config_module.Config, state_path: Path, out: TextIO) -> int:
    current = state_module.load(state_path)
    print(core.render_report(current, settings.report_limit), file=out)
    return core.EXIT_OK


def _cmd_status(
    args: argparse.Namespace,
    settings: config_module.Config,
    config_path: Path | None,
    state_path: Path,
    out: TextIO,
) -> int:
    if args.explain_config:
        print(config_module.explain(settings, config_path), file=out)
        return core.EXIT_OK

    current = state_module.load(state_path)
    text, stale = core.render_status(
        current, adapters.SystemClock().now_iso(), settings.max_age_seconds
    )
    print(text, file=out)
    # A watchdog is only useful if it can fail. Exiting non-zero on silence is
    # what lets a second, much simpler scheduled job page you when the first
    # one has stopped running at all.
    return core.EXIT_PARTIAL if stale else core.EXIT_OK


def scheduled_main(argv: Sequence[str] | None = None) -> int:
    """The entry point a scheduler invokes.

    It is a thin wrapper on purpose. The scheduled run wants a fetch, quieter
    logging by default, and no interactive help — everything else is identical,
    because a scheduled run that behaves differently from the one you tested by
    hand is a scheduled run you have not tested.
    """
    argv = list(sys.argv[1:] if argv is None else argv)
    return main(["--log-level", "info", "fetch", *argv])


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(main())

"""Configuration, resolved by a precedence that is written down and testable.

The order, weakest first:

    1. defaults baked into the code   — so the tool runs with no setup at all
    2. a configuration file           — the machine's long-lived preferences
    3. environment variables          — deployment-specific values and secrets
    4. command-line flags             — this one run, right now

Every automation has this order. Most of them have it by accident, spread over
a dozen `or` expressions, and nobody can say what wins. Here it is one pure
function over four dictionaries, so a test can assert all four levels — which
is exactly what `tests/run_tests.sh` does.

`resolve` also records WHERE each value came from. `feedkit status
--explain-config` prints that table, which turns "why is it doing that?" into a
five-second question instead of an afternoon.
"""

from __future__ import annotations

import os
import tomllib
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Mapping

#: Layer 1. Everything the toolkit needs to run at all, with no file, no
#: environment and no flags. A tool that cannot start without configuration is
#: a tool nobody tries.
DEFAULTS: dict[str, Any] = {
    "sources": ["notes", "links"],
    "max_items": 5,
    "report_limit": 10,
    "timeout_seconds": 5.0,
    "retries": 3,
    "backoff_seconds": 0.5,
    "log_level": "info",
    "state_file": "feedkit-state.json",
    "max_age_minutes": 1440,
}

#: Layer 3. Environment variable name -> (config key, type).
ENV_KEYS: dict[str, tuple[str, str]] = {
    "FEEDKIT_BASE_URL": ("base_url", "str"),
    "FEEDKIT_MAX_ITEMS": ("max_items", "int"),
    "FEEDKIT_REPORT_LIMIT": ("report_limit", "int"),
    "FEEDKIT_TIMEOUT_SECONDS": ("timeout_seconds", "float"),
    "FEEDKIT_RETRIES": ("retries", "int"),
    "FEEDKIT_BACKOFF_SECONDS": ("backoff_seconds", "float"),
    "FEEDKIT_LOG_LEVEL": ("log_level", "str"),
    "FEEDKIT_STATE_FILE": ("state_file", "str"),
    "FEEDKIT_MAX_AGE_MINUTES": ("max_age_minutes", "int"),
    "FEEDKIT_SOURCES": ("sources", "list"),
}

#: The one value that must NEVER come from a file in the repository.
SECRET_ENV = "FEEDKIT_TOKEN"


class ConfigError(ValueError):
    """The configuration is unusable. Stop the run; do not guess."""


@dataclass(frozen=True)
class Config:
    """The resolved settings for one run."""

    base_url: str
    sources: tuple[str, ...]
    max_items: int
    report_limit: int
    timeout_seconds: float
    retries: int
    backoff_seconds: float
    log_level: str
    state_file: str
    max_age_minutes: int
    token: str = ""
    provenance: Mapping[str, str] = field(default_factory=dict)

    @property
    def max_age_seconds(self) -> int:
        return self.max_age_minutes * 60


def coerce(value: Any, kind: str, key: str) -> Any:
    """Turn a string from a file, an environment variable or a flag into the
    type the rest of the program expects, and fail loudly when it cannot."""
    if kind == "list":
        if isinstance(value, (list, tuple)):
            return [str(item) for item in value]
        return [part.strip() for part in str(value).split(",") if part.strip()]
    try:
        if kind == "int":
            return int(value)
        if kind == "float":
            return float(value)
    except (TypeError, ValueError) as exc:
        raise ConfigError(f"{key}: {value!r} is not a valid {kind}") from exc
    return str(value)


def kind_of(key: str) -> str:
    """The declared type of a configuration key, from the defaults table."""
    for _, (config_key, kind) in ENV_KEYS.items():
        if config_key == key:
            return kind
    default = DEFAULTS.get(key)
    if isinstance(default, bool):
        return "str"
    if isinstance(default, int):
        return "int"
    if isinstance(default, float):
        return "float"
    if isinstance(default, list):
        return "list"
    return "str"


def read_config_file(path: Path) -> dict[str, Any]:
    """Read a TOML configuration file. Missing file is not an error; an
    unparseable one is."""
    if not path.is_file():
        return {}
    try:
        with path.open("rb") as handle:
            data = tomllib.load(handle)
    except tomllib.TOMLDecodeError as exc:
        raise ConfigError(f"{path}: not valid TOML: {exc}") from exc
    settings = data.get("feedkit", data)
    if not isinstance(settings, dict):
        raise ConfigError(f"{path}: expected a table of settings")
    return dict(settings)


def find_config_file(explicit: str | None, environ: Mapping[str, str], cwd: Path) -> Path | None:
    """Where the configuration file lives, in the order a user expects.

    A flag beats the environment, which beats the current directory, which
    beats the user's XDG configuration directory. Returning None means "no file
    anywhere", which is a perfectly normal state, not an error.
    """
    if explicit:
        return Path(explicit).expanduser()
    from_env = environ.get("FEEDKIT_CONFIG")
    if from_env:
        return Path(from_env).expanduser()
    local = cwd / "feedkit.toml"
    if local.is_file():
        return local
    base = environ.get("XDG_CONFIG_HOME")
    home = Path(base).expanduser() if base else Path(environ.get("HOME", "~")).expanduser() / ".config"
    candidate = home / "feedkit" / "feedkit.toml"
    return candidate if candidate.is_file() else None


def env_values(environ: Mapping[str, str]) -> dict[str, Any]:
    """Layer 3, extracted from an environment mapping passed in as an argument
    (never read from os.environ in here — that is a boundary, and tests need to
    supply their own)."""
    values: dict[str, Any] = {}
    for name, (key, kind) in ENV_KEYS.items():
        if name in environ and environ[name] != "":
            values[key] = coerce(environ[name], kind, key)
    return values


def resolve(
    file_values: Mapping[str, Any],
    environment: Mapping[str, Any],
    flags: Mapping[str, Any],
    token: str = "",
) -> Config:
    """Apply the four layers in order and record where each value came from.

    Pure. Four dictionaries in, one Config out. This is the function the lab's
    precedence test drives directly, and it is why "which layer wins?" is a
    question with a checked answer rather than a folk belief.
    """
    merged: dict[str, Any] = {}
    provenance: dict[str, str] = {}

    for layer_name, layer in (
        ("default", DEFAULTS),
        ("file", file_values),
        ("environment", environment),
        ("flag", {key: value for key, value in flags.items() if value is not None}),
    ):
        for key, value in layer.items():
            if key not in DEFAULTS and key != "base_url":
                # An unknown key in a config file is almost always a typo, and
                # silently ignoring it is how people lose an afternoon.
                if layer_name == "file":
                    raise ConfigError(f"unknown setting in configuration file: {key!r}")
                continue
            merged[key] = coerce(value, kind_of(key), key) if layer_name != "default" else value
            provenance[key] = layer_name

    base_url = str(merged.get("base_url", "")).rstrip("/")
    if not base_url:
        raise ConfigError(
            "no base URL configured. Set FEEDKIT_BASE_URL in the environment, "
            "or pass --base-url. Deployment-specific addresses do not belong in "
            "a file that is committed."
        )
    if merged["retries"] < 1:
        raise ConfigError("retries must be at least 1")
    if merged["max_items"] < 0:
        raise ConfigError("max-items must not be negative")

    provenance.setdefault("base_url", "environment")
    provenance["token"] = "environment" if token else "unset"

    return Config(
        base_url=base_url,
        sources=tuple(merged["sources"]),
        max_items=int(merged["max_items"]),
        report_limit=int(merged["report_limit"]),
        timeout_seconds=float(merged["timeout_seconds"]),
        retries=int(merged["retries"]),
        backoff_seconds=float(merged["backoff_seconds"]),
        log_level=str(merged["log_level"]),
        state_file=str(merged["state_file"]),
        max_age_minutes=int(merged["max_age_minutes"]),
        token=token,
        provenance=provenance,
    )


def load(
    flags: Mapping[str, Any],
    environ: Mapping[str, str] | None = None,
    cwd: Path | None = None,
) -> tuple[Config, Path | None]:
    """The impure wrapper: find the file, read the environment, then call the
    pure `resolve`. All the I/O is in these six lines."""
    environ = os.environ if environ is None else environ
    cwd = Path.cwd() if cwd is None else cwd
    config_path = find_config_file(flags.get("config"), environ, cwd)
    file_values = read_config_file(config_path) if config_path else {}
    token = environ.get(SECRET_ENV, "")
    config = resolve(file_values, env_values(environ), flags, token=token)
    return config, config_path


def explain(config: Config, config_path: Path | None) -> str:
    """The provenance table. Answers 'why is it doing that?' in one command."""
    lines = [f"configuration file: {config_path or 'none found'}", ""]
    lines.append(f"  {'setting':<20} {'value':<28} {'came from'}")
    for key in sorted(config.provenance):
        if key == "token":
            shown = "set (never printed)" if config.token else "unset"
        else:
            shown = str(getattr(config, key, ""))
        lines.append(f"  {key:<20} {shown:<28} {config.provenance[key]}")
    return "\n".join(lines)

"""The pure core of the toolkit.

Nothing in this module touches the network, the clock, the filesystem or a
subprocess. Every function here takes plain data and returns plain data, which
is exactly what makes the interesting parts of an automation testable without
starting a server or waiting a second. Day 74 argued for pushing boundaries to
the edges; this module is what the middle looks like when you do.

If you ever find yourself wanting to `import requests` or call `time.time()` in
here, that is the signal that the value belongs in a parameter instead.
"""

from __future__ import annotations

import calendar
import time
from dataclasses import dataclass
from typing import Any, Iterable, Mapping, Sequence

# `calendar` and `time` appear here only to PARSE a timestamp that was handed
# in as a string. Nothing in this module ever asks what time it is now — that
# question belongs to the clock adapter, so that every test can answer it.

STATE_VERSION = 1

#: Exit codes. These are the machine-readable half of every run, and the
#: scheduler is the thing that reads them.
EXIT_OK = 0
EXIT_FATAL = 1
EXIT_PARTIAL = 3
EXIT_LOCKED = 75


@dataclass(frozen=True)
class Entry:
    """One item collected from one source."""

    id: str
    title: str
    published: str
    source: str = ""

    def as_dict(self) -> dict[str, str]:
        return {
            "id": self.id,
            "title": self.title,
            "published": self.published,
            "source": self.source,
        }


@dataclass(frozen=True)
class SourceResult:
    """What happened to one source during one run."""

    source: str
    status: str  # "ok" | "failed"
    new_entries: tuple[Entry, ...] = ()
    error: str = ""
    attempts: int = 1


class InvalidPayload(ValueError):
    """The server answered, but not with something this toolkit understands."""


def parse_entries(payload: Any, source: str) -> tuple[Entry, ...]:
    """Validate a decoded JSON payload and turn it into Entry objects.

    A 200 response is not the same thing as a correct response. Validating at
    the edge means every function downstream can assume the shape it was given,
    which is the whole reason this is a separate step rather than a dict lookup
    buried three calls deep.
    """
    if not isinstance(payload, Mapping):
        raise InvalidPayload(f"top level is {type(payload).__name__}, expected an object")
    items = payload.get("entries")
    if not isinstance(items, Sequence) or isinstance(items, (str, bytes)):
        raise InvalidPayload("'entries' is missing or is not a list")

    parsed: list[Entry] = []
    for index, item in enumerate(items):
        if not isinstance(item, Mapping):
            raise InvalidPayload(f"entry {index} is not an object")
        missing = [field for field in ("id", "title", "published") if field not in item]
        if missing:
            raise InvalidPayload(f"entry {index} is missing {', '.join(missing)}")
        parsed.append(
            Entry(
                id=str(item["id"]),
                title=str(item["title"]),
                published=str(item["published"]),
                source=source,
            )
        )
    return tuple(parsed)


def select_new(
    entries: Iterable[Entry], seen_ids: Iterable[str], max_items: int
) -> tuple[Entry, ...]:
    """Return the entries not already recorded, newest first, capped at max_items.

    This is the idempotence rule, and it is four lines of pure logic precisely
    because the record of what has been seen arrives as an argument rather than
    being read from a file in here.
    """
    already = set(seen_ids)
    fresh = [entry for entry in entries if entry.id not in already]
    fresh.sort(key=lambda entry: (entry.published, entry.id), reverse=True)
    if max_items >= 0:
        fresh = fresh[:max_items]
    return tuple(fresh)


def empty_state() -> dict[str, Any]:
    """The state of a toolkit that has never run."""
    return {
        "version": STATE_VERSION,
        "last_run": None,
        "last_success": None,
        "sources": {},
        "entries": [],
    }


def merge_state(
    state: Mapping[str, Any],
    results: Sequence[SourceResult],
    run_id: str,
    started_at: str,
    finished_at: str,
    keep_entries: int = 200,
) -> dict[str, Any]:
    """Fold one run's results into the previous state and return the new state.

    Pure: give it the same inputs and it returns the same dictionary, every
    time, on any machine. The caller decides whether to write it.
    """
    summary = summarise(results)
    sources = {name: dict(value) for name, value in dict(state.get("sources") or {}).items()}
    collected: list[dict[str, str]] = list(state.get("entries") or [])

    for result in results:
        record = sources.setdefault(
            result.source, {"seen_ids": [], "last_success": None, "last_error": ""}
        )
        if result.status == "ok":
            seen = list(record.get("seen_ids") or [])
            seen.extend(entry.id for entry in result.new_entries if entry.id not in seen)
            record["seen_ids"] = seen
            record["last_success"] = finished_at
            record["last_error"] = ""
            collected = [entry.as_dict() for entry in result.new_entries] + collected
        else:
            record["last_error"] = result.error

    collected.sort(key=lambda item: (item["published"], item["id"]), reverse=True)

    new_state: dict[str, Any] = {
        "version": STATE_VERSION,
        "last_run": {
            "run_id": run_id,
            "started_at": started_at,
            "finished_at": finished_at,
            "status": summary["status"],
            "sources_ok": summary["sources_ok"],
            "sources_failed": summary["sources_failed"],
            "new_entries": summary["new_entries"],
        },
        "last_success": (
            finished_at if summary["status"] == "ok" else state.get("last_success")
        ),
        "sources": sources,
        "entries": collected[:keep_entries],
    }
    return new_state


def summarise(results: Sequence[SourceResult]) -> dict[str, Any]:
    """Reduce a run's results to the handful of numbers a human reads."""
    ok = [result for result in results if result.status == "ok"]
    failed = [result for result in results if result.status != "ok"]
    new_entries = sum(len(result.new_entries) for result in ok)

    if not results:
        status = "ok"
    elif not failed:
        status = "ok"
    elif not ok:
        status = "failed"
    else:
        status = "partial"

    return {
        "status": status,
        "sources_total": len(results),
        "sources_ok": len(ok),
        "sources_failed": len(failed),
        "new_entries": new_entries,
        "failures": {result.source: result.error for result in failed},
        "retried": {result.source: result.attempts for result in results if result.attempts > 1},
    }


def exit_code_for(summary: Mapping[str, Any]) -> int:
    """Map a run summary onto the exit code the scheduler will read.

    Partial success gets its own code. Reporting 0 for "most of it worked" is
    the single most common way an automation lies to the person who owns it.
    """
    status = summary.get("status")
    if status == "ok":
        return EXIT_OK
    if status == "partial":
        return EXIT_PARTIAL
    return EXIT_FATAL


def format_summary(summary: Mapping[str, Any], run_id: str, dry_run: bool = False) -> str:
    """The human-readable run summary, printed at the end of every run."""
    lines = [
        f"run {run_id}: {summary['status']}"
        + ("  (dry run — nothing was written)" if dry_run else ""),
        f"  sources: {summary['sources_ok']} ok, {summary['sources_failed']} failed,"
        f" {summary['sources_total']} total",
        f"  new entries: {summary['new_entries']}",
    ]
    for source, attempts in sorted(dict(summary.get("retried") or {}).items()):
        lines.append(f"  retried: {source} succeeded on attempt {attempts}")
    for source, error in sorted(dict(summary.get("failures") or {}).items()):
        lines.append(f"  FAILED: {source}: {error}")
    return "\n".join(lines)


def render_report(state: Mapping[str, Any], limit: int) -> str:
    """Render what has been collected. Reads state, touches nothing."""
    entries = list(state.get("entries") or [])
    if not entries:
        return "No entries collected yet. Run `feedkit fetch` first."

    shown = entries[: limit if limit >= 0 else len(entries)]
    width = max(len(entry["source"]) for entry in shown)
    lines = [f"{len(entries)} entries collected; showing {len(shown)}", ""]
    for entry in shown:
        lines.append(f"  {entry['published']}  {entry['source']:<{width}}  {entry['title']}")
    return "\n".join(lines)


def render_status(state: Mapping[str, Any], now: str, max_age_seconds: int) -> tuple[str, bool]:
    """Render the status block and say whether the toolkit has gone quiet.

    The second half of the tuple is the watchdog answer: True when the last
    successful run is older than the allowance. Alerting on silence catches the
    failure mode that alerting on errors cannot — the run that never happened.
    """
    last_success = state.get("last_success")
    last_run = state.get("last_run")
    stale = is_stale(last_success, now, max_age_seconds)

    lines = [f"last success: {last_success or 'never'}"]
    if last_run:
        lines.append(
            f"last run:     {last_run['run_id']} at {last_run['finished_at']}"
            f" ({last_run['status']}, {last_run['new_entries']} new)"
        )
    else:
        lines.append("last run:     never")
    lines.append(f"now:          {now}")
    lines.append(
        f"watchdog:     {'STALE' if stale else 'fresh'}"
        f" (allowance {max_age_seconds}s)"
    )
    for name, record in sorted(dict(state.get("sources") or {}).items()):
        note = record.get("last_error") or "ok"
        lines.append(
            f"  {name}: {len(record.get('seen_ids') or [])} seen,"
            f" last success {record.get('last_success') or 'never'} — {note}"
        )
    return "\n".join(lines), stale


def is_stale(last_success: str | None, now: str, max_age_seconds: int) -> bool:
    """True when the last success is missing or older than the allowance.

    Timestamps are ISO 8601 strings, compared by parsing them into seconds. The
    parsing lives in the caller's clock adapter; here we accept the already
    normalised comparison to keep this module free of the datetime module's
    timezone surprises. Both arguments must be UTC ISO strings.
    """
    if not last_success:
        return True
    return _iso_seconds(now) - _iso_seconds(last_success) > max_age_seconds


def _iso_seconds(value: str) -> int:
    """Seconds since the epoch for a UTC ISO 8601 timestamp such as
    2026-07-19T10:11:12Z. Deliberately small and deliberately strict."""
    cleaned = value.replace("Z", "").split(".")[0]
    parsed = time.strptime(cleaned, "%Y-%m-%dT%H:%M:%S")
    return calendar.timegm(parsed)

"""Structured logging — the thing that makes an unattended run debuggable.

Two decisions are baked in here, and both are worth arguing rather than
copying.

**One JSON object per line, on stdout.** A line of prose is readable by you at
your desk; a line of JSON is readable by you AND by `grep`, `jq`, a log
shipper, and whatever the supervisor writes it into. Writing to stdout rather
than opening a log file means the program does not have to know about log
rotation, permissions, or where the operator wants their logs — cron mails it,
systemd hands it to the journal, launchd redirects it, and a human running the
command by hand simply sees it. Fewer decisions inside the program is the
point.

**Every record carries the run id and, where it applies, the item.** An
unattended failure is a message you read hours later with no memory of the
context. "Timeout" tells you nothing. "run 8f2c1a: source=papers attempt=3
timeout after 5.0s" tells you which run, which item, how hard it tried, and
what the limit was.
"""

from __future__ import annotations

import json
import logging
import sys
from typing import Any, Iterable, TextIO

LEVELS = {
    "debug": logging.DEBUG,
    "info": logging.INFO,
    "warning": logging.WARNING,
    "error": logging.ERROR,
    "critical": logging.CRITICAL,
}

#: Extra keys that the formatter promotes to top-level JSON fields.
CONTEXT_KEYS = ("source", "attempt", "status", "count", "path", "elapsed_ms", "url")


class RedactingFilter(logging.Filter):
    """Replace known secret values with a placeholder, everywhere.

    This is a seatbelt, not a licence. The right habit is to never put a token
    into a log call in the first place; this filter exists because one day
    somebody will log a whole request object, or an exception message that
    happens to quote a URL with a token in the query string, and the difference
    between a bad afternoon and a credential rotation is whether that string
    reached the log.
    """

    PLACEHOLDER = "***REDACTED***"

    def __init__(self, secrets: Iterable[str] = ()) -> None:
        super().__init__()
        # Very short strings would redact half the alphabet; ignore them.
        self.secrets = tuple(secret for secret in secrets if secret and len(secret) >= 6)

    def _scrub(self, value: Any) -> Any:
        if isinstance(value, str):
            for secret in self.secrets:
                value = value.replace(secret, self.PLACEHOLDER)
            return value
        if isinstance(value, (list, tuple)):
            return type(value)(self._scrub(item) for item in value)
        if isinstance(value, dict):
            return {key: self._scrub(item) for key, item in value.items()}
        return value

    def filter(self, record: logging.LogRecord) -> bool:
        if not self.secrets:
            return True
        record.msg = self._scrub(record.msg)
        if record.args:
            record.args = self._scrub(record.args)
        for key in CONTEXT_KEYS:
            if hasattr(record, key):
                setattr(record, key, self._scrub(getattr(record, key)))
        if record.exc_info:
            # An exception's own text is the most common accidental leak.
            exc = record.exc_info[1]
            if exc is not None and exc.args:
                exc.args = tuple(self._scrub(arg) for arg in exc.args)
        return True


class JsonFormatter(logging.Formatter):
    """One JSON object per line, with a stable field order."""

    def __init__(self, run_id: str) -> None:
        super().__init__()
        self.run_id = run_id

    def format(self, record: logging.LogRecord) -> str:
        payload: dict[str, Any] = {
            "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"),
            "level": record.levelname.lower(),
            "run_id": self.run_id,
            "event": record.getMessage(),
        }
        for key in CONTEXT_KEYS:
            if hasattr(record, key):
                payload[key] = getattr(record, key)
        if record.exc_info:
            payload["error"] = str(record.exc_info[1])
        # ensure_ascii=False keeps the log readable to a human. Escaping every
        # non-ASCII character is the default and is nearly always wrong for a
        # file somebody has to read at three in the morning.
        return json.dumps(payload, sort_keys=False, ensure_ascii=False)


def configure(
    level: str,
    run_id: str,
    secrets: Iterable[str] = (),
    stream: TextIO | None = None,
) -> logging.Logger:
    """Build the toolkit's logger. Called once, at the start of a run."""
    logger = logging.getLogger("feedkit")
    logger.handlers.clear()
    logger.propagate = False
    logger.setLevel(LEVELS.get(level.lower(), logging.INFO))

    handler = logging.StreamHandler(stream if stream is not None else sys.stdout)
    handler.setFormatter(JsonFormatter(run_id))
    handler.addFilter(RedactingFilter(secrets))
    logger.addHandler(handler)
    return logger

"""One unattended run, start to finish.

This is the only module that knows the ORDER of things: acquire the lock, load
the state, fetch each source with retries, skip and report the ones that fail,
fold the successes into the state, write the state atomically, print the
summary, return an exit code. Everything it does with the outside world arrives
as an argument — the fetcher, the clock, the logger, the paths — which is why
the whole thing can be exercised against a fake fetcher in a millisecond.

The failure policy is the part worth reading twice, because it is the part that
distinguishes an automation from a script:

* a **transport error or a 5xx** on one source is retried with backoff and, if
  it never succeeds, is SKIPPED and REPORTED — one broken source must not stop
  the other four;
* a **payload that does not parse** is not retried, because it will not parse
  next time either; it is skipped and reported the same way;
* a **failure to acquire the lock** stops everything immediately, because the
  correct response to "a run is already happening" is to do nothing;
* an **unusable state file or an invalid configuration** stops everything,
  because continuing would mean guessing about the thing that records what has
  already been done.

Partial success is the normal case for a batch job, and it gets its own exit
code. Reporting 0 because "most of it worked" is how an automation becomes a
thing nobody can trust.
"""

from __future__ import annotations

import uuid
from pathlib import Path
from typing import Any, Protocol, Sequence

from . import core, state as state_module
from .adapters import FetchError
from .config import Config


class Fetcher(Protocol):
    def fetch(self, source: str) -> tuple[Any, int]: ...


def new_run_id() -> str:
    """A short, unique label for one run. Every log line carries it, so the
    lines belonging to a single 03:00 run can be pulled out of a month of
    output with one grep."""
    return uuid.uuid4().hex[:8]


def fetch_sources(
    sources: Sequence[str],
    fetcher: Fetcher,
    seen: dict[str, list[str]],
    max_items: int,
    logger: Any,
) -> list[core.SourceResult]:
    """Fetch every source, collecting successes and failures side by side."""
    results: list[core.SourceResult] = []
    for source in sources:
        logger.info("source started", extra={"source": source})
        try:
            payload, attempts = fetcher.fetch(source)
            entries = core.parse_entries(payload, source)
        except (FetchError, core.InvalidPayload) as exc:
            logger.error("source failed", extra={"source": source, "status": "failed"})
            results.append(core.SourceResult(source=source, status="failed", error=str(exc)))
            continue

        fresh = core.select_new(entries, seen.get(source, []), max_items)
        logger.info(
            "source finished",
            extra={
                "source": source,
                "status": "ok",
                "count": len(fresh),
                "attempt": attempts,
            },
        )
        results.append(
            core.SourceResult(
                source=source, status="ok", new_entries=fresh, attempts=attempts
            )
        )
    return results


def run_fetch(
    config: Config,
    fetcher: Fetcher,
    clock: Any,
    state_path: Path,
    lock_path: Path,
    logger: Any,
    dry_run: bool = False,
    run_id: str | None = None,
) -> tuple[dict[str, Any], int]:
    """Do one fetch run. Returns the summary and the exit code.

    The run id is passed IN rather than generated here, so that the id stamped
    on every log line and the id recorded in the state file are the same
    string. Two ids for one run is a small bug that makes an incident twice as
    slow to investigate, and it is easy to ship without noticing.
    """
    run_id = run_id or new_run_id()
    started_at = clock.now_iso()

    try:
        with state_module.Lock(lock_path):
            logger.info("run started", extra={"status": "started", "path": str(state_path)})
            current = state_module.load(state_path)
            seen = {
                name: list(record.get("seen_ids") or [])
                for name, record in dict(current.get("sources") or {}).items()
            }

            results = fetch_sources(config.sources, fetcher, seen, config.max_items, logger)
            finished_at = clock.now_iso()
            summary = core.summarise(results)
            merged = core.merge_state(current, results, run_id, started_at, finished_at)

            if dry_run:
                logger.info(
                    "dry run — state not written",
                    extra={"status": "dry-run", "count": summary["new_entries"]},
                )
            else:
                state_module.write_atomic(state_path, merged)
                logger.info(
                    "state written",
                    extra={"status": summary["status"], "path": str(state_path)},
                )
    except state_module.LockHeld as exc:
        logger.error("another run is in progress", extra={"status": "locked"})
        return (
            {
                "run_id": run_id,
                "status": "locked",
                "sources_total": 0,
                "sources_ok": 0,
                "sources_failed": 0,
                "new_entries": 0,
                "failures": {"lock": str(exc)},
                "retried": {},
            },
            core.EXIT_LOCKED,
        )

    summary["run_id"] = run_id
    exit_code = core.exit_code_for(summary)
    logger.info(
        "run finished",
        extra={"status": summary["status"], "count": summary["new_entries"]},
    )
    return summary, exit_code

"""The state file, and the atomic write that keeps it trustworthy.

State is what makes a job idempotent: it is the record of what has already been
processed, so a second run does not do the work twice. That makes it the single
most valuable file the toolkit owns, and losing it is worse than a failed run —
a failed run is visible, a corrupted state file quietly re-processes or
silently skips.

So the write is atomic, exactly as Days 64 and 65 described. Write the whole
new document to a temporary file in the SAME directory, flush it, ask the
operating system to put it on the disk, then `os.replace` it over the old name.
`os.replace` is atomic on POSIX and on Windows: any reader sees either the
complete old file or the complete new one, never a half-written mixture. If the
machine loses power between the write and the replace, the previous state is
still there and the temporary file is garbage that the next run cleans up.

The naive version — `open(path, "w")` then `json.dump` — truncates the real
file first. Interrupt it and the record of everything you have ever processed
is a zero-byte file.
"""

from __future__ import annotations

import json
import os
import tempfile
from pathlib import Path
from typing import Any, Callable, Mapping

from .core import empty_state


class StateError(RuntimeError):
    """The state file exists but cannot be used. Never guess; stop."""


def load(path: Path) -> dict[str, Any]:
    """Read the state file, or return a fresh empty state if there is none."""
    if not path.is_file():
        return empty_state()
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise StateError(
            f"{path} is not valid JSON ({exc}). Refusing to overwrite it. "
            f"Move it aside to start fresh."
        ) from exc
    if not isinstance(data, dict) or "version" not in data:
        raise StateError(f"{path} does not look like a feedkit state file")
    return data


def write_atomic(
    path: Path,
    state: Mapping[str, Any],
    crash_hook: Callable[[], None] | None = None,
) -> None:
    """Write state so that an interruption leaves the previous file intact.

    `crash_hook` exists purely so the lab can prove the property. The test
    passes a function that raises, standing in for the power cut, and then
    asserts the old file is byte-identical. Production code passes nothing.
    """
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = json.dumps(state, indent=2, sort_keys=True) + "\n"

    handle = tempfile.NamedTemporaryFile(
        mode="w",
        encoding="utf-8",
        dir=str(path.parent),
        prefix=path.name + ".",
        suffix=".tmp",
        delete=False,
    )
    tmp_path = Path(handle.name)
    try:
        with handle:
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        if crash_hook is not None:
            crash_hook()
        os.replace(tmp_path, path)
    except BaseException:
        tmp_path.unlink(missing_ok=True)
        raise


class LockHeld(RuntimeError):
    """Another run of this toolkit is already in progress."""


class Lock:
    """A lock file, so two scheduled runs never overlap.

    Created with O_CREAT | O_EXCL, which the operating system guarantees will
    succeed for exactly one caller. The file holds the process id, which is
    what lets a human decide whether a lock left behind by a crash is stale.

    This is deliberately the simplest thing that works on one machine. It is
    not a distributed lock and must not be used as one.
    """

    def __init__(self, path: Path) -> None:
        self.path = path
        self._acquired = False

    def __enter__(self) -> "Lock":
        self.path.parent.mkdir(parents=True, exist_ok=True)
        try:
            fd = os.open(self.path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
        except FileExistsError as exc:
            holder = ""
            try:
                holder = self.path.read_text(encoding="utf-8").strip()
            except OSError:
                pass
            raise LockHeld(
                f"{self.path} exists (held by pid {holder or 'unknown'}). "
                f"Another run is in progress, or a previous run was killed. "
                f"Delete the file only after checking that no such process exists."
            ) from exc
        with os.fdopen(fd, "w") as handle:
            handle.write(str(os.getpid()))
        self._acquired = True
        return self

    def __exit__(self, *exc_info: object) -> None:
        if self._acquired:
            self.path.unlink(missing_ok=True)
            self._acquired = False

# feedkit configuration — layer 2 of four.
#
# What belongs here: the long-lived preferences of THIS machine. What does not:
# anything that changes per run (that is a flag), anything that differs between
# machines or deployments (that is the environment), and above all anything
# secret. There is no token in this file and there never will be one; the
# toolkit reads FEEDKIT_TOKEN from the environment and refuses to look anywhere
# else.
#
# The address of the source server is also absent on purpose. It is a
# deployment fact, not a preference, so it arrives as FEEDKIT_BASE_URL — which
# is also what lets the test suite point the toolkit at a server it started
# itself, on a port it chose at run time.
#
# Where this file lives, in the order the toolkit looks:
#   1. the path given to --config
#   2. $FEEDKIT_CONFIG
#   3. ./feedkit.toml   (this one — handy while developing)
#   4. $XDG_CONFIG_HOME/feedkit/feedkit.toml, or ~/.config/feedkit/feedkit.toml
#
# Number 4 is the right home for a real installation on your own machine: it
# survives reinstalling the package, it is not inside a directory you might
# delete, and it is not inside a repository you might publish.

[feedkit]
sources = ["notes", "links", "papers"]

# How many previously unseen entries to accept from one source in one run.
# A cap is a safety belt: the first run against a large source should not
# process ten thousand items before you have looked at one.
max_items = 10

report_limit = 10

# Never omit a timeout. The default in `requests` is to wait forever, and a
# scheduled job that hangs is invisible to every supervisor there is.
timeout_seconds = 5.0

# Bounded retries, doubling the wait each time: 0.5s, 1.0s, 2.0s.
retries = 3
backoff_seconds = 0.5

log_level = "info"

# Relative to the working directory the scheduler starts the job in. For a real
# installation, prefer an absolute path under your home directory — a schedule
# entry rarely runs where you think it does.
state_file = "feedkit-state.json"

# The watchdog allowance, in minutes. `feedkit status` exits non-zero when the
# last SUCCESSFUL run is older than this, which is how you get alerted about
# the run that never happened rather than only about the run that failed.
max_age_minutes = 1440

# Packaging metadata for the toolkit (Day 83's mechanism, used for real).
#
# The two entries under [project.scripts] are what turn this package into
# commands on PATH. After `pip install -e .` you type `feedkit`, not
# `python /some/long/path/cli.py` — and the scheduler's job line becomes short
# enough to read, which matters more than it sounds when you are looking at a
# crontab at midnight.

[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"

[project]
name = "feedkit"
version = "1.0.0"
description = "A personal automation toolkit: collect entries from JSON sources on a schedule, without processing anything twice."
requires-python = ">=3.11"
license = { text = "MIT" }
keywords = ["automation", "cli", "scheduling", "idempotence"]
dependencies = [
  "requests>=2.31",
]

[project.optional-dependencies]
# Development-only tools. A scheduled machine installs the package without
# these; a developer installs `.[dev]`. Keeping them out of `dependencies` is
# what stops a test framework from being deployed to a server.
dev = ["pytest>=8"]

[project.scripts]
feedkit = "feedkit.cli:main"
feedkit-scheduled = "feedkit.cli:scheduled_main"

[tool.setuptools.packages.find]
where = ["src"]

[tool.pytest.ini_options]
addopts = "-q"

<?xml version="1.0" encoding="UTF-8"?>
<!--
  feedkit — launchd reference for macOS. NOT INSTALLED BY THIS LAB.

  To install on your own machine, deliberately:
      cp com.example.feedkit.plist ~/Library/LaunchAgents/
      launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/com.example.feedkit.plist
  To inspect:
      launchctl print gui/$(id -u)/com.example.feedkit
  To remove:
      launchctl bootout gui/$(id -u)/com.example.feedkit
      rm ~/Library/LaunchAgents/com.example.feedkit.plist

  Change every path before using this.

  Two launchd behaviours matter for an automation and differ from cron:

    * RunAtLoad fires the job once when it is loaded, which is usually what you
      want and is occasionally a surprise at 4pm on a Friday.
    * launchd DOES catch up. If the machine was asleep when a StartInterval
      elapsed, launchd runs the job once shortly after it wakes — not once per
      missed interval. Idempotence is what makes that harmless.
-->
<plist version="1.0">
<dict>
  <key>Label</key>
  <string>com.example.feedkit</string>

  <key>ProgramArguments</key>
  <array>
    <!-- An absolute path. launchd does not read your shell profile, so a bare
         "feedkit" will not be found. -->
    <string>/Users/you/.local/bin/feedkit-scheduled</string>
  </array>

  <key>EnvironmentVariables</key>
  <dict>
    <key>FEEDKIT_BASE_URL</key>
    <string>https://feeds.example.org</string>
    <key>FEEDKIT_STATE_FILE</key>
    <string>/Users/you/Library/Application Support/feedkit/feedkit-state.json</string>
    <key>FEEDKIT_LOG_LEVEL</key>
    <string>info</string>
    <!-- FEEDKIT_TOKEN is deliberately absent. A plist in LaunchAgents is
         readable by anything running as you and is copied by Time Machine.
         Put the token in the login keychain and have a small wrapper script
         export it, or use a mode-600 file that the wrapper sources. -->
  </dict>

  <key>WorkingDirectory</key>
  <string>/Users/you/Library/Application Support/feedkit</string>

  <!-- Every 3600 seconds. Use StartCalendarInterval instead when the job must
       happen at a wall-clock time rather than at an interval. -->
  <key>StartInterval</key>
  <integer>3600</integer>

  <key>RunAtLoad</key>
  <true/>

  <!-- Structured logs go to stdout; launchd writes them where you say. This is
       the whole argument for logging to stdout: the program stays out of the
       business of log files, rotation and permissions. -->
  <key>StandardOutPath</key>
  <string>/Users/you/Library/Logs/feedkit.log</string>
  <key>StandardErrorPath</key>
  <string>/Users/you/Library/Logs/feedkit.err.log</string>

  <!-- Do not restart on failure. A failing fetch should wait for the next
       interval, not spin. KeepAlive on a job that exits quickly is how a
       laptop fan comes on for no reason. -->
  <key>KeepAlive</key>
  <false/>

  <!-- Be a good citizen about battery and disk. -->
  <key>LowPriorityIO</key>
  <true/>
  <key>ProcessType</key>
  <string>Background</string>
</dict>
</plist>

# feedkit — cron reference. NOT INSTALLED BY THIS LAB.
#
# To install on your own machine, deliberately:   crontab -e   and paste below.
# To inspect what is installed:                   crontab -l
# To remove:                                      crontab -e   and delete the lines.
#
# Change every path before using this. None of them will be right for you.

# cron runs with a nearly empty environment: no shell profile, no PATH beyond a
# minimal default, no HOME on some systems. Set what the job needs, explicitly.
SHELL=/bin/sh
PATH=/usr/local/bin:/usr/bin:/bin
HOME=/home/you
MAILTO=you@example.org

# The base address of the source server. A deployment fact, so it lives here
# rather than in the configuration file.
FEEDKIT_BASE_URL=https://feeds.example.org

# The state file, absolute. cron's working directory is unspecified, and a
# relative path is how a job silently starts a fresh state every time.
FEEDKIT_STATE_FILE=/home/you/.local/state/feedkit/feedkit-state.json

# The token is NOT written here. crontab entries are world-readable on many
# systems and end up in backups. The wrapper below sources a file that only
# you can read (chmod 600).

# ┌───────────── minute (0 - 59)
# │ ┌─────────── hour (0 - 23)
# │ │ ┌───────── day of month (1 - 31)
# │ │ │ ┌─────── month (1 - 12)
# │ │ │ │ ┌───── day of week (0 - 6, Sunday = 0)
# │ │ │ │ │
# 17 * * * *   → seventeen minutes past every hour.
#
# Why 17 and not 0: everybody's job runs at the top of the hour, which means
# everybody's job hits the same server in the same second. Picking an odd
# minute is free and is good manners.
17 * * * * . /home/you/.config/feedkit/token.env; /home/you/.local/bin/feedkit-scheduled >> /home/you/.local/state/feedkit/feedkit.log 2>&1

# The watchdog: a second, much simpler job that checks the first one is still
# alive. It runs from a different schedule on purpose — a watchdog that shares
# a fate with the thing it watches is not a watchdog. Exits non-zero when the
# last success is older than 90 minutes, and cron mails you the output.
23 */2 * * * /home/you/.local/bin/feedkit status --max-age-minutes 90 || echo "feedkit has gone quiet on $(hostname)"

# Notes worth keeping:
#
#   * cron mails you the output of a job only when the job produces output.
#     Redirecting to a log file (as above) therefore silences the mail; that is
#     why the watchdog line deliberately does NOT redirect.
#   * cron does not catch up. If the machine is asleep at 03:17, that run does
#     not happen — it is not deferred. This is precisely why the toolkit is
#     idempotent: the next run collects everything that was missed.
#   * cron uses the system's local time, which moves twice a year. A job at
#     02:30 runs twice or not at all on those two days. Anything sensitive to
#     that should run hourly, or on UTC, or both.

# feedkit — systemd service reference. NOT INSTALLED BY THIS LAB.
#
# To install on your own machine, deliberately:
#     mkdir -p ~/.config/systemd/user
#     cp feedkit.service feedkit.timer ~/.config/systemd/user/
#     systemctl --user daemon-reload
#     systemctl --user enable --now feedkit.timer
# To inspect:
#     systemctl --user list-timers feedkit.timer
#     journalctl --user -u feedkit.service -n 50
# To remove:
#     systemctl --user disable --now feedkit.timer
#     rm ~/.config/systemd/user/feedkit.service ~/.config/systemd/user/feedkit.timer
#
# Change every path before using this.

[Unit]
Description=feedkit — collect entries from configured sources
Documentation=man:feedkit(1)
# Do not start before the network is usable. Without this the first run after a
# boot fails with a name-resolution error and looks like a bug in your code.
Wants=network-online.target
After=network-online.target

[Service]
# oneshot: this runs, finishes, and exits. It is not a daemon, and telling
# systemd so is what makes `systemctl --user status` report honestly.
Type=oneshot

ExecStart=%h/.local/bin/feedkit-scheduled
WorkingDirectory=%h/.local/state/feedkit

Environment=FEEDKIT_BASE_URL=https://feeds.example.org
Environment=FEEDKIT_STATE_FILE=%h/.local/state/feedkit/feedkit-state.json
Environment=FEEDKIT_LOG_LEVEL=info

# The token comes from a file with mode 600, not from an Environment= line.
# Unit files are readable and `systemctl show` prints Environment= values.
EnvironmentFile=%h/.config/feedkit/token.env

# Structured logs on stdout land in the journal, tagged with this identifier,
# and `journalctl --user -u feedkit.service` reads them back. The program never
# has to know any of that.
StandardOutput=journal
StandardError=journal
SyslogIdentifier=feedkit

# A job that hangs is worse than a job that fails, because nothing notices.
# This is the outer belt; the inner one is the per-request timeout in the code.
TimeoutStartSec=300

# Modest hardening. None of this is exotic and all of it is free.
PrivateTmp=true
ProtectSystem=strict
ProtectHome=read-only
ReadWritePaths=%h/.local/state/feedkit
NoNewPrivileges=true

[Install]
WantedBy=default.target

# feedkit — systemd timer reference. NOT INSTALLED BY THIS LAB.
# Installed together with feedkit.service; see that file for the commands.

[Unit]
Description=Run feedkit hourly

[Timer]
# Hourly, at a randomly chosen but STABLE offset within the hour. Every machine
# with this timer picks its own minute and keeps it, which spreads load across
# the hour instead of hammering the source at :00 from everywhere at once.
OnCalendar=hourly
RandomizedDelaySec=900
FixedRandomDelay=true

# The one line cron cannot express. If the machine was off or asleep when a run
# was due, run it once shortly after boot instead of skipping it silently.
# Catch-up plus idempotence is the combination that makes a scheduled job
# survive a closed laptop.
Persistent=true

# Give the machine a moment to finish booting before the first catch-up run.
OnBootSec=2min

# Accuracy defaults to a minute; asking for a second costs wakeups and battery
# for a job that does not care.
AccuracySec=1min

[Install]
WantedBy=timers.target

# Schedule files — shown, never installed

Everything in this directory is a **reference**. Nothing here is installed by
this lab, and nothing here is executed by the test suite. That is the same
safety rule Day 81 followed: a course must not write into your real `crontab`,
your `~/Library/LaunchAgents`, or your systemd user units, and it must not
leave a background process running when you close the terminal.

To try one of these for real on your own machine, read it first, change the
paths to yours, and install it yourself with the command noted at the top of
the file. To undo it, use the removal command noted at the bottom.

| File | Supervisor | Platform |
| --- | --- | --- |
| `feedkit.cron` | `cron` | macOS, Linux, anything Unix-like |
| `com.example.feedkit.plist` | `launchd` | macOS |
| `feedkit.service` + `feedkit.timer` | systemd | most Linux distributions |

## The three things every one of them gets right

**An absolute path to the executable.** A scheduler does not run your shell
profile, so `PATH` is nearly empty and `feedkit` will not be found. Write
`/home/you/.local/bin/feedkit`, or the path inside your virtual environment.
This is the single most common reason a job that works by hand does nothing on
a schedule.

**A working directory, set explicitly.** `state_file = "feedkit-state.json"`
is relative to wherever the job starts, and where a job starts differs between
cron, launchd and systemd. Either set the working directory in the schedule
file or use an absolute `state_file` — the examples do both, belt and braces.

**The environment, supplied explicitly.** `FEEDKIT_BASE_URL` and
`FEEDKIT_TOKEN` are not in your shell profile as far as the scheduler is
concerned. Each file below shows where its supervisor expects them, and each
one keeps the token in a file that only you can read rather than in the
schedule entry itself.

## The fourth thing, which none of them can do for you

None of these supervisors will tell you that the job **stopped running**. cron
mails output when there is output; launchd and systemd record exits. All three
are silent about a job that was never triggered at all — because the plist was
unloaded, the timer was masked, the laptop was shut, or the crontab was lost
with the machine. That is what `feedkit status --max-age-minutes` is for, and
why the watchdog belongs somewhere other than the thing it is watching.

[console_scripts]
feedkit = feedkit.cli:main
feedkit-scheduled = feedkit.cli:scheduled_main

Metadata-Version: 2.4
Name: feedkit
Version: 1.0.0
Summary: A personal automation toolkit: collect entries from JSON sources on a schedule, without processing anything twice.
License: MIT
Keywords: automation,cli,scheduling,idempotence
Requires-Python: >=3.11
Requires-Dist: requests>=2.31
Provides-Extra: dev
Requires-Dist: pytest>=8; extra == "dev"

requests>=2.31

[dev]
pytest>=8

pyproject.toml
src/feedkit/__init__.py
src/feedkit/adapters.py
src/feedkit/cli.py
src/feedkit/config.py
src/feedkit/core.py
src/feedkit/logging_setup.py
src/feedkit/runner.py
src/feedkit/state.py
src/feedkit.egg-info/PKG-INFO
src/feedkit.egg-info/SOURCES.txt
src/feedkit.egg-info/dependency_links.txt
src/feedkit.egg-info/entry_points.txt
src/feedkit.egg-info/requires.txt
src/feedkit.egg-info/top_level.txt

feedkit

"""feedkit — a small, installable, scheduled automation toolkit.

The package is deliberately layered so that the boundaries sit at the edges:

    core            pure data in, pure data out — no network, no clock, no disk
    config          the four-layer precedence, resolved by a pure function
    logging_setup   structured JSON logging with secret redaction
    state           the state file, written atomically, plus the run lock
    adapters        the network and the clock — the only impure module
    runner          the order of one unattended run
    cli             argparse subcommands and the two console entry points

Read `core.py` first. It is where the interesting decisions live, and it is
readable without knowing anything about HTTP.
"""

__all__ = ["__version__"]

__version__ = "1.0.0"

"""The edges: the network, the clock, and sleeping.

Everything in this module talks to something outside the process. That is the
whole reason it is a separate module — the core can be tested with plain data
because none of this leaks into it, and this module can be swapped for a fake
in a test because the runner receives it as an argument rather than importing
it. Day 74's rule, applied to the two boundaries that hurt most.

Note the constructor of `HttpFetcher`: it takes a session, a timeout, a retry
count, a backoff base AND a `sleeper`. Injecting the sleeper is what lets the
test suite exercise three retries in microseconds instead of seconds, without
anybody having to patch `time.sleep` globally and hope.
"""

from __future__ import annotations

import time
from datetime import datetime, timezone
from typing import Any, Callable, Protocol

import requests


class FetchError(RuntimeError):
    """A source could not be fetched, after every retry was spent."""


class Clock(Protocol):
    """The clock, as an interface, so a test can hand over a fixed time."""

    def now_iso(self) -> str: ...


class SystemClock:
    """The real clock. UTC, always — a job that runs at 02:30 local time runs
    twice or not at all on the two days a year the offset changes."""

    def now_iso(self) -> str:
        return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


class FixedClock:
    """A clock that never moves. Shipped rather than hidden in a test file,
    because it belongs to the design, not to the tests."""

    def __init__(self, value: str) -> None:
        self.value = value

    def now_iso(self) -> str:
        return self.value


#: Statuses worth trying again: the server said "not now", not "never".
RETRYABLE_STATUS = frozenset({408, 425, 429, 500, 502, 503, 504})


class HttpFetcher:
    """Fetch one source's JSON, with a timeout and bounded retries.

    Three things here are not optional in an unattended job:

    * a **timeout** on every request — the default in `requests` is no timeout
      at all, and a job with no timeout does not fail, it hangs, which is the
      one outcome no supervisor can see;
    * **bounded** retries with exponential backoff — unbounded retries turn a
      failing dependency into a self-inflicted outage;
    * a retry decision based on WHAT went wrong. A 503 is worth another go; a
      404 or a 401 will be a 404 or a 401 forever, and retrying it is just
      noise you will pay for in someone else's server logs.
    """

    def __init__(
        self,
        session: requests.Session,
        base_url: str,
        token: str = "",
        timeout: float = 5.0,
        retries: int = 3,
        backoff_seconds: float = 0.5,
        sleeper: Callable[[float], None] = time.sleep,
        logger: Any = None,
    ) -> None:
        self.session = session
        self.base_url = base_url.rstrip("/")
        self.token = token
        self.timeout = timeout
        self.retries = max(1, retries)
        self.backoff_seconds = backoff_seconds
        self.sleeper = sleeper
        self.logger = logger

    def _headers(self) -> dict[str, str]:
        headers = {
            "Accept": "application/json",
            # Identify the client honestly — Day 79's rule, and the thing that
            # lets an operator find you when your job misbehaves.
            "User-Agent": "feedkit/1.0 (personal automation toolkit)",
        }
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"
        return headers

    def fetch(self, source: str) -> tuple[Any, int]:
        """Return the decoded payload and the attempt number that succeeded."""
        url = f"{self.base_url}/feed/{source}.json"
        last_error = "no attempt was made"

        for attempt in range(1, self.retries + 1):
            try:
                response = self.session.get(url, headers=self._headers(), timeout=self.timeout)
            except requests.RequestException as exc:
                last_error = f"{type(exc).__name__}: {exc}"
                if self.logger:
                    self.logger.warning(
                        "fetch attempt failed",
                        extra={"source": source, "attempt": attempt, "status": "transport"},
                    )
            else:
                if response.status_code == 200:
                    try:
                        return response.json(), attempt
                    except ValueError as exc:
                        # A body that is not JSON will not become JSON on a
                        # retry. Fail now.
                        raise FetchError(f"response was not JSON: {exc}") from exc
                last_error = f"HTTP {response.status_code}"
                if response.status_code not in RETRYABLE_STATUS:
                    raise FetchError(f"{last_error} (not retryable)")
                if self.logger:
                    self.logger.warning(
                        "fetch attempt failed",
                        extra={
                            "source": source,
                            "attempt": attempt,
                            "status": response.status_code,
                        },
                    )

            if attempt < self.retries:
                self.sleeper(self.backoff_seconds * (2 ** (attempt - 1)))

        raise FetchError(f"{last_error} after {self.retries} attempts")


def build_session() -> requests.Session:
    """One Session for the whole run, so the connection is reused across
    sources instead of being renegotiated for each one."""
    return requests.Session()

"""The command line: three subcommands over one shared core, plus the
scheduled entry point.

`feedkit fetch` does the work. `feedkit report` renders what has been
collected. `feedkit status` says when the last successful run was and whether
the toolkit has gone quiet. `feedkit-scheduled` is what the crontab, launchd
job or systemd timer invokes — the same fetch, with the settings a machine
wants rather than the settings a human wants.

Both entry points are declared in `pyproject.toml` under
`[project.scripts]`, which is what turns them into commands on PATH when the
package is installed. That is Day 83's mechanism doing the work Day 80's
argparse designed.
"""

from __future__ import annotations

import argparse
import sys
from pathlib import Path
from typing import Any, Sequence, TextIO

from . import adapters, config as config_module, core, logging_setup, runner
from . import state as state_module

VERSION = "1.0.0"


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="feedkit",
        description=(
            "Collect entries from configured JSON sources, on a schedule, "
            "without processing anything twice."
        ),
        epilog=(
            "Settings resolve in this order, weakest first: defaults, "
            "configuration file, environment, flags. Run "
            "`feedkit status --explain-config` to see where each value came from. "
            "The access token is read from FEEDKIT_TOKEN and is never written to "
            "a file or a log."
        ),
    )
    # Global options are the CONFIGURATION SETTINGS — the fourth and strongest
    # layer of the precedence. Keeping them all on the main parser rather than
    # scattering them across subcommands means `--max-items` means the same
    # thing everywhere, and `status --explain-config` can show the effect of
    # any of them.
    parser.add_argument("--version", action="version", version=f"feedkit {VERSION}")
    parser.add_argument("--config", metavar="PATH", help="configuration file to read")
    parser.add_argument("--base-url", dest="base_url", help="root address of the source server")
    parser.add_argument("--log-level", dest="log_level", choices=sorted(logging_setup.LEVELS))
    parser.add_argument("--state-file", dest="state_file", metavar="PATH")
    parser.add_argument("--sources", help="comma-separated list, overriding the configuration")
    parser.add_argument(
        "--max-items",
        dest="max_items",
        type=int,
        metavar="N",
        help="most previously unseen entries to accept from one source in one run",
    )
    parser.add_argument("--retries", type=int, metavar="N", help="attempts per source")

    subcommands = parser.add_subparsers(dest="command", metavar="COMMAND")

    fetch = subcommands.add_parser("fetch", help="collect new entries from every source")
    fetch.add_argument(
        "--dry-run",
        action="store_true",
        help="do everything except write the state file, and say what would have changed",
    )

    report = subcommands.add_parser("report", help="show what has been collected")
    report.add_argument("--limit", dest="report_limit", type=int, metavar="N")

    status = subcommands.add_parser("status", help="show the last successful run")
    status.add_argument(
        "--max-age-minutes",
        dest="max_age_minutes",
        type=int,
        metavar="N",
        help="watchdog allowance; exit 3 when the last success is older than this",
    )
    status.add_argument(
        "--explain-config",
        action="store_true",
        help="print every setting, its value, and which layer it came from",
    )
    return parser


CONFIG_FLAGS = (
    "config",
    "base_url",
    "log_level",
    "state_file",
    "sources",
    "max_items",
    "retries",
    "report_limit",
    "max_age_minutes",
)


def flags_from(args: argparse.Namespace) -> dict[str, Any]:
    """Only the parsed arguments that are configuration settings, and only the
    ones actually supplied — argparse leaves the rest as None, and None is what
    tells the resolver 'this layer has no opinion'."""
    return {name: getattr(args, name, None) for name in CONFIG_FLAGS}


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

    if not args.command:
        parser.print_help(out)
        return core.EXIT_FATAL

    try:
        settings, config_path = config_module.load(flags_from(args))
    except config_module.ConfigError as exc:
        print(f"feedkit: configuration error: {exc}", file=sys.stderr)
        return core.EXIT_FATAL

    run_id = runner.new_run_id()
    logger = logging_setup.configure(
        settings.log_level,
        run_id=run_id,
        secrets=[settings.token] if settings.token else [],
        stream=out,
    )
    state_path = Path(settings.state_file).expanduser()
    lock_path = state_path.with_suffix(state_path.suffix + ".lock")

    try:
        if args.command == "fetch":
            return _cmd_fetch(args, settings, state_path, lock_path, logger, out, run_id)
        if args.command == "report":
            return _cmd_report(settings, state_path, out)
        if args.command == "status":
            return _cmd_status(args, settings, config_path, state_path, out)
    except state_module.StateError as exc:
        print(f"feedkit: {exc}", file=sys.stderr)
        return core.EXIT_FATAL

    parser.print_help(out)
    return core.EXIT_FATAL


def _cmd_fetch(
    args: argparse.Namespace,
    settings: config_module.Config,
    state_path: Path,
    lock_path: Path,
    logger: Any,
    out: TextIO,
    run_id: str,
) -> int:
    session = adapters.build_session()
    try:
        fetcher = adapters.HttpFetcher(
            session=session,
            base_url=settings.base_url,
            token=settings.token,
            timeout=settings.timeout_seconds,
            retries=settings.retries,
            backoff_seconds=settings.backoff_seconds,
            logger=logger,
        )
        summary, code = runner.run_fetch(
            settings,
            fetcher,
            adapters.SystemClock(),
            state_path,
            lock_path,
            logger,
            dry_run=args.dry_run,
            run_id=run_id,
        )
    finally:
        session.close()

    print(core.format_summary(summary, run_id=run_id, dry_run=args.dry_run), file=out)
    return code


def _cmd_report(settings: config_module.Config, state_path: Path, out: TextIO) -> int:
    current = state_module.load(state_path)
    print(core.render_report(current, settings.report_limit), file=out)
    return core.EXIT_OK


def _cmd_status(
    args: argparse.Namespace,
    settings: config_module.Config,
    config_path: Path | None,
    state_path: Path,
    out: TextIO,
) -> int:
    if args.explain_config:
        print(config_module.explain(settings, config_path), file=out)
        return core.EXIT_OK

    current = state_module.load(state_path)
    text, stale = core.render_status(
        current, adapters.SystemClock().now_iso(), settings.max_age_seconds
    )
    print(text, file=out)
    # A watchdog is only useful if it can fail. Exiting non-zero on silence is
    # what lets a second, much simpler scheduled job page you when the first
    # one has stopped running at all.
    return core.EXIT_PARTIAL if stale else core.EXIT_OK


def scheduled_main(argv: Sequence[str] | None = None) -> int:
    """The entry point a scheduler invokes.

    It is a thin wrapper on purpose. The scheduled run wants a fetch, quieter
    logging by default, and no interactive help — everything else is identical,
    because a scheduled run that behaves differently from the one you tested by
    hand is a scheduled run you have not tested.
    """
    argv = list(sys.argv[1:] if argv is None else argv)
    return main(["--log-level", "info", "fetch", *argv])


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(main())

"""Configuration, resolved by a precedence that is written down and testable.

The order, weakest first:

    1. defaults baked into the code   — so the tool runs with no setup at all
    2. a configuration file           — the machine's long-lived preferences
    3. environment variables          — deployment-specific values and secrets
    4. command-line flags             — this one run, right now

Every automation has this order. Most of them have it by accident, spread over
a dozen `or` expressions, and nobody can say what wins. Here it is one pure
function over four dictionaries, so a test can assert all four levels — which
is exactly what `tests/run_tests.sh` does.

`resolve` also records WHERE each value came from. `feedkit status
--explain-config` prints that table, which turns "why is it doing that?" into a
five-second question instead of an afternoon.
"""

from __future__ import annotations

import os
import tomllib
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Mapping

#: Layer 1. Everything the toolkit needs to run at all, with no file, no
#: environment and no flags. A tool that cannot start without configuration is
#: a tool nobody tries.
DEFAULTS: dict[str, Any] = {
    "sources": ["notes", "links"],
    "max_items": 5,
    "report_limit": 10,
    "timeout_seconds": 5.0,
    "retries": 3,
    "backoff_seconds": 0.5,
    "log_level": "info",
    "state_file": "feedkit-state.json",
    "max_age_minutes": 1440,
}

#: Layer 3. Environment variable name -> (config key, type).
ENV_KEYS: dict[str, tuple[str, str]] = {
    "FEEDKIT_BASE_URL": ("base_url", "str"),
    "FEEDKIT_MAX_ITEMS": ("max_items", "int"),
    "FEEDKIT_REPORT_LIMIT": ("report_limit", "int"),
    "FEEDKIT_TIMEOUT_SECONDS": ("timeout_seconds", "float"),
    "FEEDKIT_RETRIES": ("retries", "int"),
    "FEEDKIT_BACKOFF_SECONDS": ("backoff_seconds", "float"),
    "FEEDKIT_LOG_LEVEL": ("log_level", "str"),
    "FEEDKIT_STATE_FILE": ("state_file", "str"),
    "FEEDKIT_MAX_AGE_MINUTES": ("max_age_minutes", "int"),
    "FEEDKIT_SOURCES": ("sources", "list"),
}

#: The one value that must NEVER come from a file in the repository.
SECRET_ENV = "FEEDKIT_TOKEN"


class ConfigError(ValueError):
    """The configuration is unusable. Stop the run; do not guess."""


@dataclass(frozen=True)
class Config:
    """The resolved settings for one run."""

    base_url: str
    sources: tuple[str, ...]
    max_items: int
    report_limit: int
    timeout_seconds: float
    retries: int
    backoff_seconds: float
    log_level: str
    state_file: str
    max_age_minutes: int
    token: str = ""
    provenance: Mapping[str, str] = field(default_factory=dict)

    @property
    def max_age_seconds(self) -> int:
        return self.max_age_minutes * 60


def coerce(value: Any, kind: str, key: str) -> Any:
    """Turn a string from a file, an environment variable or a flag into the
    type the rest of the program expects, and fail loudly when it cannot."""
    if kind == "list":
        if isinstance(value, (list, tuple)):
            return [str(item) for item in value]
        return [part.strip() for part in str(value).split(",") if part.strip()]
    try:
        if kind == "int":
            return int(value)
        if kind == "float":
            return float(value)
    except (TypeError, ValueError) as exc:
        raise ConfigError(f"{key}: {value!r} is not a valid {kind}") from exc
    return str(value)


def kind_of(key: str) -> str:
    """The declared type of a configuration key, from the defaults table."""
    for _, (config_key, kind) in ENV_KEYS.items():
        if config_key == key:
            return kind
    default = DEFAULTS.get(key)
    if isinstance(default, bool):
        return "str"
    if isinstance(default, int):
        return "int"
    if isinstance(default, float):
        return "float"
    if isinstance(default, list):
        return "list"
    return "str"


def read_config_file(path: Path) -> dict[str, Any]:
    """Read a TOML configuration file. Missing file is not an error; an
    unparseable one is."""
    if not path.is_file():
        return {}
    try:
        with path.open("rb") as handle:
            data = tomllib.load(handle)
    except tomllib.TOMLDecodeError as exc:
        raise ConfigError(f"{path}: not valid TOML: {exc}") from exc
    settings = data.get("feedkit", data)
    if not isinstance(settings, dict):
        raise ConfigError(f"{path}: expected a table of settings")
    return dict(settings)


def find_config_file(explicit: str | None, environ: Mapping[str, str], cwd: Path) -> Path | None:
    """Where the configuration file lives, in the order a user expects.

    A flag beats the environment, which beats the current directory, which
    beats the user's XDG configuration directory. Returning None means "no file
    anywhere", which is a perfectly normal state, not an error.
    """
    if explicit:
        return Path(explicit).expanduser()
    from_env = environ.get("FEEDKIT_CONFIG")
    if from_env:
        return Path(from_env).expanduser()
    local = cwd / "feedkit.toml"
    if local.is_file():
        return local
    base = environ.get("XDG_CONFIG_HOME")
    home = Path(base).expanduser() if base else Path(environ.get("HOME", "~")).expanduser() / ".config"
    candidate = home / "feedkit" / "feedkit.toml"
    return candidate if candidate.is_file() else None


def env_values(environ: Mapping[str, str]) -> dict[str, Any]:
    """Layer 3, extracted from an environment mapping passed in as an argument
    (never read from os.environ in here — that is a boundary, and tests need to
    supply their own)."""
    values: dict[str, Any] = {}
    for name, (key, kind) in ENV_KEYS.items():
        if name in environ and environ[name] != "":
            values[key] = coerce(environ[name], kind, key)
    return values


def resolve(
    file_values: Mapping[str, Any],
    environment: Mapping[str, Any],
    flags: Mapping[str, Any],
    token: str = "",
) -> Config:
    """Apply the four layers in order and record where each value came from.

    Pure. Four dictionaries in, one Config out. This is the function the lab's
    precedence test drives directly, and it is why "which layer wins?" is a
    question with a checked answer rather than a folk belief.
    """
    merged: dict[str, Any] = {}
    provenance: dict[str, str] = {}

    for layer_name, layer in (
        ("default", DEFAULTS),
        ("file", file_values),
        ("environment", environment),
        ("flag", {key: value for key, value in flags.items() if value is not None}),
    ):
        for key, value in layer.items():
            if key not in DEFAULTS and key != "base_url":
                # An unknown key in a config file is almost always a typo, and
                # silently ignoring it is how people lose an afternoon.
                if layer_name == "file":
                    raise ConfigError(f"unknown setting in configuration file: {key!r}")
                continue
            merged[key] = coerce(value, kind_of(key), key) if layer_name != "default" else value
            provenance[key] = layer_name

    base_url = str(merged.get("base_url", "")).rstrip("/")
    if not base_url:
        raise ConfigError(
            "no base URL configured. Set FEEDKIT_BASE_URL in the environment, "
            "or pass --base-url. Deployment-specific addresses do not belong in "
            "a file that is committed."
        )
    if merged["retries"] < 1:
        raise ConfigError("retries must be at least 1")
    if merged["max_items"] < 0:
        raise ConfigError("max-items must not be negative")

    provenance.setdefault("base_url", "environment")
    provenance["token"] = "environment" if token else "unset"

    return Config(
        base_url=base_url,
        sources=tuple(merged["sources"]),
        max_items=int(merged["max_items"]),
        report_limit=int(merged["report_limit"]),
        timeout_seconds=float(merged["timeout_seconds"]),
        retries=int(merged["retries"]),
        backoff_seconds=float(merged["backoff_seconds"]),
        log_level=str(merged["log_level"]),
        state_file=str(merged["state_file"]),
        max_age_minutes=int(merged["max_age_minutes"]),
        token=token,
        provenance=provenance,
    )


def load(
    flags: Mapping[str, Any],
    environ: Mapping[str, str] | None = None,
    cwd: Path | None = None,
) -> tuple[Config, Path | None]:
    """The impure wrapper: find the file, read the environment, then call the
    pure `resolve`. All the I/O is in these six lines."""
    environ = os.environ if environ is None else environ
    cwd = Path.cwd() if cwd is None else cwd
    config_path = find_config_file(flags.get("config"), environ, cwd)
    file_values = read_config_file(config_path) if config_path else {}
    token = environ.get(SECRET_ENV, "")
    config = resolve(file_values, env_values(environ), flags, token=token)
    return config, config_path


def explain(config: Config, config_path: Path | None) -> str:
    """The provenance table. Answers 'why is it doing that?' in one command."""
    lines = [f"configuration file: {config_path or 'none found'}", ""]
    lines.append(f"  {'setting':<20} {'value':<28} {'came from'}")
    for key in sorted(config.provenance):
        if key == "token":
            shown = "set (never printed)" if config.token else "unset"
        else:
            shown = str(getattr(config, key, ""))
        lines.append(f"  {key:<20} {shown:<28} {config.provenance[key]}")
    return "\n".join(lines)

"""The pure core of the toolkit.

Nothing in this module touches the network, the clock, the filesystem or a
subprocess. Every function here takes plain data and returns plain data, which
is exactly what makes the interesting parts of an automation testable without
starting a server or waiting a second. Day 74 argued for pushing boundaries to
the edges; this module is what the middle looks like when you do.

If you ever find yourself wanting to `import requests` or call `time.time()` in
here, that is the signal that the value belongs in a parameter instead.
"""

from __future__ import annotations

import calendar
import time
from dataclasses import dataclass
from typing import Any, Iterable, Mapping, Sequence

# `calendar` and `time` appear here only to PARSE a timestamp that was handed
# in as a string. Nothing in this module ever asks what time it is now — that
# question belongs to the clock adapter, so that every test can answer it.

STATE_VERSION = 1

#: Exit codes. These are the machine-readable half of every run, and the
#: scheduler is the thing that reads them.
EXIT_OK = 0
EXIT_FATAL = 1
EXIT_PARTIAL = 3
EXIT_LOCKED = 75


@dataclass(frozen=True)
class Entry:
    """One item collected from one source."""

    id: str
    title: str
    published: str
    source: str = ""

    def as_dict(self) -> dict[str, str]:
        return {
            "id": self.id,
            "title": self.title,
            "published": self.published,
            "source": self.source,
        }


@dataclass(frozen=True)
class SourceResult:
    """What happened to one source during one run."""

    source: str
    status: str  # "ok" | "failed"
    new_entries: tuple[Entry, ...] = ()
    error: str = ""
    attempts: int = 1


class InvalidPayload(ValueError):
    """The server answered, but not with something this toolkit understands."""


def parse_entries(payload: Any, source: str) -> tuple[Entry, ...]:
    """Validate a decoded JSON payload and turn it into Entry objects.

    A 200 response is not the same thing as a correct response. Validating at
    the edge means every function downstream can assume the shape it was given,
    which is the whole reason this is a separate step rather than a dict lookup
    buried three calls deep.
    """
    if not isinstance(payload, Mapping):
        raise InvalidPayload(f"top level is {type(payload).__name__}, expected an object")
    items = payload.get("entries")
    if not isinstance(items, Sequence) or isinstance(items, (str, bytes)):
        raise InvalidPayload("'entries' is missing or is not a list")

    parsed: list[Entry] = []
    for index, item in enumerate(items):
        if not isinstance(item, Mapping):
            raise InvalidPayload(f"entry {index} is not an object")
        missing = [field for field in ("id", "title", "published") if field not in item]
        if missing:
            raise InvalidPayload(f"entry {index} is missing {', '.join(missing)}")
        parsed.append(
            Entry(
                id=str(item["id"]),
                title=str(item["title"]),
                published=str(item["published"]),
                source=source,
            )
        )
    return tuple(parsed)


def select_new(
    entries: Iterable[Entry], seen_ids: Iterable[str], max_items: int
) -> tuple[Entry, ...]:
    """Return the entries not already recorded, newest first, capped at max_items.

    This is the idempotence rule, and it is four lines of pure logic precisely
    because the record of what has been seen arrives as an argument rather than
    being read from a file in here.
    """
    already = set(seen_ids)
    fresh = [entry for entry in entries if entry.id not in already]
    fresh.sort(key=lambda entry: (entry.published, entry.id), reverse=True)
    if max_items >= 0:
        fresh = fresh[:max_items]
    return tuple(fresh)


def empty_state() -> dict[str, Any]:
    """The state of a toolkit that has never run."""
    return {
        "version": STATE_VERSION,
        "last_run": None,
        "last_success": None,
        "sources": {},
        "entries": [],
    }


def merge_state(
    state: Mapping[str, Any],
    results: Sequence[SourceResult],
    run_id: str,
    started_at: str,
    finished_at: str,
    keep_entries: int = 200,
) -> dict[str, Any]:
    """Fold one run's results into the previous state and return the new state.

    Pure: give it the same inputs and it returns the same dictionary, every
    time, on any machine. The caller decides whether to write it.
    """
    summary = summarise(results)
    sources = {name: dict(value) for name, value in dict(state.get("sources") or {}).items()}
    collected: list[dict[str, str]] = list(state.get("entries") or [])

    for result in results:
        record = sources.setdefault(
            result.source, {"seen_ids": [], "last_success": None, "last_error": ""}
        )
        if result.status == "ok":
            seen = list(record.get("seen_ids") or [])
            seen.extend(entry.id for entry in result.new_entries if entry.id not in seen)
            record["seen_ids"] = seen
            record["last_success"] = finished_at
            record["last_error"] = ""
            collected = [entry.as_dict() for entry in result.new_entries] + collected
        else:
            record["last_error"] = result.error

    collected.sort(key=lambda item: (item["published"], item["id"]), reverse=True)

    new_state: dict[str, Any] = {
        "version": STATE_VERSION,
        "last_run": {
            "run_id": run_id,
            "started_at": started_at,
            "finished_at": finished_at,
            "status": summary["status"],
            "sources_ok": summary["sources_ok"],
            "sources_failed": summary["sources_failed"],
            "new_entries": summary["new_entries"],
        },
        "last_success": (
            finished_at if summary["status"] == "ok" else state.get("last_success")
        ),
        "sources": sources,
        "entries": collected[:keep_entries],
    }
    return new_state


def summarise(results: Sequence[SourceResult]) -> dict[str, Any]:
    """Reduce a run's results to the handful of numbers a human reads."""
    ok = [result for result in results if result.status == "ok"]
    failed = [result for result in results if result.status != "ok"]
    new_entries = sum(len(result.new_entries) for result in ok)

    if not results:
        status = "ok"
    elif not failed:
        status = "ok"
    elif not ok:
        status = "failed"
    else:
        status = "partial"

    return {
        "status": status,
        "sources_total": len(results),
        "sources_ok": len(ok),
        "sources_failed": len(failed),
        "new_entries": new_entries,
        "failures": {result.source: result.error for result in failed},
        "retried": {result.source: result.attempts for result in results if result.attempts > 1},
    }


def exit_code_for(summary: Mapping[str, Any]) -> int:
    """Map a run summary onto the exit code the scheduler will read.

    Partial success gets its own code. Reporting 0 for "most of it worked" is
    the single most common way an automation lies to the person who owns it.
    """
    status = summary.get("status")
    if status == "ok":
        return EXIT_OK
    if status == "partial":
        return EXIT_PARTIAL
    return EXIT_FATAL


def format_summary(summary: Mapping[str, Any], run_id: str, dry_run: bool = False) -> str:
    """The human-readable run summary, printed at the end of every run."""
    lines = [
        f"run {run_id}: {summary['status']}"
        + ("  (dry run — nothing was written)" if dry_run else ""),
        f"  sources: {summary['sources_ok']} ok, {summary['sources_failed']} failed,"
        f" {summary['sources_total']} total",
        f"  new entries: {summary['new_entries']}",
    ]
    for source, attempts in sorted(dict(summary.get("retried") or {}).items()):
        lines.append(f"  retried: {source} succeeded on attempt {attempts}")
    for source, error in sorted(dict(summary.get("failures") or {}).items()):
        lines.append(f"  FAILED: {source}: {error}")
    return "\n".join(lines)


def render_report(state: Mapping[str, Any], limit: int) -> str:
    """Render what has been collected. Reads state, touches nothing."""
    entries = list(state.get("entries") or [])
    if not entries:
        return "No entries collected yet. Run `feedkit fetch` first."

    shown = entries[: limit if limit >= 0 else len(entries)]
    width = max(len(entry["source"]) for entry in shown)
    lines = [f"{len(entries)} entries collected; showing {len(shown)}", ""]
    for entry in shown:
        lines.append(f"  {entry['published']}  {entry['source']:<{width}}  {entry['title']}")
    return "\n".join(lines)


def render_status(state: Mapping[str, Any], now: str, max_age_seconds: int) -> tuple[str, bool]:
    """Render the status block and say whether the toolkit has gone quiet.

    The second half of the tuple is the watchdog answer: True when the last
    successful run is older than the allowance. Alerting on silence catches the
    failure mode that alerting on errors cannot — the run that never happened.
    """
    last_success = state.get("last_success")
    last_run = state.get("last_run")
    stale = is_stale(last_success, now, max_age_seconds)

    lines = [f"last success: {last_success or 'never'}"]
    if last_run:
        lines.append(
            f"last run:     {last_run['run_id']} at {last_run['finished_at']}"
            f" ({last_run['status']}, {last_run['new_entries']} new)"
        )
    else:
        lines.append("last run:     never")
    lines.append(f"now:          {now}")
    lines.append(
        f"watchdog:     {'STALE' if stale else 'fresh'}"
        f" (allowance {max_age_seconds}s)"
    )
    for name, record in sorted(dict(state.get("sources") or {}).items()):
        note = record.get("last_error") or "ok"
        lines.append(
            f"  {name}: {len(record.get('seen_ids') or [])} seen,"
            f" last success {record.get('last_success') or 'never'} — {note}"
        )
    return "\n".join(lines), stale


def is_stale(last_success: str | None, now: str, max_age_seconds: int) -> bool:
    """True when the last success is missing or older than the allowance.

    Timestamps are ISO 8601 strings, compared by parsing them into seconds. The
    parsing lives in the caller's clock adapter; here we accept the already
    normalised comparison to keep this module free of the datetime module's
    timezone surprises. Both arguments must be UTC ISO strings.
    """
    if not last_success:
        return True
    return _iso_seconds(now) - _iso_seconds(last_success) > max_age_seconds


def _iso_seconds(value: str) -> int:
    """Seconds since the epoch for a UTC ISO 8601 timestamp such as
    2026-07-19T10:11:12Z. Deliberately small and deliberately strict."""
    cleaned = value.replace("Z", "").split(".")[0]
    parsed = time.strptime(cleaned, "%Y-%m-%dT%H:%M:%S")
    return calendar.timegm(parsed)

"""Structured logging — the thing that makes an unattended run debuggable.

Two decisions are baked in here, and both are worth arguing rather than
copying.

**One JSON object per line, on stdout.** A line of prose is readable by you at
your desk; a line of JSON is readable by you AND by `grep`, `jq`, a log
shipper, and whatever the supervisor writes it into. Writing to stdout rather
than opening a log file means the program does not have to know about log
rotation, permissions, or where the operator wants their logs — cron mails it,
systemd hands it to the journal, launchd redirects it, and a human running the
command by hand simply sees it. Fewer decisions inside the program is the
point.

**Every record carries the run id and, where it applies, the item.** An
unattended failure is a message you read hours later with no memory of the
context. "Timeout" tells you nothing. "run 8f2c1a: source=papers attempt=3
timeout after 5.0s" tells you which run, which item, how hard it tried, and
what the limit was.
"""

from __future__ import annotations

import json
import logging
import sys
from typing import Any, Iterable, TextIO

LEVELS = {
    "debug": logging.DEBUG,
    "info": logging.INFO,
    "warning": logging.WARNING,
    "error": logging.ERROR,
    "critical": logging.CRITICAL,
}

#: Extra keys that the formatter promotes to top-level JSON fields.
CONTEXT_KEYS = ("source", "attempt", "status", "count", "path", "elapsed_ms", "url")


class RedactingFilter(logging.Filter):
    """Replace known secret values with a placeholder, everywhere.

    This is a seatbelt, not a licence. The right habit is to never put a token
    into a log call in the first place; this filter exists because one day
    somebody will log a whole request object, or an exception message that
    happens to quote a URL with a token in the query string, and the difference
    between a bad afternoon and a credential rotation is whether that string
    reached the log.
    """

    PLACEHOLDER = "***REDACTED***"

    def __init__(self, secrets: Iterable[str] = ()) -> None:
        super().__init__()
        # Very short strings would redact half the alphabet; ignore them.
        self.secrets = tuple(secret for secret in secrets if secret and len(secret) >= 6)

    def _scrub(self, value: Any) -> Any:
        if isinstance(value, str):
            for secret in self.secrets:
                value = value.replace(secret, self.PLACEHOLDER)
            return value
        if isinstance(value, (list, tuple)):
            return type(value)(self._scrub(item) for item in value)
        if isinstance(value, dict):
            return {key: self._scrub(item) for key, item in value.items()}
        return value

    def filter(self, record: logging.LogRecord) -> bool:
        if not self.secrets:
            return True
        record.msg = self._scrub(record.msg)
        if record.args:
            record.args = self._scrub(record.args)
        for key in CONTEXT_KEYS:
            if hasattr(record, key):
                setattr(record, key, self._scrub(getattr(record, key)))
        if record.exc_info:
            # An exception's own text is the most common accidental leak.
            exc = record.exc_info[1]
            if exc is not None and exc.args:
                exc.args = tuple(self._scrub(arg) for arg in exc.args)
        return True


class JsonFormatter(logging.Formatter):
    """One JSON object per line, with a stable field order."""

    def __init__(self, run_id: str) -> None:
        super().__init__()
        self.run_id = run_id

    def format(self, record: logging.LogRecord) -> str:
        payload: dict[str, Any] = {
            "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"),
            "level": record.levelname.lower(),
            "run_id": self.run_id,
            "event": record.getMessage(),
        }
        for key in CONTEXT_KEYS:
            if hasattr(record, key):
                payload[key] = getattr(record, key)
        if record.exc_info:
            payload["error"] = str(record.exc_info[1])
        # ensure_ascii=False keeps the log readable to a human. Escaping every
        # non-ASCII character is the default and is nearly always wrong for a
        # file somebody has to read at three in the morning.
        return json.dumps(payload, sort_keys=False, ensure_ascii=False)


def configure(
    level: str,
    run_id: str,
    secrets: Iterable[str] = (),
    stream: TextIO | None = None,
) -> logging.Logger:
    """Build the toolkit's logger. Called once, at the start of a run."""
    logger = logging.getLogger("feedkit")
    logger.handlers.clear()
    logger.propagate = False
    logger.setLevel(LEVELS.get(level.lower(), logging.INFO))

    handler = logging.StreamHandler(stream if stream is not None else sys.stdout)
    handler.setFormatter(JsonFormatter(run_id))
    handler.addFilter(RedactingFilter(secrets))
    logger.addHandler(handler)
    return logger

"""One unattended run, start to finish.

This is the only module that knows the ORDER of things: acquire the lock, load
the state, fetch each source with retries, skip and report the ones that fail,
fold the successes into the state, write the state atomically, print the
summary, return an exit code. Everything it does with the outside world arrives
as an argument — the fetcher, the clock, the logger, the paths — which is why
the whole thing can be exercised against a fake fetcher in a millisecond.

The failure policy is the part worth reading twice, because it is the part that
distinguishes an automation from a script:

* a **transport error or a 5xx** on one source is retried with backoff and, if
  it never succeeds, is SKIPPED and REPORTED — one broken source must not stop
  the other four;
* a **payload that does not parse** is not retried, because it will not parse
  next time either; it is skipped and reported the same way;
* a **failure to acquire the lock** stops everything immediately, because the
  correct response to "a run is already happening" is to do nothing;
* an **unusable state file or an invalid configuration** stops everything,
  because continuing would mean guessing about the thing that records what has
  already been done.

Partial success is the normal case for a batch job, and it gets its own exit
code. Reporting 0 because "most of it worked" is how an automation becomes a
thing nobody can trust.
"""

from __future__ import annotations

import uuid
from pathlib import Path
from typing import Any, Protocol, Sequence

from . import core, state as state_module
from .adapters import FetchError
from .config import Config


class Fetcher(Protocol):
    def fetch(self, source: str) -> tuple[Any, int]: ...


def new_run_id() -> str:
    """A short, unique label for one run. Every log line carries it, so the
    lines belonging to a single 03:00 run can be pulled out of a month of
    output with one grep."""
    return uuid.uuid4().hex[:8]


def fetch_sources(
    sources: Sequence[str],
    fetcher: Fetcher,
    seen: dict[str, list[str]],
    max_items: int,
    logger: Any,
) -> list[core.SourceResult]:
    """Fetch every source, collecting successes and failures side by side."""
    results: list[core.SourceResult] = []
    for source in sources:
        logger.info("source started", extra={"source": source})
        try:
            payload, attempts = fetcher.fetch(source)
            entries = core.parse_entries(payload, source)
        except (FetchError, core.InvalidPayload) as exc:
            logger.error("source failed", extra={"source": source, "status": "failed"})
            results.append(core.SourceResult(source=source, status="failed", error=str(exc)))
            continue

        fresh = core.select_new(entries, seen.get(source, []), max_items)
        logger.info(
            "source finished",
            extra={
                "source": source,
                "status": "ok",
                "count": len(fresh),
                "attempt": attempts,
            },
        )
        results.append(
            core.SourceResult(
                source=source, status="ok", new_entries=fresh, attempts=attempts
            )
        )
    return results


def run_fetch(
    config: Config,
    fetcher: Fetcher,
    clock: Any,
    state_path: Path,
    lock_path: Path,
    logger: Any,
    dry_run: bool = False,
    run_id: str | None = None,
) -> tuple[dict[str, Any], int]:
    """Do one fetch run. Returns the summary and the exit code.

    The run id is passed IN rather than generated here, so that the id stamped
    on every log line and the id recorded in the state file are the same
    string. Two ids for one run is a small bug that makes an incident twice as
    slow to investigate, and it is easy to ship without noticing.
    """
    run_id = run_id or new_run_id()
    started_at = clock.now_iso()

    try:
        with state_module.Lock(lock_path):
            logger.info("run started", extra={"status": "started", "path": str(state_path)})
            current = state_module.load(state_path)
            seen = {
                name: list(record.get("seen_ids") or [])
                for name, record in dict(current.get("sources") or {}).items()
            }

            results = fetch_sources(config.sources, fetcher, seen, config.max_items, logger)
            finished_at = clock.now_iso()
            summary = core.summarise(results)
            merged = core.merge_state(current, results, run_id, started_at, finished_at)

            if dry_run:
                logger.info(
                    "dry run — state not written",
                    extra={"status": "dry-run", "count": summary["new_entries"]},
                )
            else:
                state_module.write_atomic(state_path, merged)
                logger.info(
                    "state written",
                    extra={"status": summary["status"], "path": str(state_path)},
                )
    except state_module.LockHeld as exc:
        logger.error("another run is in progress", extra={"status": "locked"})
        return (
            {
                "run_id": run_id,
                "status": "locked",
                "sources_total": 0,
                "sources_ok": 0,
                "sources_failed": 0,
                "new_entries": 0,
                "failures": {"lock": str(exc)},
                "retried": {},
            },
            core.EXIT_LOCKED,
        )

    summary["run_id"] = run_id
    exit_code = core.exit_code_for(summary)
    logger.info(
        "run finished",
        extra={"status": summary["status"], "count": summary["new_entries"]},
    )
    return summary, exit_code

"""The state file, and the atomic write that keeps it trustworthy.

State is what makes a job idempotent: it is the record of what has already been
processed, so a second run does not do the work twice. That makes it the single
most valuable file the toolkit owns, and losing it is worse than a failed run —
a failed run is visible, a corrupted state file quietly re-processes or
silently skips.

So the write is atomic, exactly as Days 64 and 65 described. Write the whole
new document to a temporary file in the SAME directory, flush it, ask the
operating system to put it on the disk, then `os.replace` it over the old name.
`os.replace` is atomic on POSIX and on Windows: any reader sees either the
complete old file or the complete new one, never a half-written mixture. If the
machine loses power between the write and the replace, the previous state is
still there and the temporary file is garbage that the next run cleans up.

The naive version — `open(path, "w")` then `json.dump` — truncates the real
file first. Interrupt it and the record of everything you have ever processed
is a zero-byte file.
"""

from __future__ import annotations

import json
import os
import tempfile
from pathlib import Path
from typing import Any, Callable, Mapping

from .core import empty_state


class StateError(RuntimeError):
    """The state file exists but cannot be used. Never guess; stop."""


def load(path: Path) -> dict[str, Any]:
    """Read the state file, or return a fresh empty state if there is none."""
    if not path.is_file():
        return empty_state()
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise StateError(
            f"{path} is not valid JSON ({exc}). Refusing to overwrite it. "
            f"Move it aside to start fresh."
        ) from exc
    if not isinstance(data, dict) or "version" not in data:
        raise StateError(f"{path} does not look like a feedkit state file")
    return data


def write_atomic(
    path: Path,
    state: Mapping[str, Any],
    crash_hook: Callable[[], None] | None = None,
) -> None:
    """Write state so that an interruption leaves the previous file intact.

    `crash_hook` exists purely so the lab can prove the property. The test
    passes a function that raises, standing in for the power cut, and then
    asserts the old file is byte-identical. Production code passes nothing.
    """
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = json.dumps(state, indent=2, sort_keys=True) + "\n"

    handle = tempfile.NamedTemporaryFile(
        mode="w",
        encoding="utf-8",
        dir=str(path.parent),
        prefix=path.name + ".",
        suffix=".tmp",
        delete=False,
    )
    tmp_path = Path(handle.name)
    try:
        with handle:
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        if crash_hook is not None:
            crash_hook()
        os.replace(tmp_path, path)
    except BaseException:
        tmp_path.unlink(missing_ok=True)
        raise


class LockHeld(RuntimeError):
    """Another run of this toolkit is already in progress."""


class Lock:
    """A lock file, so two scheduled runs never overlap.

    Created with O_CREAT | O_EXCL, which the operating system guarantees will
    succeed for exactly one caller. The file holds the process id, which is
    what lets a human decide whether a lock left behind by a crash is stale.

    This is deliberately the simplest thing that works on one machine. It is
    not a distributed lock and must not be used as one.
    """

    def __init__(self, path: Path) -> None:
        self.path = path
        self._acquired = False

    def __enter__(self) -> "Lock":
        self.path.parent.mkdir(parents=True, exist_ok=True)
        try:
            fd = os.open(self.path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
        except FileExistsError as exc:
            holder = ""
            try:
                holder = self.path.read_text(encoding="utf-8").strip()
            except OSError:
                pass
            raise LockHeld(
                f"{self.path} exists (held by pid {holder or 'unknown'}). "
                f"Another run is in progress, or a previous run was killed. "
                f"Delete the file only after checking that no such process exists."
            ) from exc
        with os.fdopen(fd, "w") as handle:
            handle.write(str(os.getpid()))
        self._acquired = True
        return self

    def __exit__(self, *exc_info: object) -> None:
        if self._acquired:
            self.path.unlink(missing_ok=True)
            self._acquired = False

lesson_id: D084
day: 84
kind: python-program
languages: [python, bash, toml, xml, ini]
setup_commands:
  - cd labs/sections/programming-with-python/day-084-shipping-an-automation-toolkit
  - python3 -m venv .venv
  - .venv/bin/pip install -r requirements/requirements.txt
  - .venv/bin/pytest --version
run_commands:
  - bash tests/run_tests.sh
  - .venv/bin/python tests/fixture_server.py --token demo-token-value
  - 'FEEDKIT_BASE_URL=http://127.0.0.1:PORT FEEDKIT_TOKEN=demo-token-value .venv/bin/python -m feedkit.cli fetch'
  - 'FEEDKIT_BASE_URL=http://127.0.0.1:PORT FEEDKIT_TOKEN=demo-token-value .venv/bin/python -m feedkit.cli fetch --dry-run'
  - 'FEEDKIT_BASE_URL=http://127.0.0.1:PORT FEEDKIT_TOKEN=demo-token-value .venv/bin/python -m feedkit.cli --sources notes,broken,papers fetch'
  - 'FEEDKIT_BASE_URL=http://127.0.0.1:PORT .venv/bin/python -m feedkit.cli status --explain-config'
  - 'FEEDKIT_BASE_URL=http://127.0.0.1:PORT .venv/bin/python -m feedkit.cli report --limit 5'
  - 'FEEDKIT_BASE_URL=http://127.0.0.1:PORT .venv/bin/python -m feedkit.cli status --max-age-minutes 60'
  - .venv/bin/pip install -e examples --no-build-isolation --no-deps
  - cat examples/schedule/feedkit.cron
test_commands:
  - bash tests/run_tests.sh
cleanup_commands:
  - rm -f examples/feedkit-state.json examples/feedkit-state.json.lock
  - rm -rf examples/src/feedkit.egg-info
  - find . -type d -name __pycache__ -prune -exec rm -rf -- {} +
  - '.venv/bin/pip uninstall -y feedkit  # optional: remove the console scripts'
  - '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, arm64), Python 3.14.0, bash 3.2.57, requests 2.34.2, pytest 9.1.1, setuptools 83.0.0 — bash tests/run_tests.sh -> 52 checks, 0 failure(s), exit 0'

# Dependencies

Three packages, all free and open source, all pinned to an exact version.

| Package | Version | Why this lab needs it | Licence |
| --- | --- | --- | --- |
| `requests` | 2.34.2 | The HTTP client the toolkit's fetch adapter uses (Day 78). Everything it does here could be done with the standard library's `urllib.request`; `requests` is used because sessions, timeouts and JSON decoding are one line each, and because it is what you will meet in other people's code | Apache-2.0 |
| `pytest` | 9.1.1 | Runs the property suite in `tests/test_toolkit.py` (Days 71–73) | MIT |
| `setuptools` | 83.0.0 | The build backend named in `examples/pyproject.toml`. It is what turns `pip install -e examples` into two working console scripts (Day 83). Python's `venv` no longer installs it by default, so it has to be asked for | MIT |

Install them:

```bash
python3 -m venv .venv
.venv/bin/pip install -r requirements/requirements.txt
```

## What is deliberately absent

**No scheduling library.** cron, launchd and systemd timers are already on your
machine, and a Python process that sleeps in a loop to schedule itself is a
process that has to be kept alive, monitored and restarted — which is the job
you just gave yourself instead of the one the operating system already does.
Day 81 argued this at length; the schedule files under `examples/schedule/` are
the result.

**No configuration library.** `tomllib` has been in the standard library since
Python 3.11, and the four-layer precedence in `config.py` is thirty lines. A
dependency for that is a dependency you will still be maintaining in a year.

**No logging library.** `logging` is in the standard library, the JSON
formatter is fifteen lines, and the redaction filter is the part that actually
matters and would have needed writing regardless.

**No retry library.** The bounded retry with exponential backoff is twelve
lines in `adapters.py`. Reach for a library when you need jitter, circuit
breaking, or per-exception policies — not before.

## The network

The install above is the only step that needs the internet. **The tests
themselves need no network at all**: they start a fixture server on 127.0.0.1
on a port the operating system chooses, and nothing in this lab ever resolves a
name or opens a socket to anywhere else.

# Exact pins. An automation you will not look at for six months must not change
# underneath you because a dependency released on a Tuesday.
requests==2.34.2
pytest==9.1.1
setuptools==83.0.0

# feedkit configuration — layer 2 of four.
#
# What belongs here: the long-lived preferences of THIS machine. What does not:
# anything that changes per run (that is a flag), anything that differs between
# machines or deployments (that is the environment), and above all anything
# secret. There is no token in this file and there never will be one; the
# toolkit reads FEEDKIT_TOKEN from the environment and refuses to look anywhere
# else.
#
# The address of the source server is also absent on purpose. It is a
# deployment fact, not a preference, so it arrives as FEEDKIT_BASE_URL — which
# is also what lets the test suite point the toolkit at a server it started
# itself, on a port it chose at run time.
#
# Where this file lives, in the order the toolkit looks:
#   1. the path given to --config
#   2. $FEEDKIT_CONFIG
#   3. ./feedkit.toml   (this one — handy while developing)
#   4. $XDG_CONFIG_HOME/feedkit/feedkit.toml, or ~/.config/feedkit/feedkit.toml
#
# Number 4 is the right home for a real installation on your own machine: it
# survives reinstalling the package, it is not inside a directory you might
# delete, and it is not inside a repository you might publish.

[feedkit]
sources = ["notes", "links", "papers"]

# How many previously unseen entries to accept from one source in one run.
# A cap is a safety belt: the first run against a large source should not
# process ten thousand items before you have looked at one.
max_items = 10

report_limit = 10

# Never omit a timeout. The default in `requests` is to wait forever, and a
# scheduled job that hangs is invisible to every supervisor there is.
timeout_seconds = 5.0

# Bounded retries, doubling the wait each time: 0.5s, 1.0s, 2.0s.
retries = 3
backoff_seconds = 0.5

log_level = "info"

# Relative to the working directory the scheduler starts the job in. For a real
# installation, prefer an absolute path under your home directory — a schedule
# entry rarely runs where you think it does.
state_file = "feedkit-state.json"

# The watchdog allowance, in minutes. `feedkit status` exits non-zero when the
# last SUCCESSFUL run is older than this, which is how you get alerted about
# the run that never happened rather than only about the run that failed.
max_age_minutes = 1440

# Packaging metadata for the toolkit (Day 83's mechanism, used for real).
#
# The two entries under [project.scripts] are what turn this package into
# commands on PATH. After `pip install -e .` you type `feedkit`, not
# `python /some/long/path/cli.py` — and the scheduler's job line becomes short
# enough to read, which matters more than it sounds when you are looking at a
# crontab at midnight.

[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"

[project]
name = "feedkit"
version = "1.0.0"
description = "A personal automation toolkit: collect entries from JSON sources on a schedule, without processing anything twice."
requires-python = ">=3.11"
license = { text = "MIT" }
keywords = ["automation", "cli", "scheduling", "idempotence"]
dependencies = [
  "requests>=2.31",
]

[project.optional-dependencies]
# Development-only tools. A scheduled machine installs the package without
# these; a developer installs `.[dev]`. Keeping them out of `dependencies` is
# what stops a test framework from being deployed to a server.
dev = ["pytest>=8"]

[project.scripts]
feedkit = "feedkit.cli:main"
feedkit-scheduled = "feedkit.cli:scheduled_main"

[tool.setuptools.packages.find]
where = ["src"]

[tool.pytest.ini_options]
addopts = "-q"

"""feedkit — a small, installable, scheduled automation toolkit.

The package is deliberately layered so that the boundaries sit at the edges:

    core            pure data in, pure data out — no network, no clock, no disk
    config          the four-layer precedence, resolved by a pure function
    logging_setup   structured JSON logging with secret redaction
    state           the state file, written atomically, plus the run lock
    adapters        the network and the clock — the only impure module
    runner          the order of one unattended run
    cli             argparse subcommands and the two console entry points

Read `core.py` first. It is where the interesting decisions live, and it is
readable without knowing anything about HTTP.
"""

__all__ = ["__version__"]

__version__ = "1.0.0"

"""The edges: the network, the clock, and sleeping.

Everything in this module talks to something outside the process. That is the
whole reason it is a separate module — the core can be tested with plain data
because none of this leaks into it, and this module can be swapped for a fake
in a test because the runner receives it as an argument rather than importing
it. Day 74's rule, applied to the two boundaries that hurt most.

Note the constructor of `HttpFetcher`: it takes a session, a timeout, a retry
count, a backoff base AND a `sleeper`. Injecting the sleeper is what lets the
test suite exercise three retries in microseconds instead of seconds, without
anybody having to patch `time.sleep` globally and hope.
"""

from __future__ import annotations

import time
from datetime import datetime, timezone
from typing import Any, Callable, Protocol

import requests


class FetchError(RuntimeError):
    """A source could not be fetched, after every retry was spent."""


class Clock(Protocol):
    """The clock, as an interface, so a test can hand over a fixed time."""

    def now_iso(self) -> str: ...


class SystemClock:
    """The real clock. UTC, always — a job that runs at 02:30 local time runs
    twice or not at all on the two days a year the offset changes."""

    def now_iso(self) -> str:
        return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


class FixedClock:
    """A clock that never moves. Shipped rather than hidden in a test file,
    because it belongs to the design, not to the tests."""

    def __init__(self, value: str) -> None:
        self.value = value

    def now_iso(self) -> str:
        return self.value


#: Statuses worth trying again: the server said "not now", not "never".
RETRYABLE_STATUS = frozenset({408, 425, 429, 500, 502, 503, 504})


class HttpFetcher:
    """Fetch one source's JSON, with a timeout and bounded retries.

    Three things here are not optional in an unattended job:

    * a **timeout** on every request — the default in `requests` is no timeout
      at all, and a job with no timeout does not fail, it hangs, which is the
      one outcome no supervisor can see;
    * **bounded** retries with exponential backoff — unbounded retries turn a
      failing dependency into a self-inflicted outage;
    * a retry decision based on WHAT went wrong. A 503 is worth another go; a
      404 or a 401 will be a 404 or a 401 forever, and retrying it is just
      noise you will pay for in someone else's server logs.
    """

    def __init__(
        self,
        session: requests.Session,
        base_url: str,
        token: str = "",
        timeout: float = 5.0,
        retries: int = 3,
        backoff_seconds: float = 0.5,
        sleeper: Callable[[float], None] = time.sleep,
        logger: Any = None,
    ) -> None:
        self.session = session
        self.base_url = base_url.rstrip("/")
        self.token = token
        self.timeout = timeout
        self.retries = max(1, retries)
        self.backoff_seconds = backoff_seconds
        self.sleeper = sleeper
        self.logger = logger

    def _headers(self) -> dict[str, str]:
        headers = {
            "Accept": "application/json",
            # Identify the client honestly — Day 79's rule, and the thing that
            # lets an operator find you when your job misbehaves.
            "User-Agent": "feedkit/1.0 (personal automation toolkit)",
        }
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"
        return headers

    def fetch(self, source: str) -> tuple[Any, int]:
        """Return the decoded payload and the attempt number that succeeded."""
        # Exercise 6 — retry the failures worth retrying, and only those.
        #
        # Loop `self.retries` times. On each attempt:
        #   * GET f"{self.base_url}/feed/{source}.json" with self._headers()
        #     and timeout=self.timeout — never omit the timeout;
        #   * a requests.RequestException is a transport failure: record it and
        #     try again;
        #   * status 200: return (response.json(), attempt). A body that is not
        #     JSON will not become JSON on a retry, so raise FetchError instead
        #     of looping;
        #   * a status in RETRYABLE_STATUS means "not now": try again;
        #   * any other status means "never" — a 404 will be a 404 forever, and
        #     retrying it is noise you pay for in someone else's server logs.
        #     Raise FetchError immediately;
        #   * between attempts, self.sleeper(self.backoff_seconds * 2 ** (attempt - 1)).
        #     Doubling is what stops a struggling server being hammered by the
        #     very client that noticed it was struggling.
        # After the last attempt, raise FetchError naming the source, the last
        # error, and how many attempts were spent.
        #
        # Check it with:  pytest tests/test_toolkit.py -k retry
        raise NotImplementedError("Exercise 6: implement HttpFetcher.fetch in adapters.py")

        url = f"{self.base_url}/feed/{source}.json"
        last_error = "no attempt was made"

        for attempt in range(1, self.retries + 1):
            try:
                response = self.session.get(url, headers=self._headers(), timeout=self.timeout)
            except requests.RequestException as exc:
                last_error = f"{type(exc).__name__}: {exc}"
                if self.logger:
                    self.logger.warning(
                        "fetch attempt failed",
                        extra={"source": source, "attempt": attempt, "status": "transport"},
                    )
            else:
                if response.status_code == 200:
                    try:
                        return response.json(), attempt
                    except ValueError as exc:
                        # A body that is not JSON will not become JSON on a
                        # retry. Fail now.
                        raise FetchError(f"response was not JSON: {exc}") from exc
                last_error = f"HTTP {response.status_code}"
                if response.status_code not in RETRYABLE_STATUS:
                    raise FetchError(f"{last_error} (not retryable)")
                if self.logger:
                    self.logger.warning(
                        "fetch attempt failed",
                        extra={
                            "source": source,
                            "attempt": attempt,
                            "status": response.status_code,
                        },
                    )

            if attempt < self.retries:
                self.sleeper(self.backoff_seconds * (2 ** (attempt - 1)))

        raise FetchError(f"{last_error} after {self.retries} attempts")


def build_session() -> requests.Session:
    """One Session for the whole run, so the connection is reused across
    sources instead of being renegotiated for each one."""
    return requests.Session()

"""The command line: three subcommands over one shared core, plus the
scheduled entry point.

`feedkit fetch` does the work. `feedkit report` renders what has been
collected. `feedkit status` says when the last successful run was and whether
the toolkit has gone quiet. `feedkit-scheduled` is what the crontab, launchd
job or systemd timer invokes — the same fetch, with the settings a machine
wants rather than the settings a human wants.

Both entry points are declared in `pyproject.toml` under
`[project.scripts]`, which is what turns them into commands on PATH when the
package is installed. That is Day 83's mechanism doing the work Day 80's
argparse designed.
"""

from __future__ import annotations

import argparse
import sys
from pathlib import Path
from typing import Any, Sequence, TextIO

from . import adapters, config as config_module, core, logging_setup, runner
from . import state as state_module

VERSION = "1.0.0"


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="feedkit",
        description=(
            "Collect entries from configured JSON sources, on a schedule, "
            "without processing anything twice."
        ),
        epilog=(
            "Settings resolve in this order, weakest first: defaults, "
            "configuration file, environment, flags. Run "
            "`feedkit status --explain-config` to see where each value came from. "
            "The access token is read from FEEDKIT_TOKEN and is never written to "
            "a file or a log."
        ),
    )
    # Global options are the CONFIGURATION SETTINGS — the fourth and strongest
    # layer of the precedence. Keeping them all on the main parser rather than
    # scattering them across subcommands means `--max-items` means the same
    # thing everywhere, and `status --explain-config` can show the effect of
    # any of them.
    parser.add_argument("--version", action="version", version=f"feedkit {VERSION}")
    parser.add_argument("--config", metavar="PATH", help="configuration file to read")
    parser.add_argument("--base-url", dest="base_url", help="root address of the source server")
    parser.add_argument("--log-level", dest="log_level", choices=sorted(logging_setup.LEVELS))
    parser.add_argument("--state-file", dest="state_file", metavar="PATH")
    parser.add_argument("--sources", help="comma-separated list, overriding the configuration")
    parser.add_argument(
        "--max-items",
        dest="max_items",
        type=int,
        metavar="N",
        help="most previously unseen entries to accept from one source in one run",
    )
    parser.add_argument("--retries", type=int, metavar="N", help="attempts per source")

    subcommands = parser.add_subparsers(dest="command", metavar="COMMAND")

    fetch = subcommands.add_parser("fetch", help="collect new entries from every source")
    fetch.add_argument(
        "--dry-run",
        action="store_true",
        help="do everything except write the state file, and say what would have changed",
    )

    report = subcommands.add_parser("report", help="show what has been collected")
    report.add_argument("--limit", dest="report_limit", type=int, metavar="N")

    status = subcommands.add_parser("status", help="show the last successful run")
    status.add_argument(
        "--max-age-minutes",
        dest="max_age_minutes",
        type=int,
        metavar="N",
        help="watchdog allowance; exit 3 when the last success is older than this",
    )
    status.add_argument(
        "--explain-config",
        action="store_true",
        help="print every setting, its value, and which layer it came from",
    )
    return parser


CONFIG_FLAGS = (
    "config",
    "base_url",
    "log_level",
    "state_file",
    "sources",
    "max_items",
    "retries",
    "report_limit",
    "max_age_minutes",
)


def flags_from(args: argparse.Namespace) -> dict[str, Any]:
    """Only the parsed arguments that are configuration settings, and only the
    ones actually supplied — argparse leaves the rest as None, and None is what
    tells the resolver 'this layer has no opinion'."""
    return {name: getattr(args, name, None) for name in CONFIG_FLAGS}


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

    if not args.command:
        parser.print_help(out)
        return core.EXIT_FATAL

    try:
        settings, config_path = config_module.load(flags_from(args))
    except config_module.ConfigError as exc:
        print(f"feedkit: configuration error: {exc}", file=sys.stderr)
        return core.EXIT_FATAL

    run_id = runner.new_run_id()
    logger = logging_setup.configure(
        settings.log_level,
        run_id=run_id,
        secrets=[settings.token] if settings.token else [],
        stream=out,
    )
    state_path = Path(settings.state_file).expanduser()
    lock_path = state_path.with_suffix(state_path.suffix + ".lock")

    try:
        if args.command == "fetch":
            return _cmd_fetch(args, settings, state_path, lock_path, logger, out, run_id)
        if args.command == "report":
            return _cmd_report(settings, state_path, out)
        if args.command == "status":
            return _cmd_status(args, settings, config_path, state_path, out)
    except state_module.StateError as exc:
        print(f"feedkit: {exc}", file=sys.stderr)
        return core.EXIT_FATAL

    parser.print_help(out)
    return core.EXIT_FATAL


def _cmd_fetch(
    args: argparse.Namespace,
    settings: config_module.Config,
    state_path: Path,
    lock_path: Path,
    logger: Any,
    out: TextIO,
    run_id: str,
) -> int:
    session = adapters.build_session()
    try:
        fetcher = adapters.HttpFetcher(
            session=session,
            base_url=settings.base_url,
            token=settings.token,
            timeout=settings.timeout_seconds,
            retries=settings.retries,
            backoff_seconds=settings.backoff_seconds,
            logger=logger,
        )
        summary, code = runner.run_fetch(
            settings,
            fetcher,
            adapters.SystemClock(),
            state_path,
            lock_path,
            logger,
            dry_run=args.dry_run,
            run_id=run_id,
        )
    finally:
        session.close()

    print(core.format_summary(summary, run_id=run_id, dry_run=args.dry_run), file=out)
    return code


def _cmd_report(settings: config_module.Config, state_path: Path, out: TextIO) -> int:
    current = state_module.load(state_path)
    print(core.render_report(current, settings.report_limit), file=out)
    return core.EXIT_OK


def _cmd_status(
    args: argparse.Namespace,
    settings: config_module.Config,
    config_path: Path | None,
    state_path: Path,
    out: TextIO,
) -> int:
    if args.explain_config:
        print(config_module.explain(settings, config_path), file=out)
        return core.EXIT_OK

    current = state_module.load(state_path)
    text, stale = core.render_status(
        current, adapters.SystemClock().now_iso(), settings.max_age_seconds
    )
    print(text, file=out)
    # A watchdog is only useful if it can fail. Exiting non-zero on silence is
    # what lets a second, much simpler scheduled job page you when the first
    # one has stopped running at all.
    return core.EXIT_PARTIAL if stale else core.EXIT_OK


def scheduled_main(argv: Sequence[str] | None = None) -> int:
    """The entry point a scheduler invokes.

    It is a thin wrapper on purpose. The scheduled run wants a fetch, quieter
    logging by default, and no interactive help — everything else is identical,
    because a scheduled run that behaves differently from the one you tested by
    hand is a scheduled run you have not tested.
    """
    argv = list(sys.argv[1:] if argv is None else argv)
    return main(["--log-level", "info", "fetch", *argv])


if __name__ == "__main__":  # pragma: no cover
    raise SystemExit(main())

"""Configuration, resolved by a precedence that is written down and testable.

The order, weakest first:

    1. defaults baked into the code   — so the tool runs with no setup at all
    2. a configuration file           — the machine's long-lived preferences
    3. environment variables          — deployment-specific values and secrets
    4. command-line flags             — this one run, right now

Every automation has this order. Most of them have it by accident, spread over
a dozen `or` expressions, and nobody can say what wins. Here it is one pure
function over four dictionaries, so a test can assert all four levels — which
is exactly what `tests/run_tests.sh` does.

`resolve` also records WHERE each value came from. `feedkit status
--explain-config` prints that table, which turns "why is it doing that?" into a
five-second question instead of an afternoon.
"""

from __future__ import annotations

import os
import tomllib
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Mapping

#: Layer 1. Everything the toolkit needs to run at all, with no file, no
#: environment and no flags. A tool that cannot start without configuration is
#: a tool nobody tries.
DEFAULTS: dict[str, Any] = {
    "sources": ["notes", "links"],
    "max_items": 5,
    "report_limit": 10,
    "timeout_seconds": 5.0,
    "retries": 3,
    "backoff_seconds": 0.5,
    "log_level": "info",
    "state_file": "feedkit-state.json",
    "max_age_minutes": 1440,
}

#: Layer 3. Environment variable name -> (config key, type).
ENV_KEYS: dict[str, tuple[str, str]] = {
    "FEEDKIT_BASE_URL": ("base_url", "str"),
    "FEEDKIT_MAX_ITEMS": ("max_items", "int"),
    "FEEDKIT_REPORT_LIMIT": ("report_limit", "int"),
    "FEEDKIT_TIMEOUT_SECONDS": ("timeout_seconds", "float"),
    "FEEDKIT_RETRIES": ("retries", "int"),
    "FEEDKIT_BACKOFF_SECONDS": ("backoff_seconds", "float"),
    "FEEDKIT_LOG_LEVEL": ("log_level", "str"),
    "FEEDKIT_STATE_FILE": ("state_file", "str"),
    "FEEDKIT_MAX_AGE_MINUTES": ("max_age_minutes", "int"),
    "FEEDKIT_SOURCES": ("sources", "list"),
}

#: The one value that must NEVER come from a file in the repository.
SECRET_ENV = "FEEDKIT_TOKEN"


class ConfigError(ValueError):
    """The configuration is unusable. Stop the run; do not guess."""


@dataclass(frozen=True)
class Config:
    """The resolved settings for one run."""

    base_url: str
    sources: tuple[str, ...]
    max_items: int
    report_limit: int
    timeout_seconds: float
    retries: int
    backoff_seconds: float
    log_level: str
    state_file: str
    max_age_minutes: int
    token: str = ""
    provenance: Mapping[str, str] = field(default_factory=dict)

    @property
    def max_age_seconds(self) -> int:
        return self.max_age_minutes * 60


def coerce(value: Any, kind: str, key: str) -> Any:
    """Turn a string from a file, an environment variable or a flag into the
    type the rest of the program expects, and fail loudly when it cannot."""
    if kind == "list":
        if isinstance(value, (list, tuple)):
            return [str(item) for item in value]
        return [part.strip() for part in str(value).split(",") if part.strip()]
    try:
        if kind == "int":
            return int(value)
        if kind == "float":
            return float(value)
    except (TypeError, ValueError) as exc:
        raise ConfigError(f"{key}: {value!r} is not a valid {kind}") from exc
    return str(value)


def kind_of(key: str) -> str:
    """The declared type of a configuration key, from the defaults table."""
    for _, (config_key, kind) in ENV_KEYS.items():
        if config_key == key:
            return kind
    default = DEFAULTS.get(key)
    if isinstance(default, bool):
        return "str"
    if isinstance(default, int):
        return "int"
    if isinstance(default, float):
        return "float"
    if isinstance(default, list):
        return "list"
    return "str"


def read_config_file(path: Path) -> dict[str, Any]:
    """Read a TOML configuration file. Missing file is not an error; an
    unparseable one is."""
    if not path.is_file():
        return {}
    try:
        with path.open("rb") as handle:
            data = tomllib.load(handle)
    except tomllib.TOMLDecodeError as exc:
        raise ConfigError(f"{path}: not valid TOML: {exc}") from exc
    settings = data.get("feedkit", data)
    if not isinstance(settings, dict):
        raise ConfigError(f"{path}: expected a table of settings")
    return dict(settings)


def find_config_file(explicit: str | None, environ: Mapping[str, str], cwd: Path) -> Path | None:
    """Where the configuration file lives, in the order a user expects.

    A flag beats the environment, which beats the current directory, which
    beats the user's XDG configuration directory. Returning None means "no file
    anywhere", which is a perfectly normal state, not an error.
    """
    if explicit:
        return Path(explicit).expanduser()
    from_env = environ.get("FEEDKIT_CONFIG")
    if from_env:
        return Path(from_env).expanduser()
    local = cwd / "feedkit.toml"
    if local.is_file():
        return local
    base = environ.get("XDG_CONFIG_HOME")
    home = Path(base).expanduser() if base else Path(environ.get("HOME", "~")).expanduser() / ".config"
    candidate = home / "feedkit" / "feedkit.toml"
    return candidate if candidate.is_file() else None


def env_values(environ: Mapping[str, str]) -> dict[str, Any]:
    """Layer 3, extracted from an environment mapping passed in as an argument
    (never read from os.environ in here — that is a boundary, and tests need to
    supply their own)."""
    values: dict[str, Any] = {}
    for name, (key, kind) in ENV_KEYS.items():
        if name in environ and environ[name] != "":
            values[key] = coerce(environ[name], kind, key)
    return values


def resolve(
    file_values: Mapping[str, Any],
    environment: Mapping[str, Any],
    flags: Mapping[str, Any],
    token: str = "",
) -> Config:
    """Apply the four layers in order and record where each value came from.

    Pure. Four dictionaries in, one Config out. This is the function the lab's
    precedence test drives directly, and it is why "which layer wins?" is a
    question with a checked answer rather than a folk belief.
    """
    merged: dict[str, Any] = {}
    provenance: dict[str, str] = {}

    # Exercise 3 — the four-layer precedence, written down once.
    #
    # Walk the four layers in order, WEAKEST FIRST:
    #     ("default", DEFAULTS), ("file", file_values),
    #     ("environment", environment), and finally the flags.
    # For each key in each layer, put the value into `merged` and record the
    # layer's name in `provenance[key]`. Later layers overwrite earlier ones,
    # which is the whole mechanism.
    #
    # Three details decide whether this is correct or merely plausible:
    #   a) a flag whose value is None means "no opinion" — argparse leaves
    #      every unsupplied option as None, so filter those out BEFORE the
    #      loop, or every default will silently beat the config file;
    #   b) values from a file, the environment or a flag are strings or raw
    #      TOML values and must go through coerce(value, kind_of(key), key);
    #      the defaults are already the right types;
    #   c) an unrecognised key in a FILE is a typo — raise ConfigError naming
    #      it. The same key from the environment or the flags is just another
    #      setting the resolver does not own, so skip it quietly.
    #
    # Check it with:  bash tests/run_tests.sh   (section 5 asserts all four)
    raise NotImplementedError("Exercise 3: implement the precedence loop in config.py")

    base_url = str(merged.get("base_url", "")).rstrip("/")
    if not base_url:
        raise ConfigError(
            "no base URL configured. Set FEEDKIT_BASE_URL in the environment, "
            "or pass --base-url. Deployment-specific addresses do not belong in "
            "a file that is committed."
        )
    if merged["retries"] < 1:
        raise ConfigError("retries must be at least 1")
    if merged["max_items"] < 0:
        raise ConfigError("max-items must not be negative")

    provenance.setdefault("base_url", "environment")
    provenance["token"] = "environment" if token else "unset"

    return Config(
        base_url=base_url,
        sources=tuple(merged["sources"]),
        max_items=int(merged["max_items"]),
        report_limit=int(merged["report_limit"]),
        timeout_seconds=float(merged["timeout_seconds"]),
        retries=int(merged["retries"]),
        backoff_seconds=float(merged["backoff_seconds"]),
        log_level=str(merged["log_level"]),
        state_file=str(merged["state_file"]),
        max_age_minutes=int(merged["max_age_minutes"]),
        token=token,
        provenance=provenance,
    )


def load(
    flags: Mapping[str, Any],
    environ: Mapping[str, str] | None = None,
    cwd: Path | None = None,
) -> tuple[Config, Path | None]:
    """The impure wrapper: find the file, read the environment, then call the
    pure `resolve`. All the I/O is in these six lines."""
    environ = os.environ if environ is None else environ
    cwd = Path.cwd() if cwd is None else cwd
    config_path = find_config_file(flags.get("config"), environ, cwd)
    file_values = read_config_file(config_path) if config_path else {}
    token = environ.get(SECRET_ENV, "")
    config = resolve(file_values, env_values(environ), flags, token=token)
    return config, config_path


def explain(config: Config, config_path: Path | None) -> str:
    """The provenance table. Answers 'why is it doing that?' in one command."""
    lines = [f"configuration file: {config_path or 'none found'}", ""]
    lines.append(f"  {'setting':<20} {'value':<28} {'came from'}")
    for key in sorted(config.provenance):
        if key == "token":
            shown = "set (never printed)" if config.token else "unset"
        else:
            shown = str(getattr(config, key, ""))
        lines.append(f"  {key:<20} {shown:<28} {config.provenance[key]}")
    return "\n".join(lines)

"""The pure core of the toolkit.

Nothing in this module touches the network, the clock, the filesystem or a
subprocess. Every function here takes plain data and returns plain data, which
is exactly what makes the interesting parts of an automation testable without
starting a server or waiting a second. Day 74 argued for pushing boundaries to
the edges; this module is what the middle looks like when you do.

If you ever find yourself wanting to `import requests` or call `time.time()` in
here, that is the signal that the value belongs in a parameter instead.
"""

from __future__ import annotations

import calendar
import time
from dataclasses import dataclass
from typing import Any, Iterable, Mapping, Sequence

# `calendar` and `time` appear here only to PARSE a timestamp that was handed
# in as a string. Nothing in this module ever asks what time it is now — that
# question belongs to the clock adapter, so that every test can answer it.

STATE_VERSION = 1

#: Exit codes. These are the machine-readable half of every run, and the
#: scheduler is the thing that reads them.
EXIT_OK = 0
EXIT_FATAL = 1
EXIT_PARTIAL = 3
EXIT_LOCKED = 75


@dataclass(frozen=True)
class Entry:
    """One item collected from one source."""

    id: str
    title: str
    published: str
    source: str = ""

    def as_dict(self) -> dict[str, str]:
        return {
            "id": self.id,
            "title": self.title,
            "published": self.published,
            "source": self.source,
        }


@dataclass(frozen=True)
class SourceResult:
    """What happened to one source during one run."""

    source: str
    status: str  # "ok" | "failed"
    new_entries: tuple[Entry, ...] = ()
    error: str = ""
    attempts: int = 1


class InvalidPayload(ValueError):
    """The server answered, but not with something this toolkit understands."""


def parse_entries(payload: Any, source: str) -> tuple[Entry, ...]:
    """Validate a decoded JSON payload and turn it into Entry objects.

    A 200 response is not the same thing as a correct response. Validating at
    the edge means every function downstream can assume the shape it was given,
    which is the whole reason this is a separate step rather than a dict lookup
    buried three calls deep.
    """
    if not isinstance(payload, Mapping):
        raise InvalidPayload(f"top level is {type(payload).__name__}, expected an object")
    items = payload.get("entries")
    if not isinstance(items, Sequence) or isinstance(items, (str, bytes)):
        raise InvalidPayload("'entries' is missing or is not a list")

    parsed: list[Entry] = []
    for index, item in enumerate(items):
        if not isinstance(item, Mapping):
            raise InvalidPayload(f"entry {index} is not an object")
        missing = [field for field in ("id", "title", "published") if field not in item]
        if missing:
            raise InvalidPayload(f"entry {index} is missing {', '.join(missing)}")
        parsed.append(
            Entry(
                id=str(item["id"]),
                title=str(item["title"]),
                published=str(item["published"]),
                source=source,
            )
        )
    return tuple(parsed)


def select_new(
    entries: Iterable[Entry], seen_ids: Iterable[str], max_items: int
) -> tuple[Entry, ...]:
    """Return the entries not already recorded, newest first, capped at max_items.

    This is the idempotence rule, and it is four lines of pure logic precisely
    because the record of what has been seen arrives as an argument rather than
    being read from a file in here.
    """
    # Exercise 1 — the idempotence rule, and the reason a second run is cheap.
    #
    # Return only the entries whose `id` is NOT in `seen_ids`, sorted newest
    # first by (published, id), and no more than `max_items` of them. Treat a
    # negative `max_items` as "no cap".
    #
    # Do it in four lines and resist the urge to open a file in here: the
    # record of what has been seen arrives as an argument precisely so this
    # function stays pure and its test needs nothing but two lists.
    #
    # Check it with:  pytest tests/test_toolkit.py -k select_new
    raise NotImplementedError("Exercise 1: implement select_new in core.py")


def empty_state() -> dict[str, Any]:
    """The state of a toolkit that has never run."""
    return {
        "version": STATE_VERSION,
        "last_run": None,
        "last_success": None,
        "sources": {},
        "entries": [],
    }


def merge_state(
    state: Mapping[str, Any],
    results: Sequence[SourceResult],
    run_id: str,
    started_at: str,
    finished_at: str,
    keep_entries: int = 200,
) -> dict[str, Any]:
    """Fold one run's results into the previous state and return the new state.

    Pure: give it the same inputs and it returns the same dictionary, every
    time, on any machine. The caller decides whether to write it.
    """
    summary = summarise(results)
    sources = {name: dict(value) for name, value in dict(state.get("sources") or {}).items()}
    collected: list[dict[str, str]] = list(state.get("entries") or [])

    for result in results:
        record = sources.setdefault(
            result.source, {"seen_ids": [], "last_success": None, "last_error": ""}
        )
        if result.status == "ok":
            seen = list(record.get("seen_ids") or [])
            seen.extend(entry.id for entry in result.new_entries if entry.id not in seen)
            record["seen_ids"] = seen
            record["last_success"] = finished_at
            record["last_error"] = ""
            collected = [entry.as_dict() for entry in result.new_entries] + collected
        else:
            record["last_error"] = result.error

    collected.sort(key=lambda item: (item["published"], item["id"]), reverse=True)

    new_state: dict[str, Any] = {
        "version": STATE_VERSION,
        "last_run": {
            "run_id": run_id,
            "started_at": started_at,
            "finished_at": finished_at,
            "status": summary["status"],
            "sources_ok": summary["sources_ok"],
            "sources_failed": summary["sources_failed"],
            "new_entries": summary["new_entries"],
        },
        "last_success": (
            finished_at if summary["status"] == "ok" else state.get("last_success")
        ),
        "sources": sources,
        "entries": collected[:keep_entries],
    }
    return new_state


def summarise(results: Sequence[SourceResult]) -> dict[str, Any]:
    """Reduce a run's results to the handful of numbers a human reads."""
    ok = [result for result in results if result.status == "ok"]
    failed = [result for result in results if result.status != "ok"]
    new_entries = sum(len(result.new_entries) for result in ok)

    if not results:
        status = "ok"
    elif not failed:
        status = "ok"
    elif not ok:
        status = "failed"
    else:
        status = "partial"

    return {
        "status": status,
        "sources_total": len(results),
        "sources_ok": len(ok),
        "sources_failed": len(failed),
        "new_entries": new_entries,
        "failures": {result.source: result.error for result in failed},
        "retried": {result.source: result.attempts for result in results if result.attempts > 1},
    }


def exit_code_for(summary: Mapping[str, Any]) -> int:
    """Map a run summary onto the exit code the scheduler will read.

    Partial success gets its own code. Reporting 0 for "most of it worked" is
    the single most common way an automation lies to the person who owns it.
    """
    # Exercise 2 — make partial success visible to the machine.
    #
    # Map summary["status"] onto an exit code: "ok" -> EXIT_OK,
    # "partial" -> EXIT_PARTIAL, anything else -> EXIT_FATAL.
    #
    # Three lines. The temptation to return 0 for "partial" because most of it
    # worked is exactly the habit this exercise exists to break: the scheduler
    # reads the exit code and nothing else, and an automation that reports
    # success while dropping items is one nobody can ever trust again.
    #
    # Check it with:  pytest tests/test_toolkit.py -k exit_codes
    raise NotImplementedError("Exercise 2: implement exit_code_for in core.py")


def format_summary(summary: Mapping[str, Any], run_id: str, dry_run: bool = False) -> str:
    """The human-readable run summary, printed at the end of every run."""
    lines = [
        f"run {run_id}: {summary['status']}"
        + ("  (dry run — nothing was written)" if dry_run else ""),
        f"  sources: {summary['sources_ok']} ok, {summary['sources_failed']} failed,"
        f" {summary['sources_total']} total",
        f"  new entries: {summary['new_entries']}",
    ]
    for source, attempts in sorted(dict(summary.get("retried") or {}).items()):
        lines.append(f"  retried: {source} succeeded on attempt {attempts}")
    for source, error in sorted(dict(summary.get("failures") or {}).items()):
        lines.append(f"  FAILED: {source}: {error}")
    return "\n".join(lines)


def render_report(state: Mapping[str, Any], limit: int) -> str:
    """Render what has been collected. Reads state, touches nothing."""
    entries = list(state.get("entries") or [])
    if not entries:
        return "No entries collected yet. Run `feedkit fetch` first."

    shown = entries[: limit if limit >= 0 else len(entries)]
    width = max(len(entry["source"]) for entry in shown)
    lines = [f"{len(entries)} entries collected; showing {len(shown)}", ""]
    for entry in shown:
        lines.append(f"  {entry['published']}  {entry['source']:<{width}}  {entry['title']}")
    return "\n".join(lines)


def render_status(state: Mapping[str, Any], now: str, max_age_seconds: int) -> tuple[str, bool]:
    """Render the status block and say whether the toolkit has gone quiet.

    The second half of the tuple is the watchdog answer: True when the last
    successful run is older than the allowance. Alerting on silence catches the
    failure mode that alerting on errors cannot — the run that never happened.
    """
    last_success = state.get("last_success")
    last_run = state.get("last_run")
    stale = is_stale(last_success, now, max_age_seconds)

    lines = [f"last success: {last_success or 'never'}"]
    if last_run:
        lines.append(
            f"last run:     {last_run['run_id']} at {last_run['finished_at']}"
            f" ({last_run['status']}, {last_run['new_entries']} new)"
        )
    else:
        lines.append("last run:     never")
    lines.append(f"now:          {now}")
    lines.append(
        f"watchdog:     {'STALE' if stale else 'fresh'}"
        f" (allowance {max_age_seconds}s)"
    )
    for name, record in sorted(dict(state.get("sources") or {}).items()):
        note = record.get("last_error") or "ok"
        lines.append(
            f"  {name}: {len(record.get('seen_ids') or [])} seen,"
            f" last success {record.get('last_success') or 'never'} — {note}"
        )
    return "\n".join(lines), stale


def is_stale(last_success: str | None, now: str, max_age_seconds: int) -> bool:
    """True when the last success is missing or older than the allowance.

    Timestamps are ISO 8601 strings, compared by parsing them into seconds. The
    parsing lives in the caller's clock adapter; here we accept the already
    normalised comparison to keep this module free of the datetime module's
    timezone surprises. Both arguments must be UTC ISO strings.
    """
    if not last_success:
        return True
    return _iso_seconds(now) - _iso_seconds(last_success) > max_age_seconds


def _iso_seconds(value: str) -> int:
    """Seconds since the epoch for a UTC ISO 8601 timestamp such as
    2026-07-19T10:11:12Z. Deliberately small and deliberately strict."""
    cleaned = value.replace("Z", "").split(".")[0]
    parsed = time.strptime(cleaned, "%Y-%m-%dT%H:%M:%S")
    return calendar.timegm(parsed)

"""Structured logging — the thing that makes an unattended run debuggable.

Two decisions are baked in here, and both are worth arguing rather than
copying.

**One JSON object per line, on stdout.** A line of prose is readable by you at
your desk; a line of JSON is readable by you AND by `grep`, `jq`, a log
shipper, and whatever the supervisor writes it into. Writing to stdout rather
than opening a log file means the program does not have to know about log
rotation, permissions, or where the operator wants their logs — cron mails it,
systemd hands it to the journal, launchd redirects it, and a human running the
command by hand simply sees it. Fewer decisions inside the program is the
point.

**Every record carries the run id and, where it applies, the item.** An
unattended failure is a message you read hours later with no memory of the
context. "Timeout" tells you nothing. "run 8f2c1a: source=papers attempt=3
timeout after 5.0s" tells you which run, which item, how hard it tried, and
what the limit was.
"""

from __future__ import annotations

import json
import logging
import sys
from typing import Any, Iterable, TextIO

LEVELS = {
    "debug": logging.DEBUG,
    "info": logging.INFO,
    "warning": logging.WARNING,
    "error": logging.ERROR,
    "critical": logging.CRITICAL,
}

#: Extra keys that the formatter promotes to top-level JSON fields.
CONTEXT_KEYS = ("source", "attempt", "status", "count", "path", "elapsed_ms", "url")


class RedactingFilter(logging.Filter):
    """Replace known secret values with a placeholder, everywhere.

    This is a seatbelt, not a licence. The right habit is to never put a token
    into a log call in the first place; this filter exists because one day
    somebody will log a whole request object, or an exception message that
    happens to quote a URL with a token in the query string, and the difference
    between a bad afternoon and a credential rotation is whether that string
    reached the log.
    """

    PLACEHOLDER = "***REDACTED***"

    def __init__(self, secrets: Iterable[str] = ()) -> None:
        super().__init__()
        # Very short strings would redact half the alphabet; ignore them.
        self.secrets = tuple(secret for secret in secrets if secret and len(secret) >= 6)

    def _scrub(self, value: Any) -> Any:
        # Exercise 4 — the leak guard.
        #
        # Return `value` with every string in self.secrets replaced by
        # self.PLACEHOLDER. Recurse into lists, tuples and dicts, because a
        # careless log call passes a whole request or response object and the
        # token is three levels down inside it. Anything else, return
        # unchanged.
        #
        # This is a seatbelt, not a licence: the right habit is never to log a
        # credential in the first place. Write it anyway, because one day
        # somebody will.
        #
        # Check it with:  pytest tests/test_toolkit.py -k redacting
        raise NotImplementedError("Exercise 4: implement _scrub in logging_setup.py")

    def filter(self, record: logging.LogRecord) -> bool:
        if not self.secrets:
            return True
        record.msg = self._scrub(record.msg)
        if record.args:
            record.args = self._scrub(record.args)
        for key in CONTEXT_KEYS:
            if hasattr(record, key):
                setattr(record, key, self._scrub(getattr(record, key)))
        if record.exc_info:
            # An exception's own text is the most common accidental leak.
            exc = record.exc_info[1]
            if exc is not None and exc.args:
                exc.args = tuple(self._scrub(arg) for arg in exc.args)
        return True


class JsonFormatter(logging.Formatter):
    """One JSON object per line, with a stable field order."""

    def __init__(self, run_id: str) -> None:
        super().__init__()
        self.run_id = run_id

    def format(self, record: logging.LogRecord) -> str:
        payload: dict[str, Any] = {
            "ts": self.formatTime(record, "%Y-%m-%dT%H:%M:%S"),
            "level": record.levelname.lower(),
            "run_id": self.run_id,
            "event": record.getMessage(),
        }
        for key in CONTEXT_KEYS:
            if hasattr(record, key):
                payload[key] = getattr(record, key)
        if record.exc_info:
            payload["error"] = str(record.exc_info[1])
        # ensure_ascii=False keeps the log readable to a human. Escaping every
        # non-ASCII character is the default and is nearly always wrong for a
        # file somebody has to read at three in the morning.
        return json.dumps(payload, sort_keys=False, ensure_ascii=False)


def configure(
    level: str,
    run_id: str,
    secrets: Iterable[str] = (),
    stream: TextIO | None = None,
) -> logging.Logger:
    """Build the toolkit's logger. Called once, at the start of a run."""
    logger = logging.getLogger("feedkit")
    logger.handlers.clear()
    logger.propagate = False
    logger.setLevel(LEVELS.get(level.lower(), logging.INFO))

    handler = logging.StreamHandler(stream if stream is not None else sys.stdout)
    handler.setFormatter(JsonFormatter(run_id))
    handler.addFilter(RedactingFilter(secrets))
    logger.addHandler(handler)
    return logger

"""One unattended run, start to finish.

This is the only module that knows the ORDER of things: acquire the lock, load
the state, fetch each source with retries, skip and report the ones that fail,
fold the successes into the state, write the state atomically, print the
summary, return an exit code. Everything it does with the outside world arrives
as an argument — the fetcher, the clock, the logger, the paths — which is why
the whole thing can be exercised against a fake fetcher in a millisecond.

The failure policy is the part worth reading twice, because it is the part that
distinguishes an automation from a script:

* a **transport error or a 5xx** on one source is retried with backoff and, if
  it never succeeds, is SKIPPED and REPORTED — one broken source must not stop
  the other four;
* a **payload that does not parse** is not retried, because it will not parse
  next time either; it is skipped and reported the same way;
* a **failure to acquire the lock** stops everything immediately, because the
  correct response to "a run is already happening" is to do nothing;
* an **unusable state file or an invalid configuration** stops everything,
  because continuing would mean guessing about the thing that records what has
  already been done.

Partial success is the normal case for a batch job, and it gets its own exit
code. Reporting 0 because "most of it worked" is how an automation becomes a
thing nobody can trust.
"""

from __future__ import annotations

import uuid
from pathlib import Path
from typing import Any, Protocol, Sequence

from . import core, state as state_module
from .adapters import FetchError
from .config import Config


class Fetcher(Protocol):
    def fetch(self, source: str) -> tuple[Any, int]: ...


def new_run_id() -> str:
    """A short, unique label for one run. Every log line carries it, so the
    lines belonging to a single 03:00 run can be pulled out of a month of
    output with one grep."""
    return uuid.uuid4().hex[:8]


def fetch_sources(
    sources: Sequence[str],
    fetcher: Fetcher,
    seen: dict[str, list[str]],
    max_items: int,
    logger: Any,
) -> list[core.SourceResult]:
    """Fetch every source, collecting successes and failures side by side."""
    # Exercise 7 — skip and report, rather than stop.
    #
    # Build a list of core.SourceResult, one per source. For each source:
    #   * log that it started, with extra={"source": source} so the line names
    #     WHICH item this is — an unattended log without that is unreadable;
    #   * call fetcher.fetch(source), then core.parse_entries(payload, source);
    #   * catch FetchError and core.InvalidPayload TOGETHER: log at ERROR with
    #     the source name, append a SourceResult with status="failed" and
    #     str(exc) as the error, and CONTINUE to the next source. One broken
    #     source must not cost you the other four;
    #   * on success, call core.select_new(entries, seen.get(source, []),
    #     max_items), log the count and the attempt number, and append a
    #     SourceResult with status="ok".
    #
    # Do not catch bare Exception here. A KeyboardInterrupt or a programming
    # error is not a source failure, and swallowing it turns a bug into a
    # mysteriously empty run.
    #
    # Check it with:  pytest tests/test_toolkit.py -k skipped_and_reported
    raise NotImplementedError("Exercise 7: implement fetch_sources in runner.py")


def run_fetch(
    config: Config,
    fetcher: Fetcher,
    clock: Any,
    state_path: Path,
    lock_path: Path,
    logger: Any,
    dry_run: bool = False,
    run_id: str | None = None,
) -> tuple[dict[str, Any], int]:
    """Do one fetch run. Returns the summary and the exit code.

    The run id is passed IN rather than generated here, so that the id stamped
    on every log line and the id recorded in the state file are the same
    string. Two ids for one run is a small bug that makes an incident twice as
    slow to investigate, and it is easy to ship without noticing.
    """
    run_id = run_id or new_run_id()
    started_at = clock.now_iso()

    try:
        with state_module.Lock(lock_path):
            logger.info("run started", extra={"status": "started", "path": str(state_path)})
            current = state_module.load(state_path)
            seen = {
                name: list(record.get("seen_ids") or [])
                for name, record in dict(current.get("sources") or {}).items()
            }

            results = fetch_sources(config.sources, fetcher, seen, config.max_items, logger)
            finished_at = clock.now_iso()
            summary = core.summarise(results)
            merged = core.merge_state(current, results, run_id, started_at, finished_at)

            if dry_run:
                logger.info(
                    "dry run — state not written",
                    extra={"status": "dry-run", "count": summary["new_entries"]},
                )
            else:
                state_module.write_atomic(state_path, merged)
                logger.info(
                    "state written",
                    extra={"status": summary["status"], "path": str(state_path)},
                )
    except state_module.LockHeld as exc:
        logger.error("another run is in progress", extra={"status": "locked"})
        return (
            {
                "run_id": run_id,
                "status": "locked",
                "sources_total": 0,
                "sources_ok": 0,
                "sources_failed": 0,
                "new_entries": 0,
                "failures": {"lock": str(exc)},
                "retried": {},
            },
            core.EXIT_LOCKED,
        )

    summary["run_id"] = run_id
    exit_code = core.exit_code_for(summary)
    logger.info(
        "run finished",
        extra={"status": summary["status"], "count": summary["new_entries"]},
    )
    return summary, exit_code

"""The state file, and the atomic write that keeps it trustworthy.

State is what makes a job idempotent: it is the record of what has already been
processed, so a second run does not do the work twice. That makes it the single
most valuable file the toolkit owns, and losing it is worse than a failed run —
a failed run is visible, a corrupted state file quietly re-processes or
silently skips.

So the write is atomic, exactly as Days 64 and 65 described. Write the whole
new document to a temporary file in the SAME directory, flush it, ask the
operating system to put it on the disk, then `os.replace` it over the old name.
`os.replace` is atomic on POSIX and on Windows: any reader sees either the
complete old file or the complete new one, never a half-written mixture. If the
machine loses power between the write and the replace, the previous state is
still there and the temporary file is garbage that the next run cleans up.

The naive version — `open(path, "w")` then `json.dump` — truncates the real
file first. Interrupt it and the record of everything you have ever processed
is a zero-byte file.
"""

from __future__ import annotations

import json
import os
import tempfile
from pathlib import Path
from typing import Any, Callable, Mapping

from .core import empty_state


class StateError(RuntimeError):
    """The state file exists but cannot be used. Never guess; stop."""


def load(path: Path) -> dict[str, Any]:
    """Read the state file, or return a fresh empty state if there is none."""
    if not path.is_file():
        return empty_state()
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise StateError(
            f"{path} is not valid JSON ({exc}). Refusing to overwrite it. "
            f"Move it aside to start fresh."
        ) from exc
    if not isinstance(data, dict) or "version" not in data:
        raise StateError(f"{path} does not look like a feedkit state file")
    return data


def write_atomic(
    path: Path,
    state: Mapping[str, Any],
    crash_hook: Callable[[], None] | None = None,
) -> None:
    """Write state so that an interruption leaves the previous file intact.

    `crash_hook` exists purely so the lab can prove the property. The test
    passes a function that raises, standing in for the power cut, and then
    asserts the old file is byte-identical. Production code passes nothing.
    """
    # Exercise 5 — the atomic write.
    #
    # Steps, in this exact order:
    #   1. make sure path.parent exists;
    #   2. serialise `state` to JSON (indent=2, sort_keys=True, trailing
    #      newline — a stable byte-for-byte rendering is what makes the
    #      dry-run test possible at all);
    #   3. write it to a NamedTemporaryFile in the SAME directory as `path`
    #      with delete=False. The same directory matters: os.replace is only
    #      atomic within one filesystem, and /tmp is often a different one;
    #   4. flush the handle and os.fsync its file descriptor;
    #   5. if crash_hook is not None, call it — this is where the lab injects
    #      the simulated power cut;
    #   6. os.replace(tmp_path, path);
    #   7. on ANY exception, delete the temporary file and re-raise. Catch
    #      BaseException, not Exception, so a KeyboardInterrupt still cleans up.
    #
    # What you must NOT write is open(path, "w") followed by json.dump. That
    # truncates the real file first, and an interruption there destroys the
    # record of everything you have ever processed.
    #
    # Check it with:  pytest tests/test_toolkit.py -k interrupted
    raise NotImplementedError("Exercise 5: implement write_atomic in state.py")


class LockHeld(RuntimeError):
    """Another run of this toolkit is already in progress."""


class Lock:
    """A lock file, so two scheduled runs never overlap.

    Created with O_CREAT | O_EXCL, which the operating system guarantees will
    succeed for exactly one caller. The file holds the process id, which is
    what lets a human decide whether a lock left behind by a crash is stale.

    This is deliberately the simplest thing that works on one machine. It is
    not a distributed lock and must not be used as one.
    """

    def __init__(self, path: Path) -> None:
        self.path = path
        self._acquired = False

    def __enter__(self) -> "Lock":
        self.path.parent.mkdir(parents=True, exist_ok=True)
        try:
            fd = os.open(self.path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
        except FileExistsError as exc:
            holder = ""
            try:
                holder = self.path.read_text(encoding="utf-8").strip()
            except OSError:
                pass
            raise LockHeld(
                f"{self.path} exists (held by pid {holder or 'unknown'}). "
                f"Another run is in progress, or a previous run was killed. "
                f"Delete the file only after checking that no such process exists."
            ) from exc
        with os.fdopen(fd, "w") as handle:
            handle.write(str(os.getpid()))
        self._acquired = True
        return self

    def __exit__(self, *exc_info: object) -> None:
        if self._acquired:
            self.path.unlink(missing_ok=True)
            self._acquired = False

"""Make the reference package importable without installing it.

The lab runs its pytest suite against `examples/src/feedkit` directly, so the
unit tests work before you have installed anything. The separate question — does
the INSTALLED console script work — is checked by `tests/run_tests.sh`, which
does a real `pip install -e` and then runs the command by name.
"""

from __future__ import annotations

import os
import sys
from pathlib import Path

# FEEDKIT_SRC lets the harness point this suite at a DIFFERENT copy of the
# package — which is how it proves the suite is not vacuous: it breaks one line
# in a temporary copy and demands that these tests go red.
override = os.environ.get("FEEDKIT_SRC")
SRC = Path(override) if override else Path(__file__).resolve().parent.parent / "examples" / "src"
if str(SRC) not in sys.path:
    sys.path.insert(0, str(SRC))

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

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

Behaviour, chosen so the harness can prove the toolkit's failure design:

    /feed/notes.json      200, from tests/fixtures/feed/
    /feed/links.json      200
    /feed/papers.json     200
    /feed/malformed.json  200 with a body that is valid JSON but the wrong
                          shape — the case a status code cannot warn you about
    /feed/broken.json     500 every single time — the source that must be
                          skipped and reported while the others still succeed
    /feed/flaky.json      503, 503, then 200 — the source that proves retry
                          with backoff actually recovers
    /health               200 "ok", used only for the readiness loop

Every request requires `Authorization: Bearer <token>` when a token was given
on the command line, and answers 401 otherwise. That is what makes the secret
in this lab real rather than decorative: if redaction were achieved by simply
never sending the token, the leak test would prove nothing.

Run it by hand if you like:

    python3 tests/fixture_server.py --token demo-token-value

Stop it with Ctrl-C. It is single-threaded, serves only 127.0.0.1, and exits
when its parent harness kills it.
"""

from __future__ import annotations

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

FIXTURES = Path(__file__).resolve().parent / "fixtures" / "feed"

#: How many times /feed/flaky.json has been asked for, this process.
_flaky_hits = 0
_lock = threading.Lock()


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

    def log_message(self, fmt: str, *args: object) -> None:
        """Silence the default per-request line on stderr; the harness has its
        own output and a wall of request logs helps nobody."""

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

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

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

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

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

        if self.path == "/feed/broken.json":
            self._json(500, {"error": "this source is broken on purpose"})
            return

        if self.path == "/feed/flaky.json":
            with _lock:
                _flaky_hits += 1
                hits = _flaky_hits
            if hits < 3:
                self._json(503, {"error": f"temporarily unavailable (attempt {hits})"})
                return
            self._json(
                200,
                {
                    "source": "flaky",
                    "entries": [
                        {
                            "id": "f-001",
                            "title": "Recovered on the third attempt",
                            "published": "2026-07-11T07:00:00Z",
                        }
                    ],
                },
            )
            return

        if self.path.startswith("/feed/") and self.path.endswith(".json"):
            name = self.path[len("/feed/") : -len(".json")]
            candidate = FIXTURES / f"{name}.json"
            # Refuse anything that escapes the fixture directory.
            if candidate.resolve().parent != FIXTURES.resolve() or not candidate.is_file():
                self._json(404, {"error": f"no such source: {name}"})
                return
            self._send(200, candidate.read_bytes())
            return

        self._json(404, {"error": "not found"})


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

    FixtureHandler.token = args.token
    # Port 0 means "any free port"; the kernel picks and we read it back.
    server = HTTPServer(("127.0.0.1", 0), FixtureHandler)
    port = server.server_address[1]

    # First line of stdout is the contract with the harness.
    print(port, flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()
    return 0


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

{
  "source": "links",
  "generated": "2026-07-19T09:00:00Z",
  "entries": [
    {"id": "l-001", "title": "Atomic writes on POSIX", "published": "2026-07-13T11:30:00Z"},
    {"id": "l-002", "title": "Why stdout beats a log file", "published": "2026-07-17T11:30:00Z"}
  ]
}

{
  "source": "malformed",
  "generated": "2026-07-19T09:00:00Z",
  "entries": [
    {"id": "m-001", "headline": "This entry has no title field and no published field"}
  ]
}

{
  "source": "notes",
  "generated": "2026-07-19T09:00:00Z",
  "entries": [
    {"id": "n-001", "title": "Backoff is a courtesy, not a trick", "published": "2026-07-14T08:00:00Z"},
    {"id": "n-002", "title": "Idempotence in one paragraph", "published": "2026-07-15T08:00:00Z"},
    {"id": "n-003", "title": "What a run summary should say", "published": "2026-07-16T08:00:00Z"}
  ]
}

{
  "source": "papers",
  "generated": "2026-07-19T09:00:00Z",
  "entries": [
    {"id": "p-001", "title": "Partial failure in batch systems", "published": "2026-07-12T16:45:00Z"},
    {"id": "p-002", "title": "Scheduling under clock changes", "published": "2026-07-18T16:45:00Z"}
  ]
}

#!/usr/bin/env bash
# Tests for the Day 084 lab. Run from the lab directory:
#   bash tests/run_tests.sh
#
# This suite checks the OPERATIONAL properties of an automation, not its happy
# path. The happy path is one check out of the whole file. The rest ask the
# questions that decide whether a tool can be left running unattended:
#
#   * does running it twice process each item once?
#   * does one broken item get skipped and REPORTED while the others succeed —
#     and does the exit code say "partial", not "fine"?
#   * does --dry-run leave the state file byte-identical?
#   * does the four-layer configuration precedence actually resolve that way?
#   * does a secret supplied in the environment stay out of the log? (This is
#     the leak check, and it is the most important assertion in the lab.)
#   * does an interrupted state write leave the previous state intact?
#   * does the INSTALLED console script run?
#
# Nothing here touches the internet. A fixture server is started on 127.0.0.1
# on an ephemeral port, waited for, and killed in a trap. Nothing is installed
# into any real scheduler and no background process outlives this script.
set -u

export PYTHONDONTWRITEBYTECODE=1

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

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

cleanup() {
  if [ -n "${server_pid}" ] && kill -0 "${server_pid}" 2>/dev/null; then
    kill "${server_pid}" 2>/dev/null || true
    wait "${server_pid}" 2>/dev/null || true
  fi
  [ -n "${work_root}" ] && [ -d "${work_root}" ] && rm -rf "${work_root}"
}
trap cleanup EXIT INT TERM

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

install_hint() {
  echo "  Install this lab's dependencies with:" >&2
  echo "    python3 -m venv .venv" >&2
  echo "    .venv/bin/pip install -r requirements/requirements.txt" >&2
}

pytest_bin="$(resolve_tool pytest "${PYTEST:-}")" || {
  echo "FAIL: pytest not found." >&2
  install_hint
  echo "  Or point this suite at an existing pytest: PYTEST=/path/to/pytest bash tests/run_tests.sh" >&2
  exit 1
}
python_bin="$(resolve_tool python "${PYTHON:-}")" || python_bin="$(command -v python3 || true)"
if [ -z "${python_bin}" ]; then
  echo "FAIL: python3 not found on PATH." >&2
  exit 1
fi
pip_bin="$(resolve_tool pip "${PIP:-}")" || {
  echo "FAIL: pip not found." >&2
  install_hint
  exit 1
}

if ! "${python_bin}" -c "import requests" >/dev/null 2>&1; then
  echo "FAIL: the 'requests' package is not importable by ${python_bin}." >&2
  install_hint
  exit 1
fi

work_root="$(mktemp -d "${TMPDIR:-/tmp}/feedkit-tests.XXXXXX")"

# The secret. Invented here, used for real (the fixture server rejects any
# request without it), and never written to any file in this repository.
export FEEDKIT_TEST_TOKEN="lab-token-9f2b7c41d0"

echo "Day 084 — Ship the Toolkit"
echo

# --------------------------------------------------------------------------
echo "1. The local fixture server (127.0.0.1, ephemeral port, no internet)"
# --------------------------------------------------------------------------

server_out="${work_root}/server.port"
"${python_bin}" "${lab_dir}/tests/fixture_server.py" --token "${FEEDKIT_TEST_TOKEN}" \
  > "${server_out}" 2>"${work_root}/server.err" &
server_pid=$!

# Wait for readiness by polling, not by sleeping a fixed time. A fixed sleep is
# either too short on a slow machine or wasted time on a fast one.
port=""
for _ in $(seq 1 100); do
  if [ -s "${server_out}" ]; then
    port="$(head -1 "${server_out}" | tr -d '[:space:]')"
    [ -n "${port}" ] && break
  fi
  sleep 0.05
done

if [ -n "${port}" ]; then
  check "the fixture server chose an ephemeral port (${port}, not a hard-coded one)" "yes"
else
  check "the fixture server started and reported its port" "no"
  echo "${checks} checks, ${failures} failure(s)."
  exit 1
fi

export FEEDKIT_TEST_BASE_URL="http://127.0.0.1:${port}"
export FEEDKIT_BASE_URL="${FEEDKIT_TEST_BASE_URL}"

ready="no"
for _ in $(seq 1 100); do
  if "${python_bin}" - "${port}" <<'PY' >/dev/null 2>&1
import sys, urllib.request
port = sys.argv[1]
with urllib.request.urlopen(f"http://127.0.0.1:{port}/health", timeout=1) as response:
    sys.exit(0 if response.read() == b"ok" else 1)
PY
  then ready="yes"; break; fi
  sleep 0.05
done
check "the fixture server answers /health before any test runs" "${ready}"

auth_status="$("${python_bin}" - "${port}" <<'PY'
import sys, urllib.error, urllib.request
port = sys.argv[1]
try:
    urllib.request.urlopen(f"http://127.0.0.1:{port}/feed/notes.json", timeout=2)
    print("200")
except urllib.error.HTTPError as exc:
    print(exc.code)
PY
)"
if [ "${auth_status}" = "401" ]; then
  check "the fixture server really requires the token (401 without it)" "yes"
else
  check "the fixture server really requires the token (got ${auth_status})" "no"
fi

# --------------------------------------------------------------------------
echo
echo "2. The property suite (pytest over examples/src/feedkit)"
# --------------------------------------------------------------------------

pytest_out="$(cd "${lab_dir}" && "${pytest_bin}" tests/test_toolkit.py -q 2>&1)"
pytest_exit=$?
if [ "${pytest_exit}" -eq 0 ]; then
  check "the property suite passes (exit 0)" "yes"
else
  check "the property suite passes (exit ${pytest_exit})" "no"
  echo "${pytest_out}" | tail -30
fi
pytest_tail="$(printf '%s\n' "${pytest_out}" | grep -E '[0-9]+ passed' | tail -1)"
case "${pytest_tail}" in
  *passed*) check "pytest reports: ${pytest_tail}" "yes" ;;
  *) check "pytest reports a passed count" "no" ;;
esac

# The suite must not be vacuous: break the idempotence rule and demand red.
broken_src="${work_root}/broken-src"
mkdir -p "${broken_src}"
cp -R "${lab_dir}/examples/src/feedkit" "${broken_src}/feedkit"
"${python_bin}" - "${broken_src}/feedkit/core.py" <<'PY'
import sys
from pathlib import Path
path = Path(sys.argv[1])
text = path.read_text()
old = "fresh = [entry for entry in entries if entry.id not in already]"
new = "fresh = list(entries)  # deliberately broken: ignores what has been seen"
assert old in text, "the line to break was not found"
path.write_text(text.replace(old, new))
PY
broken_out="$(cd "${lab_dir}" && FEEDKIT_SRC="${broken_src}" "${pytest_bin}" \
  tests/test_toolkit.py -q -p no:cacheprovider 2>&1)"
broken_exit=$?
if [ "${broken_exit}" -ne 0 ]; then
  check "breaking the idempotence rule makes the suite FAIL (exit ${broken_exit})" "yes"
else
  check "breaking the idempotence rule makes the suite FAIL — it did not, so the suite is vacuous" "no"
fi
case "${broken_out}" in
  *test_running_fetch_twice_processes_each_entry_once*)
    check "the failing run names the idempotence test" "yes" ;;
  *) check "the failing run names the idempotence test" "no" ;;
esac

# --------------------------------------------------------------------------
echo
echo "3. The command line, end to end, as a real process"
# --------------------------------------------------------------------------

run_dir="${work_root}/run"
mkdir -p "${run_dir}"
cp "${lab_dir}/examples/feedkit.toml" "${run_dir}/feedkit.toml"

feedkit() {
  (cd "${run_dir}" && PYTHONPATH="${lab_dir}/examples/src" \
     FEEDKIT_TOKEN="${FEEDKIT_TEST_TOKEN}" \
     "${python_bin}" -m feedkit.cli "$@")
}

first_log="${work_root}/fetch-1.log"
feedkit --log-level info fetch > "${first_log}" 2>&1
first_exit=$?
if [ "${first_exit}" -eq 0 ]; then
  check "a first fetch of three good sources exits 0" "yes"
else
  check "a first fetch of three good sources exits 0 (got ${first_exit})" "no"
  tail -20 "${first_log}"
fi
if grep -q 'new entries: 7' "${first_log}"; then
  check "the first run collects 7 entries from notes, links and papers" "yes"
else
  check "the first run collects 7 entries from notes, links and papers" "no"
fi

second_log="${work_root}/fetch-2.log"
feedkit fetch > "${second_log}" 2>&1
second_exit=$?
if [ "${second_exit}" -eq 0 ] && grep -q 'new entries: 0' "${second_log}"; then
  check "running fetch again collects 0 new entries and exits 0 (idempotence)" "yes"
else
  check "running fetch again collects 0 new entries and exits 0 (idempotence)" "no"
  tail -10 "${second_log}"
fi

# Dry run must not write. Compare the bytes.
state_file="${run_dir}/feedkit-state.json"
before_hash="$("${python_bin}" -c "import hashlib,sys;print(hashlib.sha256(open(sys.argv[1],'rb').read()).hexdigest())" "${state_file}")"
dry_log="${work_root}/dry.log"
feedkit --sources notes,links,papers,flaky fetch --dry-run > "${dry_log}" 2>&1
dry_exit=$?
after_hash="$("${python_bin}" -c "import hashlib,sys;print(hashlib.sha256(open(sys.argv[1],'rb').read()).hexdigest())" "${state_file}")"
if [ "${before_hash}" = "${after_hash}" ]; then
  check "--dry-run leaves the state file byte-identical" "yes"
else
  check "--dry-run leaves the state file byte-identical" "no"
fi
if grep -q 'dry run — nothing was written' "${dry_log}"; then
  check "--dry-run says plainly that nothing was written" "yes"
else
  check "--dry-run says plainly that nothing was written" "no"
fi
if [ "${dry_exit}" -eq 0 ]; then
  check "--dry-run exits 0" "yes"
else
  check "--dry-run exits 0 (got ${dry_exit})" "no"
fi
if [ -z "$(find "${run_dir}" -name '*.tmp' -print -quit)" ]; then
  check "no temporary state files are left behind anywhere" "yes"
else
  check "no temporary state files are left behind anywhere" "no"
fi

# Partial success: one source is broken, the rest still work, exit code is 2.
partial_log="${work_root}/partial.log"
feedkit --sources notes,broken,papers fetch > "${partial_log}" 2>&1
partial_exit=$?
if [ "${partial_exit}" -eq 3 ]; then
  check "partial success exits 3, not 0 (it does not pretend everything worked)" "yes"
else
  check "partial success exits 3, not 0 (got ${partial_exit})" "no"
fi
if grep -q 'FAILED: broken' "${partial_log}"; then
  check "the failing source is named in the run summary" "yes"
else
  check "the failing source is named in the run summary" "no"
fi
if grep -q 'sources: 2 ok, 1 failed' "${partial_log}"; then
  check "the summary counts 2 ok and 1 failed" "yes"
else
  check "the summary counts 2 ok and 1 failed" "no"
fi
if grep -q '"level": "error"' "${partial_log}" && grep -q '"source": "broken"' "${partial_log}"; then
  check "the structured log records WHICH source failed, at error level" "yes"
else
  check "the structured log records WHICH source failed, at error level" "no"
fi

# Total failure: every source broken.
total_log="${work_root}/total.log"
feedkit --sources broken fetch > "${total_log}" 2>&1
total_exit=$?
if [ "${total_exit}" -eq 1 ]; then
  check "a run where every source fails exits 1" "yes"
else
  check "a run where every source fails exits 1 (got ${total_exit})" "no"
fi

# --------------------------------------------------------------------------
echo
echo "4. The leak check — a supplied secret must never reach the log"
# --------------------------------------------------------------------------

leak_log="${work_root}/leak.log"
feedkit --log-level debug --sources notes,broken fetch > "${leak_log}" 2>&1 || true
if grep -q "${FEEDKIT_TEST_TOKEN}" "${leak_log}"; then
  check "the token supplied in FEEDKIT_TOKEN never appears in the log" "no"
else
  check "the token supplied in FEEDKIT_TOKEN never appears in the log" "yes"
fi
if grep -rq "${FEEDKIT_TEST_TOKEN}" "${run_dir}" 2>/dev/null; then
  check "the token never reaches the state file or the config file" "no"
else
  check "the token never reaches the state file or the config file" "yes"
fi
explain_log="${work_root}/explain.log"
feedkit status --explain-config > "${explain_log}" 2>&1
if grep -q "${FEEDKIT_TEST_TOKEN}" "${explain_log}"; then
  check "even --explain-config does not print the token" "no"
else
  check "even --explain-config does not print the token" "yes"
fi
if grep -q 'set (never printed)' "${explain_log}"; then
  check "--explain-config says the token is set without showing it" "yes"
else
  check "--explain-config says the token is set without showing it" "no"
fi
# And the token is genuinely required: without it every source 401s.
noauth_log="${work_root}/noauth.log"
(cd "${run_dir}" && PYTHONPATH="${lab_dir}/examples/src" \
   "${python_bin}" -m feedkit.cli --sources notes fetch > "${noauth_log}" 2>&1)
noauth_exit=$?
if [ "${noauth_exit}" -eq 1 ] && grep -q 'HTTP 401' "${noauth_log}"; then
  check "without the token every request is refused — the secret is real, not decorative" "yes"
else
  check "without the token every request is refused (exit ${noauth_exit})" "no"
fi

# --------------------------------------------------------------------------
echo
echo "5. Configuration precedence: flag beats environment beats file beats default"
# --------------------------------------------------------------------------

prec_dir="${work_root}/precedence"
mkdir -p "${prec_dir}"
printf '[feedkit]\nmax_items = 10\n' > "${prec_dir}/feedkit.toml"

# One command, one setting, four different sources for its value. Each call
# below adds exactly one layer to the one before it, so the four assertions
# together pin the whole precedence rather than just its ends.
precedence_value() {
  local where="$1"
  shift
  (cd "${where}" && PYTHONPATH="${lab_dir}/examples/src" "${python_bin}" -m feedkit.cli \
     "$@" status --explain-config 2>/dev/null | awk '$1=="max_items"{print $2, $3}')
}

empty_dir="${work_root}/no-config"
mkdir -p "${empty_dir}"

got="$(precedence_value "${empty_dir}")"
if [ "${got}" = "5 default" ]; then
  check "layer 1: with no file, no environment and no flag, max_items is 5 (default)" "yes"
else
  check "layer 1: max_items is 5 from the default (got '${got}')" "no"
fi

got="$(precedence_value "${prec_dir}")"
if [ "${got}" = "10 file" ]; then
  check "layer 2: the configuration file beats the default (10, file)" "yes"
else
  check "layer 2: the configuration file beats the default (got '${got}')" "no"
fi

got="$(FEEDKIT_MAX_ITEMS=20 precedence_value "${prec_dir}")"
if [ "${got}" = "20 environment" ]; then
  check "layer 3: the environment beats the file (20, environment)" "yes"
else
  check "layer 3: the environment beats the file (got '${got}')" "no"
fi

got="$(FEEDKIT_MAX_ITEMS=20 precedence_value "${prec_dir}" --max-items 40)"
if [ "${got}" = "40 flag" ]; then
  check "layer 4: a flag beats the environment (40, flag) — all four confirmed" "yes"
else
  check "layer 4: a flag beats the environment (got '${got}')" "no"
fi

# An absent flag must mean "no opinion", not "override with nothing" — the bug
# that makes every default silently win over the configuration file.
got="$(precedence_value "${prec_dir}")"
if [ "${got}" = "10 file" ]; then
  check "an unsupplied flag does not override the file" "yes"
else
  check "an unsupplied flag does not override the file (got '${got}')" "no"
fi

# A typo in a config file is an error, not a shrug.
typo_dir="${work_root}/typo"
mkdir -p "${typo_dir}"
printf '[feedkit]\nmax_itmes = 10\n' > "${typo_dir}/feedkit.toml"
typo_out="$( (cd "${typo_dir}" && PYTHONPATH="${lab_dir}/examples/src" \
  "${python_bin}" -m feedkit.cli status 2>&1) )"
typo_exit=$?
if [ "${typo_exit}" -ne 0 ] && printf '%s' "${typo_out}" | grep -q "unknown setting"; then
  check "a misspelled setting in the configuration file stops the run" "yes"
else
  check "a misspelled setting in the configuration file stops the run" "no"
fi

# --------------------------------------------------------------------------
echo
echo "6. status, report, and the watchdog that notices silence"
# --------------------------------------------------------------------------

report_log="${work_root}/report.log"
feedkit report --limit 3 > "${report_log}" 2>&1
if grep -q 'entries collected; showing 3' "${report_log}"; then
  check "report renders the collected entries" "yes"
else
  check "report renders the collected entries" "no"
fi

status_log="${work_root}/status.log"
feedkit status > "${status_log}" 2>&1
status_exit=$?
if [ "${status_exit}" -eq 0 ] && grep -q 'watchdog:     fresh' "${status_log}"; then
  check "status exits 0 while the last success is recent" "yes"
else
  check "status exits 0 while the last success is recent (got ${status_exit})" "no"
fi

stale_log="${work_root}/stale.log"
feedkit status --max-age-minutes 0 > "${stale_log}" 2>&1
stale_exit=$?
if [ "${stale_exit}" -eq 3 ] && grep -q 'STALE' "${stale_log}"; then
  check "the watchdog exits non-zero when the last success is too old" "yes"
else
  check "the watchdog exits non-zero when the last success is too old (got ${stale_exit})" "no"
fi

fresh_dir="${work_root}/never-run"
mkdir -p "${fresh_dir}"
never_log="${work_root}/never.log"
(cd "${fresh_dir}" && PYTHONPATH="${lab_dir}/examples/src" "${python_bin}" -m feedkit.cli status \
   > "${never_log}" 2>&1)
never_exit=$?
if [ "${never_exit}" -eq 3 ] && grep -q 'last success: never' "${never_log}"; then
  check "a toolkit that has never run reports 'never' and exits non-zero" "yes"
else
  check "a toolkit that has never run reports 'never' and exits non-zero (got ${never_exit})" "no"
fi

# --------------------------------------------------------------------------
echo
echo "7. Two runs must not overlap"
# --------------------------------------------------------------------------

touch "${run_dir}/feedkit-state.json.lock"
locked_log="${work_root}/locked.log"
feedkit fetch > "${locked_log}" 2>&1
locked_exit=$?
rm -f "${run_dir}/feedkit-state.json.lock"
if [ "${locked_exit}" -eq 75 ]; then
  check "a run that finds the lock held exits 75 and does nothing" "yes"
else
  check "a run that finds the lock held exits 75 (got ${locked_exit})" "no"
fi
if grep -q '"status": "locked"' "${locked_log}"; then
  check "the overlapping run says so in the log" "yes"
else
  check "the overlapping run says so in the log" "no"
fi

# --------------------------------------------------------------------------
echo
echo "8. The installed console script"
# --------------------------------------------------------------------------

# The install goes into a THROWAWAY environment under the work directory, never
# into whatever pip happens to be on PATH. Running a lab's tests must never
# install a package into the caller's environment as a side effect — and
# resolving the console script from PATH afterwards would only find it because
# of that pollution.
#
# The package is built into a wheel FIRST, using the interpreter that already
# has a build backend, and only the finished wheel is installed into the
# throwaway environment. Installing a wheel needs no build backend at all, so
# the throwaway environment can be a plain empty venv and the whole step stays
# offline. (A --system-site-packages venv would NOT help here: it inherits the
# base interpreter's site-packages, not those of the virtualenv it was created
# from, so setuptools would still be missing.)
install_env="${work_root}/install-env"
wheel_dir="${work_root}/wheels"
install_log="${work_root}/install.log"
mkdir -p "${wheel_dir}"

built_wheel=""
if "${python_bin}" -m pip wheel --no-deps --no-build-isolation --no-index \
     -w "${wheel_dir}" "${lab_dir}/examples" -q > "${install_log}" 2>&1; then
  built_wheel="$(find "${wheel_dir}" -maxdepth 1 -name 'feedkit-*.whl' -print -quit)"
fi
if [ -n "${built_wheel}" ] && [ -f "${built_wheel}" ]; then
  check "the package builds into a wheel offline (--no-build-isolation --no-index)" "yes"
else
  check "the package builds into a wheel offline (--no-build-isolation --no-index)" "no"
  tail -15 "${install_log}"
fi

if [ -n "${built_wheel}" ] &&
   "${python_bin}" -m venv "${install_env}" >> "${install_log}" 2>&1 &&
   "${install_env}/bin/pip" install --no-index --no-deps "${built_wheel}" -q \
     >> "${install_log}" 2>&1; then
  check "the wheel installs into a throwaway environment, not the caller's" "yes"
else
  check "the wheel installs into a throwaway environment, not the caller's" "no"
  tail -15 "${install_log}"
fi

console_bin="${install_env}/bin/feedkit"
if [ -x "${console_bin}" ]; then
  check "the console script 'feedkit' is created by the installation" "yes"
else
  check "the console script 'feedkit' is created by the installation" "no"
  console_bin=""
fi

# The wheel was installed with --no-deps, so the throwaway environment has
# feedkit but not its runtime dependency. Rather than reach for an index, the
# dependency is made importable from the environment that already has it. That
# keeps the step offline while still proving the INSTALLED console script — the
# one pip generated from the entry point — is what runs.
deps_site="$("${python_bin}" -c 'import site; print(site.getsitepackages()[0])' 2>/dev/null || true)"

if [ -n "${console_bin}" ]; then
  console_log="${work_root}/console.log"
  (cd "${run_dir}" && PYTHONPATH="${deps_site}" FEEDKIT_TOKEN="${FEEDKIT_TEST_TOKEN}" "${console_bin}" fetch \
     > "${console_log}" 2>&1)
  console_exit=$?
  if [ "${console_exit}" -eq 0 ]; then
    check "the installed console script runs a real fetch and exits 0" "yes"
  else
    check "the installed console script runs a real fetch and exits 0 (got ${console_exit})" "no"
    tail -10 "${console_log}"
  fi

  version_line="$(PYTHONPATH="${deps_site}" "${console_bin}" --version 2>&1 | head -1)"
  case "${version_line}" in
    "feedkit 1.0.0") check "feedkit --version reports 1.0.0" "yes" ;;
    *) check "feedkit --version reports 1.0.0 (got '${version_line}')" "no" ;;
  esac

  scheduled_bin="${install_env}/bin/feedkit-scheduled"
  [ -x "${scheduled_bin}" ] || scheduled_bin=""
  if [ -n "${scheduled_bin}" ]; then
    (cd "${run_dir}" && PYTHONPATH="${deps_site}" FEEDKIT_TOKEN="${FEEDKIT_TEST_TOKEN}" "${scheduled_bin}" \
       > "${work_root}/scheduled.log" 2>&1)
    scheduled_exit=$?
    if [ "${scheduled_exit}" -eq 0 ]; then
      check "the scheduled entry point 'feedkit-scheduled' runs and exits 0" "yes"
    else
      check "the scheduled entry point runs and exits 0 (got ${scheduled_exit})" "no"
    fi
  else
    check "the scheduled entry point 'feedkit-scheduled' exists" "no"
  fi
fi

# --------------------------------------------------------------------------
echo
echo "9. The starter is runnable, and the shipped files behave"
# --------------------------------------------------------------------------

starter_help="$(cd "${lab_dir}/starter" && PYTHONPATH="${lab_dir}/starter/src" \
  "${python_bin}" -m feedkit.cli --help 2>&1)"
starter_exit=$?
if [ "${starter_exit}" -eq 0 ]; then
  check "the starter's --help works before you write a line" "yes"
else
  check "the starter's --help works before you write a line (got ${starter_exit})" "no"
fi
exercise_count="$(grep -rc 'Exercise ' "${lab_dir}/starter/src/feedkit/" 2>/dev/null | \
  awk -F: '{total += $2} END {print total+0}')"
if [ "${exercise_count}" -ge 6 ]; then
  check "the starter carries its numbered exercises (${exercise_count} markers)" "yes"
else
  check "the starter carries its numbered exercises (found ${exercise_count})" "no"
fi
if (cd "${lab_dir}/starter" && PYTHONPATH="${lab_dir}/starter/src" \
    "${python_bin}" -c "import feedkit.core, feedkit.cli" >/dev/null 2>&1); then
  check "the starter package imports cleanly" "yes"
else
  check "the starter package imports cleanly" "no"
fi

# The schedule files are references and must never be installed. The strongest
# available proof is that nothing in this lab can spawn a process at all: no
# Python file imports subprocess or calls os.system, so no code path exists
# that could reach crontab, launchctl or systemctl.
if "${python_bin}" - "${lab_dir}" <<'PY' >/dev/null 2>&1
import re, sys
from pathlib import Path

root = Path(sys.argv[1])
banned = re.compile(r"^\s*import subprocess|^\s*from subprocess|os\.system\(|os\.exec")
offenders = []
for directory in ("examples/src", "starter/src", "tests"):
    for path in (root / directory).rglob("*.py"):
        for number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
            if banned.search(line):
                offenders.append(f"{path.name}:{number}")
for line in offenders:
    print(line, file=sys.stderr)
sys.exit(1 if offenders else 0)
PY
then
  check "no Python file can spawn a process, so nothing can touch a scheduler" "yes"
else
  check "no Python file can spawn a process, so nothing can touch a scheduler" "no"
fi
for schedule_file in feedkit.cron com.example.feedkit.plist feedkit.service feedkit.timer; do
  schedule_path="${lab_dir}/examples/schedule/${schedule_file}"
  if [ -f "${schedule_path}" ] && grep -q 'NOT INSTALLED BY THIS LAB' "${schedule_path}"; then
    check "examples/schedule/${schedule_file} ships as a reference and says so" "yes"
  else
    check "examples/schedule/${schedule_file} ships as a reference and says so" "no"
  fi
done

# --------------------------------------------------------------------------
echo
echo "10. Nothing in this lab reaches the internet"
# --------------------------------------------------------------------------

if "${python_bin}" - "${lab_dir}" <<'PY' >/dev/null 2>&1
import re, sys
from pathlib import Path

root = Path(sys.argv[1])
pattern = re.compile(r"https?://(?!127\.0\.0\.1)[A-Za-z0-9.-]+")
offenders = []
for directory in ("examples/src", "starter/src", "tests"):
    for path in (root / directory).rglob("*.py"):
        for match in pattern.finditer(path.read_text(encoding="utf-8")):
            offenders.append(f"{path.name}: {match.group(0)}")
for line in offenders:
    print(line, file=sys.stderr)
sys.exit(1 if offenders else 0)
PY
then
  check "no executable file names any host but 127.0.0.1" "yes"
else
  check "no executable file names any host but 127.0.0.1" "no"
fi

if "${python_bin}" - "${lab_dir}" <<'PY' >/dev/null 2>&1
import sys
from pathlib import Path

root = Path(sys.argv[1])
offenders = []
for directory in ("examples", "starter"):
    for path in (root / directory).rglob("*"):
        if path.is_file() and ".venv" not in path.parts and "8000" in path.read_text(
            encoding="utf-8", errors="ignore"
        ):
            offenders.append(path.name)
for candidate in ("fixture_server.py", "test_toolkit.py", "conftest.py"):
    path = root / "tests" / candidate
    if "8000" in path.read_text(encoding="utf-8"):
        offenders.append(candidate)
for name in offenders:
    print(name, file=sys.stderr)
sys.exit(1 if offenders else 0)
PY
then
  check "nothing hard-codes port 8000 — the port everyone already has in use" "yes"
else
  check "nothing hard-codes port 8000" "no"
fi

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

"""The properties an automation has to have, asserted one at a time.

Notice what is NOT tested here: that a happy-path fetch returns three entries.
That is the easy half and it is checked once, in passing. Everything else in
this file is about the parts that are not the happy path — running twice,
failing partway, being interrupted mid-write, being told four different things
by four configuration layers, and being trusted with a secret.

Two of these tests need no server at all, because the core is pure and the
fetcher arrives as an argument. The rest talk to the local fixture server on
127.0.0.1 that `run_tests.sh` started; the address is handed over in
FEEDKIT_TEST_BASE_URL. Nothing here reaches the internet.
"""

from __future__ import annotations

import json
import os
from pathlib import Path

import pytest
import requests

from feedkit import adapters, config as config_module, core, logging_setup, runner
from feedkit import state as state_module

TOKEN = os.environ.get("FEEDKIT_TEST_TOKEN", "")
BASE_URL = os.environ.get("FEEDKIT_TEST_BASE_URL", "")


# --------------------------------------------------------------------------
# Fixtures
# --------------------------------------------------------------------------


@pytest.fixture
def logger(capsys):
    """A logger writing structured JSON to captured stdout."""
    return logging_setup.configure("debug", run_id="test0000", secrets=[TOKEN] if TOKEN else [])


@pytest.fixture
def live_fetcher():
    """A real HttpFetcher pointed at the local fixture server, with a no-op
    sleeper so three retries cost microseconds instead of 1.5 seconds."""
    if not BASE_URL:
        pytest.fail("FEEDKIT_TEST_BASE_URL is not set; run this suite through tests/run_tests.sh")
    session = requests.Session()
    fetcher = adapters.HttpFetcher(
        session=session,
        base_url=BASE_URL,
        token=TOKEN,
        timeout=5.0,
        retries=3,
        backoff_seconds=0.0,
        sleeper=lambda _seconds: None,
    )
    yield fetcher
    session.close()


def settings(**overrides):
    """Build a Config without touching os.environ or any file."""
    values = {
        "base_url": BASE_URL or "http://127.0.0.1:1",
        "sources": ["notes", "links"],
        "max_items": 50,
        "retries": 3,
    }
    values.update(overrides)
    return config_module.resolve(
        file_values={},
        environment={key: value for key, value in values.items()},
        flags={},
        token=TOKEN,
    )


def do_run(tmp_path: Path, fetcher, logger, sources, dry_run=False, clock_value="2026-07-19T10:00:00Z"):
    state_path = tmp_path / "state.json"
    return runner.run_fetch(
        settings(sources=list(sources)),
        fetcher,
        adapters.FixedClock(clock_value),
        state_path,
        tmp_path / "state.json.lock",
        logger,
        dry_run=dry_run,
    ), state_path


# --------------------------------------------------------------------------
# 1. The core is pure, so these need nothing at all
# --------------------------------------------------------------------------


def test_select_new_skips_ids_already_seen():
    entries = [
        core.Entry("a", "first", "2026-07-01T00:00:00Z", "notes"),
        core.Entry("b", "second", "2026-07-02T00:00:00Z", "notes"),
    ]
    assert [e.id for e in core.select_new(entries, ["a"], 10)] == ["b"]
    assert core.select_new(entries, ["a", "b"], 10) == ()


def test_select_new_caps_at_max_items_newest_first():
    entries = [
        core.Entry(str(n), f"t{n}", f"2026-07-0{n}T00:00:00Z", "notes") for n in range(1, 6)
    ]
    picked = core.select_new(entries, [], 2)
    assert [e.id for e in picked] == ["5", "4"]


@pytest.mark.parametrize(
    "statuses, expected",
    [
        (["ok", "ok"], "ok"),
        (["ok", "failed"], "partial"),
        (["failed", "failed"], "failed"),
        ([], "ok"),
    ],
)
def test_summary_status(statuses, expected):
    results = [core.SourceResult(source=f"s{n}", status=s) for n, s in enumerate(statuses)]
    assert core.summarise(results)["status"] == expected


def test_exit_codes_distinguish_partial_from_total_success():
    assert core.exit_code_for({"status": "ok"}) == 0
    assert core.exit_code_for({"status": "partial"}) == 3
    assert core.exit_code_for({"status": "failed"}) == 1


def test_parse_entries_rejects_the_wrong_shape():
    with pytest.raises(core.InvalidPayload):
        core.parse_entries({"entries": [{"id": "x"}]}, "notes")
    with pytest.raises(core.InvalidPayload):
        core.parse_entries({"items": []}, "notes")
    with pytest.raises(core.InvalidPayload):
        core.parse_entries([1, 2, 3], "notes")


def test_watchdog_reports_silence_not_only_errors():
    assert core.is_stale(None, "2026-07-19T10:00:00Z", 3600) is True
    assert core.is_stale("2026-07-19T09:30:00Z", "2026-07-19T10:00:00Z", 3600) is False
    assert core.is_stale("2026-07-19T08:00:00Z", "2026-07-19T10:00:00Z", 3600) is True


# --------------------------------------------------------------------------
# 2. Configuration precedence — all four layers
# --------------------------------------------------------------------------


def test_configuration_precedence_default_file_environment_flag():
    base = {"base_url": "http://127.0.0.1:9"}

    only_default = config_module.resolve({}, base, {})
    assert only_default.max_items == 5
    assert only_default.provenance["max_items"] == "default"

    with_file = config_module.resolve({"max_items": 10}, base, {})
    assert with_file.max_items == 10
    assert with_file.provenance["max_items"] == "file"

    with_env = config_module.resolve({"max_items": 10}, {**base, "max_items": "20"}, {})
    assert with_env.max_items == 20
    assert with_env.provenance["max_items"] == "environment"

    with_flag = config_module.resolve(
        {"max_items": 10}, {**base, "max_items": "20"}, {"max_items": 40}
    )
    assert with_flag.max_items == 40
    assert with_flag.provenance["max_items"] == "flag"


def test_an_unsupplied_flag_does_not_override_anything():
    """argparse leaves absent options as None. None must mean 'no opinion',
    not 'set it to nothing' — the bug that makes every flag override the file."""
    resolved = config_module.resolve(
        {"max_items": 10},
        {"base_url": "http://127.0.0.1:9"},
        {"max_items": None, "retries": None},
    )
    assert resolved.max_items == 10
    assert resolved.provenance["max_items"] == "file"


def test_a_typo_in_the_configuration_file_is_an_error_not_a_shrug():
    with pytest.raises(config_module.ConfigError):
        config_module.resolve({"max_itmes": 10}, {"base_url": "http://127.0.0.1:9"}, {})


def test_a_missing_base_url_stops_the_run():
    with pytest.raises(config_module.ConfigError):
        config_module.resolve({}, {}, {})


def test_the_token_never_comes_from_the_configuration_file():
    resolved = config_module.resolve(
        {}, {"base_url": "http://127.0.0.1:9"}, {}, token="secret-value-123456"
    )
    assert resolved.token == "secret-value-123456"
    assert "secret-value-123456" not in config_module.explain(resolved, None)


# --------------------------------------------------------------------------
# 3. Idempotence, partial failure, dry run — against the local server
# --------------------------------------------------------------------------


def test_running_fetch_twice_processes_each_entry_once(tmp_path, live_fetcher, logger):
    (first, first_code), state_path = do_run(tmp_path, live_fetcher, logger, ["notes", "links"])
    assert first["status"] == "ok"
    assert first["new_entries"] == 5
    assert first_code == 0

    (second, code), _ = do_run(tmp_path, live_fetcher, logger, ["notes", "links"])
    assert second["new_entries"] == 0, "a second run must find nothing new"
    assert code == 0

    stored = json.loads(state_path.read_text())
    assert len(stored["entries"]) == 5
    ids = [entry["id"] for entry in stored["entries"]]
    assert len(ids) == len(set(ids)), "no entry may be recorded twice"


def test_one_broken_source_is_skipped_and_reported_while_others_succeed(
    tmp_path, live_fetcher, logger
):
    (summary, code), state_path = do_run(
        tmp_path, live_fetcher, logger, ["notes", "broken", "links"]
    )
    assert summary["status"] == "partial"
    assert summary["sources_ok"] == 2
    assert summary["sources_failed"] == 1
    assert "broken" in summary["failures"]
    assert code == core.EXIT_PARTIAL, "partial success must not exit 0"

    # The successful sources still did their work.
    stored = json.loads(state_path.read_text())
    assert stored["sources"]["notes"]["seen_ids"]
    assert stored["sources"]["links"]["seen_ids"]
    assert stored["sources"]["broken"]["last_error"]
    assert stored["last_success"] is None, "a partial run is not a success"

    # And the failure is in the human-readable summary, not only in the state.
    assert "FAILED: broken" in core.format_summary(summary, "r1")


def test_a_malformed_payload_is_not_retried_and_is_reported(tmp_path, live_fetcher, logger):
    (summary, code), _ = do_run(tmp_path, live_fetcher, logger, ["notes", "malformed"])
    assert summary["status"] == "partial"
    assert "missing title, published" in summary["failures"]["malformed"]
    assert code == core.EXIT_PARTIAL


def test_retry_with_backoff_recovers_a_temporarily_failing_source(
    tmp_path, live_fetcher, logger
):
    """The fixture server answers /feed/flaky.json with 503, 503, then 200."""
    (summary, code), _ = do_run(tmp_path, live_fetcher, logger, ["flaky"])
    assert summary["status"] == "ok"
    assert summary["new_entries"] == 1
    assert summary["retried"] == {"flaky": 3}, "it should have taken three attempts"
    assert code == 0


def test_dry_run_leaves_the_state_file_byte_identical(tmp_path, live_fetcher, logger):
    (_, _), state_path = do_run(tmp_path, live_fetcher, logger, ["notes"])
    before = state_path.read_bytes()

    (summary, code), _ = do_run(
        tmp_path, live_fetcher, logger, ["notes", "links"], dry_run=True
    )
    assert summary["new_entries"] == 2, "a dry run must still say what it would do"
    assert state_path.read_bytes() == before, "a dry run must not write"
    assert code == 0
    assert not list(tmp_path.glob("*.tmp")), "and must leave no temporary files"


def test_a_second_run_cannot_start_while_one_is_in_progress(tmp_path, live_fetcher, logger):
    lock_path = tmp_path / "state.json.lock"
    with state_module.Lock(lock_path):
        summary, code = runner.run_fetch(
            settings(sources=["notes"]),
            live_fetcher,
            adapters.FixedClock("2026-07-19T10:00:00Z"),
            tmp_path / "state.json",
            lock_path,
            logger,
        )
    assert summary["status"] == "locked"
    assert code == core.EXIT_LOCKED
    assert not (tmp_path / "state.json").exists()


# --------------------------------------------------------------------------
# 4. The state file survives being interrupted
# --------------------------------------------------------------------------


def test_an_interrupted_write_leaves_the_previous_state_intact(tmp_path):
    path = tmp_path / "state.json"
    state_module.write_atomic(path, {"version": 1, "marker": "original"})
    before = path.read_bytes()

    def power_cut():
        raise KeyboardInterrupt("the machine went away mid-write")

    with pytest.raises(KeyboardInterrupt):
        state_module.write_atomic(path, {"version": 1, "marker": "replacement"}, power_cut)

    assert path.read_bytes() == before, "the old state must survive untouched"
    assert json.loads(path.read_text())["marker"] == "original"
    assert not list(tmp_path.glob("*.tmp")), "and the temporary file must be cleaned up"


def test_a_corrupt_state_file_stops_the_run_rather_than_being_overwritten(tmp_path):
    path = tmp_path / "state.json"
    path.write_text("{ this is not json")
    with pytest.raises(state_module.StateError):
        state_module.load(path)
    assert path.read_text() == "{ this is not json"


def test_state_is_written_whole_or_not_at_all_under_a_reader(tmp_path):
    """os.replace is atomic: a reader sees the old file or the new one."""
    path = tmp_path / "state.json"
    state_module.write_atomic(path, {"version": 1, "n": 1})
    for n in range(2, 6):
        state_module.write_atomic(path, {"version": 1, "n": n})
        assert json.loads(path.read_text())["n"] == n


# --------------------------------------------------------------------------
# 5. The leak check — the most important assertion in this file
# --------------------------------------------------------------------------


def test_the_secret_never_reaches_the_log(tmp_path, capsys):
    """The token is supplied, is genuinely used (the fixture server rejects a
    request without it), and must appear nowhere in the structured output."""
    if not TOKEN:
        pytest.fail("FEEDKIT_TEST_TOKEN is not set; run this suite through tests/run_tests.sh")

    log = logging_setup.configure("debug", run_id="leak0001", secrets=[TOKEN])
    session = requests.Session()
    try:
        fetcher = adapters.HttpFetcher(
            session=session,
            base_url=BASE_URL,
            token=TOKEN,
            retries=2,
            backoff_seconds=0.0,
            sleeper=lambda _s: None,
            logger=log,
        )
        summary, _ = runner.run_fetch(
            settings(sources=["notes", "broken"]),
            fetcher,
            adapters.FixedClock("2026-07-19T10:00:00Z"),
            tmp_path / "state.json",
            tmp_path / "state.json.lock",
            log,
        )
    finally:
        session.close()

    captured = capsys.readouterr().out
    assert captured.strip(), "the run must actually have logged something"
    assert TOKEN not in captured
    assert TOKEN not in json.dumps(summary)
    assert TOKEN not in (tmp_path / "state.json").read_text()


def test_the_redacting_filter_catches_a_deliberate_leak(capsys):
    """Even when somebody logs the token by mistake — and one day somebody
    will — the filter replaces it before the line is written."""
    secret = "not-a-real-token-abcdef"
    log = logging_setup.configure("debug", run_id="leak0002", secrets=[secret])
    log.info("careless message containing %s", secret, extra={"url": f"?token={secret}"})
    captured = capsys.readouterr().out
    assert secret not in captured
    assert "***REDACTED***" in captured


def test_every_log_line_is_json_carrying_the_run_id_and_the_item(tmp_path, live_fetcher, capsys):
    log = logging_setup.configure("info", run_id="abc12345", secrets=[TOKEN] if TOKEN else [])
    runner.run_fetch(
        settings(sources=["notes", "broken"]),
        live_fetcher,
        adapters.FixedClock("2026-07-19T10:00:00Z"),
        tmp_path / "state.json",
        tmp_path / "state.json.lock",
        log,
    )
    lines = [line for line in capsys.readouterr().out.splitlines() if line.strip()]
    records = [json.loads(line) for line in lines]
    assert records, "an unattended run that logs nothing cannot be debugged"
    assert all(record["run_id"] == "abc12345" for record in records)
    assert all({"ts", "level", "event"} <= set(record) for record in records)

    failures = [r for r in records if r["level"] == "error"]
    assert failures, "the failed source must be logged at error level"
    assert failures[0]["source"] == "broken", "the log must name WHICH item failed"


# --------------------------------------------------------------------------
# 6. Rendering
# --------------------------------------------------------------------------


def test_report_renders_what_was_collected(tmp_path, live_fetcher, logger):
    (_, _), state_path = do_run(tmp_path, live_fetcher, logger, ["notes", "links"])
    stored = json.loads(state_path.read_text())
    rendered = core.render_report(stored, limit=3)
    assert "5 entries collected; showing 3" in rendered
    assert "Why stdout beats a log file" in rendered


def test_status_says_never_before_the_first_run():
    text, stale = core.render_status(core.empty_state(), "2026-07-19T10:00:00Z", 3600)
    assert "last success: never" in text
    assert stale is True