214 lines
10 KiB
Python
214 lines
10 KiB
Python
"""Business-scoped acceptance, separate from reusable implementation receipts."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
LEDGER = ROOT / "docs/current/business-acceptance.json"
|
|
|
|
|
|
def read(path: Path) -> dict:
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
return value if isinstance(value, dict) else {}
|
|
except (OSError, ValueError):
|
|
return {}
|
|
|
|
|
|
def requirements() -> list[dict]:
|
|
rows = list(read(ROOT / "docs/current/feature-capability-registry.json").get("records", []))
|
|
formats = {}
|
|
for row in read(ROOT / "docs/p6/format-coverage-kernel.json").get("records", []):
|
|
if row.get("direction") in {"import", "export"}:
|
|
record_id = f"business-format:{row['module']}:{row['direction']}:{row['extension']}"
|
|
passed = row.get("status") == "passed" or formats.get(record_id, {}).get("status") == "passed"
|
|
formats[record_id] = {**row, "id": record_id, "command": f"{row['direction']} {row['extension']}",
|
|
"sourcePath": row.get("path"), "sourceLine": row.get("line", 1),
|
|
"status": "passed" if passed else row.get("status")}
|
|
rows.extend(formats.values())
|
|
rows.extend({**row, "module": "Web", "command": row["name"]}
|
|
for row in read(ROOT / "docs/current/product-business-backlog.json").get("records", []))
|
|
policy = read(ROOT / "docs/current/execution-policy.json")
|
|
if len(rows) < int(policy.get("minimumOriginalRequirements", 0)):
|
|
raise ValueError("Original acceptance scope was reduced; completion cannot be evaluated")
|
|
supplemental = read(ROOT / "docs/current/business-acceptance-supplement.json").get("records", [])
|
|
known = {row["id"] for row in rows}
|
|
for row in supplemental:
|
|
if (not isinstance(row, dict) or not all(isinstance(row.get(key), str) and row[key]
|
|
for key in ("id", "module", "command", "sourcePath")) or row["id"] in known):
|
|
raise ValueError("Supplemental business requirements must have unique IDs and explicit owners/sources")
|
|
rows.append(row)
|
|
known.add(row["id"])
|
|
if not set(policy.get("requiredSupplementalIds", [])) <= known:
|
|
raise ValueError("A required supplemental business contract is missing")
|
|
return rows
|
|
|
|
|
|
@lru_cache(maxsize=8192)
|
|
def _file_digest(path: Path, signature: tuple) -> str:
|
|
with path.open("rb") as stream:
|
|
value = hashlib.file_digest(stream, "sha256").hexdigest()
|
|
if file_signature(path) != signature:
|
|
raise OSError(f"File changed during evidence hashing: {path}")
|
|
return value
|
|
|
|
|
|
def file_signature(path: Path) -> tuple:
|
|
stat = path.stat()
|
|
return stat.st_dev, stat.st_ino, stat.st_size, stat.st_mtime_ns, stat.st_ctime_ns
|
|
|
|
|
|
def digest(path: Path) -> str:
|
|
if not path.is_file():
|
|
return "missing"
|
|
return _file_digest(path, file_signature(path))
|
|
|
|
|
|
digest.cache_clear = _file_digest.cache_clear
|
|
digest.cache_info = _file_digest.cache_info
|
|
|
|
|
|
def case_dependency_digest(path: Path) -> str:
|
|
if path.parent == ROOT / "docs/business-acceptance" and path.suffix == ".json":
|
|
contract = read(path)
|
|
execution = {key: value for key, value in contract.items() if key not in {"scope", "remainingBoundaries"}}
|
|
return hashlib.sha256(json.dumps(execution, sort_keys=True, ensure_ascii=False).encode()).hexdigest()
|
|
return digest(path)
|
|
|
|
|
|
def evidence_files_current(files: dict, *, case_dependencies: bool = False) -> bool:
|
|
if not isinstance(files, dict):
|
|
return False
|
|
for name, expected in files.items():
|
|
if not isinstance(name, str) or not isinstance(expected, str):
|
|
return False
|
|
path = (ROOT / name).resolve()
|
|
if not path.is_relative_to(ROOT):
|
|
return False
|
|
if not path.is_file():
|
|
if case_dependencies and expected == "missing":
|
|
continue
|
|
return False
|
|
actual = case_dependency_digest(path) if case_dependencies else digest(path)
|
|
if actual != expected:
|
|
return False
|
|
return True
|
|
|
|
|
|
@lru_cache(maxsize=64)
|
|
def module_inputs(module: str) -> dict:
|
|
slug = "".join(character for character in module.lower() if character.isalnum())
|
|
paths = set((ROOT / "web/src").glob(f"{slug}-*.ts*"))
|
|
paths.update((ROOT / "web/src").glob(f"{slug}.*.ts"))
|
|
modules = read(ROOT / "web/public/modules/manifest.json").get("modules", {})
|
|
def owns(path: Path) -> bool:
|
|
matches = [name for name in modules if path.stem.lower().startswith(name.lower())]
|
|
return max(matches, key=len, default=module) == module and path.stem.lower().startswith(slug)
|
|
paths.update(path for path in (ROOT / "freecad-source/src/WebAssembly/FreeCADCoreProbe").glob("*.*") if owns(path))
|
|
artifact = modules.get(module, {})
|
|
for worker in [artifact, *artifact.get("workers", {}).values()] if isinstance(artifact, dict) else []:
|
|
runtime = worker.get("runtimeUrl")
|
|
if isinstance(runtime, str):
|
|
stem = Path(urlsplit(runtime).path).stem.removeprefix("FreeCADWeb")
|
|
source = ROOT / "freecad-source/src/WebAssembly/FreeCADCoreProbe" / f"{stem}.cpp"
|
|
if source.is_file():
|
|
paths.add(source)
|
|
paths.update(ROOT / f"web/src/{name}" for name in (
|
|
"module-router.ts", "native-source-policy.ts", "desktop-command-dispatcher.ts", "facade-envelope.ts"))
|
|
if module == "Web":
|
|
paths.update((ROOT / "web/src").glob("*-shell.tsx"))
|
|
paths.update((ROOT / "web/src").glob("*-desktop-adapter.ts"))
|
|
return {
|
|
"sources": {str(path.relative_to(ROOT)): digest(path) for path in sorted(paths) if path.is_file()},
|
|
"artifact": read(ROOT / "web/public/modules/manifest.json").get("modules", {}).get(module),
|
|
}
|
|
|
|
|
|
def fingerprint(row: dict) -> str:
|
|
route = row.get("route") or {}
|
|
module = (route.get("module") if isinstance(route, dict) else None) or row["module"]
|
|
source = str(row.get("sourcePath") or row.get("source") or "")
|
|
source_path = ROOT / source
|
|
if source and not source_path.is_file():
|
|
source_path = ROOT / "freecad-source" / source
|
|
value = {"id": row["id"], "route": route, "inputs": module_inputs(module), "source": digest(source_path)}
|
|
if row.get("contractDocument"):
|
|
value["acceptanceContract"] = {"document": digest(ROOT / row["contractDocument"]),
|
|
"requiredBehavior": row.get("requiredBehavior", []),
|
|
"requiredProofAreas": row.get("requiredProofAreas", []),
|
|
"independentVerification": row.get("independentVerification"),
|
|
"verifierSha256": digest(ROOT / row["independentVerification"][1]) if row.get("independentVerification") else None}
|
|
return hashlib.sha256(json.dumps(value, sort_keys=True).encode()).hexdigest()
|
|
|
|
|
|
def missing_proof_areas(result: dict) -> list[str]:
|
|
required = next((row.get("requiredProofAreas", []) for row in
|
|
read(ROOT / "docs/current/business-acceptance-supplement.json").get("records", [])
|
|
if row.get("id") == result.get("capabilityId")), [])
|
|
covered = set()
|
|
for case in result.get("cases", []):
|
|
if not isinstance(case, dict):
|
|
continue
|
|
assertions = case.get("assertions", [])
|
|
actual_present = case.get("actual") is not None or any(
|
|
isinstance(item, dict) and item.get("actual") is not None for item in assertions)
|
|
areas = case.get("covers", [])
|
|
if actual_present and isinstance(areas, list) and assertions and all(
|
|
isinstance(item, dict) and item.get("passed") is True for item in assertions):
|
|
covered.update(area for area in areas if isinstance(area, str))
|
|
return sorted(set(required) - covered)
|
|
|
|
|
|
def successful_result(result: dict) -> bool:
|
|
cases = result.get("cases", [])
|
|
return bool(result.get("status") == "passed" and cases and all(
|
|
isinstance(case, dict) and "input" in case and case.get("assertions") and all(
|
|
isinstance(item, dict) and item.get("name") and item.get("passed") is True
|
|
for item in case["assertions"])
|
|
for case in cases)) and not missing_proof_areas(result)
|
|
|
|
|
|
def accepted_ids(rows: list[dict] | None = None) -> set[str]:
|
|
accepted = set()
|
|
evidence_cache = {}
|
|
entries = read(LEDGER).get("records", {})
|
|
for row in rows if rows is not None else requirements():
|
|
if "implementationStatus" in row and row["implementationStatus"] != "routed":
|
|
continue
|
|
entry = entries.get(row["id"], {})
|
|
if entry.get("status") != "passed" or entry.get("fingerprint") != fingerprint(row):
|
|
continue
|
|
path = (ROOT / str(entry.get("evidencePath", ""))).resolve()
|
|
if not path.is_relative_to(ROOT) or not path.is_file() or digest(path) != entry.get("evidenceSha256"):
|
|
continue
|
|
if path not in evidence_cache:
|
|
report = read(path)
|
|
inputs_current = evidence_files_current(report.get("caseInputs", {}).get("files", {}), case_dependencies=True)
|
|
evidence_cache[path] = {item.get("capabilityId"): item for item in report.get("results", [])} if report.get("schema") == "freecad-web-business-cases/v1" and inputs_current else {}
|
|
result = evidence_cache[path].get(row["id"], {})
|
|
if row.get("independentVerification"):
|
|
verification = result.get("independentVerification", {})
|
|
if (verification.get("command") != row["independentVerification"]
|
|
or verification.get("exitCode") != 0
|
|
or verification.get("scriptSha256") != digest(ROOT / row["independentVerification"][1])
|
|
or verification.get("report", {}).get("status") != "passed"):
|
|
continue
|
|
if successful_result(result) and evidence_files_current(result.get("evidenceFiles", {})):
|
|
accepted.add(row["id"])
|
|
return accepted
|
|
|
|
|
|
def business_complete() -> bool:
|
|
from failure_reconciliation import summary_current
|
|
audit = read(ROOT / "docs/current/web-business-completion-audit.json")
|
|
if audit.get("businessFunctionalComplete") is not True:
|
|
return False
|
|
digest.cache_clear()
|
|
module_inputs.cache_clear()
|
|
rows = requirements()
|
|
return bool(rows) and len(accepted_ids(rows)) == len(rows) and summary_current()["complete"]
|