004 — Router: it returns a name, not a module

{
  "name": "router",
  "what_it_is": "Decides which capability should handle a request, without loading or calling it.",
  "why_it_exists": "MSSP defines the Router and gives no implementation pattern, no scale guidance, and no way to test one. 開發區 缺點 3 puts it first on the example roadmap.",
  "core_task_loop": "request -> rule match -> permission check -> a named capability",

  "capabilities": {
    "SMS": {
      "model": "Request and Route. A Route names a capability; it does not hold one.",
      "router": "Rule evaluation, permission check, and the counters that make coverage measurable."
    },
    "SCL": { "policy": "Which actor may use which capability. Checked before a Route is returned, not after." },
    "TMS": {
      "handlers/markdown": "Summarises a markdown document.",
      "handlers/csv": "Validates CSV row widths."
    },
    "DMS": { "ledger": "The decisions, plus rules that never fired and requests nothing matched." }
  },

  "decisions": [
    {
      "id": "D-001",
      "date": "2026-08-04",
      "decision": "the router returns an identifier, never a module",
      "because": "A router that imports what it selects can only be tested with every candidate present, so routing itself has no island test — and the router stops being a subset the moment it holds references to all of them."
    },
    {
      "id": "D-002",
      "date": "2026-08-04",
      "decision": "a refused permission ends the routing decision rather than falling through to the next rule",
      "because": "Falling through would let a caller reach a capability by being denied a better-matching one, which turns a permission check into a preference."
    }
  ],

  "non_goals": [
    "Model-based routing. The scale question this example makes measurable is the input to that decision, not the decision.",
    "Loading. Resolving a name to code is done in main.py in four lines, deliberately outside the router."
  ]
}

{
  "policy_version": "1.0",
  "_comment": [
    "Selecting a capability and being allowed to use it are different questions,",
    "and the router asks the second one before it answers the first. A router",
    "that only matched would let permission be enforced by whoever loads the",
    "result — which means it is enforced by every caller, separately, forever."
  ],
  "actors": {
    "app/editor":  { "capabilities": ["handlers/markdown", "handlers/csv"] },
    "app/importer": { "capabilities": ["handlers/csv"] },
    "anonymous":   { "capabilities": [] }
  },
  "rules": [
    {
      "id": "route-requires-permission",
      "statement": "A route may only name a capability the requesting actor is permitted to use.",
      "enforced_by": "SCL/policy.py::may_use, called by SMS/router.py before a Route is returned"
    }
  ]
}

"""Who may use what.

Read by the router before it returns a decision, so a Route naming a capability
is already a permitted one. The alternative — return the match and let the
loader check — pushes the same check into every caller, and a check that lives
in every caller lives properly in none of them.
"""
from __future__ import annotations

import json
from pathlib import Path

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


def may_use(actor: str, capability: str) -> tuple[bool, str]:
    """Permitted, and the reason when not.

    Returning the reason rather than a bare False is what lets the router put
    "refused, and by what" into the Route — a caller that only learns "no"
    cannot tell a missing permission from a missing rule.
    """
    entry = ACTORS.get(actor)
    if entry is None:
        return False, f"unknown actor {actor!r}"
    if capability not in entry["capabilities"]:
        return False, f"{actor} is not permitted {capability}"
    return True, ""

"""The shapes routing is decided over.

`Route` is the decision this example turns on. It names a capability; it does
not hold one. That single choice is what makes the router testable alone — a
Route can be produced, inspected and asserted about with none of the capabilities
it names present in the process.
"""
from __future__ import annotations

from dataclasses import dataclass, field


@dataclass(frozen=True)
class Request:
    """What the caller asked for."""
    kind: str                 # the artifact: "markdown", "csv", "image"
    intent: str               # what to do with it: "summarise", "validate"
    actor: str = "anonymous"  # who is asking; SCL reads this
    size_kb: int = 0


@dataclass(frozen=True)
class Route:
    """A decision, as a value.

    `capability` is a name. Nothing here imports, loads, or holds the module it
    refers to. Resolving a name to code is the caller's job, and keeping it the
    caller's job is the whole point.
    """
    capability: str | None
    rule: str | None
    why: str
    evidence: list[str] = field(default_factory=list)

    @property
    def matched(self) -> bool:
        return self.capability is not None

"""Choosing a capability for a request.

The router is SMS: remove it and the system cannot decide what to do with
anything, so the loop does not close. It is *not* a dispatcher — it never calls
what it selects, and it imports nothing from TMS.

Rules are data. That matters for two reasons beyond taste:

  - A rule set can be handed in, so the island test hands in one rule and
    asserts on the decision without any capability existing at all.
  - Rules that never fire and requests that match nothing can be counted, which
    is the only honest signal for "the rule set has stopped being enough".
    Without it, that judgement is a feeling, and 開發區 缺點 3 says the method
    currently has nothing better than a feeling.
"""
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
from SMS.model import Request, Route        # noqa: E402


class Rule:
    def __init__(self, name: str, capability: str, *, when) -> None:
        self.name = name
        self.capability = capability
        self._when = when

    def matches(self, request: Request) -> bool:
        return bool(self._when(request))


class Router:
    def __init__(self, rules: list[Rule]) -> None:
        self.rules = rules
        # Counted per rule rather than derived from decisions afterwards: a rule
        # that never fires produces no decision to count, and that is the fact
        # worth surfacing.
        self.fired: dict[str, int] = {rule.name: 0 for rule in rules}
        self.unmatched: list[Request] = []
        self.refused: list[tuple[Request, str]] = []

    def route(self, request: Request) -> Route:
        for rule in self.rules:
            if not rule.matches(request):
                continue

            allowed, why = policy.may_use(request.actor, rule.capability)
            if not allowed:
                # A refusal is a decision, not an absence of one. Falling
                # through to the next rule here would let a caller reach a
                # capability by being denied a better-matching one.
                self.refused.append((request, why))
                return Route(None, rule.name, f"refused: {why}",
                             [f"{request.actor} -> {rule.capability}"])

            self.fired[rule.name] += 1
            return Route(rule.capability, rule.name, "matched",
                         [f"{rule.name} matched {request.kind}/{request.intent}"])

        self.unmatched.append(request)
        return Route(None, None, "no rule matched", [f"{request.kind}/{request.intent}"])

    def coverage(self) -> dict:
        """What this rule set did and did not cover.

        The two numbers 開發區 缺點 3 asks for. `never_fired` says the rule set
        carries weight it is not using; `unmatched` says it is not reaching
        requests that arrive. Both grow as a rule set ages, and both are
        countable, which a judgement about "scale" is not.
        """
        return {
            "rules": len(self.rules),
            "never_fired": sorted(name for name, n in self.fired.items() if n == 0),
            "unmatched": [f"{r.kind}/{r.intent}" for r in self.unmatched],
            "refused": [f"{r.actor}: {why}" for r, why in self.refused],
        }

"""Validates a CSV document: every row the width of the header."""
from __future__ import annotations

name = "handlers/csv"


def handle(text: str) -> str:
    rows = [line.split(",") for line in text.strip().splitlines() if line.strip()]
    if not rows:
        return "empty"
    width = len(rows[0])
    bad = [i for i, row in enumerate(rows[1:], start=2) if len(row) != width]
    return f"{len(rows) - 1} row(s), width {width}, ragged at {bad}" if bad else f"{len(rows) - 1} row(s), all width {width}"

"""Summarises a markdown document.

Loaded only when a Route names it. Nothing imports this file at module scope
except the resolver in main, which is the point: the router decided this was
the right capability without this file being on disk, let alone imported.
"""
from __future__ import annotations

name = "handlers/markdown"


def handle(text: str) -> str:
    headings = [line.lstrip("# ").strip() for line in text.splitlines() if line.startswith("#")]
    return f"{len(headings)} heading(s): " + "; ".join(headings[:3])

"""What the routing did, and what it did not reach.

The second half is the one 開發區 缺點 3 asked for. "When does a rule-based
router stop being enough?" has been a judgement call; these are the two numbers
that make it a measurement:

  never_fired   the rule set is carrying weight it is not using
  unmatched     requests are arriving that no rule reaches

Neither is an error. A young rule set has unmatched requests because it is
young; an old one has never-fired rules because the world moved. What matters is
the direction over time, and you cannot have a direction without a number.
"""
from __future__ import annotations


def render(coverage: dict, decisions: list) -> str:
    lines = ["  decisions"]
    for request, route in decisions:
        target = route.capability or "(none)"
        asked = f"{request.kind}/{request.intent}"
        lines.append(f"    {request.actor:<14} {asked:<20} -> {target:<20} {route.why}")

    lines.append("")
    lines.append("  coverage")
    lines.append(f"    rules             {coverage['rules']}")

    if coverage["never_fired"]:
        lines.append(f"    never fired       {', '.join(coverage['never_fired'])}")
        lines.append("                      the rule set carries weight it did not use")
    else:
        lines.append("    never fired       none - every rule was reached")

    if coverage["unmatched"]:
        lines.append(f"    unmatched         {', '.join(coverage['unmatched'])}")
        lines.append("                      requests arrived that no rule reaches")
    else:
        lines.append("    unmatched         none")

    if coverage["refused"]:
        lines.append(f"    refused           {'; '.join(coverage['refused'])}")
    return "\n".join(lines)

"""The island test for the router itself, plus four ways to break it.

    python src/island-test.py

Section 1 is the one that matters and it is unusual: it exercises the router
with **zero TMS modules importable at all** — the package directory is hidden
from the interpreter for the duration. A router that returned modules could not
survive that line. A router that returns names does not notice.

Sections 3-5 exist because of 改良點 6: every claim here rests on the router
refusing things, and a refusal nobody has watched is not a refusal.
"""
from __future__ import annotations

import sys
from pathlib import Path

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

from SCL.policy import may_use                   # noqa: E402
from SMS.model import Request                    # noqa: E402
from SMS.router import Router, Rule              # noqa: E402

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. the router, with no capability present at all")
{
}
# Nothing under TMS/ has been imported, and for this section nothing can be.
_tms = HERE / "TMS"
_hidden = HERE / "_TMS_hidden_for_the_island_test"
_tms.rename(_hidden)
try:
    rules = [
        Rule("md", "handlers/markdown", when=lambda r: r.kind == "markdown"),
        Rule("csv", "handlers/csv", when=lambda r: r.kind == "csv"),
        Rule("ghost", "handlers/does-not-exist", when=lambda r: r.kind == "ghost"),
    ]
    router = Router(rules)

    decided = router.route(Request(kind="markdown", intent="summarise", actor="app/editor"))
    report("routes with the TMS directory removed from disk", decided.capability == "handlers/markdown",
           f"chose {decided.capability!r}")

    ghost = router.route(Request(kind="ghost", intent="anything", actor="app/editor"))
    report("names a capability that has no file and never will",
           ghost.capability is None or ghost.rule == "ghost",
           f"rule={ghost.rule!r} why={ghost.why!r}")

    imported = [m for m in sys.modules if m.startswith("TMS")]
    report("no TMS module was imported by routing", imported == [], f"imported: {imported}")
finally:
    _hidden.rename(_tms)

print("\n== 2. a refusal is a decision, not a fall-through")
{
}
# app/importer is permitted handlers/csv and nothing else. The specific rule
# names markdown, which it may not use; the catch-all names csv, which it may.
# So a fall-through would hand it a capability by way of being denied a better
# match — which is the shape this section exists to forbid.
router = Router([
    Rule("specific", "handlers/markdown", when=lambda r: r.kind == "markdown"),
    Rule("catch-all", "handlers/csv", when=lambda r: True),
])
denied = router.route(Request(kind="markdown", intent="summarise", actor="app/importer"))
report("a denied match does not fall through to a later rule",
       denied.capability is None and denied.rule == "specific",
       f"capability={denied.capability!r} rule={denied.rule!r}")
report("and the refusal says which permission was missing",
       "not permitted" in denied.why, f'why="{denied.why}"')
report("the caller it WAS permitted stays reachable by its own request",
       Router([Rule("catch-all", "handlers/csv", when=lambda r: True)])
       .route(Request(kind="csv", intent="validate", actor="app/importer")).capability == "handlers/csv",
       "otherwise this section would pass by refusing everything")

print("\n== 3. coverage reports what the rule set did not reach")
{
}
router = Router([
    Rule("used", "handlers/markdown", when=lambda r: r.kind == "markdown"),
    Rule("unused", "handlers/csv", when=lambda r: r.kind == "csv"),
])
router.route(Request(kind="markdown", intent="summarise", actor="app/editor"))
router.route(Request(kind="pdf", intent="extract", actor="app/editor"))
c = router.coverage()
report("a rule that never fired is named", c["never_fired"] == ["unused"], str(c["never_fired"]))
report("a request nothing matched is named", c["unmatched"] == ["pdf/extract"], str(c["unmatched"]))

# and the inverse, or the warning is decoration
full = Router([Rule("used", "handlers/markdown", when=lambda r: True)])
full.route(Request(kind="markdown", intent="summarise", actor="app/editor"))
report("and both go quiet when the rule set covers everything",
       full.coverage()["never_fired"] == [] and full.coverage()["unmatched"] == [])

print("\n== 4. SCL is consulted before a Route exists, not after")
{
}
ok_editor, _ = may_use("app/editor", "handlers/markdown")
ok_anon, why_anon = may_use("anonymous", "handlers/markdown")
report("policy itself distinguishes the two actors", ok_editor and not ok_anon, why_anon)

unknown_ok, why_unknown = may_use("app/nobody", "handlers/markdown")
report("an unknown actor is refused rather than defaulted", not unknown_ok, why_unknown)

routed = Router([Rule("md", "handlers/markdown", when=lambda r: True)]).route(
    Request(kind="markdown", intent="summarise", actor="app/nobody"))
report("and the router never returns a capability the actor may not use",
       routed.capability is None, f"capability={routed.capability!r}")

print("\n== 5. the router holds no reference to anything it names")
{
}
source = (HERE / "SMS" / "router.py").read_text(encoding="utf-8")
# Import lines only. The first version of this check grepped the whole file for
# "TMS" and failed on the docstring sentence saying it imports nothing from TMS
# — a check reading its own documentation and reporting on that.
import_lines = [ln.strip() for ln in source.splitlines()
                if ln.strip().startswith(("import ", "from "))]
report("router.py imports nothing from TMS",
       not any("TMS" in ln for ln in import_lines),
       f"{len(import_lines)} import line(s): {', '.join(import_lines)}")
report("router.py does not import importlib either",
       not any("importlib" in ln for ln in import_lines),
       "resolution belongs to the caller")

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

"""Route some requests, then load only what was selected.

    python src/main.py

The load step is four lines at the bottom and it is not part of the router. That
separation is the example.
"""
from __future__ import annotations

import importlib
import sys
from pathlib import Path

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

from DMS.ledger import render                    # noqa: E402
from SMS.model import Request                    # noqa: E402
from SMS.router import Router, Rule              # noqa: E402

RULES = [
    Rule("markdown-summary", "handlers/markdown",
         when=lambda r: r.kind == "markdown" and r.intent == "summarise"),
    Rule("csv-validate", "handlers/csv",
         when=lambda r: r.kind == "csv" and r.intent == "validate"),
    # Deliberately present and unreachable by the requests below: an ageing rule
    # set accumulates these, and the ledger is what makes that visible.
    Rule("image-thumbnail", "handlers/image",
         when=lambda r: r.kind == "image"),
]

REQUESTS = [
    Request(kind="markdown", intent="summarise", actor="app/editor"),
    Request(kind="csv", intent="validate", actor="app/importer"),
    Request(kind="csv", intent="validate", actor="anonymous"),      # refused
    Request(kind="pdf", intent="extract", actor="app/editor"),      # nothing matches
]

DOCUMENTS = {
    "markdown": "# Title\n\ntext\n\n## Section\n\nmore\n",
    "csv": "a,b,c\n1,2,3\n4,5\n",
}


def main() -> int:
    router = Router(RULES)
    decisions = [(request, router.route(request)) for request in REQUESTS]

    print("\n== router")
    print(render(router.coverage(), decisions))

    print("\n  loading only what was selected")
    for request, route in decisions:
        if not route.matched:
            continue
        # The whole resolution step. It lives here rather than in the router so
        # that the router never holds a reference to a capability.
        module = importlib.import_module("TMS." + route.capability.replace("/", ".").replace("handlers.", "handlers."))
        print(f"    {route.capability:<20} {module.handle(DOCUMENTS[request.kind])}")

    loaded = [m for m in sys.modules if m.startswith("TMS.")]
    print(f"\n  TMS modules in sys.modules: {len(loaded)}")
    print(f"    {', '.join(sorted(loaded))}")

    # The assertion worth making: the router named a capability that was never
    # loaded, which is only possible because a Route is a name.
    named = {r.capability for _, r in decisions if r.matched} | {"handlers/image"}
    unloaded = [c for c in named if "TMS." + c.replace("/", ".") not in sys.modules]
    if "handlers/image" not in unloaded:
        print("\n  EXAMPLE FAILED: handlers/image was loaded, and it does not exist")
        return 1
    print(f"    named but never loaded: {', '.join(sorted(unloaded))}")
    print("\n  the router decided about a capability that has no file at all.")
    return 0


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