002 — Link checker: promoting a TMS to SMS, and paying for it

{
  "name": "link-checker",
  "what_it_is": "Reads a list of pages, extracts their links, and reports which ones are reachable.",
  "why_it_exists": "To show what happens when a capability declared optional turns out to be the reason the main loop can finish.",
  "core_task_loop": "pages -> links -> verdict per link -> report",

  "capabilities": {
    "SMS": {
      "model": "Link, Verdict, Report. The shapes every other set agrees on.",
      "pacing": "Waits between requests to the same host. Promoted from TMS on 2026-08-02 - see the README.",
      "pipeline": "The loop. Selects a checker per link and collects verdicts."
    },
    "SCL": {
      "policy": "Which capability may reach the network, may wait, and may fail the run."
    },
    "TMS": {
      "checkers/http": "Checks an absolute URL. Reaches the network, so it paces.",
      "checkers/anchor": "Checks an in-page #fragment against the page's own headings. Touches nothing outside the document.",
      "reporters/text": "Renders a report for a terminal."
    },
    "DMS": {
      "trace": "What the run actually did, including how many waits each checker incurred."
    }
  },

  "terminology": {
    "pacing wait": "A delay this run would have taken before contacting a host it has contacted recently.",
    "island test": "Running one TMS with only the minimal core loaded and every other TMS absent."
  },

  "decisions": [
    {
      "id": "D-001",
      "date": "2026-08-02",
      "decision": "pacing moves from TMS to SMS",
      "because": "Without it the run does not finish: the transport begins refusing and links are left with no verdict. A capability the loop cannot close without is core, whatever it was called yesterday.",
      "cost": "Stated in the README. It is not zero and it does not fall on the module that caused the promotion."
    }
  ],

  "non_goals": [
    "Real network I/O. The transport is injected and the clock is fake, so the run is deterministic and offline.",
    "Being a usable link checker. It checks enough to make one structural point."
  ]
}

{
  "policy_version": "1.0",

  "_comment": [
    "Knowledge and permission are separate axes. checkers/anchor knows exactly",
    "how to resolve a fragment and holds no network permission at all, which is",
    "what makes the pacing question below answerable by something other than",
    "an opinion: a capability that may not reach the network cannot need pacing."
  ],

  "capabilities": {
    "checkers/http": {
      "network": true,
      "may_wait": true,
      "may_fail_run": true
    },
    "checkers/anchor": {
      "network": false,
      "may_wait": false,
      "may_fail_run": true
    },
    "reporters/text": {
      "network": false,
      "may_wait": false,
      "may_fail_run": false
    }
  },

  "rules": [
    {
      "id": "pace-requires-network",
      "statement": "A capability with network:false must never incur a pacing wait.",
      "enforced_by": "SCL/policy.py::assert_pacing_is_earned, called by DMS/trace.py at the end of every run"
    },
    {
      "id": "fail-requires-permission",
      "statement": "Only a capability with may_fail_run:true may cause a non-zero exit.",
      "enforced_by": "SCL/policy.py::may_fail_run, called by SMS/pipeline.py"
    }
  ]
}

"""What each capability is allowed to do.

Permission lives here rather than inside the capability that holds it, because a
module that grants itself a permission is describing a hope. `policy.json` is
data a runtime can read; these functions are the checks it can call.

The rule worth the trouble is `pace-requires-network`. It turns "the promotion
should not cost the anchor checker anything" from a design intention into an
assertion that fails a run — see the README's account of what promoting pacing
to SMS actually broke.
"""
from __future__ import annotations

import json
from pathlib import Path

_POLICY = json.loads((Path(__file__).with_name("policy.json")).read_text(encoding="utf-8"))
CAPABILITIES: dict[str, dict] = _POLICY["capabilities"]


class PermissionError_(Exception):
    """Raised when a capability does something its policy entry does not allow."""


def may_use_network(capability: str) -> bool:
    return bool(CAPABILITIES.get(capability, {}).get("network", False))


def may_fail_run(capability: str) -> bool:
    return bool(CAPABILITIES.get(capability, {}).get("may_fail_run", False))


def assert_pacing_is_earned(waits_by_capability: dict[str, int]) -> None:
    """A capability that may not reach the network must not have waited for one.

    This is the check that caught the first attempt at the promotion. Pacing had
    been moved into SMS and the pipeline called it once per link, before dispatch
    — which is the shape that reads cleanest and is wrong, because it charges the
    delay to whichever capability happens to be next rather than to the one that
    is about to open a connection.
    """
    offenders = [
        f"{cap} waited {n}x with network=false"
        for cap, n in sorted(waits_by_capability.items())
        if n and not may_use_network(cap)
    ]
    if offenders:
        raise PermissionError_(
            "SCL rule 'pace-requires-network' violated: " + "; ".join(offenders)
        )

"""The shapes every other set agrees on.

Nothing here reaches outside itself. A TMS binds to these types, not to another
TMS's idea of them, which is what lets a checker be replaced without any other
module being edited.
"""
from __future__ import annotations

from dataclasses import dataclass, field


@dataclass(frozen=True)
class Link:
    """One link found on one page."""
    page: str
    target: str

    @property
    def host(self) -> str | None:
        """The host to pace against, or None when the link never leaves the page.

        This is the whole reason an in-page fragment and an absolute URL can share
        a pipeline: the link itself says whether contacting anyone is involved.
        """
        if self.target.startswith("#"):
            return None
        rest = self.target.split("://", 1)[-1]
        return rest.split("/", 1)[0] or None


@dataclass(frozen=True)
class Verdict:
    link: Link
    ok: bool
    detail: str
    checked_by: str


@dataclass
class Report:
    verdicts: list[Verdict] = field(default_factory=list)
    unchecked: list[Link] = field(default_factory=list)

    @property
    def closed(self) -> bool:
        """Whether the loop finished its job: every link got a verdict.

        The identity test for pacing turns on this property and not on a
        stopwatch. A run that is merely slow is still a link checker; a run that
        leaves links unchecked is not.
        """
        return not self.unchecked

"""Waits between requests to the same host.

**This module was TMS until 2026-08-02.** It moved here because of the identity
test, not because moving it silenced a build error — the build error is what
made anyone look, and looking is what found the reason.

    If pacing is removed, is the system still the system it was?

Run `python src/main.py --without-pacing` and read the answer: the transport
starts refusing, links are left with no verdict, and `Report.closed` is False.
The loop does not close. A capability the loop cannot close without is core.

The delay is recorded rather than slept, so a run is instant and the cost is
still visible. What matters structurally is who is charged for it, and that is
answered by `Link.host` being None for a link that never leaves the page — not
by pacing knowing anything about checkers.
"""
from __future__ import annotations


class Pacer:
    def __init__(self, min_gap_ticks: int = 2) -> None:
        self.min_gap = min_gap_ticks
        self.clock = 0
        self._last_seen: dict[str, int] = {}
        self.waits: dict[str, int] = {}

    def tick(self) -> None:
        self.clock += 1

    def pace(self, host: str | None, *, on_behalf_of: str) -> int:
        """Charge whatever wait this host needs to the capability asking for it.

        `host=None` costs nothing and is recorded as nothing. That is the entire
        mechanism keeping the promotion from spreading to modules that do not
        reach the network — and it works because the caller is the capability
        about to open a connection, not the pipeline dispatching to it.
        """
        if host is None:
            return 0

        last = self._last_seen.get(host)
        waited = 0
        if last is not None and self.clock - last < self.min_gap:
            waited = self.min_gap - (self.clock - last)
            self.clock += waited
            self.waits[on_behalf_of] = self.waits.get(on_behalf_of, 0) + 1

        self._last_seen[host] = self.clock
        return waited

"""The loop: pages in, a verdict per link out.

The pipeline knows the registered checkers by the interface they publish here.
It does not import any of them, and it does not know which ones exist until it
is handed them — which is what makes running one alone possible.

Note what the pipeline does *not* do: it does not call `pace()`. That was the
first shape after the promotion and it is the reason the SCL rule exists. See
`SCL/policy.py::assert_pacing_is_earned`.
"""
from __future__ import annotations

import sys
from pathlib import Path
from typing import Callable, Protocol

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

from SCL import policy                       # noqa: E402
from SMS.model import Link, Report, Verdict  # noqa: E402
from SMS.pacing import Pacer                 # noqa: E402


class Checker(Protocol):
    """What a TMS checker must publish. The only contract between core and TMS."""

    name: str

    def handles(self, link: Link) -> bool: ...

    def check(self, link: Link, pacer: Pacer) -> Verdict: ...


def run(
    pages: dict[str, str],
    checkers: list[Checker],
    *,
    pacer: Pacer | None = None,
    extract: Callable[[str, str], list[Link]],
    on_event: Callable[..., None] = lambda **_: None,
) -> Report:
    report = Report()
    pacer = pacer or Pacer()

    for page, body in pages.items():
        for link in extract(page, body):
            pacer.tick()
            checker = next((c for c in checkers if c.handles(link)), None)
            if checker is None:
                # No verdict is a real outcome, not an error to swallow. It is
                # what `Report.closed` reads to decide whether the loop finished.
                report.unchecked.append(link)
                on_event(kind="unchecked", link=link.target, reason="no checker loaded handles it")
                continue

            verdict = checker.check(link, pacer)
            report.verdicts.append(verdict)
            on_event(kind="verdict", link=link.target, ok=verdict.ok,
                     by=verdict.checked_by, detail=verdict.detail)

    return report


def exit_code(report: Report) -> int:
    """Non-zero only where a capability that is allowed to fail the run found a fault."""
    for verdict in report.verdicts:
        if not verdict.ok and policy.may_fail_run(verdict.checked_by):
            return 1
    return 0 if report.closed else 2

"""Checks an in-page `#fragment` against the headings of the page it sits on.

Nothing here leaves the document. SCL gives it `network: false`, and it is the
module that proves the promotion did not spread: it takes a `Pacer` because the
interface says so, and never calls it, so its wait count stays at zero.

If a later change makes this module wait — most likely by moving `pace()` up
into the pipeline where it reads more tidily — `SCL/policy.py` fails the run.
That check exists because the first version of the promotion did exactly that.
"""
from __future__ import annotations

import re
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from SMS.model import Link, Verdict  # noqa: E402
from SMS.pacing import Pacer         # noqa: E402

name = "checkers/anchor"

_HEADING = re.compile(r"^#{1,6}\s+(.+)$", re.M)


def _slugs(body: str) -> set[str]:
    return {
        re.sub(r"[^a-z0-9]+", "-", h.strip().lower()).strip("-")
        for h in _HEADING.findall(body)
    }


class AnchorChecker:
    name = name

    def __init__(self, pages: dict[str, str]) -> None:
        self._pages = pages

    def handles(self, link: Link) -> bool:
        return link.host is None and link.target.startswith("#")

    def check(self, link: Link, pacer: Pacer) -> Verdict:
        # `pacer` is accepted and deliberately unused: the contract is the same
        # for every checker, and this one has nobody to be polite to.
        del pacer
        wanted = link.target[1:]
        available = _slugs(self._pages.get(link.page, ""))
        ok = wanted in available
        detail = "found" if ok else f"no heading yields #{wanted}"
        return Verdict(link=link, ok=ok, detail=detail, checked_by=self.name)

"""Checks an absolute URL by asking an injected transport.

This is the capability that caused the promotion, and it is the one that pays
for it. It calls `pace()` itself, immediately before contacting the host,
because it is the module that knows a connection is about to open.

The transport is injected rather than imported so the example runs offline and
deterministically. A real one would go here and nothing else would change.
"""
from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from SMS.model import Link, Verdict  # noqa: E402
from SMS.pacing import Pacer         # noqa: E402

name = "checkers/http"


class HttpChecker:
    name = name

    def __init__(self, transport) -> None:
        self._transport = transport

    def handles(self, link: Link) -> bool:
        return link.host is not None

    def check(self, link: Link, pacer: Pacer) -> Verdict:
        waited = pacer.pace(link.host, on_behalf_of=self.name)
        status = self._transport(link.target)
        detail = f"HTTP {status}" + (f", paced {waited}" if waited else "")
        return Verdict(link=link, ok=200 <= status < 400, detail=detail, checked_by=self.name)

"""Renders a report for a terminal.

Reads only SMS types. It never asks which checker produced a verdict in order to
format it differently, which is what keeps adding a checker from being a change
here as well.
"""
from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from SMS.model import Report  # noqa: E402

name = "reporters/text"


def render(report: Report) -> str:
    lines = []
    for v in report.verdicts:
        mark = "ok  " if v.ok else "FAIL"
        lines.append(f"  {mark} {v.link.page} -> {v.link.target}  ({v.detail})")
    for link in report.unchecked:
        lines.append(f"  ????  {link.page} -> {link.target}  (no verdict)")

    failed = sum(1 for v in report.verdicts if not v.ok)
    lines.append("")
    lines.append(
        f"  {len(report.verdicts)} checked, {failed} failed, "
        f"{len(report.unchecked)} left without a verdict"
    )
    lines.append(f"  loop closed: {report.closed}")
    return "\n".join(lines)

"""What the run actually did.

The number this example exists to make visible is the last block: how many
pacing waits each capability incurred. Prose can claim the promotion did not
spread; a count per capability shows it, and `SCL/policy.py` turns the same
number into something that fails a run.
"""
from __future__ import annotations

import sys
from pathlib import Path

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

from SCL import policy  # noqa: E402


class Trace:
    def __init__(self) -> None:
        self.events: list[dict] = []

    def record(self, **event) -> None:
        self.events.append(event)

    def render(self, pacer, checkers) -> str:
        lines = ["  events:"]
        for e in self.events:
            kind = e.pop("kind")
            body = " ".join(f"{k}={v}" for k, v in e.items())
            lines.append(f"    {kind:<10} {body}")

        lines.append("")
        lines.append("  pacing waits by capability:")
        for checker in checkers:
            n = pacer.waits.get(checker.name, 0)
            allowed = "network" if policy.may_use_network(checker.name) else "no network"
            lines.append(f"    {checker.name:<18} {n:>3}   ({allowed})")
        lines.append(f"    {'clock ticks':<18} {pacer.clock:>3}")
        return "\n".join(lines)

"""The island test, plus a check that the SCL rule can fail.

    python src/island-test.py

Four things, in the order they are worth doing:

1. Each TMS runs alone, with the minimal core and no sibling loaded.
2. No TMS imports a sibling TMS (grep, not eyeballs).
3. The promotion changed what "minimal core" means, and this says by how much.
4. The wrong shape is built on purpose and the SCL rule is required to reject it.

Point 4 is the one that matters. `assert_pacing_is_earned` passes every ordinary
run, so on its own it is indistinguishable from a function that returns None.
"""
from __future__ import annotations

import re
import sys
from pathlib import Path

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

from SCL import policy                          # noqa: E402
from SCL.policy import PermissionError_         # noqa: E402
from SMS.model import Link                      # noqa: E402
from SMS.pacing import Pacer                    # noqa: E402
from SMS.pipeline import run                    # noqa: E402

PAGES = {
    "solo.md": (
        "# Solo\n\n## Second heading\n\n"
        "[a](https://one.example.com/a) [b](https://one.example.com/b)\n"
        "[here](#second-heading) [gone](#nowhere)\n"
    )
}


def extract(page: str, body: str) -> list[Link]:
    return [Link(page=page, target=t) for t in re.findall(r"\]\(([^)]+)\)", body)]


failures: list[str] = []


def report(label: str, ok: bool, detail: str = "") -> None:
    print(f"  {'PASS' if ok else 'FAIL'}  {label}" + (f" - {detail}" if detail else ""))
    if not ok:
        failures.append(label)


print("\n== 1. each TMS alone, minimal core, no sibling loaded")

# --- anchor alone ------------------------------------------------------------
from TMS.checkers.anchor import AnchorChecker    # noqa: E402

pacer = Pacer()
anchor_only = run(PAGES, [AnchorChecker(pages=PAGES)], pacer=pacer, extract=extract)
handled = [v for v in anchor_only.verdicts]
report("checkers/anchor runs with no other checker loaded", len(handled) == 2,
       f"{len(handled)} verdict(s), {len(anchor_only.unchecked)} link(s) left to the absent http checker")
report("checkers/anchor incurred no pacing wait", pacer.waits.get("checkers/anchor", 0) == 0,
       f"waits={pacer.waits.get('checkers/anchor', 0)}")
report("the loop reports the links it could not cover", not anchor_only.closed,
       "unchecked links are recorded rather than silently dropped")

# --- http alone --------------------------------------------------------------
from TMS.checkers.http import HttpChecker        # noqa: E402

pacer = Pacer()


def stub(url: str) -> int:
    return 200


http_only = run(PAGES, [HttpChecker(transport=stub)], pacer=pacer, extract=extract)
report("checkers/http runs with no other checker loaded", len(http_only.verdicts) == 2,
       f"{len(http_only.verdicts)} verdict(s)")
report("checkers/http paced itself", pacer.waits.get("checkers/http", 0) > 0,
       f"waits={pacer.waits.get('checkers/http', 0)}")

print("\n== 2. no TMS imports a sibling TMS")
tms_root = HERE / "TMS"
units = sorted(p for p in tms_root.rglob("*.py"))
violations = []
for path in units:
    text = path.read_text(encoding="utf-8")
    for other in units:
        if other == path:
            continue
        module = other.relative_to(tms_root).with_suffix("").as_posix().replace("/", ".")
        if re.search(rf"\bfrom\s+TMS\.{re.escape(module)}\b|\bimport\s+TMS\.{re.escape(module)}\b", text):
            violations.append(f"{path.name} imports {module}")
report("no sibling TMS import", not violations, "; ".join(violations) or f"{len(units)} unit(s) scanned")

print("\n== 3. what the promotion did to 'minimal core'")
core = sorted(p.name for p in (HERE / "SMS").glob("*.py"))
report("pacing is part of the minimal core now", "pacing.py" in core,
       f"SMS = {', '.join(core)} - before 2026-08-02 this was model.py, pipeline.py")

print("\n== 4. the SCL rule can fail")
# The tidy-looking shape: pace once per link in the pipeline, before dispatch.
# It reads better and it charges the wait to whichever capability is next.
pacer = Pacer()
charged_wrongly = {"checkers/anchor": 1}
try:
    policy.assert_pacing_is_earned(charged_wrongly)
    report("SCL rejects a network-free capability that waited", False,
           "it accepted checkers/anchor waiting, so the rule is decoration")
except PermissionError_ as exc:
    report("SCL rejects a network-free capability that waited", True, str(exc)[:78])

# and it must still accept the real shape
try:
    policy.assert_pacing_is_earned({"checkers/http": 3, "checkers/anchor": 0})
    report("SCL accepts the published shape", True)
except PermissionError_ as exc:
    report("SCL accepts the published shape", False, str(exc))

print()
if failures:
    print(f"  {len(failures)} check(s) failed: {', '.join(failures)}")
    raise SystemExit(1)
print("  island test passed")

"""Run the checker.

    python src/main.py                  the structure as published
    python src/main.py --without-pacing the evidence for the promotion

The second one is the point. It removes pacing and shows the loop failing to
close, which is what makes "pacing is core" a measured claim and not a taste.
"""
from __future__ import annotations

import re
import sys
from pathlib import Path

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

from DMS.trace import Trace                       # noqa: E402
from SCL import policy                            # noqa: E402
from SMS.model import Link                        # noqa: E402
from SMS.pacing import Pacer                      # noqa: E402
from SMS.pipeline import exit_code, run           # noqa: E402
from TMS.checkers.anchor import AnchorChecker     # noqa: E402
from TMS.checkers.http import HttpChecker         # noqa: E402
from TMS.reporters.text import render             # noqa: E402

# Consecutive links to the same host, which is the ordinary shape of a docs page
# and the condition under which pacing has to do something. Spread the same
# links out and pacing never fires — which is how the first version of this
# example demonstrated nothing while printing a tidy report.
PAGES = {
    "index.md": (
        "# Getting started\n\n"
        "See [setup](https://docs.example.com/setup), [config](https://docs.example.com/config) "
        "and [tuning](https://docs.example.com/tuning).\n"
        "Jump to [#getting-started](#getting-started) or [#missing](#missing).\n"
    ),
    "guide.md": (
        "# Guide\n\n## Deep dive\n\n"
        "More at [v1](https://api.example.com/v1) and [v2](https://api.example.com/v2).\n"
        "Back to [#deep-dive](#deep-dive).\n"
    ),
}


def extract(page: str, body: str) -> list[Link]:
    return [Link(page=page, target=t) for t in re.findall(r"\]\(([^)]+)\)", body)]


def make_transport(clock, *, min_gap: int = 2):
    """A host that refuses when contacted again too soon.

    It does not know whether pacing exists; it reads the same clock everything
    else reads and answers on the gap it observes. An earlier version took a
    `paced` flag, so the two runs differed because they were told to.
    """
    last: dict[str, int] = {}

    def transport(url: str) -> int:
        host = url.split("://", 1)[-1].split("/", 1)[0]
        now = clock()
        previous, last[host] = last.get(host), now
        if previous is not None and now - previous < min_gap:
            return 429      # refused: no usable verdict for this link
        return 200 if "/v2" not in url else 404

    return transport


def main(argv: list[str]) -> int:
    without_pacing = "--without-pacing" in argv

    class NoPacer(Pacer):
        """What "this project has no pacing" looks like from every other module."""
        def pace(self, host, *, on_behalf_of):
            return 0

    trace = Trace()
    pacer: Pacer = NoPacer() if without_pacing else Pacer()
    checkers = [
        HttpChecker(transport=make_transport(lambda: pacer.clock)),
        AnchorChecker(pages=PAGES),
    ]

    report = run(PAGES, checkers, pacer=pacer, extract=extract, on_event=trace.record)

    # A refusal leaves the link with no usable answer, so it counts as unchecked
    # rather than as a link that is broken. That distinction is what
    # `Report.closed` — and therefore the identity test — turns on.
    for verdict in [v for v in report.verdicts if "429" in v.detail]:
        report.verdicts.remove(verdict)
        report.unchecked.append(verdict.link)

    print(f"\n== link-checker {'WITHOUT pacing' if without_pacing else ''}".rstrip())
    print(render(report))
    print()
    print(trace.render(pacer, checkers))

    if without_pacing:
        if report.closed:
            print("\n  pacing removed and the loop still closed — the promotion is NOT justified.")
            return 1
        print("\n  pacing removed: the loop did not close, so pacing is not optional.")
        return 0

    policy.assert_pacing_is_earned(pacer.waits)     # SCL rule the first promotion broke
    print("\n  SCL rule 'pace-requires-network' holds: no network-free capability waited.")

    # This program's exit code reports whether the *structure* held. Two links in
    # the fixture are broken on purpose, so what a real link checker would return
    # is printed rather than exited with.
    problems = []
    if not report.closed:
        problems.append("the loop did not close with pacing present")
    if not pacer.waits.get("checkers/http"):
        problems.append("pacing never fired, so this run demonstrates nothing about it")

    print(f"  a real run would exit {exit_code(report)} (2 fixture links are broken on purpose)")
    if problems:
        print("\n  STRUCTURE FAILED: " + "; ".join(problems))
        return 1
    return 0


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