015 — Working and absent are not the two options

{
  "name": "015-present-and-failing",
  "what_it_is": "A gather step over three sources, where a source that is present and failing is a third outcome rather than an empty one.",
  "the_structural_decision": "Working and absent are not the two options. A capability can be present, resolved, called, and failing — and the island test, which is MSSP's own criterion for a unit, only ever removes things. A unit must declare what it can fail WITH, and a report must keep failed apart from empty.",
  "why_this_one": "The island test proves a unit can be taken away. It says nothing about a unit that is there and broken, and that is the state real applications spend their time in. Five days from the switch, this is a gap in the method's core criterion rather than in an example.",
  "status": "candidate",

  "outcomes": {
    "worked": "returned records and did not fail",
    "empty": "returned nothing and did not fail — a legitimate quiet day",
    "failed": "was called and could not do its job",
    "absent": "not loaded at all — what the island test produces"
  },

  "the_collapse_this_prevents": "failed and empty both produce zero records. Counting records cannot tell them apart, and neither can absent. Three different situations, one number.",

  "sources": {
    "local-files":  {"can_fail_with": ["unreadable-path"]},
    "remote-index": {"can_fail_with": ["unreachable", "timeout"]},
    "archive-dump": {"can_fail_with": ["corrupt-archive"], "note": "the control — it returns zero records and has NOT failed, so `empty` is a real category and not a synonym"}
  },

  "the_finding": "Removing remote-index and breaking it produce the same record total. Archaeology 015 measures the same shape upstream: os.walk on a path that does not exist yields nothing and raises nothing, and a walk whose subdirectory vanishes midway returns a partial result with no exception — the same files whether or not anyone was told.",

  "sets": {
    "FMS": "this file: the four outcomes, what each source can fail with, and the units map",
    "SCL": "what this deployment does when a source fails — fatal, degrade or ignore",
    "SMS": "loading, running, and classifying each outcome into one of four",
    "TMS": "one file per source — each declares what it can fail with, and reaches no sibling set",
    "DMS": "the outcome per source, the total, and whether the run was degraded"
  },

  "units": {"TMS/sources": ["archive_dump.py", "local_files.py", "remote_index.py"]},

  "non_goals": [
    "Retries, backoff or circuit breaking. This is about naming the state, not recovering from it.",
    "Deciding whether a degraded run should be served. SCL is where that lives and this deployment picks one; the example takes no position on which is right.",
    "Claiming the four outcomes are complete. Partial failure — a source that returned some records and then broke — is a fifth and is not modelled here."
  ]
}

{
  "on_source_failure": "degrade",
  "a_degraded_run_must_say_so": true,
  "_note": "island_test.py runs every policy regardless of what this says. Options: fatal, degrade, ignore."
}

"""What this deployment does when a source fails."""
import json
import pathlib

_C = json.loads((pathlib.Path(__file__).parent / "policy.json").read_text(encoding="utf-8"))

on_failure = lambda: _C["on_source_failure"]                          # noqa: E731
must_say_so = lambda: bool(_C["a_degraded_run_must_say_so"])          # noqa: E731

"""Load the sources, run them, and keep the three outcomes apart.

The one rule this example adds: a run has three outcomes per unit, not two.
worked / empty / failed. Collapsing `failed` into `empty` is what makes a
degraded run report as a successful one.
"""
import importlib

SOURCES = ["local_files", "remote_index", "archive_dump"]


def load():
    loaded, problems = {}, []
    for module_name in SOURCES:
        try:
            module = importlib.import_module(f"TMS.sources.{module_name}")
        except ModuleNotFoundError:
            problems.append(f'source "{module_name}" has no module - fail closed')
            continue
        for attribute in ("NAME", "CAN_FAIL_WITH", "collect"):
            if not hasattr(module, attribute):
                problems.append(f"{module_name} does not declare {attribute}")
        if not getattr(module, "CAN_FAIL_WITH", None):
            problems.append(f"{module_name}: CAN_FAIL_WITH is empty - a unit that cannot say "
                            f"what a bad day looks like cannot be reported as degraded")
        loaded[module.NAME] = module
    return loaded, problems


def resolve(name, loaded):
    module = loaded.get(name)
    if module is None:
        return None, f'source "{name}" has no implementation - fail closed (known: {", ".join(sorted(loaded))})'
    return module, None


def gather(loaded, absent=()):
    """Run every source that has not been removed, and classify each outcome."""
    rows = []
    for name in sorted(loaded):
        if name in absent:
            rows.append({"source": name, "outcome": "absent", "records": 0, "failed": None})
            continue
        result = loaded[name].collect()
        outcome = "failed" if result["failed"] else ("worked" if result["records"] else "empty")
        rows.append({"source": name, "outcome": outcome,
                     "records": len(result["records"]), "failed": result["failed"]})
    return rows


def total(rows):
    return sum(row["records"] for row in rows)


def degraded(rows):
    return [row for row in rows if row["outcome"] == "failed"]

"""A source that legitimately has nothing today.

The control. Without a unit that returns zero records and has NOT failed, an
empty result and a failure are the same observation, and the report cannot be
shown to distinguish them.
"""
NAME = "archive-dump"
CAN_FAIL_WITH = ["corrupt-archive"]


def collect():
    return {"records": [], "failed": None}

"""A source that works.

Every unit here declares what it can fail WITH, not merely that it might. The
island test removes units; nothing in the method until now asked a unit what a
bad day looks like.
"""
NAME = "local-files"
CAN_FAIL_WITH = ["unreadable-path"]


def collect():
    return {"records": [{"id": "L-1"}, {"id": "L-2"}], "failed": None}

"""A source that is present and failing.

Not absent — loaded, resolved, called, and returning nothing because the thing
it depends on is unavailable. This is the state the island test cannot produce:
removing this unit and breaking it give the same record count.
"""
NAME = "remote-index"
CAN_FAIL_WITH = ["unreachable", "timeout"]


def collect():
    return {"records": [], "failed": "unreachable"}

"""The outcome per source, and whether the run was degraded."""

MARK = {"worked": "ok ", "empty": "-- ", "failed": "!! ", "absent": "   "}


def outcomes(rows, out):
    out(f"\n  {'source':<15} {'outcome':<9} {'records':<9} failed with")
    for row in rows:
        out(f"  {MARK[row['outcome']]}{row['source']:<12} {row['outcome']:<9} "
            f"{row['records']:<9} {row['failed'] or ''}")


def headline(rows, total, degraded, out, must_say_so):
    """The line a caller reads. It is the whole point that it cannot be just a number."""
    if degraded and must_say_so:
        out(f"\n  {total} record(s) — DEGRADED: {', '.join(r['source'] for r in degraded)} failed")
    else:
        out(f"\n  {total} record(s)")


def collapse(rows, out):
    out("\n  what a count alone would have said:")
    for row in rows:
        if row["records"] == 0:
            out(f"    {row['source']:<15} 0 records — and this is {row['outcome']}")
    out("    three different situations, one number")


def gaps(out):
    out("\n  measurable, not measured here:")
    out("    - how often a real source is present-and-failing rather than absent")
    out("    - what a degraded run costs a caller who served it as complete")
    out("\n  not measurable by this program at all:")
    out("    - whether a degraded run should be served. fatal, degrade and ignore")
    out("      are all defensible; this deployment picks one and says which.")
    out("    - partial failure. A source that returned some records and THEN broke")
    out("      is a fifth outcome and is not modelled — the classifier here would")
    out("      call it `worked`.")

"""The island test, and the state the island test cannot produce.

    python src/island_test.py

Section 2 is the point: the drill this method has always used — remove the unit
and see what happens — and the state next to it that the drill cannot reach.
"""
import json
import pathlib
import re
import sys

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

from DMS import report  # noqa: E402
from SCL import policy  # noqa: E402
from SMS import gather  # noqa: E402

CONTRACT = json.loads((HERE / "FMS" / "contract.json").read_text(encoding="utf-8"))
FAILURES = []


def check(label, ok, detail=""):
    print(f"  {'PASS' if ok else 'FAIL'}  {label}{' - ' + detail if detail else ''}")
    if not ok:
        FAILURES.append(label)


LOADED, PROBLEMS = gather.load()
ROWS = gather.gather(LOADED)
row = lambda name: next(r for r in ROWS if r["source"] == name)  # noqa: E731

print("\n== 1. every source is an island, says what a bad day looks like, and FMS matches the tree")
check("all three sources loaded with no problems", not PROBLEMS, "; ".join(PROBLEMS))
source_dir = HERE / "TMS" / "sources"
files = sorted(f.name for f in source_dir.iterdir()
               if f.suffix == ".py" and f.name != "__init__.py")
for name in files:
    reaches = re.findall(r"^\s*(?:from|import)\s+(\S+)", (source_dir / name).read_text(encoding="utf-8"), re.M)
    check(f"{name} reaches no sibling set", not reaches, ", ".join(reaches) or "no imports at all")
for where, expected in CONTRACT["units"].items():
    on_disk = sorted(f.name for f in (HERE / where).iterdir()
                     if f.suffix == ".py" and f.name != "__init__.py")
    check(f"{where}: FMS declares {len(expected)}, on disk {len(on_disk)}",
          on_disk == sorted(expected), ", ".join(on_disk))
for name, module in sorted(LOADED.items()):
    check(f"{name} declares what it can fail with", bool(module.CAN_FAIL_WITH),
          ", ".join(module.CAN_FAIL_WITH))

print("\n== 2. the drill this method has, and the state it cannot produce")
removed = gather.gather(LOADED, absent={"remote-index"})
broken = ROWS
check("removing remote-index and breaking it give the same total",
      gather.total(removed) == gather.total(broken),
      f"{gather.total(removed)} both ways")
check("and the record count is the only thing they agree on",
      next(r for r in removed if r["source"] == "remote-index")["outcome"]
      != row("remote-index")["outcome"],
      f"absent vs {row('remote-index')['outcome']}")
check("the island test can produce `absent`",
      next(r for r in removed if r["source"] == "remote-index")["outcome"] == "absent")
check("it cannot produce `failed` - removal is not breakage",
      all(r["outcome"] != "failed" for r in removed if r["source"] == "remote-index"),
      "so MSSP's own structural drill has never seen this state")

print("\n== 3. `empty` is a category, not a synonym for `failed`")
check("archive-dump returned zero records", row("archive-dump")["records"] == 0)
check("and it did NOT fail", row("archive-dump")["failed"] is None,
      "the control - without it, zero records and failure are one observation")
check("remote-index also returned zero records", row("remote-index")["records"] == 0)
check("and it DID fail", row("remote-index")["failed"] == "unreachable")
check("so a count cannot separate them, and the outcome field can",
      row("archive-dump")["records"] == row("remote-index")["records"]
      and row("archive-dump")["outcome"] != row("remote-index")["outcome"],
      "0 == 0, empty != failed")

print("\n== 4. a unit that cannot say what a bad day looks like is refused")


class Silent:
    NAME = "silent-source"
    CAN_FAIL_WITH = []

    @staticmethod
    def collect():
        return {"records": [], "failed": None}


problems = []
if not Silent.CAN_FAIL_WITH:
    problems.append("CAN_FAIL_WITH is empty")
check("a source declaring no failure modes is a problem, not a default",
      bool(problems), problems[0])

print("\n== 5. the report says degraded, and it can be made not to")
lines = []
report.headline(ROWS, gather.total(ROWS), gather.degraded(ROWS), lines.append, True)
check("with must_say_so, the headline names the failure",
      "DEGRADED" in lines[-1], lines[-1].strip())
quiet = []
report.headline(ROWS, gather.total(ROWS), gather.degraded(ROWS), quiet.append, False)
check("without it, the same run reports as a plain count",
      "DEGRADED" not in quiet[-1], quiet[-1].strip())
check("and the two differ, which is what makes the first one evidence",
      lines[-1] != quiet[-1])
check("SCL currently requires it", policy.must_say_so(), str(policy.must_say_so()))

print("\n== 6. fail closed")
_, problem = gather.resolve("s3-bucket", LOADED)
check("an unresolvable source stops the run", problem is not None, problem or "resolved anyway")
check("SCL names a policy the report understands",
      policy.on_failure() in {"fatal", "degrade", "ignore"}, policy.on_failure())

print("\n== 7. what this example does not solve")
print("        MEASURABLE, NOT MEASURED")
print("          - how often a real source is present-and-failing rather than absent")
print("          - what a degraded run costs a caller who served it as complete")
print("        NOT MEASURABLE HERE")
print("          - whether a degraded run should be served. fatal, degrade and ignore")
print("            are all defensible; this takes no position on which.")
print("          - PARTIAL failure. A source that returned some records and then broke")
print("            is a fifth outcome, and the classifier here would call it `worked`.")
print("            That is a known hole, not an omission I noticed afterwards.")

print(f"\n{len(FAILURES)} failure(s)" if FAILURES else "\nall checks passed")
for f in FAILURES:
    print(f"  - {f}")
sys.exit(1 if FAILURES else 0)

"""Three sources, four outcomes, and one number that cannot tell them apart.

    python src/main.py            the run under the policy SCL names
    python src/main.py --strict   exit 1 when a source failed and the policy is fatal
    python src/main.py --compare  the same task with remote-index absent, and broken
"""
import json
import pathlib
import sys

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

from DMS import report  # noqa: E402
from SCL import policy  # noqa: E402
from SMS import gather  # noqa: E402

CONTRACT = json.loads((HERE / "FMS" / "contract.json").read_text(encoding="utf-8"))


def main(argv):
    out = lambda line="": sys.stdout.write(line + "\n")  # noqa: E731
    loaded, problems = gather.load()
    if problems:
        for problem in problems:
            out(f"  !! {problem}")
        return 1

    if "--compare" in argv:
        out("\n== the same task, twice: remote-index removed, and remote-index broken")
        removed = gather.gather(loaded, absent={"remote-index"})
        broken = gather.gather(loaded)
        out(f"\n  {'':<24} {'records':<9} outcome for remote-index")
        for label, rows in (("removed (island test)", removed), ("present and failing", broken)):
            row = next(r for r in rows if r["source"] == "remote-index")
            out(f"  {label:<24} {gather.total(rows):<9} {row['outcome']}")
        out("\n  Same total. The island test can produce the first row and not the second,")
        out("  so a method whose only structural drill is removal has never seen the")
        out("  state a running system spends its bad days in.")
        return 0

    rows = gather.gather(loaded)
    degraded = gather.degraded(rows)
    out(f"\n== gather, policy on failure: {policy.on_failure()}")
    report.outcomes(rows, out)
    report.headline(rows, gather.total(rows), degraded, out, policy.must_say_so())
    report.collapse(rows, out)
    report.gaps(out)

    if "--strict" in argv and degraded and policy.on_failure() == "fatal":
        return 1
    return 0


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