017 — Finishing and being complete are two different things

{
  "name": "017-finished-is-not-complete",
  "what_it_is": "A gather step over four sources where every one of them finishes cleanly and they are not all complete.",
  "the_structural_decision": "Finishing and being complete are two axes, and the second one is not observable from outside the unit. So it has to be declared — and example 016 had just established that a fact a unit could get wrong should not be the unit's to state. The resolution is directional: a unit may declare itself INCOMPLETE and may not declare itself COMPLETE. A declaration that can only make the report worse for the declarer is taken on trust; one that makes it better is refused, because nothing outside can check it.",
  "why_this_one": "Example 016 removed `finished` from the unit and named the cost on the board the same day — a source that KNOWS it is about to break, or knows it did not return everything, now has nowhere to say so. This is that cost paid back, and it does not come free: what you get is a floor, not a count.",
  "status": "candidate",

  "the_two_axes": {
    "finished": "observed by the collector — did the call complete without raising (example 016)",
    "complete": "not observable here at all — did the unit return everything behind it"
  },

  "the_completeness_column_has_two_values": {
    "no - declared": "the unit said so, and this example takes that on trust",
    "not known to be otherwise": "nothing said so — which is NOT the same as complete, and the report never renders it as complete"
  },
  "there_is_no_verified_complete_state": "Nothing in this example can put a source in one. A report that offers 'complete' as a value is claiming something it did not measure.",

  "sources": {
    "full-page":        {"records": 3, "declares": null},
    "short-page":       {"records": 2, "declares": null, "note": "the control — the same record count as truncated-page, finishes just as cleanly, and genuinely has nothing more"},
    "truncated-page":   {"records": 2, "declares": "more-after-cursor"},
    "quiet-truncation": {"records": 2, "declares": null, "note": "the limit, kept as a running unit — it IS truncated and says nothing, so it is reported identically to short-page. Island test section 6 asserts that indistinguishability so a change in it goes red."}
  },

  "the_finding": "All four sources finish. None of them fails. The record counts are 3, 2, 2, 2 — and one of the three twos is missing records, one is not, and one is missing records without saying so. Archaeology 017 measures the same thing in zlib: a truncated stream and a complete one decompress to BYTE-IDENTICAL output, and the only field that separates them is `.eof`.",

  "sets": {
    "FMS": "this file: the two axes, the two values completeness may take, what each source declares, and the units map",
    "SCL": "what a declared-incomplete source means in this deployment, and what this deployment cannot refuse",
    "SMS": "running each source, observing `finished`, accepting the self-penalising declaration and refusing the self-serving one",
    "TMS": "one file per source — each declares what it can fail with, and none of them declares itself complete",
    "DMS": "the two axes side by side, and the floor stated as a floor"
  },

  "units": {"TMS/sources": ["full_page.py", "quiet_truncation.py", "short_page.py", "truncated_page.py"]},

  "non_goals": [
    "Detecting a truncated source that says nothing. That is what the floor is a floor because of, and quiet_truncation is kept in the tree so the gap is a running unit rather than a sentence.",
    "Deciding whether a declared-incomplete export should be served. SCL picks one and the example takes no position.",
    "Following the cursor. Resumption is a different problem, and it was already deferred by example 016."
  ]
}

{
  "deployment": "compliance-export",
  "a_declared_incomplete_source_is": "fatal",
  "why": "An export that goes to a regulator may be smaller than expected, but it may not be smaller than expected without saying so. This deployment refuses to publish while any source has declared itself incomplete.",
  "what_this_deployment_cannot_refuse": "A source that is truncated and silent. The policy can only act on what was declared, and the report says so rather than implying the export is complete.",
  "not_a_general_rule": "A search page would serve the truncated result and show a 'more results' control. SCL is where the position lives; the example takes none."
}

"""Deployment policy: what a declared-incomplete source means here."""
import json
import pathlib

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


def is_fatal():
    return POLICY["a_declared_incomplete_source_is"] == "fatal"


def describe():
    return f'{POLICY["deployment"]}: a declared-incomplete source is {POLICY["a_declared_incomplete_source_is"]}'

"""Run the sources, and keep `finished` and `complete` apart.

Example 016 established that `finished` is observed here and never declared by
the unit. Completeness cannot be observed here at all — nothing outside a
source can see the cursor it did not follow.

So this module applies one rule, and it is the whole example:

    a unit may declare itself INCOMPLETE, and may not declare itself COMPLETE.

A declaration that can only make the report worse for the declarer is taken on
trust. One that makes it better is refused, because there is nothing to check
it against. The consequence is that the resulting number is a FLOOR and the
report has to say so.
"""
import importlib

SOURCES = ["full_page", "quiet_truncation", "short_page", "truncated_page"]

# The two values the completeness column may hold. There is no third: nothing
# here can put a source in a "verified complete" state, so the report never
# claims one.
DECLARED_INCOMPLETE = "no - declared"
NOT_KNOWN_OTHERWISE = "not known to be otherwise"


def load(extra=()):
    loaded, problems = {}, []
    for module_name in [*SOURCES, *extra]:
        module = module_name if not isinstance(module_name, str) else (
            importlib.import_module(f"TMS.sources.{module_name}"))
        for attribute in ("NAME", "CAN_FAIL_WITH", "collect"):
            if not hasattr(module, attribute):
                problems.append(f"a source 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")
        # The self-serving direction, refused. A unit vouching for its own
        # completeness is a declaration nothing outside it can check.
        for vouch in ("COMPLETE", "IS_COMPLETE", "RETURNS_EVERYTHING"):
            if getattr(module, vouch, None) is True:
                problems.append(f"{module.NAME}: declares {vouch} - a unit may declare itself "
                                f"incomplete and may not declare itself complete")
        loaded[module.NAME] = module
    return loaded, problems


def run(module, absent=False):
    if absent:
        return {"source": module.NAME, "outcome": "absent", "records": [],
                "finished": None, "completeness": None, "reason": None}

    records, error = [], None
    reason = None
    try:
        result = module.collect()
        for record in result["records"]:
            if not isinstance(record, dict) or record.get("from") != module.NAME:
                raise ValueError(f'a record without a usable "from" reached the collector '
                                 f'(got {record.get("from")!r}, expected {module.NAME!r})')
            records.append(record)
        reason = result.get("incomplete_because")
    except Exception as raised:  # noqa: BLE001 - the collector observes, it does not re-raise
        error = str(raised)

    finished = error is None
    if not finished and records:
        outcome = "partial"
    elif not finished:
        outcome = "failed"
    elif records:
        outcome = "worked"
    else:
        outcome = "empty"

    completeness = DECLARED_INCOMPLETE if reason else NOT_KNOWN_OTHERWISE
    return {"source": module.NAME, "outcome": outcome, "records": records,
            "finished": finished, "completeness": completeness,
            "reason": reason or error}


def run_all(loaded, absent=()):
    return [run(loaded[name], absent=name in absent) for name in sorted(loaded)]


def at_least_incomplete(rows):
    """A FLOOR, never a count.

    Records known to come from a source that said it was incomplete. A source
    that is truncated and silent lands in the same column as a complete one, so
    the true number is greater than or equal to this and nothing here can say
    by how much.
    """
    return sum(len(row["records"]) for row in rows
               if row["completeness"] == DECLARED_INCOMPLETE)

"""A page that returns everything behind it.

It imports nothing and knows nothing about the other sources. Note what it
does NOT do: it does not declare itself complete. No unit in this example is
allowed to, and refusing that is the point of section 4.
"""
NAME = "full-page"
CAN_FAIL_WITH = ["unreadable-page"]


def collect():
    return {
        "records": [{"from": NAME, "id": f"f-{n}"} for n in range(1, 4)],
        "incomplete_because": None,
    }

"""A page that is truncated and says nothing. Deliberately in the example.

This is the limit, kept as a running unit rather than as a sentence in the
README. It is truncated in exactly the way `truncated_page` is, and it declares
nothing — so the collector reports it identically to `short_page`, which is
complete.

Section 6 of the island test ASSERTS that indistinguishability. If that check
ever goes red, the limit changed and the documents are wrong.
"""
NAME = "quiet-truncation"
CAN_FAIL_WITH = ["unreadable-page", "cursor-expired"]

# Kept as data so the test can state what is true of this unit and unobservable
# from outside it. Nothing in SMS reads it.
TRUTH_THE_COLLECTOR_CANNOT_SEE = "there is a cursor after these two records"


def collect():
    return {
        "records": [{"from": NAME, "id": f"q-{n}"} for n in range(1, 3)],
        "incomplete_because": None,
    }

"""The control.

Two records, finishes cleanly, and genuinely has nothing more behind it.
`truncated_page` also returns two records and also finishes cleanly. Without
this file, "two records and finished" and "two records, finished, and there is
more" would be one observation, and section 3 could not come out badly.

Same role as `short-batch` in example 016 and `archive-dump` in example 015.
"""
NAME = "short-page"
CAN_FAIL_WITH = ["unreadable-page"]


def collect():
    return {
        "records": [{"from": NAME, "id": f"s-{n}"} for n in range(1, 3)],
        "incomplete_because": None,
    }

"""A page that finishes cleanly and knows it did not return everything.

Nothing goes wrong here. No exception, no empty result, no early return. The
call completes and the unit holds one fact the collector cannot observe from
outside: there is a cursor it was not allowed to follow.

`incomplete_because` is a declaration, and it is the only kind this example
takes on trust — it can only make the report worse for the unit that made it.
"""
NAME = "truncated-page"
CAN_FAIL_WITH = ["unreadable-page", "cursor-expired"]


def collect():
    return {
        "records": [{"from": NAME, "id": f"t-{n}"} for n in range(1, 3)],
        "incomplete_because": "more-after-cursor",
    }

"""What a person is shown.

The one thing this report may never print is an exact count of incomplete
records, because it does not have one. It has a floor. Section 5 of the island
test checks that the word "at least" is in the sentence, and that no exact
count is offered anywhere near it.
"""


def rows(runs):
    lines = ["  source             outcome   records   finished   complete"]
    mark = {"worked": "ok", "empty": "--", "partial": "~~", "failed": "!!", "absent": "  "}
    for row in runs:
        detail = f' ({row["reason"]})' if row["reason"] else ""
        lines.append(
            f'  {mark[row["outcome"]]} {row["source"]:<16} {row["outcome"]:<9} '
            f'{len(row["records"]):<9} {str(row["finished"]) if row["finished"] is not None else "-":<10} '
            f'{(row["completeness"] or "-")}{detail}')
    return "\n".join(lines)


def floor(at_least, total):
    return "\n".join([
        f"  at least {at_least} of {total} records come from a source that declared itself incomplete.",
        "  This is a FLOOR and not a count: a source that is truncated and says nothing is",
        "  reported in the same column as one that is complete.",
    ])


def two_axes(runs):
    finished = [row for row in runs if row["finished"]]
    lines = ["  what `finished` alone would have said:"]
    for row in finished:
        lines.append(f'    {row["source"]:<16} finished, no error, {len(row["records"])} records')
    lines.append(f"    {len(finished)} sources, all of them clean, and they are not all complete")
    return "\n".join(lines)

"""The island test.

    python src/island_test.py

Section 3 is the control. Section 4 is the rule the example exists for: a unit
may declare itself incomplete and may not declare itself complete. Section 6
asserts what this cannot see, so that a change in the limit goes red.
"""
import json
import pathlib
import re
import sys

from SMS import gather
from SCL import policy
from DMS import report
from TMS.sources import quiet_truncation, short_page, truncated_page

HERE = pathlib.Path(__file__).parent
CONTRACT = json.loads((HERE / "FMS" / "contract.json").read_text(encoding="utf-8"))
FAILURES = []
RAN = [0]


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


loaded, problems = gather.load()
runs = gather.run_all(loaded)
row = {r["source"]: r for r in runs}

print("\n== 1. every source is an island, and FMS matches the tree")
check("loading raised no problems", not problems, "; ".join(problems))
for unit, declared in CONTRACT["units"].items():
    directory = HERE.joinpath(*unit.split("/"))
    on_disk = sorted(p.name for p in directory.glob("*.py") if p.name != "__init__.py")
    check(f"{unit}: FMS declares what is on disk", on_disk == sorted(declared),
          f'disk {", ".join(on_disk)} | FMS {", ".join(sorted(declared))}')
    for name in on_disk:
        body = directory.joinpath(name).read_text(encoding="utf-8")
        siblings = [n[:-3] for n in on_disk if n != name]
        reached = [s for s in siblings if re.search(rf"\bimport\b.*\b{s}\b", body)]
        check(f"{unit}/{name} reaches no sibling", not reached, ", ".join(reached))
        check(f"{unit}/{name} reaches no other set",
              not re.search(r"\b(from|import)\s+(SMS|DMS|SCL|FMS)\b", body))

print("\n== 2. finished and complete are two axes")
check("every source finished", all(r["finished"] for r in runs))
check("and none of them failed", all(r["outcome"] == "worked" for r in runs))
check("so the outcome column separates nothing today",
      len({r["outcome"] for r in runs}) == 1, ", ".join(sorted({r["outcome"] for r in runs})))
check("the completeness column does",
      len({r["completeness"] for r in runs}) == 2,
      " | ".join(sorted({r["completeness"] for r in runs})))
check("truncated-page is the one that declared", row["truncated-page"]["reason"] == "more-after-cursor")
absent_row = gather.run(loaded["truncated-page"], absent=True)
check("removed, it is absent and declares nothing",
      absent_row["outcome"] == "absent" and absent_row["completeness"] is None)

print("\n== 3. the control - the same number, finished just as cleanly")
short, trunc = row["short-page"], row["truncated-page"]
check("short-page returned two records", len(short["records"]) == 2)
check("and it finished", short["finished"] is True)
check("and it is genuinely complete - the control", short["reason"] is None)
check("truncated-page returned two records as well", len(trunc["records"]) == 2)
check("and it finished just as cleanly", trunc["finished"] is True)
check("so neither the count nor `finished` separates them",
      len(short["records"]) == len(trunc["records"]) and short["finished"] == trunc["finished"])
check("only the declaration does",
      short["completeness"] != trunc["completeness"],
      f'{short["completeness"]} vs {trunc["completeness"]}')

print("\n== 4. a unit may declare itself incomplete and may not declare itself complete")


def drill_source(name, **attributes):
    module = type(sys)(name)
    module.NAME = name
    module.CAN_FAIL_WITH = ["x"]
    module.collect = lambda: {"records": [{"from": name, "id": "d-1"}], "incomplete_because": None}
    for key, value in attributes.items():
        setattr(module, key, value)
    return module


for vouch in ("COMPLETE", "IS_COMPLETE", "RETURNS_EVERYTHING"):
    _, drilled = gather.load(extra=[drill_source(f"drill-{vouch.lower()}", **{vouch: True})])
    check(f"DRILL: a source declaring {vouch} is refused",
          any(f"declares {vouch}" in p for p in drilled))
_, honest = gather.load(extra=[drill_source("drill-honest")])
check("and a source that declares neither is accepted", not honest)
_, silent = gather.load(extra=[drill_source("drill-silent", CAN_FAIL_WITH=[])])
check("DRILL: an empty CAN_FAIL_WITH is still refused (improvement 13)",
      any("CAN_FAIL_WITH is empty" in p for p in silent))
check("the self-penalising direction is taken on trust, and it is the only one",
      trunc["completeness"] == gather.DECLARED_INCOMPLETE
      and gather.NOT_KNOWN_OTHERWISE != "complete")
check("there is no verified-complete value anywhere in the vocabulary",
      "complete" not in {gather.DECLARED_INCOMPLETE, gather.NOT_KNOWN_OTHERWISE})

print("\n== 5. the number is a floor and the report says so")
total = sum(len(r["records"]) for r in runs)
at_least = gather.at_least_incomplete(runs)
sentence = report.floor(at_least, total)
check("the floor counts only what was declared", at_least == 2, f"{at_least} of {total}")
check('the report says "at least"', "at least" in sentence)
check('and says the word "FLOOR"', "FLOOR" in sentence)
check("and states why it is not a count", "truncated and says nothing" in sentence)
check("the true number is greater - quiet-truncation is truncated too",
      at_least < at_least + len(row["quiet-truncation"]["records"]))
check("removing the declaring source removes the warning, not the truncation",
      gather.at_least_incomplete(gather.run_all(loaded, absent=["truncated-page"])) == 0)

print("\n== 6. what this cannot see, asserted so it stays measured")
quiet = row["quiet-truncation"]
check("quiet-truncation is truncated in the same way truncated-page is",
      quiet_truncation.TRUTH_THE_COLLECTOR_CANNOT_SEE.startswith("there is a cursor"))
check("it declares nothing", quiet["reason"] is None)
check("and every field the collector reads matches short-page, which IS complete",
      (quiet["outcome"], quiet["finished"], len(quiet["records"]), quiet["completeness"])
      == (short["outcome"], short["finished"], len(short["records"]), short["completeness"]),
      "the limit named in FMS non_goals - if this goes red, the limit changed and the text must too")
check("the two source files differ, so the sameness is in the observation and not the tree",
      truncated_page.collect()["incomplete_because"] != quiet_truncation.collect()["incomplete_because"])
check("SCL cannot refuse it, and says that in words",
      "truncated and silent" in json.loads(
          (HERE / "SCL" / "policy.json").read_text(encoding="utf-8"))["what_this_deployment_cannot_refuse"])
check("and the deployment does refuse the declared one", policy.is_fatal())

print()
if FAILURES:
    print(f'  {len(FAILURES)} FAILED: {" | ".join(FAILURES)}')
    sys.exit(1)
print(f"  {RAN[0]} checks passed - {len(runs)} sources, all finished, not all complete")

"""Four sources, all of them finish, and they are not all complete.

    python src/main.py            the run under the deployment SCL names
    python src/main.py --axes     finished and complete, side by side
    python src/main.py --strict   exit 1 when a source declared itself incomplete and that is fatal
"""
import sys

from DMS import report
from SCL import policy
from SMS import gather


def main(argv):
    loaded, problems = gather.load()
    if problems:
        for problem in problems:
            print(f"  REFUSED: {problem}")
        return 1

    runs = gather.run_all(loaded)
    total = sum(len(row["records"]) for row in runs)
    at_least = gather.at_least_incomplete(runs)

    print(f"\n  {policy.describe()}\n")
    print(report.rows(runs))
    print()
    print(report.floor(at_least, total))

    if "--axes" in argv:
        print()
        print(report.two_axes(runs))
        print()
        print("  removed (island test), for contrast:")
        removed = gather.run_all(loaded, absent=["truncated-page"])
        print(report.rows(removed))
        print(f'\n  at least {gather.at_least_incomplete(removed)} of '
              f'{sum(len(row["records"]) for row in removed)} - removing the one source that '
              f'declared\n  anything takes the warning away and leaves the other truncation behind.')

    declared = [row for row in runs if row["completeness"] == gather.DECLARED_INCOMPLETE]
    if "--strict" in argv and declared and policy.is_fatal():
        print(f"\n  --strict: {len(declared)} source(s) declared themselves incomplete and this "
              f"deployment calls that fatal")
        return 1
    return 0


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