Investment Plans workspace
Open raw ↗
# -*- coding: utf-8 -*-
"""
build_source_manifest.py — the session-end source manifest (GOV-D1.13).

OWNER: 01_System/build_source_manifest.py
REVIEW: run at every session-end gate; compare at every session start.

GOV-D1.13: the session-end gate MUST hash every source of truth into a manifest, and the
session-start action MUST compare the live files with it and report, in plain words, every source
that changed since the last gate. "A change nobody gated is a finding, not a surprise."

    python 01_System/build_source_manifest.py --gate      write the manifest (session end)
    python 01_System/build_source_manifest.py --compare   report what changed (session start)

The manifest deliberately covers SOURCES OF TRUTH and DELIVERABLES, not generated intermediates:
a projection that changed because its source changed is not news, and a manifest that reports
everything reports nothing.
"""
import os, sys, json, hashlib, datetime

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
OUT = os.path.join(HERE, "SOURCE_MANIFEST.json")

# (path relative to the project root, what it is, why losing track of it would matter)
TRACKED = [
 ("01_System/project_data.py", "source of truth", "Every register, surface and deliverable is a projection of this file."),
 ("01_System/model_params.py", "source of truth", "The NDIS financial model. Every NDIS figure in every artefact comes from here."),
 ("01_System/model_agedcare.py", "source of truth", "The aged care financial model."),
 ("01_System/trade_study.py", "source of truth", "The scores and weights behind the recommendation."),
 ("01_System/business_plan.json", "source of truth", "The PIL-BZ business plan twin (REQ-BUS-001)."),
 ("01_System/portable/FILE_PLACEMENT_MAP.md", "portable core", "Defines which folder takes which file."),
 ("01_System/portable/templates/business_plan/build_business_plan.py", "portable core", "The business plan generator and its twelve checks."),
 ("01_System/portable/templates/business_plan/business_plan.schema.json", "portable core", "The business plan schema."),
 ("01_System/tidy.py", "portable core", "Placement and debris enforcement."),
 ("01_System/checkers/run_checks.py", "checker", "The gate. If this is altered without a change record, every green result is worthless."),
 ("01_System/build_all.py", "checker", "The build order and the gate invocation."),
 ("01_System/SYSTEM_BREAKDOWN.md", "process drawing", "The parts-and-interfaces drawing GOV-B4.10 requires."),
 ("05_Outputs/AI_Project_Governance_Standard_v3.9.md", "governing file", "The Standard this project obeys."),
 ("05_Outputs/AI_Project_Governance_Standard_v4.0_Business_Layer.md", "governing file", "The business layer."),
 ("05_Outputs/NDIS_and_Aged_Care_Business_Enabling_Study_v2.0.docx", "deliverable", "The study."),
 ("05_Outputs/NDIS_and_Aged_Care_Financial_Model_v2.0.xlsx", "deliverable", "The financial workbook."),
 ("05_Outputs/NDIS_and_Aged_Care_Decision_Pack_v2.0.pptx", "deliverable", "The decision pack."),
 ("01_System/BUSINESS_PLAN_A3.html", "deliverable", "The one-sheet business plan."),
 ("02_Work/scratch/T3_trade_study_weights.md", "evidence", "The weights, timestamped before scoring. If this changed after the scores were set, the trade study is void."),
 ("02_Work/scratch/V5_negative_tests_business_layer.md", "evidence", "Proof that the new checks can fail."),
]


def digest(path):
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(65536), b""):
            h.update(chunk)
    return h.hexdigest()


def snapshot():
    out = {}
    for rel, kind, why in TRACKED:
        full = os.path.join(ROOT, rel)
        if os.path.exists(full):
            out[rel] = {"sha256": digest(full), "bytes": os.path.getsize(full),
                        "kind": kind, "why_it_is_tracked": why}
        else:
            out[rel] = {"sha256": None, "bytes": None, "kind": kind, "why_it_is_tracked": why,
                        "note": "MISSING at the time of this gate"}
    return out


def gate():
    data = {"gated_at": datetime.datetime.now().isoformat(timespec="seconds"),
            "rule": "GOV-D1.13 — a change nobody gated is a finding, not a surprise.",
            "files": snapshot()}
    with open(OUT, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2)
    return data


def compare():
    if not os.path.exists(OUT):
        return None, ["No manifest exists yet, so nothing can be compared. Run --gate to establish one."]
    prev = json.load(open(OUT, encoding="utf-8"))
    now = snapshot()
    lines = []
    for rel, rec in now.items():
        old = prev["files"].get(rel)
        if old is None:
            lines.append("NEW SINCE THE LAST GATE   %s — %s" % (rel, rec["why_it_is_tracked"]))
        elif old["sha256"] != rec["sha256"]:
            if rec["sha256"] is None:
                lines.append("GONE SINCE THE LAST GATE  %s — %s" % (rel, rec["why_it_is_tracked"]))
            elif old["sha256"] is None:
                lines.append("APPEARED SINCE THE GATE   %s — %s" % (rel, rec["why_it_is_tracked"]))
            else:
                lines.append("CHANGED SINCE THE GATE    %s (%s bytes -> %s) — %s"
                             % (rel, old["bytes"], rec["bytes"], rec["why_it_is_tracked"]))
    for rel in prev["files"]:
        if rel not in now:
            lines.append("NO LONGER TRACKED         %s" % rel)
    return prev.get("gated_at"), lines


if __name__ == "__main__":
    args = set(sys.argv[1:]) or {"--compare"}
    if "--gate" in args:
        d = gate()
        n = sum(1 for v in d["files"].values() if v["sha256"])
        missing = [k for k, v in d["files"].items() if not v["sha256"]]
        print("gated %d of %d sources of truth at %s" % (n, len(d["files"]), d["gated_at"]))
        if missing:
            print("  MISSING at this gate: %s" % ", ".join(missing))
    else:
        when, lines = compare()
        if when:
            print("Last gate: %s" % when)
        if not lines:
            print("Nothing has changed since the last gate. Every source of truth matches its recorded hash.")
        else:
            print("%d source(s) changed since the last gate:" % len(lines))
            for l in lines:
                print("  " + l)
        sys.exit(1 if lines else 0)