Added updated files from dev repo

Author: Wright-Shawn

governance_ledger/cli.py

@@ -16,6 +16,11 @@
     list_contracts,
     show_artifact,
 )
+from governance_ledger.replay import (
+    replay_admissibility,
+    replay_governance_compilation,
+    verify_authority_lineage,
+)
 from governance_ledger.summary import (
     build_pr_summary,
     format_publish_summary,
@@ -69,6 +74,31 @@ def main(argv: Sequence[str] | None = None) -> int:
     show_parser.add_argument("path")
     show_parser.add_argument("--json", action="store_true")
 
+    replay_authority_parser = subparsers.add_parser(
+        "replay-authority",
+        help="replay governance source into compilation report and authority contract",
+    )
+    replay_authority_parser.add_argument("--source", required=True)
+    replay_authority_parser.add_argument("--report")
+    replay_authority_parser.add_argument("--contract")
+    replay_authority_parser.add_argument("--json", action="store_true")
+
+    replay_execution_parser = subparsers.add_parser(
+        "replay-execution",
+        help="replay deterministic admissibility from authority contract and execution state",
+    )
+    replay_execution_parser.add_argument("--contract", required=True)
+    replay_execution_parser.add_argument("--execution-state", required=True)
+    replay_execution_parser.add_argument("--json", action="store_true")
+
+    verify_lineage_parser = subparsers.add_parser(
+        "verify-lineage",
+        help="verify authority provenance lineage against an optional compilation report",
+    )
+    verify_lineage_parser.add_argument("--contract", required=True)
+    verify_lineage_parser.add_argument("--report")
+    verify_lineage_parser.add_argument("--json", action="store_true")
+
     args = parser.parse_args(argv)
 
     if args.command == "run":
@@ -115,11 +145,29 @@ def main(argv: Sequence[str] | None = None) -> int:
         result = check_validation_directory(args.generated_dir)
     elif args.command == "list":
         result = list_contracts(args.contracts_dir)
+    elif args.command == "replay-authority":
+        result = replay_governance_compilation(
+            source_text=Path(args.source).read_text(encoding="utf-8"),
+            expected_report=_read_json_arg(args.report),
+            expected_contract=_read_json_arg(args.contract),
+        )
+    elif args.command == "replay-execution":
+        result = replay_admissibility(
+            authority_contract=_read_json_arg(args.contract),
+            execution_state=_read_json_arg(args.execution_state),
+        )
+    elif args.command == "verify-lineage":
+        result = verify_authority_lineage(
+            authority_contract=_read_json_arg(args.contract),
+            compilation_report=_read_json_arg(args.report),
+        )
     else:
         result = show_artifact(args.path)
 
     if getattr(args, "json", False):
         print(json.dumps(result, indent=2, sort_keys=True))
+        if args.command in {"replay-authority", "replay-execution", "verify-lineage"}:
+            return 0 if _replay_ok(result) else 1
     elif args.command == "run":
         print(format_run_summary(result))
     elif args.command == "publish":
@@ -131,6 +179,9 @@ def main(argv: Sequence[str] | None = None) -> int:
         print(format_contract_list(result))
     elif args.command == "show":
         print(format_artifact(args.path, result))
+    elif args.command in {"replay-authority", "replay-execution", "verify-lineage"}:
+        print(_format_replay_summary(result))
+        return 0 if _replay_ok(result) else 1
     else:
         print(
             "\n".join(
@@ -148,5 +199,37 @@ def main(argv: Sequence[str] | None = None) -> int:
     return 0
 
 
+def _read_json_arg(path: str | None) -> dict | None:
+    if path is None:
+        return None
+    return json.loads(Path(path).read_text(encoding="utf-8"))
+
+
+def _replay_ok(result: dict) -> bool:
+    if "replay_verified" in result:
+        return bool(result["replay_verified"])
+    if "lineage_verified" in result:
+        return bool(result["lineage_verified"])
+    return not result.get("diagnostics")
+
+
+def _format_replay_summary(result: dict) -> str:
+    status = "VERIFIED" if _replay_ok(result) else "FAILED"
+    lines = [
+        "[Governance Replay]",
+        "",
+        f"Status: {status}",
+    ]
+    if result.get("authority_ref"):
+        lines.append(f"Authority: {result['authority_ref']}")
+    if result.get("contract_hash"):
+        lines.append(f"Contract hash: {result['contract_hash']}")
+    diagnostics = result.get("diagnostics") or []
+    if diagnostics:
+        lines.extend(["", "Diagnostics:"])
+        lines.extend(f"  {item['text']}" for item in diagnostics)
+    return "\n".join(lines)
+
+
 if __name__ == "__main__":
     raise SystemExit(main())

governance_ledger/diagnostics.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+GOVERNANCE_DIAGNOSTIC_V1 = "governance_diagnostic.v1"
+
+DIAGNOSTIC_DOMAINS = {
+    "G1": "authority_threshold",
+    "G3": "lifecycle",
+    "G4": "artifact",
+    "G5": "role_integrity",
+    "G7": "governance_lint",
+    "G8": "lineage",
+}
+
+DIAGNOSTIC_SEVERITIES = {"error", "warning", "info"}
+
+
+def build_diagnostic(
+    *,
+    code: str,
+    severity: str,
+    title: str,
+    detail: str,
+    recommendation: str | None = None,
+) -> dict[str, Any]:
+    if severity not in DIAGNOSTIC_SEVERITIES:
+        raise ValueError(f"unsupported diagnostic severity: {severity}")
+    diagnostic = {
+        "schema_version": GOVERNANCE_DIAGNOSTIC_V1,
+        "type": "compiler_diagnostic",
+        "severity": severity,
+        "code": code,
+        "domain": diagnostic_domain(code),
+        "title": title,
+        "detail": detail,
+        "blocks_publication": severity == "error",
+        "text": f"{severity.upper()} {code}: {title}. {detail}",
+    }
+    if recommendation:
+        diagnostic["recommendation"] = recommendation
+    validate_diagnostic(diagnostic)
+    return diagnostic
+
+
+def validate_diagnostic(diagnostic: Any) -> None:
+    if not isinstance(diagnostic, dict):
+        raise ValueError("diagnostic must be an object")
+    if diagnostic.get("schema_version") != GOVERNANCE_DIAGNOSTIC_V1:
+        raise ValueError("unsupported diagnostic schema_version")
+    for field in ["type", "severity", "code", "domain", "title", "detail", "text"]:
+        if not isinstance(diagnostic.get(field), str) or not diagnostic[field]:
+            raise ValueError(f"diagnostic {field} must be a non-empty string")
+    if diagnostic["type"] != "compiler_diagnostic":
+        raise ValueError("diagnostic type must be compiler_diagnostic")
+    if diagnostic["severity"] not in DIAGNOSTIC_SEVERITIES:
+        raise ValueError("unsupported diagnostic severity")
+    if not isinstance(diagnostic.get("blocks_publication"), bool):
+        raise ValueError("diagnostic blocks_publication must be boolean")
+    if diagnostic["blocks_publication"] != (diagnostic["severity"] == "error"):
+        raise ValueError("diagnostic publication impact does not match severity")
+
+
+def diagnostic_domain(code: str) -> str:
+    return DIAGNOSTIC_DOMAINS.get(code[:2], "general")

governance_ledger/extract.py

@@ -2,14 +2,14 @@
 
 from __future__ import annotations
 
-import re
-from copy import deepcopy
 from typing import Any
 
-from governance_ledger.patterns import (
-    ROLE_PATTERNS,
-    SEPARATION_PATTERNS,
-    THRESHOLD_PATTERNS,
+from governance_ledger.statement_normalizer import (
+    normalize_operator,
+    normalize_policy_text,
+    normalize_role,
+    normalize_text,
+    parse_amount,
 )
 
 DEFAULT_CONTRACT_ID = "finance-policy"
@@ -18,86 +18,20 @@
 
 def extract_constraints(text: str) -> dict[str, Any]:
     """Extract v0.1 governance constraints from policy text."""
-    policy: dict[str, Any] = {
-        "contract_id": DEFAULT_CONTRACT_ID,
-        "contract_version": DEFAULT_CONTRACT_VERSION,
-        "authority": {"required_roles": []},
-        "approvals": {"thresholds": []},
-    }
-
-    normalized_text = _normalize_text(text)
-
-    for pattern in ROLE_PATTERNS:
-        for match in re.finditer(pattern, normalized_text, flags=re.IGNORECASE):
-            _append_unique(policy["authority"]["required_roles"], _normalize_role(match["role"]))
-
-    if any(re.search(pattern, normalized_text, flags=re.IGNORECASE) for pattern in SEPARATION_PATTERNS):
-        policy["authority"]["separation_of_duties"] = True
-
-    for pattern in THRESHOLD_PATTERNS:
-        for match in re.finditer(pattern, normalized_text, flags=re.IGNORECASE):
-            requires_role = match.groupdict().get("requires_role")
-            if requires_role:
-                requires_role = _normalize_role(requires_role)
-            elif policy["authority"]["required_roles"]:
-                requires_role = policy["authority"]["required_roles"][0]
-            else:
-                requires_role = "manager"
-            _append_unique_threshold(
-                policy["approvals"]["thresholds"],
-                {
-                    "field": "amount",
-                    "operator": ">",
-                    "value": _parse_amount(
-                        match["amount"],
-                        match.groupdict().get("suffix"),
-                    ),
-                    "requires_role": requires_role,
-                },
-            )
-
-    return _without_empty_sections(policy)
+    return normalize_policy_text(text)
 
 
 def _normalize_text(text: str) -> str:
-    return " ".join(text.strip().split())
+    return normalize_text(text)
 
 
 def _normalize_role(role: str) -> str:
-    normalized = role.strip().lower().replace("-", "_")
-    if normalized.endswith("s") and not normalized.endswith("ss"):
-        return normalized[:-1]
-    return normalized
+    return normalize_role(role)
 
 
 def _parse_amount(amount: str, suffix: str | None = None) -> int:
-    value = float(amount.replace(",", ""))
-    if suffix and suffix.lower() in {"m", "million"}:
-        value *= 1_000_000
-    return int(value)
-
-
-def _append_unique(values: list[str], value: str) -> None:
-    if value and value not in values:
-        values.append(value)
-
-
-def _append_unique_threshold(thresholds: list[dict[str, Any]], threshold: dict[str, Any]) -> None:
-    if threshold not in thresholds:
-        thresholds.append(threshold)
-
-
-def _without_empty_sections(policy: dict[str, Any]) -> dict[str, Any]:
-    cleaned = deepcopy(policy)
+    return parse_amount(amount, suffix)
 
-    if (
-        not cleaned["authority"].get("required_roles")
-        and "separation_of_duties" not in cleaned["authority"]
-    ):
-        cleaned.pop("authority")
-    elif not cleaned["authority"].get("required_roles"):
-        cleaned["authority"].pop("required_roles")
-    if not cleaned["approvals"]["thresholds"]:
-        cleaned.pop("approvals")
 
-    return cleaned
+def _normalize_operator(phrase: str) -> str:
+    return normalize_operator(phrase)