798 lines
40 KiB
Python
798 lines
40 KiB
Python
#!/usr/bin/env python3
|
|
"""P24-02 Operations and Notification Acceptance Evidence Validator.
|
|
|
|
Strict verify-only validator enforcing schema, scope source manifest,
|
|
concrete condition-specific evidence artifacts, OS execution receipts,
|
|
cross-process restart proof, and all 8 required conditions:
|
|
- placement_and_invalid_drop
|
|
- modal_and_pause_resume
|
|
- unit_bubbles
|
|
- event_log
|
|
- important_warning_cutin
|
|
- click_alternatives
|
|
- japanese_input_and_scale
|
|
- final_validation_review
|
|
"""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import importlib.util
|
|
from pathlib import Path
|
|
|
|
# This helper ships in S; it has only stdlib dependencies (no Main-only imports).
|
|
_HELPER = Path(__file__).resolve().parents[1] / 'docs/plans/original-completion-2026-10-02/execution/p24-02/source-ready-repair5-20261017/composition.py'
|
|
_spec = importlib.util.spec_from_file_location('p24_composition', _HELPER)
|
|
composition = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(composition)
|
|
_spec = importlib.util.spec_from_file_location('p24_correspondence', _HELPER.with_name('correspondence.py'))
|
|
correspondence = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(correspondence)
|
|
ROOT = Path(composition.git_bytes(Path(__file__).resolve().parents[1], 'rev-parse', '--show-toplevel').decode().strip())
|
|
VERIFIED_ROOT = ROOT
|
|
|
|
REQUIRED_CONDITIONS = [
|
|
"placement_and_invalid_drop",
|
|
"modal_and_pause_resume",
|
|
"unit_bubbles",
|
|
"event_log",
|
|
"important_warning_cutin",
|
|
"click_alternatives",
|
|
"japanese_input_and_scale",
|
|
"final_validation_review",
|
|
]
|
|
|
|
REQUIRED_SCOPE_FILES = [
|
|
"scenes/main/main.gd",
|
|
"scenes/main/world_preview.gd",
|
|
"scripts/ui/event_log_dialog.gd",
|
|
"scripts/ui/manual_save_slots_dialog.gd",
|
|
"scripts/ui/operation_feedback.gd",
|
|
"scripts/ui/operation_feedback.gd.uid",
|
|
"scripts/ui/operations_camera_controls.gd",
|
|
"scripts/ui/operations_camera_controls.gd.uid",
|
|
"tests/ui/test_operations_notification.gd",
|
|
"tests/ui/test_operations_notification.gd.uid",
|
|
"tests/ui/fixtures/operations_os_observer.gd",
|
|
"tests/ui/fixtures/operations_os_observer.gd.uid",
|
|
"tools/check_p24_operations_evidence.py",
|
|
"tools/run_all_tests.py",
|
|
"tools/regression/suite-policy.json",
|
|
"docs/plans/original-completion-2026-10-02/execution/completion_marker_contracts.json",
|
|
]
|
|
|
|
EXPECTED_TASK_ID = "P24-02"
|
|
EXPECTED_BASE_COMMIT = "1a5abc65163097d4b36a060d68f58c92093eb4e9"
|
|
EXPECTED_SPEC_SHA256 = "738268b406e381ccab415b426d4a547b02e7cb2a481cefce65579a1e346834c7"
|
|
HEX64_PATTERN = re.compile(r"^[0-9a-f]{64}$")
|
|
HEX_COMMIT_PATTERN = re.compile(r"^[0-9a-f]{40}$|^[0-9a-f]{64}$")
|
|
JAPANESE_CHAR_PATTERN = re.compile(r"[\u3040-\u309F\u30A0-\u30FF\u4E00-\u9FFF]")
|
|
MAX_VALID_PID = 4_194_304
|
|
|
|
|
|
def compute_sha256(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def get_actual_git_head(root: Path = None) -> str:
|
|
root = root if root is not None else ROOT
|
|
try:
|
|
composition.root_identity(root, root)
|
|
return composition.git_bytes(root, 'rev-parse', 'HEAD').decode().strip()
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def get_required_suite_roster(root: Path) -> list[str]:
|
|
import importlib.util
|
|
spec = importlib.util.spec_from_file_location("p24_required_roster", root / "tools/run_all_tests.py")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module.roster(root)
|
|
|
|
|
|
def find_index_file(custom_path: str = None) -> Path:
|
|
if custom_path:
|
|
p = Path(custom_path) if Path(custom_path).is_absolute() else ROOT / custom_path
|
|
if p.is_file():
|
|
return p.resolve()
|
|
if p.is_dir():
|
|
candidate = p / "index.json"
|
|
if candidate.is_file():
|
|
return candidate.resolve()
|
|
candidate2 = p / "p24-operations-evidence" / "index.json"
|
|
if candidate2.is_file():
|
|
return candidate2.resolve()
|
|
|
|
build_dir = os.environ.get("BUILD_OUTPUT_DIR")
|
|
if build_dir:
|
|
candidate = Path(build_dir) / "p24-operations-evidence" / "index.json"
|
|
if candidate.is_file():
|
|
return candidate.resolve()
|
|
candidate2 = Path(build_dir) / "index.json"
|
|
if candidate2.is_file():
|
|
return candidate2.resolve()
|
|
|
|
candidates = [
|
|
ROOT / Path(".runtime/P24-02/build/p24-operations-evidence/index.json"),
|
|
ROOT / Path("docs/plans/original-completion-2026-10-02/execution/p24-02/evidence/index.json"),
|
|
ROOT / Path("artifacts/p24-02-operations/index.json"),
|
|
]
|
|
for c in candidates:
|
|
if c.is_file():
|
|
return c.resolve()
|
|
|
|
return None
|
|
|
|
|
|
def _is_bare_or_dummy_artifact(data: dict) -> bool:
|
|
if not isinstance(data, dict):
|
|
return True
|
|
if len(data) <= 1:
|
|
return True
|
|
trivial_keys = {"pid", "status", "passed", "id", "timestamp", "name", "contract_id"}
|
|
if set(data.keys()).issubset(trivial_keys):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _validate_condition_artifact_content(
|
|
cond: str,
|
|
art_data: dict,
|
|
art_type: str,
|
|
evidence_dir: Path,
|
|
root_data: dict,
|
|
validated_artifacts: dict,
|
|
validated_roster: list = None,
|
|
) -> tuple[bool, str]:
|
|
if not isinstance(art_data, dict):
|
|
return False, f"Artifact content for condition '{cond}' must be a JSON object"
|
|
|
|
if _is_bare_or_dummy_artifact(art_data):
|
|
return False, f"Artifact for condition '{cond}' is a bare self-declaration/PID-only JSON without substantive evidence details"
|
|
|
|
if cond == "placement_and_invalid_drop":
|
|
if "grid_snap" not in art_data:
|
|
return False, "Missing 'grid_snap' in placement evidence"
|
|
if "connection_warning" not in art_data:
|
|
return False, "Missing 'connection_warning' in placement evidence"
|
|
if not art_data.get("valid_placement"):
|
|
return False, "Missing 'valid_placement' in placement evidence"
|
|
if not art_data.get("invalid_placement_rejection"):
|
|
return False, "Missing 'invalid_placement_rejection' in placement evidence"
|
|
if not art_data.get("click_assignment"):
|
|
return False, "Missing 'click_assignment' in placement evidence"
|
|
|
|
elif cond == "modal_and_pause_resume":
|
|
if not art_data.get("modal_input_blocked"):
|
|
return False, "Missing 'modal_input_blocked': true in modal/pause evidence"
|
|
if not art_data.get("behind_dialog_suppressed") and not (
|
|
art_data.get("behind_dialog_build_suppressed") and art_data.get("behind_dialog_drop_suppressed")
|
|
):
|
|
return False, "Missing behind-dialog input suppression verification in modal evidence"
|
|
if not art_data.get("clock_pause"):
|
|
return False, "Missing 'clock_pause' (freezing simulation on real delta) in modal/pause evidence"
|
|
if not art_data.get("clock_resume"):
|
|
return False, "Missing 'clock_resume' (advancing exactly once on real delta) in modal/pause evidence"
|
|
if not art_data.get("save_identity_preserved"):
|
|
return False, "Missing 'save_identity_preserved': true in modal/pause evidence"
|
|
|
|
# Separate process restart verification (別process restart・保存再開とPID)
|
|
cross_proc = art_data.get("cross_process", art_data.get("process_restart", root_data.get("cross_process")))
|
|
if not cross_proc or not isinstance(cross_proc, dict):
|
|
return False, "Missing verified separate process restart evidence ('cross_process' or 'process_restart' with distinct producer and consumer PIDs)"
|
|
|
|
producer_pid = cross_proc.get("producer_pid")
|
|
consumer_pid = cross_proc.get("consumer_pid")
|
|
|
|
# Support artifact-linked cross-process structure (P24-03 pattern)
|
|
if (producer_pid is None or consumer_pid is None) and "producer" in cross_proc and "consumer" in cross_proc:
|
|
prod_art_id = cross_proc["producer"]
|
|
cons_art_id = cross_proc["consumer"]
|
|
if prod_art_id in validated_artifacts and cons_art_id in validated_artifacts:
|
|
try:
|
|
prod_json = json.loads(validated_artifacts[prod_art_id]["file"].read_text(encoding="utf-8"))
|
|
cons_json = json.loads(validated_artifacts[cons_art_id]["file"].read_text(encoding="utf-8"))
|
|
producer_pid = prod_json.get("pid")
|
|
consumer_pid = cons_json.get("pid")
|
|
if prod_json.get("manifest_sha256") and cons_json.get("manifest_sha256"):
|
|
if prod_json["manifest_sha256"] != cons_json["manifest_sha256"]:
|
|
return False, "Cross-process restart manifest_sha256 mismatch between producer and consumer"
|
|
if prod_json.get("world_sha256") and cons_json.get("world_sha256"):
|
|
if prod_json["world_sha256"] != cons_json["world_sha256"]:
|
|
return False, "Cross-process restart world_sha256 mismatch between producer and consumer"
|
|
except Exception as e:
|
|
return False, f"Failed to parse cross-process producer/consumer artifacts: {e}"
|
|
|
|
if isinstance(producer_pid, bool) or not isinstance(producer_pid, int) or producer_pid <= 1:
|
|
return False, f"Invalid or missing cross-process producer_pid: {producer_pid!r}"
|
|
if isinstance(consumer_pid, bool) or not isinstance(consumer_pid, int) or consumer_pid <= 1:
|
|
return False, f"Invalid or missing cross-process consumer_pid: {consumer_pid!r}"
|
|
if producer_pid == consumer_pid:
|
|
return False, f"Cross-process restart requires distinct processes; producer and consumer share identical PID {producer_pid}"
|
|
|
|
state_preserved = (
|
|
cross_proc.get("same_manifest_world") is True
|
|
or cross_proc.get("whole_model_and_screen_values_equal") is True
|
|
or cross_proc.get("state_preserved") is True
|
|
or ("manifest_sha256" in cross_proc and "world_sha256" in cross_proc)
|
|
or ("producer" in cross_proc and "consumer" in cross_proc)
|
|
)
|
|
if not state_preserved:
|
|
return False, "Cross-process restart evidence missing proof of save/world consistency across distinct processes"
|
|
|
|
elif cond == "unit_bubbles":
|
|
if not art_data.get("zzz_bubble"):
|
|
return False, "Missing 'zzz_bubble' in unit bubbles evidence"
|
|
if not art_data.get("anger_bubble"):
|
|
return False, "Missing 'anger_bubble' in unit bubbles evidence"
|
|
if not art_data.get("heart_bubble"):
|
|
return False, "Missing 'heart_bubble' with authoritative relationship model in unit bubbles evidence"
|
|
if not art_data.get("stale_removal"):
|
|
return False, "Missing 'stale_removal' state transition clearing in unit bubbles evidence"
|
|
|
|
tracked = art_data.get("bubbles_tracked", art_data.get("bubble_events_count", 0))
|
|
if isinstance(tracked, bool) or not isinstance(tracked, int) or tracked < 3:
|
|
return False, f"Unit bubbles evidence must track at least 3 bubble events, got: {tracked!r}"
|
|
|
|
elif cond == "event_log":
|
|
limit = art_data.get("recent_limit")
|
|
if isinstance(limit, bool) or not isinstance(limit, int) or limit > 256 or limit <= 0:
|
|
return False, f"Event log recent limit must be positive int <= 256, got: {limit!r}"
|
|
if not art_data.get("readable_japanese"):
|
|
return False, "Missing 'readable_japanese': true in event log evidence"
|
|
if not art_data.get("deduplicated"):
|
|
return False, "Missing 'deduplicated': true in event log evidence"
|
|
events = art_data.get("events")
|
|
if not isinstance(events, list) or not events:
|
|
return False, "Event log evidence must include non-empty 'events' list"
|
|
for i, ev in enumerate(events):
|
|
if not isinstance(ev, dict):
|
|
return False, f"Event log row {i} must be a dictionary"
|
|
if not ev.get("event_id") or not ev.get("event_type"):
|
|
return False, f"Event log row {i} missing event_id or event_type"
|
|
title = ev.get("title", "")
|
|
if not isinstance(title, str) or not JAPANESE_CHAR_PATTERN.search(title):
|
|
return False, f"Event log row {i} title must contain readable Japanese, got: {title!r}"
|
|
|
|
elif cond == "important_warning_cutin":
|
|
if not art_data.get("defense_cutin"):
|
|
return False, "Missing 'defense_cutin' in cutin evidence"
|
|
if not art_data.get("shield_vanishing_cutin"):
|
|
return False, "Missing 'shield_vanishing_cutin' in cutin evidence"
|
|
if not art_data.get("simulation_stop") or not art_data.get("simulation_resume"):
|
|
return False, "Missing simulation stop/resume behavior in cutin evidence"
|
|
|
|
elif cond == "click_alternatives":
|
|
for k in ["zoom_in", "zoom_out", "pan", "reset", "grid_toggle", "click_assignment"]:
|
|
if not art_data.get(k):
|
|
return False, f"Missing '{k}' click alternative in click alternatives evidence"
|
|
if not art_data.get("presentation_only"):
|
|
return False, "Missing 'presentation_only': true (no model corruption) in click evidence"
|
|
|
|
elif cond == "japanese_input_and_scale":
|
|
# IME composition verification
|
|
ime_comp = art_data.get("ime_composition")
|
|
if not isinstance(ime_comp, dict):
|
|
return False, "Missing or invalid 'ime_composition' evidence object (must be JSON object with composition details)"
|
|
comp_text = ime_comp.get("text", ime_comp.get("composition_string", ime_comp.get("value", "")))
|
|
if not isinstance(comp_text, str) or not JAPANESE_CHAR_PATTERN.search(comp_text):
|
|
return False, f"'ime_composition' must contain Japanese text, got: {comp_text!r}"
|
|
cursor = ime_comp.get("cursor_position", ime_comp.get("cursor", ime_comp.get("position")))
|
|
if cursor is not None and (isinstance(cursor, bool) or not isinstance(cursor, int) or cursor < 0):
|
|
return False, f"'ime_composition' cursor position must be non-negative integer, got: {cursor!r}"
|
|
|
|
# IME confirm verification
|
|
ime_conf = art_data.get("ime_confirm")
|
|
if not isinstance(ime_conf, dict):
|
|
return False, "Missing or invalid 'ime_confirm' evidence object (must be JSON object with confirmation details)"
|
|
conf_text = ime_conf.get("text", ime_conf.get("confirmed_text", ime_conf.get("value", "")))
|
|
if not isinstance(conf_text, str) or not JAPANESE_CHAR_PATTERN.search(conf_text):
|
|
return False, f"'ime_confirm' must contain confirmed Japanese text, got: {conf_text!r}"
|
|
if not (ime_conf.get("applied") is True or ime_conf.get("confirmed") is True or ime_conf.get("status") in ("confirmed", "applied")):
|
|
return False, "Missing 'applied': true or 'confirmed': true in 'ime_confirm' evidence"
|
|
|
|
# IME cancel verification
|
|
ime_canc = art_data.get("ime_cancel")
|
|
if not isinstance(ime_canc, dict):
|
|
return False, "Missing or invalid 'ime_cancel' evidence object (must be JSON object with cancel details)"
|
|
if not (
|
|
ime_canc.get("canceled") is True
|
|
or ime_canc.get("reverted") is True
|
|
or ime_canc.get("cleared") is True
|
|
or ime_canc.get("status") in ("canceled", "reverted", "cleared")
|
|
):
|
|
return False, "Missing cancel/revert confirmation in 'ime_cancel' evidence"
|
|
|
|
# Cryptographic owned save slot path
|
|
path_str = art_data.get("cryptographic_owned_path")
|
|
if not path_str or not isinstance(path_str, str):
|
|
return False, f"Missing 'cryptographic_owned_path' in Japanese input evidence: {path_str!r}"
|
|
slot_filename = Path(path_str).name
|
|
if not re.match(r"^manual_ui_[0-9a-f]{32}$", slot_filename):
|
|
return False, f"Invalid cryptographic owned save slot filename (must match ^manual_ui_[0-9a-f]{{32}}$): {slot_filename!r}"
|
|
|
|
if not art_data.get("invalid_input_feedback"):
|
|
return False, "Missing 'invalid_input_feedback' in Japanese input evidence"
|
|
|
|
# Window scales inspection list
|
|
window_scales = art_data.get("window_scales")
|
|
if not isinstance(window_scales, list) or not window_scales:
|
|
return False, "Missing 'window_scales' inspection list in scale evidence"
|
|
for i, entry in enumerate(window_scales):
|
|
if not isinstance(entry, dict):
|
|
return False, f"window_scales entry {i} must be a dictionary"
|
|
ws = entry.get("window_size", entry.get("size"))
|
|
if ws is not None:
|
|
if not isinstance(ws, (list, tuple)) or len(ws) != 2:
|
|
return False, f"window_scales entry {i} window_size must be 2-element [w, h]"
|
|
ew, eh = ws
|
|
if isinstance(ew, bool) or isinstance(eh, bool) or not isinstance(ew, int) or not isinstance(eh, int) or ew <= 0 or eh <= 0:
|
|
return False, f"window_scales entry {i} dimensions must be positive integers: {ws!r}"
|
|
readability = entry.get("readability")
|
|
if readability is not True and (not isinstance(readability, str) or not readability.strip()):
|
|
return False, f"window_scales entry {i} lacks verified 'readability': true or legibility description"
|
|
clipping = entry.get("clipping")
|
|
if clipping is not False and clipping not in ("none", "no_clipping"):
|
|
return False, f"window_scales entry {i} has unexpected clipping: {clipping!r} (expected false or 'none')"
|
|
focus = entry.get("focus")
|
|
if focus is not True:
|
|
return False, f"window_scales entry {i} missing 'focus': true"
|
|
sc = entry.get("screenshot")
|
|
if sc:
|
|
if not isinstance(sc, str) or not sc.endswith(".png"):
|
|
return False, f"window_scales entry {i} screenshot must be a .png file: {sc!r}"
|
|
sc_path = evidence_dir / sc if evidence_dir else Path(sc)
|
|
if not sc_path.is_file() or sc_path.stat().st_size == 0:
|
|
return False, f"window_scales entry {i} screenshot missing or empty on disk: {sc!r}"
|
|
|
|
# Windows acceptance remains pending; macOS verified
|
|
if art_data.get("windows_acceptance") is True:
|
|
return False, "Premature claim of Windows acceptance; Windows release acceptance must remain separate pending condition"
|
|
|
|
elif cond == "final_validation_review":
|
|
suite = art_data.get("test_suite")
|
|
if suite != "tests/ui/test_operations_notification.gd":
|
|
return False, f"Unexpected test suite in final validation review: {suite!r}"
|
|
failures = art_data.get("failures")
|
|
if isinstance(failures, bool) or not isinstance(failures, int) or failures != 0:
|
|
return False, f"Test failures must be 0, got: {failures!r}"
|
|
checks = art_data.get("checks")
|
|
if isinstance(checks, bool) or not isinstance(checks, int) or checks <= 0:
|
|
return False, f"Test checks must be positive integer, got: {checks!r}"
|
|
|
|
# Full roster verification (全rosterの実スイート集合/ソース一致/結果を照合する)
|
|
full_roster = art_data.get("full_roster")
|
|
if not full_roster or not isinstance(full_roster, dict):
|
|
# Check if root index provides full_regression dict
|
|
full_roster = root_data.get("full_regression")
|
|
if not full_roster or not isinstance(full_roster, dict):
|
|
return False, "Missing 'full_roster' summary in final validation review evidence"
|
|
|
|
repo_root = VERIFIED_ROOT
|
|
required_suites = get_required_suite_roster(repo_root)
|
|
total_required = len(required_suites)
|
|
|
|
for field in ("total_suites", "suites_count"):
|
|
if field in full_roster and (type(full_roster[field]) is not int or full_roster[field] != total_required):
|
|
return False, f"Full roster {field} must equal {total_required}"
|
|
|
|
roster_failures = full_roster.get("failures")
|
|
if "failures" in full_roster and (type(roster_failures) is not int or roster_failures != 0):
|
|
return False, f"Full roster regression reported failures: {roster_failures!r}"
|
|
|
|
# Check suite entries if provided in full_roster or referenced all-tests artifact
|
|
suite_entries = full_roster.get("results", full_roster.get("rows", full_roster.get("all_tests")))
|
|
if isinstance(suite_entries, str) and suite_entries in validated_artifacts:
|
|
rows_art = validated_artifacts[suite_entries]
|
|
try:
|
|
suite_entries = json.loads(rows_art["file"].read_text(encoding="utf-8"))
|
|
except Exception:
|
|
suite_entries = None
|
|
|
|
if not isinstance(suite_entries, list) or not suite_entries:
|
|
return False, "Full roster requires a non-empty suite list"
|
|
if isinstance(suite_entries, list) and suite_entries:
|
|
seen_suites = set()
|
|
for entry in suite_entries:
|
|
if not isinstance(entry, dict):
|
|
return False, "Full roster row must be a dict"
|
|
s_path = entry.get("test", entry.get("suite", entry.get("path")))
|
|
if not isinstance(s_path, str) or not s_path:
|
|
return False, "Full roster row missing suite path"
|
|
if s_path not in required_suites:
|
|
return False, f"Unknown full roster suite: {s_path}"
|
|
if s_path in seen_suites:
|
|
return False, f"Duplicate full roster suite: {s_path}"
|
|
if s_path:
|
|
seen_suites.add(s_path)
|
|
p = repo_root / s_path
|
|
if not p.is_file():
|
|
return False, f"Full roster suite does not exist on disk: {s_path}"
|
|
expected_sha = entry.get("source_sha256")
|
|
if not isinstance(expected_sha, str) or not HEX64_PATTERN.fullmatch(expected_sha):
|
|
return False, f"Full roster missing or invalid source_sha256: {s_path}"
|
|
if expected_sha:
|
|
actual_sha = compute_sha256(p)
|
|
if actual_sha != expected_sha:
|
|
return False, f"Full roster source SHA256 mismatch for {s_path}: expected {expected_sha}, got {actual_sha}"
|
|
e_fails = entry.get("failures")
|
|
if type(e_fails) is not int or e_fails != 0:
|
|
return False, f"Full roster suite {s_path} failed with {e_fails} failures"
|
|
if entry.get("passed") is not True:
|
|
return False, f"Full roster suite {s_path} passed must be True"
|
|
|
|
missing_from_roster = set(required_suites) - seen_suites
|
|
if seen_suites != set(required_suites):
|
|
return False, f"Full roster execution missing required suites: {sorted(list(missing_from_roster))[:5]}"
|
|
|
|
# Hand the exact successfully validated objects to correspondence; never
|
|
# re-read a different list from the index or a coverage receipt.
|
|
if validated_roster is not None:
|
|
validated_roster.extend(suite_entries)
|
|
|
|
return True, ""
|
|
|
|
|
|
def validate_condition_stage(condition, row, mode):
|
|
correspondence.stage_report(mode)
|
|
if condition == 'final_validation_review' and mode == 'normal_validation':
|
|
correspondence.require(row.get('passed') is False and row.get('status') == 'prerequisites-only',
|
|
'normal_validation final group must be prerequisites-only, passed=false')
|
|
else:
|
|
correspondence.require(row.get('passed') is True, 'condition requires passed=true plus actual evidence: '+condition)
|
|
|
|
|
|
def _verify_evidence(index_path: Path, roles: dict, composition_result: dict) -> bool:
|
|
print(f"Reading evidence index from: {index_path}")
|
|
try:
|
|
data = json.loads(index_path.read_text(encoding="utf-8"))
|
|
except Exception as e:
|
|
print(f"ERROR: Failed to read/parse index JSON: {e}", file=sys.stderr)
|
|
return False
|
|
|
|
if not isinstance(data, dict):
|
|
print("ERROR: Evidence index root must be a JSON object", file=sys.stderr)
|
|
return False
|
|
|
|
# Requirement A: Task ID, base commit, spec SHA256, and candidate version
|
|
task_id = data.get("task_id", data.get("contract_id", data.get("id")))
|
|
if task_id != EXPECTED_TASK_ID:
|
|
print(f"ERROR: Unexpected task ID: {task_id!r} (expected {EXPECTED_TASK_ID!r})", file=sys.stderr)
|
|
return False
|
|
|
|
base_commit = data.get("base_commit", data.get("commit"))
|
|
if base_commit != EXPECTED_BASE_COMMIT:
|
|
print(f"ERROR: Base commit mismatch: {base_commit!r} (expected {EXPECTED_BASE_COMMIT!r})", file=sys.stderr)
|
|
return False
|
|
|
|
# spec_sha256 is strictly mandatory
|
|
spec_sha = data.get("spec_sha256")
|
|
if not spec_sha or not isinstance(spec_sha, str) or spec_sha != EXPECTED_SPEC_SHA256:
|
|
print(f"ERROR: Missing or invalid spec_sha256: {spec_sha!r} (expected {EXPECTED_SPEC_SHA256!r})", file=sys.stderr)
|
|
return False
|
|
|
|
# Candidate commit must be strictly valid 40/64 hex and match actual git HEAD
|
|
candidate_commit = data.get("candidate_commit", data.get("candidate_head"))
|
|
if not candidate_commit or not isinstance(candidate_commit, str):
|
|
print(f"ERROR: Missing candidate_commit in evidence index: {candidate_commit!r}", file=sys.stderr)
|
|
return False
|
|
|
|
candidate_commit_norm = candidate_commit.strip().lower()
|
|
if not HEX_COMMIT_PATTERN.match(candidate_commit_norm):
|
|
print(f"ERROR: candidate_commit must be a valid 40 or 64 character hex SHA, got: {candidate_commit!r}", file=sys.stderr)
|
|
return False
|
|
|
|
if set(candidate_commit_norm) <= {"0"} or set(candidate_commit_norm) <= {"f"}:
|
|
print(f"ERROR: candidate_commit is a dummy placeholder hash: {candidate_commit!r}", file=sys.stderr)
|
|
return False
|
|
|
|
actual_head = get_actual_git_head()
|
|
if not actual_head:
|
|
print("ERROR: Unable to verify real repository git HEAD; git failed or CANDIDATE_COMMIT/TARGET_HEAD mismatched", file=sys.stderr)
|
|
return False
|
|
|
|
if candidate_commit_norm != actual_head:
|
|
print(f"ERROR: Candidate commit mismatch: index specifies {candidate_commit_norm!r}, but verified actual candidate git HEAD is {actual_head!r}", file=sys.stderr)
|
|
return False
|
|
|
|
# Scope source manifest verification: all REQUIRED_SCOPE_FILES must be present in source_hashes
|
|
sources = data.get("source_hashes")
|
|
if not isinstance(sources, dict) or not sources:
|
|
print("ERROR: 'source_hashes' dictionary is required in evidence index", file=sys.stderr)
|
|
return False
|
|
|
|
missing_scope_sources = [f for f in REQUIRED_SCOPE_FILES if f not in sources and f not in data.get("inputs", {})]
|
|
if missing_scope_sources:
|
|
print(f"ERROR: 'source_hashes' omits required scope files: {missing_scope_sources}", file=sys.stderr)
|
|
return False
|
|
|
|
for rel_path, expected_hash in sources.items():
|
|
p = composition.safe_path(VERIFIED_ROOT, rel_path)
|
|
if not p.is_file():
|
|
print(f"ERROR: Required source file does not exist on disk: {rel_path}", file=sys.stderr)
|
|
return False
|
|
|
|
if not isinstance(expected_hash, str) or not HEX64_PATTERN.match(expected_hash) or expected_hash == "0" * 64:
|
|
print(f"ERROR: Invalid or dummy SHA256 format for {rel_path}: {expected_hash!r}", file=sys.stderr)
|
|
return False
|
|
|
|
actual_hash = compute_sha256(p)
|
|
if actual_hash != expected_hash:
|
|
print(f"ERROR: Source hash mismatch for {rel_path}: expected {expected_hash}, got {actual_hash}", file=sys.stderr)
|
|
return False
|
|
|
|
# Requirement B: Artifacts are non-optional and strictly validated
|
|
artifacts = data.get("artifacts")
|
|
if not isinstance(artifacts, dict) or not artifacts:
|
|
print("ERROR: 'artifacts' must be a non-empty JSON object mapping artifact IDs to artifact details", file=sys.stderr)
|
|
return False
|
|
|
|
evidence_dir = index_path.parent
|
|
validated_artifacts = {}
|
|
|
|
for art_id, art_info in artifacts.items():
|
|
if not isinstance(art_info, dict):
|
|
print(f"ERROR: Artifact '{art_id}' must be a dictionary", file=sys.stderr)
|
|
return False
|
|
|
|
rel_art_path = art_info.get("path")
|
|
if not rel_art_path or not isinstance(rel_art_path, str):
|
|
print(f"ERROR: Artifact '{art_id}' missing valid 'path'", file=sys.stderr)
|
|
return False
|
|
|
|
if ".." in rel_art_path or os.path.isabs(rel_art_path):
|
|
print(f"ERROR: Artifact '{art_id}' path traversal or absolute path forbidden: {rel_art_path}", file=sys.stderr)
|
|
return False
|
|
|
|
art_file = composition.safe_path(evidence_dir, rel_art_path)
|
|
if not art_file.is_file():
|
|
art_file = composition.safe_path(ROOT, rel_art_path)
|
|
if not art_file.is_file():
|
|
print(f"ERROR: Referenced artifact file missing on disk: {rel_art_path}", file=sys.stderr)
|
|
return False
|
|
|
|
if art_file.stat().st_size == 0:
|
|
print(f"ERROR: Artifact file is empty: {rel_art_path}", file=sys.stderr)
|
|
return False
|
|
|
|
expected_art_hash = art_info.get("sha256")
|
|
if not expected_art_hash or not isinstance(expected_art_hash, str) or not HEX64_PATTERN.match(expected_art_hash) or expected_art_hash == "0" * 64:
|
|
print(f"ERROR: Invalid or dummy SHA256 for artifact '{art_id}': {expected_art_hash!r}", file=sys.stderr)
|
|
return False
|
|
|
|
actual_art_hash = compute_sha256(art_file)
|
|
if actual_art_hash != expected_art_hash:
|
|
print(f"ERROR: SHA256 mismatch for artifact '{art_id}': expected {expected_art_hash}, got {actual_art_hash}", file=sys.stderr)
|
|
return False
|
|
|
|
art_type = art_info.get("type")
|
|
if not art_type or not isinstance(art_type, str):
|
|
print(f"ERROR: Artifact '{art_id}' missing required 'type'", file=sys.stderr)
|
|
return False
|
|
|
|
validated_artifacts[art_id] = {
|
|
"file": art_file,
|
|
"type": art_type,
|
|
"info": art_info,
|
|
}
|
|
|
|
# Requirement C: OS Proof, PID validation, and binding to actual execution receipt
|
|
os_proof = data.get("os_environment")
|
|
if not isinstance(os_proof, dict) or not os_proof:
|
|
print("ERROR: 'os_environment' is required and cannot be omitted or empty", file=sys.stderr)
|
|
return False
|
|
|
|
pid = os_proof.get("pid")
|
|
if isinstance(pid, bool) or not isinstance(pid, int):
|
|
print(f"ERROR: 'pid' must be a valid integer, not boolean or other type: {pid!r}", file=sys.stderr)
|
|
return False
|
|
|
|
if pid <= 1 or pid > MAX_VALID_PID or pid == 999999999:
|
|
print(f"ERROR: Invalid, dummy, or init/launchd PID: {pid}", file=sys.stderr)
|
|
return False
|
|
|
|
platform_str = os_proof.get("platform")
|
|
if not platform_str or not isinstance(platform_str, str):
|
|
print("ERROR: Missing or invalid OS platform string in os_environment", file=sys.stderr)
|
|
return False
|
|
|
|
window_size = os_proof.get("window_size")
|
|
if not isinstance(window_size, (list, tuple)) or len(window_size) != 2:
|
|
print(f"ERROR: Missing or invalid window_size in os_environment: {window_size!r}", file=sys.stderr)
|
|
return False
|
|
|
|
w, h = window_size
|
|
if isinstance(w, bool) or isinstance(h, bool) or not isinstance(w, int) or not isinstance(h, int) or w <= 0 or h <= 0:
|
|
print(f"ERROR: Window dimensions must be positive integers: {window_size!r}", file=sys.stderr)
|
|
return False
|
|
|
|
receipt_art_id = os_proof.get("receipt_artifact_id")
|
|
if not receipt_art_id or receipt_art_id not in validated_artifacts:
|
|
print(f"ERROR: os_environment lacks verified 'receipt_artifact_id' linking to an artifact: {receipt_art_id!r}", file=sys.stderr)
|
|
return False
|
|
|
|
receipt_art = validated_artifacts[receipt_art_id]
|
|
try:
|
|
receipt_data = json.loads(receipt_art["file"].read_text(encoding="utf-8"))
|
|
except Exception as e:
|
|
print(f"ERROR: Failed to parse execution receipt artifact JSON: {e}", file=sys.stderr)
|
|
return False
|
|
|
|
receipt_pid = receipt_data.get("pid")
|
|
if receipt_pid != pid:
|
|
print(f"ERROR: OS environment PID {pid} does not match receipt artifact PID {receipt_pid}", file=sys.stderr)
|
|
return False
|
|
|
|
receipt_ws = receipt_data.get("window_size")
|
|
if receipt_ws != window_size:
|
|
print(f"ERROR: OS environment window_size {window_size!r} does not match receipt artifact window_size {receipt_ws!r}", file=sys.stderr)
|
|
return False
|
|
|
|
# Optional OS input event verification
|
|
os_input_ref = data.get("os_input")
|
|
if os_input_ref:
|
|
if isinstance(os_input_ref, str) and os_input_ref in validated_artifacts:
|
|
try:
|
|
inp_data = json.loads(validated_artifacts[os_input_ref]["file"].read_text(encoding="utf-8"))
|
|
evs = inp_data.get("events")
|
|
if not isinstance(evs, list) or not evs:
|
|
print("ERROR: OS input artifact missing non-empty 'events' list", file=sys.stderr)
|
|
return False
|
|
for ev in evs:
|
|
if not isinstance(ev, dict) or not ev.get("action") or not ev.get("monotonic_ns"):
|
|
print("ERROR: OS input event missing required fields (action, monotonic_ns)", file=sys.stderr)
|
|
return False
|
|
except Exception as e:
|
|
print(f"ERROR: Failed to validate os_input artifact: {e}", file=sys.stderr)
|
|
return False
|
|
|
|
# Requirement D: Conditions and distinct artifact content verification
|
|
conditions = data.get("conditions")
|
|
if not isinstance(conditions, dict):
|
|
print("ERROR: 'conditions' must be a JSON object mapping condition ID to detailed result object", file=sys.stderr)
|
|
return False
|
|
|
|
expected_set = set(REQUIRED_CONDITIONS)
|
|
actual_set = set(conditions.keys())
|
|
|
|
missing = expected_set - actual_set
|
|
if missing:
|
|
print(f"ERROR: Missing required conditions in evidence index: {sorted(missing)}", file=sys.stderr)
|
|
return False
|
|
|
|
invented = actual_set - expected_set
|
|
if invented:
|
|
print(f"ERROR: Unknown invented conditions present in evidence index: {sorted(invented)}", file=sys.stderr)
|
|
return False
|
|
|
|
validated_roster = []
|
|
condition_to_artifacts = {}
|
|
for cond in REQUIRED_CONDITIONS:
|
|
cond_data = conditions[cond]
|
|
if not isinstance(cond_data, dict):
|
|
print(f"ERROR: Condition '{cond}' must be a structured JSON object with evidence details, not bare boolean or primitive", file=sys.stderr)
|
|
return False
|
|
|
|
try:
|
|
validate_condition_stage(cond, cond_data, roles['mode'])
|
|
except ValueError as exc:
|
|
print(f"ERROR: {exc}", file=sys.stderr)
|
|
return False
|
|
|
|
cond_art_ids = cond_data.get("artifact_ids")
|
|
if not isinstance(cond_art_ids, list) or not cond_art_ids:
|
|
print(f"ERROR: Condition '{cond}' has no referenced 'artifact_ids'; claims without evidence are strictly rejected", file=sys.stderr)
|
|
return False
|
|
|
|
for aid in cond_art_ids:
|
|
if aid not in validated_artifacts:
|
|
print(f"ERROR: Condition '{cond}' references unvalidated or missing artifact ID: {aid!r}", file=sys.stderr)
|
|
return False
|
|
|
|
# Validate substantive details
|
|
details = cond_data.get("details", cond_data.get("evidence"))
|
|
if not details or not isinstance(details, str) or len(details.strip()) < 10:
|
|
print(f"ERROR: Condition '{cond}' lacks substantive verification details", file=sys.stderr)
|
|
return False
|
|
|
|
condition_to_artifacts[cond] = cond_art_ids
|
|
|
|
# Validate content inside the referenced artifacts for this condition
|
|
content_verified = False
|
|
last_err = "No artifact content checked"
|
|
for aid in cond_art_ids:
|
|
art = validated_artifacts[aid]
|
|
try:
|
|
art_content = json.loads(art["file"].read_text(encoding="utf-8"))
|
|
except Exception:
|
|
art_content = {}
|
|
ok, err_msg = _validate_condition_artifact_content(
|
|
cond,
|
|
art_content,
|
|
art["type"],
|
|
evidence_dir,
|
|
data,
|
|
validated_artifacts,
|
|
validated_roster,
|
|
)
|
|
if ok:
|
|
content_verified = True
|
|
break
|
|
else:
|
|
last_err = err_msg
|
|
|
|
if not content_verified:
|
|
print(f"ERROR: Condition '{cond}' artifact content verification failed: {last_err}", file=sys.stderr)
|
|
return False
|
|
|
|
try:
|
|
correspondence.verify(data, validated_artifacts, roles, ROOT,
|
|
validated_roster=validated_roster, composition_result=composition_result)
|
|
except (ValueError, KeyError, TypeError, OSError, IndexError) as exc:
|
|
print(f"ERROR: correspondence: {exc}", file=sys.stderr)
|
|
return False
|
|
|
|
if roles['mode'] == 'normal_validation':
|
|
print("SUCCESS: Source/runtime/OS/full-roster prerequisites verified. P24-02 NOT complete; independent runtime review and integration-final are not-yet-applicable. All 8 remain unaccepted.")
|
|
else:
|
|
print("SUCCESS: All 8 required condition evidence groups including typed review/integration records strictly verified. Independent formal acceptance remains separate.")
|
|
return True
|
|
|
|
|
|
def verify_evidence(index_path: Path) -> bool:
|
|
global VERIFIED_ROOT
|
|
try:
|
|
composition.root_identity(ROOT, Path.cwd())
|
|
data = composition.strict_json(index_path.read_bytes())
|
|
roles = composition.root_context(index_path.parent, data)
|
|
print(json.dumps(correspondence.stage_report(roles.get('mode')), sort_keys=True))
|
|
composition.require(data.get('inputs', {}).get(composition.SPEC) == EXPECTED_SPEC_SHA256, 'frozen spec input SHA mismatch')
|
|
required_helpers = {_HELPER.relative_to(ROOT).as_posix(), _HELPER.with_name('correspondence.py').relative_to(ROOT).as_posix()}
|
|
composition.require(required_helpers <= set(data['inputs']), 'missing checker helper input pins')
|
|
composition_result = composition.verify_composition(ROOT, roles, data.get('source_hashes'), data.get('inputs'),
|
|
EXPECTED_BASE_COMMIT, REQUIRED_SCOPE_FILES + ['docs/plans/original-completion-2026-10-02/execution/p24-02/**'])
|
|
print("Source/composition prerequisites: verified; runtime/OS/roster prerequisites: verification pending; P24-02 NOT complete.")
|
|
with composition.tested_checkout(ROOT, roles['T']) as view:
|
|
VERIFIED_ROOT = view
|
|
verified = _verify_evidence(index_path, roles, composition_result)
|
|
if not verified:
|
|
print("Runtime/OS/roster or applicable later prerequisites: NOT VERIFIED; P24-02 NOT complete.")
|
|
return verified
|
|
except (ValueError, KeyError, TypeError, OSError, IndexError) as exc:
|
|
print(f"ERROR: composition: {exc}", file=sys.stderr)
|
|
return False
|
|
finally:
|
|
VERIFIED_ROOT = ROOT
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Check P24-02 Operations Evidence")
|
|
parser.add_argument("--verify-only", action="store_true", help="Strict verification of existing evidence index only")
|
|
parser.add_argument("--index", type=str, help="Path to evidence index.json", default=None)
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
composition.root_identity(ROOT, Path.cwd())
|
|
except (ValueError, OSError) as exc:
|
|
print(f"ERROR: {exc}", file=sys.stderr)
|
|
return 1
|
|
index_file = find_index_file(args.index)
|
|
if not index_file:
|
|
print("ERROR: Could not locate p24-operations-evidence index.json.", file=sys.stderr)
|
|
print("Ensure BUILD_OUTPUT_DIR is set or provide --index <path>.", file=sys.stderr)
|
|
print("Source/runtime prerequisites NOT VERIFIED; stage not established; P24-02 NOT complete.")
|
|
return 1
|
|
|
|
if not verify_evidence(index_file):
|
|
return 1
|
|
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|