#!/usr/bin/env python3 """Verify route artifacts, schemas, hashes, lint verdict, and run status.""" from __future__ import annotations import argparse import sys import unicodedata from collections import Counter from pathlib import Path from typing import Any, Callable from harness_common import ( DEFAULT_CONTRACT_PATH, DEFAULT_RULES_PATH, InputError, decode_utf8, json_text, load_json, load_rules, paths_alias, prepare_report_output, publish_report_json, read_text, require_file, require_regular_nonsymlink, rule_index, run_lock, schema_version, sha256_file, sha256_text, utc_now, validate_with_schema, ) from update_run import transition_manifest, validate_run_history from init_run import choose_route, measure_route_inputs from lint_document import lint as evaluate_lint JSON_ARTIFACTS = { "00_run.json", "01_sources.json", "02_reader_contract.json", "03_evidence_map.json", "04_logic_map.json", "05_term_ledger.json", "08_logic_review.json", "08_reader_review.json", "08_lint.json", "09_final_report.json", } ARTIFACT_SCHEMAS = { "00_run.json": "run.schema.json", "01_sources.json": "sources.schema.json", "02_reader_contract.json": "reader-contract.schema.json", "03_evidence_map.json": "evidence-map.schema.json", "04_logic_map.json": "logic-map.schema.json", "05_term_ledger.json": "term-ledger.schema.json", "08_logic_review.json": "review.schema.json", "08_reader_review.json": "review.schema.json", "08_lint.json": "lint-report.schema.json", "09_final_report.json": "final-report.schema.json", } def check(checks: list[dict[str, Any]], check_id: str, ok: bool, message: str, artifact: str | None = None) -> None: checks.append( { "id": check_id, "status": "pass" if ok else "fail", "message": message, "artifact": artifact, } ) def fields(value: dict[str, Any], names: tuple[str, ...], label: str) -> None: missing = [name for name in names if name not in value] if missing: raise InputError(f"{label} 필수 필드가 없습니다: {', '.join(missing)}") def nonempty_string(value: Any, label: str) -> None: if not isinstance(value, str) or not value.strip(): raise InputError(f"{label}는 비어 있지 않은 문자열이어야 합니다.") def string_list(value: Any, label: str) -> None: if not isinstance(value, list) or any( not isinstance(item, str) or not item.strip() for item in value ): raise InputError(f"{label}는 비어 있지 않은 문자열 배열이어야 합니다.") def normalized_label(value: str) -> str: return unicodedata.normalize("NFKC", " ".join(value.split())).casefold() def validate_run(value: dict[str, Any]) -> None: fields( value, ( "schema_version", "run_id", "mode", "document_kind", "route_requested", "route_hint", "route_reason", "route_metrics", "status", "error", "contract_sha256", "rules_version", "rules_sha256", "omissions", "inputs", "history", ), "00_run.json", ) if value["mode"] not in {"write", "revise", "review"}: raise InputError("00_run.json mode가 잘못되었습니다.") if value["route_hint"] not in {"light", "standard", "deep"}: raise InputError("00_run.json route_hint가 잘못되었습니다.") if value["route_requested"] not in {"auto", "light", "standard", "deep"}: raise InputError("00_run.json route_requested가 잘못되었습니다.") nonempty_string(value["route_reason"], "00_run.json route_reason") metrics = value["route_metrics"] if not isinstance(metrics, dict): raise InputError("00_run.json route_metrics는 객체여야 합니다.") fields( metrics, ("total_chars", "source_count", "total_headings"), "00_run.json route_metrics", ) if any( not isinstance(metrics[name], int) or isinstance(metrics[name], bool) or metrics[name] < 0 for name in ("total_chars", "source_count", "total_headings") ): raise InputError("00_run.json route_metrics 값은 0 이상의 정수여야 합니다.") allowed_status = { "initialized", "evidence_ready", "planned", "drafted", "reviewed", "finalized", "verified", "hold_for_review", "failed", "incomplete", } if value["status"] not in allowed_status: raise InputError("00_run.json status가 잘못되었습니다.") if value["document_kind"] not in {"explanation", "decision", "how-to", "reference"}: raise InputError("00_run.json document_kind가 잘못되었습니다.") if not isinstance(value["omissions"], list): raise InputError("00_run.json omissions는 배열이어야 합니다.") if not isinstance(value["history"], list) or not value["history"]: raise InputError("00_run.json history는 비어 있지 않은 배열이어야 합니다.") validate_run_history(value) inputs = value["inputs"] if not isinstance(inputs, dict): raise InputError("00_run.json inputs는 객체여야 합니다.") fields( inputs, ( "brief", "draft", "source_count", "brief_sha256", "draft_sha256", "sources_manifest_sha256", "input_sha256", ), "00_run.json inputs", ) def validate_sources(value: dict[str, Any]) -> None: fields(value, ("schema_version", "brief", "draft", "sources"), "01_sources.json") if not isinstance(value["sources"], list): raise InputError("01_sources.json sources는 배열이어야 합니다.") values = [value["brief"], value["draft"], *value["sources"]] ids: set[str] = set() for index, item in enumerate(values): if item is None: continue if not isinstance(item, dict): raise InputError(f"01_sources.json inventory[{index}]는 객체여야 합니다.") fields(item, ("id", "path", "sha256"), f"01_sources.json inventory[{index}]") if item["id"] in ids: raise InputError(f"source id가 중복됩니다: {item['id']}") ids.add(item["id"]) if not isinstance(item["sha256"], str) or len(item["sha256"]) != 64: raise InputError(f"source sha256이 잘못되었습니다: {item['id']}") def validate_reader(value: dict[str, Any]) -> None: fields( value, ( "schema_version", "document_kind", "primary_audience", "purpose", "reader_question", "reader_outcome", "prerequisites", "assumed_known", "must_explain", "non_goals", ), "02_reader_contract.json", ) for name in ("primary_audience", "purpose", "reader_question", "reader_outcome"): nonempty_string(value[name], f"reader contract {name}") for name in ("prerequisites", "assumed_known", "must_explain", "non_goals"): string_list(value[name], f"reader contract {name}") def source_ids(sources: dict[str, Any]) -> set[str]: values = [sources.get("brief"), sources.get("draft"), *sources.get("sources", [])] return {item["id"] for item in values if isinstance(item, dict) and isinstance(item.get("id"), str)} def validate_evidence(value: dict[str, Any], known_sources: set[str]) -> None: fields(value, ("schema_version", "claims"), "03_evidence_map.json") if not isinstance(value["claims"], list): raise InputError("evidence claims는 배열이어야 합니다.") claims: dict[str, dict[str, Any]] = {} for index, claim in enumerate(value["claims"]): if not isinstance(claim, dict): raise InputError(f"claim[{index}]는 객체여야 합니다.") fields( claim, ( "id", "statement", "status", "source_ids", "source_locations", "does_not_support", "load_bearing", ), f"claim[{index}]", ) nonempty_string(claim["id"], f"claim[{index}].id") nonempty_string(claim["statement"], f"claim[{index}].statement") if claim["id"] in claims: raise InputError(f"claim id가 중복됩니다: {claim['id']}") if claim["status"] not in { "source_backed", "observed", "measured", "derived", "recommended", "assumption", }: raise InputError(f"claim status가 잘못되었습니다: {claim['id']}") string_list(claim["source_ids"], f"claim {claim['id']}.source_ids") string_list(claim["does_not_support"], f"claim {claim['id']}.does_not_support") locations = claim["source_locations"] if not isinstance(locations, list): raise InputError(f"claim {claim['id']}.source_locations는 배열이어야 합니다.") location_pairs: set[tuple[str, str]] = set() location_source_ids: set[str] = set() for location_index, location in enumerate(locations): if not isinstance(location, dict): raise InputError( f"claim {claim['id']}.source_locations[{location_index}]는 객체여야 합니다." ) fields( location, ("source_id", "locator"), f"claim {claim['id']}.source_locations[{location_index}]", ) nonempty_string( location["source_id"], f"claim {claim['id']}.source_locations[{location_index}].source_id", ) nonempty_string( location["locator"], f"claim {claim['id']}.source_locations[{location_index}].locator", ) pair = (location["source_id"], location["locator"]) if pair in location_pairs: raise InputError(f"claim {claim['id']}의 source locator가 중복됩니다: {pair}") location_pairs.add(pair) location_source_ids.add(location["source_id"]) if not isinstance(claim["load_bearing"], bool): raise InputError(f"claim {claim['id']}.load_bearing은 boolean이어야 합니다.") unknown = set(claim["source_ids"]) - known_sources unknown_locations = location_source_ids - known_sources if unknown or unknown_locations: raise InputError( f"claim {claim['id']}가 모르는 source id를 참조합니다: " f"{sorted(unknown | unknown_locations)}" ) if location_source_ids != set(claim["source_ids"]): raise InputError( f"claim {claim['id']}의 source_ids와 source_locations가 일치하지 않습니다." ) claims[claim["id"]] = claim for claim in claims.values(): status = claim["status"] load_bearing = claim["load_bearing"] if status in {"source_backed", "observed", "measured"} and ( not claim["source_ids"] or not claim["source_locations"] ): raise InputError( f"{status} claim {claim['id']}에는 source_ids와 source_locations가 필요합니다." ) if load_bearing and status in {"source_backed", "observed", "measured", "derived"} and not claim["does_not_support"]: raise InputError(f"load-bearing fact claim {claim['id']}에는 does_not_support가 필요합니다.") if status == "measured": for field_name in ("method", "environment", "result"): nonempty_string(claim.get(field_name), f"measured claim {claim['id']}.{field_name}") if not claim["does_not_support"]: raise InputError(f"measured claim {claim['id']}에는 does_not_support가 필요합니다.") if status == "derived": premise_ids = claim.get("premise_ids") string_list(premise_ids, f"derived claim {claim['id']}.premise_ids") if not premise_ids: raise InputError(f"derived claim {claim['id']}에는 premise_ids가 필요합니다.") unknown = set(premise_ids) - set(claims) if unknown or claim["id"] in premise_ids: raise InputError(f"derived claim {claim['id']}의 premise_ids가 잘못되었습니다.") if status in {"recommended", "assumption"}: nonempty_string(claim.get("label"), f"claim {claim['id']}.label") visiting: set[str] = set() visited: set[str] = set() def visit(claim_id: str, trail: list[str]) -> None: if claim_id in visiting: cycle_start = trail.index(claim_id) if claim_id in trail else 0 cycle = [*trail[cycle_start:], claim_id] raise InputError(f"derived claim premise cycle이 있습니다: {' -> '.join(cycle)}") if claim_id in visited: return visiting.add(claim_id) trail.append(claim_id) claim = claims[claim_id] if claim["status"] == "derived": for premise_id in claim.get("premise_ids", []): visit(premise_id, trail) trail.pop() visiting.remove(claim_id) visited.add(claim_id) for claim_id in claims: visit(claim_id, []) def validate_logic(value: dict[str, Any]) -> None: fields( value, ( "schema_version", "title", "document_kind", "core_claim", "sections", "closure", ), "04_logic_map.json", ) nonempty_string(value["title"], "logic map title") nonempty_string(value["core_claim"], "logic map core_claim") nonempty_string(value["closure"], "logic map closure") if not isinstance(value["sections"], list) or not value["sections"]: raise InputError("logic map sections는 비어 있지 않은 배열이어야 합니다.") ids: set[str] = set() required = ( "id", "heading", "role", "depends_on", "reader_state_before", "question", "answer_plain", "claim_ids", "new_terms", "transition_to", "reader_state_after", ) for index, section in enumerate(value["sections"]): if not isinstance(section, dict): raise InputError(f"logic section[{index}]는 객체여야 합니다.") fields(section, required, f"logic section[{index}]") if section["id"] in ids: raise InputError(f"logic section id가 중복됩니다: {section['id']}") ids.add(section["id"]) for name in ( "id", "heading", "role", "reader_state_before", "question", "answer_plain", "reader_state_after", ): nonempty_string(section[name], f"logic section[{index}].{name}") for name in ("depends_on", "claim_ids", "new_terms"): string_list(section[name], f"logic section {section['id']}.{name}") if index > 0 and not section["depends_on"]: raise InputError( f"logic section {section['id']}에는 앞선 절을 가리키는 depends_on이 필요합니다." ) transition = section["transition_to"] if transition is not None: nonempty_string(transition, f"logic section {section['id']}.transition_to") def validate_terms(value: dict[str, Any]) -> None: fields(value, ("schema_version", "assumed_known", "budgets", "terms"), "05_term_ledger.json") string_list(value["assumed_known"], "term ledger assumed_known") if not isinstance(value["budgets"], dict): raise InputError("term ledger budgets는 객체여야 합니다.") if not isinstance(value["terms"], list): raise InputError("term ledger terms는 배열이어야 합니다.") ids: set[str] = set() name_owners: dict[str, tuple[str, str]] = {} for index, term in enumerate(value["terms"]): if not isinstance(term, dict): raise InputError(f"term[{index}]는 객체여야 합니다.") fields(term, ("id", "canonical", "plain_definition", "why_needed", "aliases", "first_section", "first_use"), f"term[{index}]") if term["id"] in ids: raise InputError(f"term id가 중복됩니다: {term['id']}") ids.add(term["id"]) for name in ( "id", "canonical", "plain_definition", "why_needed", "first_section", "first_use", ): nonempty_string(term[name], f"term[{index}].{name}") string_list(term["aliases"], f"term {term['id']}.aliases") for field_name, name in ( ("canonical", term["canonical"]), *(("alias", alias) for alias in term["aliases"]), ("english", term.get("english")), ("abbreviation", term.get("abbreviation")), ): if not isinstance(name, str) or not name.strip(): continue normalized = normalized_label(name) previous = name_owners.get(normalized) if previous is not None: raise InputError( f"용어 이름 {name!r}이 둘 이상에 배정되었습니다: " f"{previous[0]}.{previous[1]}, {term['id']}.{field_name}" ) name_owners[normalized] = (term["id"], field_name) def validate_review(value: dict[str, Any]) -> None: fields(value, ("schema_version", "review_type", "document", "inputs", "verdict", "findings"), "review") if value["review_type"] not in {"logic", "reader"}: raise InputError("review_type이 잘못되었습니다.") if value["verdict"] not in {"pass", "revise", "hold_for_review"}: raise InputError("review verdict가 잘못되었습니다.") if not isinstance(value["findings"], list): raise InputError("review findings는 배열이어야 합니다.") finding_ids: set[str] = set() blocking = 0 for finding in value["findings"]: if not isinstance(finding, dict): raise InputError("review finding은 객체여야 합니다.") fields(finding, ("id", "severity", "location", "reader_impact", "suggestion"), "review finding") if finding["id"] in finding_ids: raise InputError(f"review finding id가 중복됩니다: {finding['id']}") finding_ids.add(finding["id"]) if finding["severity"] not in {"critical", "high", "medium", "low"}: raise InputError("review finding severity가 잘못되었습니다.") if finding["severity"] in {"critical", "high"}: blocking += 1 verdict = value["verdict"] if verdict == "pass" and blocking: raise InputError("review verdict=pass에는 critical/high blocking finding이 있을 수 없습니다.") if verdict in {"revise", "hold_for_review"} and not blocking: raise InputError(f"review verdict={verdict}에는 critical/high blocking finding이 필요합니다.") def validate_lint(value: dict[str, Any]) -> None: fields(value, ("schema_version", "rules_version", "rules_sha256", "tool", "verdict", "summary", "findings", "fidelity"), "08_lint.json") if value["tool"] != "lint_document": raise InputError("08_lint.json tool이 lint_document가 아닙니다.") if value["verdict"] not in {"pass", "fail", "input_error"}: raise InputError("lint verdict가 잘못되었습니다.") def validate_lint_semantics(value: dict[str, Any], rules: dict[str, Any]) -> None: """Recompute every lint aggregate and gate from canonical findings.""" configured = rule_index(rules) counts = Counter() for index, finding in enumerate(value["findings"]): rule_id = finding["rule_id"] rule = configured.get(rule_id) if rule is None: raise InputError(f"lint finding[{index}]에 모르는 rule_id가 있습니다: {rule_id}") expected_severity = rule["severity"] if finding["severity"] != expected_severity: raise InputError( f"lint finding[{index}] severity가 현재 규칙과 다릅니다: " f"{rule_id}={finding['severity']!r}, expected={expected_severity!r}" ) if finding["path"] != value["document"]["path"]: raise InputError( f"lint finding[{index}] path가 lint 대상 document와 다릅니다." ) counts[finding["severity"]] += 1 expected_summary = { "errors": counts["error"], "warnings": counts["warning"], "info": counts["info"], "total": len(value["findings"]), } if value["summary"] != expected_summary: raise InputError( f"lint summary가 findings 재계산 결과와 다릅니다: " f"recorded={value['summary']!r}, expected={expected_summary!r}" ) expected_verdict = ( "fail" if counts["error"] > 0 or (value["fail_on"] == "warning" and counts["warning"] > 0) else "pass" ) if value["verdict"] != expected_verdict: raise InputError( f"lint verdict가 fail_on과 findings에 맞지 않습니다: " f"recorded={value['verdict']!r}, expected={expected_verdict!r}" ) fidelity = value["fidelity"] by_type = fidelity["by_type"] protected_types = { "fenced_code", "indented_code", "inline_code", "url", "link_destination", "number_unit", "number_range", "number", "date", "version", "quote", } expected_types = protected_types if fidelity["baseline"] is not None else set() if set(by_type) != expected_types: raise InputError( "lint fidelity.by_type 종류가 baseline 계약과 다릅니다: " f"recorded={sorted(by_type)}, expected={sorted(expected_types)}" ) for name, item in by_type.items(): if item["total"] != item["preserved"] + item["missing"]: raise InputError(f"lint fidelity.by_type.{name} 합계가 맞지 않습니다.") expected_total = sum(item["total"] for item in by_type.values()) expected_preserved = sum(item["preserved"] for item in by_type.values()) expected_missing = sum(item["missing"] for item in by_type.values()) if ( fidelity["protected_total"] != expected_total or fidelity["preserved"] != expected_preserved or fidelity["missing"] != expected_missing or expected_total != expected_preserved + expected_missing ): raise InputError("lint fidelity 합계가 by_type 재계산 결과와 다릅니다.") configured_change_rate = float( rules["thresholds"]["finalization"]["max_change_rate"] ) if fidelity["max_finalization_change_rate"] != configured_change_rate: raise InputError( "lint fidelity max_finalization_change_rate가 현재 quality rules와 다릅니다." ) def lint_report_summary(value: dict[str, Any] | None) -> dict[str, Any] | None: if value is None or value.get("verdict") not in {"pass", "fail"}: return None finding_rule_ids = list( dict.fromkeys(finding["rule_id"] for finding in value["findings"]) ) return { "verdict": value["verdict"], "fail_on": value["fail_on"], "rules_version": value["rules_version"], "rules_sha256": value["rules_sha256"], "document_sha256": value["document"]["sha256"], "finding_counts": value["summary"], "finding_rule_ids": finding_rule_ids, "fidelity": value["fidelity"], "limitations": value.get("limitations", []), } def review_report_summaries(artifacts: dict[str, dict[str, Any]]) -> list[dict[str, Any]]: summaries: list[dict[str, Any]] = [] for artifact_name in ("08_logic_review.json", "08_reader_review.json"): review = artifacts.get(artifact_name) if review is None: continue counts = Counter(finding["severity"] for finding in review["findings"]) summaries.append( { "artifact": artifact_name, "review_type": review["review_type"], "verdict": review["verdict"], "document_sha256": review["document"]["sha256"], "input_sha256s": review["inputs"], "finding_counts": { "critical": counts["critical"], "high": counts["high"], "medium": counts["medium"], "low": counts["low"], "total": len(review["findings"]), }, "finding_ids": [finding["id"] for finding in review["findings"]], } ) return summaries def review_mode_document_verdict( lint_report: dict[str, Any] | None, artifacts: dict[str, dict[str, Any]], ) -> str: reviews = [ artifacts.get("08_logic_review.json"), artifacts.get("08_reader_review.json"), ] if ( lint_report is None or lint_report.get("verdict") not in {"pass", "fail"} or any(review is None for review in reviews) ): return "not_evaluated" review_verdicts = {review["verdict"] for review in reviews if review is not None} if "hold_for_review" in review_verdicts: return "not_evaluated" if lint_report["verdict"] == "fail" or "revise" in review_verdicts: return "revise" return "pass" VALIDATORS: dict[str, Callable[[dict[str, Any]], None]] = { "00_run.json": validate_run, "01_sources.json": validate_sources, "02_reader_contract.json": validate_reader, "04_logic_map.json": validate_logic, "05_term_ledger.json": validate_terms, "08_logic_review.json": validate_review, "08_reader_review.json": validate_review, "08_lint.json": validate_lint, } def required_artifacts(contract: dict[str, Any], manifest: dict[str, Any]) -> set[str]: artifacts = contract.get("artifacts") if not isinstance(artifacts, dict): raise InputError("runtime contract artifacts가 없습니다.") base = artifacts.get("always") if not isinstance(base, list): raise InputError("runtime contract artifacts.always가 배열이 아닙니다.") mode = manifest["mode"] route = manifest["route_hint"] route_branch = artifacts.get(route) if not isinstance(route_branch, list): raise InputError(f"runtime contract artifacts.{route}가 배열이 아닙니다.") required = set(base) | set(route_branch) if mode == "review": review_branch = artifacts.get("review_mode") if not isinstance(review_branch, list): raise InputError("runtime contract artifacts.review_mode가 배열이 아닙니다.") required.update(review_branch) required.discard("final.md") return required def artifact_universe(contract: dict[str, Any]) -> set[str]: artifacts = contract["artifacts"] universe: set[str] = set() for branch_name in ("always", "light", "standard", "deep", "review_mode"): universe.update(artifacts[branch_name]) return universe def path_lexists(path: Path) -> bool: """Return true for ordinary entries and dangling symbolic links.""" return path.exists() or path.is_symlink() def capture_paths(paths: dict[str, Path]) -> dict[str, str | None]: snapshot: dict[str, str | None] = {} for name, path in paths.items(): try: snapshot[name] = sha256_file(path) if path.is_file() else None except InputError: snapshot[name] = None return snapshot def snapshot_drift( expected: dict[str, str | None], paths: dict[str, Path] ) -> list[str]: current = capture_paths(paths) return sorted(name for name, digest in expected.items() if current.get(name) != digest) def source_paths(sources: dict[str, Any]) -> dict[str, Path]: paths: dict[str, Path] = {} values = [sources.get("brief"), sources.get("draft"), *sources.get("sources", [])] for item in values: if not isinstance(item, dict) or not isinstance(item.get("id"), str): continue raw_path = item.get("resolved_path") or item.get("path") if isinstance(raw_path, str): paths[f"source:{item['id']}"] = Path(raw_path) return paths def verify_route_selection( manifest: dict[str, Any], sources: dict[str, Any], rules: dict[str, Any], checks: list[dict[str, Any]], ) -> None: inventories = [sources.get("brief"), sources.get("draft"), *sources.get("sources", [])] texts: list[str] = [] for item in inventories: if item is None: continue path_value = item.get("resolved_path") or item.get("path") if not isinstance(path_value, str): raise InputError("route 계산 source 경로가 없습니다.") path = require_file(Path(path_value), f"route source {item.get('id')}") try: data = path.read_bytes() except OSError as exc: raise InputError(f"route source를 읽을 수 없습니다: {path}: {exc}") from exc texts.append(decode_utf8(data, f"route source {item.get('id')}")) actual_metrics = measure_route_inputs(texts, len(sources.get("sources", []))) metrics_ok = manifest.get("route_metrics") == actual_metrics check( checks, "route-metrics", metrics_ok, "brief, draft, 모든 source를 포함한 route 규모 지표가 일치합니다." if metrics_ok else ( f"route metrics drift: recorded={manifest.get('route_metrics')!r}, " f"actual={actual_metrics!r}" ), "00_run.json", ) expected_route, expected_reason = choose_route( manifest["route_requested"], mode=manifest["mode"], has_draft=sources.get("draft") is not None, metrics=actual_metrics, rules=rules, ) selection_ok = ( manifest.get("route_hint") == expected_route and manifest.get("route_reason") == expected_reason ) check( checks, "route-selection", selection_ok, "route 요청, 전체 입력 지표, 선택 이유가 정본 알고리즘과 일치합니다." if selection_ok else ( f"expected route={expected_route!r}, reason={expected_reason!r}; " f"recorded route={manifest.get('route_hint')!r}, " f"reason={manifest.get('route_reason')!r}" ), "00_run.json", ) def immutable_manifest_fields(manifest: dict[str, Any]) -> dict[str, Any]: return { key: value for key, value in manifest.items() if key not in {"status", "error", "updated_at", "history"} } def verify_omissions( run_dir: Path, contract: dict[str, Any], manifest: dict[str, Any], required: set[str], checks: list[dict[str, Any]], ) -> None: omissions = manifest["omissions"] universe = artifact_universe(contract) declared = [item["artifact"] for item in omissions] duplicates = sorted(name for name, count in Counter(declared).items() if count > 1) unknown = sorted(set(declared) - universe) required_omitted = sorted(set(declared) & required) present_omitted = sorted( name for name in set(declared) if path_lexists(run_dir / name) ) invalid_parts: list[str] = [] if duplicates: invalid_parts.append(f"duplicate={duplicates}") if unknown: invalid_parts.append(f"unknown={unknown}") if required_omitted: invalid_parts.append(f"required={required_omitted}") if present_omitted: invalid_parts.append(f"present={present_omitted}") check( checks, "omissions-valid", not invalid_parts, "omission 선언이 유효합니다." if not invalid_parts else "잘못된 omission 선언: " + "; ".join(invalid_parts), "00_run.json", ) present = { name for name in universe if ( (run_dir / name).is_file() and not (run_dir / name).is_symlink() and (run_dir / name).stat().st_size > 0 ) } optional_missing = universe - required - present undeclared = sorted(optional_missing - set(declared)) check( checks, "omissions-complete", not undeclared, "생략한 optional artifact와 이유가 모두 기록되었습니다." if not undeclared else f"omission 이유가 없는 optional artifact: {undeclared}", "00_run.json", ) def verify_source_hashes( run_dir: Path, manifest: dict[str, Any], sources: dict[str, Any], checks: list[dict[str, Any]], ) -> None: input_path = run_dir / "01_input.md" actual_input = sha256_file(input_path) expected_input = manifest["inputs"]["input_sha256"] check(checks, "hash-input", actual_input == expected_input, "01_input.md SHA256 검증", "01_input.md") inventory = [sources.get("brief"), sources.get("draft"), *sources.get("sources", [])] drift: list[str] = [] for item in inventory: if item is None: continue resolved_value = item.get("resolved_path") or item.get("path") try: path = require_file(Path(resolved_value), f"source {item.get('id')}") actual = sha256_file(path) except InputError: drift.append(str(item.get("id"))) continue if actual != item.get("sha256"): drift.append(str(item.get("id"))) check( checks, "hash-sources", not drift, "source SHA256 검증" if not drift else f"source hash drift: {', '.join(drift)}", "01_sources.json", ) actual_manifest_hash = sha256_text(json_text(sources)) expected_manifest_hash = manifest["inputs"]["sources_manifest_sha256"] check( checks, "hash-source-set", actual_manifest_hash == expected_manifest_hash, "source id/hash 집합 검증", "01_sources.json", ) inputs = manifest["inputs"] brief = sources.get("brief") draft = sources.get("draft") brief_sha = brief.get("sha256") if isinstance(brief, dict) else None draft_sha = draft.get("sha256") if isinstance(draft, dict) else None manifest_brief = inputs.get("brief") manifest_draft = inputs.get("draft") inventory_ok = ( inputs.get("brief_sha256") == brief_sha and isinstance(manifest_brief, dict) and manifest_brief.get("sha256") == brief_sha and inputs.get("draft_sha256") == draft_sha and ( (draft is None and manifest_draft is None) or ( isinstance(draft, dict) and isinstance(manifest_draft, dict) and manifest_draft.get("sha256") == draft_sha ) ) and inputs.get("source_count") == len(sources.get("sources", [])) ) check( checks, "hash-input-inventory", inventory_ok, "run input 요약과 01_sources inventory 정합성", "00_run.json", ) def verify(args: argparse.Namespace) -> tuple[dict[str, Any], int, dict[str, Any] | None]: run_dir = Path(args.run_dir).expanduser().resolve(strict=True) if not run_dir.is_dir(): raise InputError(f"run-dir은 디렉터리여야 합니다: {run_dir}") if args.contract and not paths_alias(Path(args.contract), DEFAULT_CONTRACT_PATH): raise InputError("custom runtime contract override는 허용되지 않습니다.") contract_path = require_file(DEFAULT_CONTRACT_PATH, "runtime contract") contract_sha256 = sha256_file(contract_path) contract = load_json(contract_path) validate_with_schema(contract, "runtime-contract.schema.json", str(contract_path)) schema_version(contract, contract_path) if sha256_file(contract_path) != contract_sha256: raise InputError("runtime contract가 읽는 동안 변경되었습니다.") rules_path = require_file( Path(args.rules) if args.rules else DEFAULT_RULES_PATH, "quality rules", ) rules_sha256 = sha256_file(rules_path) rules = load_rules(rules_path) if sha256_file(rules_path) != rules_sha256: raise InputError("quality rules가 읽는 동안 변경되었습니다.") manifest_path = require_regular_nonsymlink( run_dir / "00_run.json", "00_run.json" ) manifest = load_json(manifest_path) validate_with_schema(manifest, "run.schema.json", str(manifest_path)) schema_version(manifest, manifest_path) validate_run(manifest) initial_manifest_sha256 = sha256_file(manifest_path) initial_manifest_immutable = immutable_manifest_fields(manifest) checks: list[dict[str, Any]] = [] contract_name = contract.get("name") check(checks, "contract-name", contract_name == "technical-doc-flow", f"runtime contract name={contract_name!r}") run_contract_hash_ok = manifest.get("contract_sha256") == contract_sha256 check( checks, "contract-sha256", run_contract_hash_ok, "run이 고정한 canonical runtime contract byte hash가 일치합니다.", "00_run.json", ) run_rules_version_ok = manifest.get("rules_version") == rules["rules_version"] check( checks, "rules-version", run_rules_version_ok, f"run rules_version={manifest.get('rules_version')!r}, runtime={rules['rules_version']!r}", "00_run.json", ) run_rules_hash_ok = manifest.get("rules_sha256") == rules_sha256 check( checks, "rules-sha256", run_rules_hash_ok, "run이 고정한 quality rules byte hash와 runtime rules가 일치합니다.", "00_run.json", ) required = required_artifacts(contract, manifest) universe = artifact_universe(contract) canonical_paths = { name: run_dir / name for name in universe if name not in {"00_run.json", "09_final_report.json"} } canonical_snapshot = capture_paths(canonical_paths) runtime_paths = {"runtime:contract": contract_path, "runtime:rules": rules_path} runtime_snapshot = { "runtime:contract": contract_sha256, "runtime:rules": rules_sha256, } if manifest["mode"] == "review" and path_lexists(run_dir / "final.md"): check( checks, "review-no-final", False, "review mode에서는 final.md를 만들지 않아야 합니다.", "final.md", ) verify_omissions(run_dir, contract, manifest, required, checks) output_path = Path(args.output).expanduser().resolve() if args.output else run_dir / "09_final_report.json" artifacts: dict[str, dict[str, Any]] = {"00_run.json": manifest} present_optional = { name for name in universe - required if path_lexists(run_dir / name) } for name in sorted(required | present_optional): path = run_dir / name if name == "09_final_report.json" and path.resolve() == output_path.resolve(): check(checks, "artifact-final-report", True, "verifier가 final report를 생성합니다.", name) continue if path.is_symlink(): check( checks, f"artifact-{name}", False, "canonical artifact는 symbolic link일 수 없습니다.", name, ) continue if not path.is_file() or path.stat().st_size == 0: check(checks, f"artifact-{name}", False, "필수 산출물이 없거나 비어 있습니다.", name) continue presence_label = "필수" if name in required else "선택" check(checks, f"artifact-{name}", True, f"{presence_label} 산출물이 존재합니다.", name) if name in JSON_ARTIFACTS: try: value = load_json(path) schema_version(value, path) schema_name = ARTIFACT_SCHEMAS.get(name) if schema_name: validate_with_schema(value, schema_name, str(path)) validator = VALIDATORS.get(name) if validator: validator(value) artifacts[name] = value check(checks, f"schema-{name}", True, "schema_version과 필수 필드가 유효합니다.", name) except InputError as exc: check(checks, f"schema-{name}", False, str(exc), name) if manifest["mode"] == "review": review_draft_path = run_dir / "07_draft.md" initial_draft = manifest.get("inputs", {}).get("draft") expected_draft_hash = ( initial_draft.get("sha256") if isinstance(initial_draft, dict) else None ) review_draft_immutable = ( isinstance(expected_draft_hash, str) and review_draft_path.is_file() and not review_draft_path.is_symlink() and sha256_file(review_draft_path) == expected_draft_hash ) check( checks, "review-draft-immutable", review_draft_immutable, "review mode의 07_draft.md가 run 초기화 때 고정한 원본과 byte-identical합니다." if review_draft_immutable else ( "review mode의 07_draft.md SHA-256이 초기 원본과 다릅니다: " f"expected={expected_draft_hash!r}" ), "07_draft.md", ) sources = artifacts.get("01_sources.json") source_snapshot: dict[str, str | None] = {} current_source_paths: dict[str, Path] = {} if sources: current_source_paths = source_paths(sources) inventory_values = [ sources.get("brief"), sources.get("draft"), *sources.get("sources", []), ] source_snapshot = { f"source:{item['id']}": item["sha256"] for item in inventory_values if isinstance(item, dict) and isinstance(item.get("id"), str) and isinstance(item.get("sha256"), str) } try: verify_source_hashes(run_dir, manifest, sources, checks) verify_route_selection(manifest, sources, rules, checks) except InputError as exc: check(checks, "hash-inputs", False, str(exc), "01_sources.json") reader = artifacts.get("02_reader_contract.json") logic = artifacts.get("04_logic_map.json") terms = artifacts.get("05_term_ledger.json") evidence = artifacts.get("03_evidence_map.json") if reader and logic: kind_ok = reader.get("document_kind") == manifest["document_kind"] == logic.get("document_kind") check(checks, "document-kind", kind_ok, "run/reader/logic document_kind 정합성") requested_audience = manifest.get("audience") audience_ok = requested_audience is None or normalized_label( requested_audience ) == normalized_label(reader.get("primary_audience", "")) check( checks, "requested-audience", audience_ok, "명시적으로 요청한 audience와 reader contract primary_audience가 일치합니다." if audience_ok else ( f"requested audience={requested_audience!r}, " f"reader primary_audience={reader.get('primary_audience')!r}" ), "02_reader_contract.json", ) if reader and terms: assumed_ok = set(reader.get("assumed_known", [])) == set(terms.get("assumed_known", [])) check(checks, "assumed-known", assumed_ok, "reader/term assumed_known 정합성") assumed = { normalized_label(item) for item in reader.get("assumed_known", []) if isinstance(item, str) } must_explain = { normalized_label(item) for item in reader.get("must_explain", []) if isinstance(item, str) } overlap = sorted(assumed & must_explain) check( checks, "must-explain-disjoint", not overlap, "assumed_known와 must_explain의 역할이 분리되어 있습니다." if not overlap else f"동시에 assumed_known와 must_explain인 항목: {overlap}", "02_reader_contract.json", ) explainable: set[str] = set() for term in terms.get("terms", []): if not isinstance(term, dict): continue for value in ( term.get("canonical"), *term.get("aliases", []), term.get("english"), term.get("abbreviation"), ): if isinstance(value, str) and value.strip(): explainable.add(normalized_label(value)) missing_explanations = sorted( item for item in reader.get("must_explain", []) if isinstance(item, str) and normalized_label(item) not in explainable ) check( checks, "must-explain-terms", not missing_explanations, "must_explain 항목이 assumed_known 또는 term ledger에 연결됩니다." if not missing_explanations else f"term ledger에 없는 must_explain 항목: {missing_explanations}", "02_reader_contract.json", ) if evidence and sources: try: validate_evidence(evidence, source_ids(sources)) check(checks, "evidence-contract", True, "claim 근거 경계가 유효합니다.", "03_evidence_map.json") except InputError as exc: check(checks, "evidence-contract", False, str(exc), "03_evidence_map.json") if evidence and logic: claim_ids = {claim["id"] for claim in evidence.get("claims", []) if isinstance(claim, dict) and "id" in claim} referenced = { claim_id for section in logic.get("sections", []) if isinstance(section, dict) for claim_id in section.get("claim_ids", []) } unknown = referenced - claim_ids check( checks, "logic-claims", not unknown, "logic claim 연결 검증" if not unknown else f"모르는 claim id: {sorted(unknown)}", "04_logic_map.json", ) if logic and terms: term_values = [ term for term in terms.get("terms", []) if isinstance(term, dict) and isinstance(term.get("id"), str) ] term_ids = {term["id"] for term in term_values} term_sections: dict[str, list[str]] = {} for section in logic.get("sections", []): if not isinstance(section, dict) or not isinstance(section.get("id"), str): continue for term_id in section.get("new_terms", []): if isinstance(term_id, str): term_sections.setdefault(term_id, []).append(section["id"]) referenced_terms = set(term_sections) unknown_terms = referenced_terms - term_ids misplaced_terms = { term["id"]: { "first_section": term.get("first_section"), "new_terms_sections": term_sections.get(term["id"], []), } for term in term_values if term_sections.get(term["id"], []) != [term.get("first_section")] } terminology_ok = not unknown_terms and not misplaced_terms check( checks, "logic-terms", terminology_ok, "각 term이 first_section의 new_terms에 정확히 한 번 연결됩니다." if terminology_ok else ( f"모르는 term id={sorted(unknown_terms)}; " f"first_section 불일치={misplaced_terms}" ), "05_term_ledger.json", ) lint_report = artifacts.get("08_lint.json") trusted_lint: dict[str, Any] | None = None if lint_report: lint_trusted = ( run_contract_hash_ok and run_rules_version_ok and run_rules_hash_ok ) lint_verdict = lint_report.get("verdict") if manifest["mode"] == "review": lint_ok = lint_verdict in {"pass", "fail"} lint_message = ( f"review 진단 lint verdict={lint_verdict!r}; pass/fail은 유효한 문서 판정입니다." ) else: lint_ok = lint_verdict == "pass" lint_message = f"publish lint verdict={lint_verdict!r}; pass가 필요합니다." check(checks, "lint-verdict", lint_ok, lint_message, "08_lint.json") version_ok = lint_report.get("rules_version") == rules["rules_version"] check( checks, "lint-rules-version", version_ok, "lint rules_version 정합성", "08_lint.json", ) lint_trusted = lint_trusted and version_ok rules_hash_ok = lint_report.get("rules_sha256") == rules_sha256 check( checks, "lint-rules-sha256", rules_hash_ok, "lint가 사용한 quality rules byte hash가 현재 run과 일치합니다.", "08_lint.json", ) lint_trusted = lint_trusted and rules_hash_ok try: validate_lint_semantics(lint_report, rules) check( checks, "lint-semantics", True, "lint findings, severity, summary, fidelity, verdict 재계산이 일치합니다.", "08_lint.json", ) except InputError as exc: lint_trusted = False check(checks, "lint-semantics", False, str(exc), "08_lint.json") document_name = "07_draft.md" if manifest["mode"] == "review" else "final.md" document_path = run_dir / document_name if document_path.is_file() and not document_path.is_symlink(): document_record = lint_report.get("document", {}) expected_hash = document_record.get("sha256") recorded_path = document_record.get("path") if isinstance(recorded_path, str): candidate = Path(recorded_path).expanduser() if not candidate.is_absolute(): candidate = run_dir / candidate document_path_ok = paths_alias(candidate, document_path) else: document_path_ok = False document_hash_ok = expected_hash == sha256_file(document_path) check( checks, "lint-document-hash", document_path_ok and document_hash_ok, f"lint 대상 path/hash와 {document_name} 정합성", "08_lint.json", ) lint_trusted = lint_trusted and document_path_ok and document_hash_ok else: lint_trusted = False for artifact_name, hash_field in ( ("04_logic_map.json", "logic_map_sha256"), ("05_term_ledger.json", "term_ledger_sha256"), ("02_reader_contract.json", "reader_contract_sha256"), ): artifact_path = run_dir / artifact_name if artifact_path.is_file() and not artifact_path.is_symlink(): input_hash_ok = lint_report.get(hash_field) == sha256_file(artifact_path) check( checks, f"lint-input-hash-{artifact_name}", input_hash_ok, f"lint 입력 hash와 현재 {artifact_name} 정합성", "08_lint.json", ) lint_trusted = lint_trusted and input_hash_ok else: lint_trusted = False for fidelity_name in ("baseline", "draft_baseline"): fidelity_record = lint_report.get("fidelity", {}).get(fidelity_name) if fidelity_record is None: continue recorded_path = Path(fidelity_record["path"]).expanduser() if not recorded_path.is_absolute(): recorded_path = run_dir / recorded_path try: current_path = require_file(recorded_path, f"lint {fidelity_name}") fidelity_ok = fidelity_record["sha256"] == sha256_file(current_path) except InputError: fidelity_ok = False check( checks, f"lint-fidelity-hash-{fidelity_name}", fidelity_ok, f"lint fidelity {fidelity_name} path/hash 정합성", "08_lint.json", ) lint_trusted = lint_trusted and fidelity_ok fidelity = lint_report.get("fidelity", {}) def fidelity_record_matches( record: Any, expected_path: Path, expected_sha256: str ) -> bool: if not isinstance(record, dict): return False recorded_path = record.get("path") return ( isinstance(recorded_path, str) and paths_alias(Path(recorded_path), expected_path) and record.get("sha256") == expected_sha256 ) mode = manifest["mode"] invocation_errors: list[str] = [] if mode in {"write", "revise"}: draft_path = run_dir / "07_draft.md" draft_record = fidelity.get("draft_baseline") if not draft_path.is_file() or not fidelity_record_matches( draft_record, draft_path, sha256_file(draft_path) ): invocation_errors.append("draft_baseline은 현재 07_draft.md여야 합니다") elif lint_report.get("document", {}).get("sha256") != draft_record.get( "sha256" ): invocation_errors.append( "final.md는 review 대상 07_draft.md와 byte 단위로 같아야 합니다" ) if fidelity.get("finalization_change_rate") is None: invocation_errors.append("finalization_change_rate가 필요합니다") elif fidelity.get("draft_baseline") is not None: invocation_errors.append("review mode에는 draft_baseline을 사용하지 않습니다") original_draft = manifest.get("inputs", {}).get("draft") baseline_record = fidelity.get("baseline") if mode == "revise": if not isinstance(original_draft, dict) or not fidelity_record_matches( baseline_record, Path(original_draft.get("resolved_path", "")), original_draft.get("sha256", ""), ): invocation_errors.append("revise baseline은 원본 draft여야 합니다") elif mode == "write" and baseline_record is not None: invocation_errors.append("write mode에는 원본 baseline을 사용하지 않습니다") elif mode == "review" and baseline_record is not None: if not isinstance(original_draft, dict) or not fidelity_record_matches( baseline_record, Path(original_draft.get("resolved_path", "")), original_draft.get("sha256", ""), ): invocation_errors.append("review baseline은 원본 draft여야 합니다") if manifest["route_hint"] == "deep" and lint_report.get("fail_on") != "warning": invocation_errors.append("deep route는 fail_on=warning이어야 합니다") invocation_ok = not invocation_errors check( checks, "lint-invocation-contract", invocation_ok, "mode/route별 lint baseline과 severity gate가 유효합니다." if invocation_ok else "; ".join(invocation_errors), "08_lint.json", ) lint_trusted = lint_trusted and invocation_ok if invocation_ok: baseline_path: Path | None = None if mode == "revise" or ( mode == "review" and baseline_record is not None ): baseline_path = Path(original_draft["resolved_path"]) replay_args = argparse.Namespace( document=str(document_path), logic_map=str(run_dir / "04_logic_map.json"), term_ledger=str(run_dir / "05_term_ledger.json"), reader_contract=str(run_dir / "02_reader_contract.json"), baseline=str(baseline_path) if baseline_path is not None else None, draft_baseline=( str(run_dir / "07_draft.md") if mode in {"write", "revise"} else None ), fail_on=lint_report["fail_on"], rules=str(rules_path), ) try: replay_report, _ = evaluate_lint(replay_args) recorded_comparable = { key: value for key, value in lint_report.items() if key != "generated_at" } replay_comparable = { key: value for key, value in replay_report.items() if key != "generated_at" } replay_ok = recorded_comparable == replay_comparable replay_message = ( "현재 document와 계약으로 canonical lint를 재실행한 결과가 일치합니다." if replay_ok else ( "저장된 lint 결과가 canonical 재실행과 다릅니다: " f"recorded verdict={lint_report.get('verdict')!r}, " f"rules={[item.get('rule_id') for item in lint_report.get('findings', [])]}; " f"replayed verdict={replay_report.get('verdict')!r}, " f"rules={[item.get('rule_id') for item in replay_report.get('findings', [])]}" ) ) except (InputError, OSError, TypeError, ValueError, KeyError) as exc: replay_ok = False replay_message = f"canonical lint 재실행 실패: {exc}" check( checks, "lint-canonical-replay", replay_ok, replay_message, "08_lint.json", ) lint_trusted = lint_trusted and replay_ok if lint_trusted: trusted_lint = lint_report trusted_reviews: dict[str, dict[str, Any]] = {} def current_artifact_hash(name: str) -> str | None: path = run_dir / name return sha256_file(path) if path.is_file() and not path.is_symlink() else None expected_review_inputs = { "input_sha256": current_artifact_hash("01_input.md"), "sources_sha256": current_artifact_hash("01_sources.json"), "reader_contract_sha256": current_artifact_hash("02_reader_contract.json"), "evidence_map_sha256": current_artifact_hash("03_evidence_map.json"), "logic_map_sha256": current_artifact_hash("04_logic_map.json"), "term_ledger_sha256": current_artifact_hash("05_term_ledger.json"), } for review_name in ("08_logic_review.json", "08_reader_review.json"): review = artifacts.get(review_name) if review is not None: expected_type = "logic" if review_name == "08_logic_review.json" else "reader" type_ok = review.get("review_type") == expected_type check( checks, f"review-type-{review_name}", type_ok, f"filename에 필요한 review_type={expected_type!r}", review_name, ) draft_path = run_dir / "07_draft.md" document = review.get("document") recorded_path = document.get("path") if isinstance(document, dict) else None if isinstance(recorded_path, str): candidate = Path(recorded_path).expanduser() if not candidate.is_absolute(): candidate = run_dir / candidate path_ok = candidate.resolve() == draft_path.resolve() else: path_ok = False hash_ok = ( draft_path.is_file() and not draft_path.is_symlink() and isinstance(document, dict) and document.get("sha256") == sha256_file(draft_path) ) check( checks, f"review-document-{review_name}", path_ok and hash_ok, "review 대상 path/hash와 현재 07_draft.md 정합성", review_name, ) inputs_ok = review.get("inputs") == expected_review_inputs check( checks, f"review-inputs-{review_name}", inputs_ok, "review 입력 hash 묶음과 현재 upstream artifact가 일치합니다.", review_name, ) if type_ok and path_ok and hash_ok and inputs_ok: trusted_reviews[review_name] = review review_verdict = review.get("verdict") if manifest["mode"] == "review": review_ok = review_verdict in {"pass", "revise"} review_message = ( f"review 진단 verdict={review_verdict!r}; pass/revise는 완료된 문서 판정이며 " "hold_for_review만 실행을 차단합니다." ) else: review_ok = review_verdict == "pass" if review_verdict == "revise": review_message = ( "review verdict='revise'; finalizer 전에 Phase 3에서 draft를 수정하고 " "두 독립 review를 모두 다시 실행해야 합니다." ) elif review_verdict == "hold_for_review": review_message = ( "review verdict='hold_for_review'; 외부 입력 또는 구조적 blocker를 " "해결하기 전에는 finalizer를 실행할 수 없습니다." ) else: review_message = "review verdict='pass'; blocking finding이 없습니다." check( checks, f"review-verdict-{review_name}", review_ok, review_message, review_name, ) history_reviewed = any( isinstance(entry, dict) and entry.get("to") == "reviewed" for entry in manifest.get("history", []) ) current_review_names = { name for name in ("08_logic_review.json", "08_reader_review.json") if name in artifacts } if manifest["mode"] in {"write", "revise"}: trusted_pass_reviews = { name for name, review in trusted_reviews.items() if review.get("verdict") == "pass" } review_stage_ok = ( history_reviewed and trusted_pass_reviews == {"08_logic_review.json", "08_reader_review.json"} ) or (not history_reviewed and not current_review_names) check( checks, "review-stage-consistency", review_stage_ok, "reviewed 이력과 현재 두 pass review가 서로 일치합니다." if review_stage_ok else ( f"history_reviewed={history_reviewed}, " f"current_reviews={sorted(current_review_names)}, " f"trusted_pass_reviews={sorted(trusted_pass_reviews)}" ), "00_run.json", ) pre_transition_drift = [ *snapshot_drift(canonical_snapshot, canonical_paths), *snapshot_drift(runtime_snapshot, runtime_paths), *snapshot_drift(source_snapshot, current_source_paths), ] manifest_changed = sha256_file(manifest_path) != initial_manifest_sha256 if manifest_changed: pre_transition_drift.append("00_run.json") check( checks, "verification-snapshot-pre-transition", not pre_transition_drift, "검증에 사용한 파일 snapshot이 상태 전이 직전까지 유지되었습니다." if not pre_transition_drift else f"검증 도중 변경된 파일: {sorted(set(pre_transition_drift))}", ) status = manifest["status"] expected_status = "reviewed" if manifest["mode"] == "review" else {"finalized", "verified"} status_ok = status == expected_status if isinstance(expected_status, str) else status in expected_status check(checks, "run-status", status_ok, f"검증 전 status={status!r}", "00_run.json") failures = [item for item in checks if item["status"] != "pass"] verdict = "fail" if failures else "pass" if verdict == "pass" and manifest["mode"] != "review" and status == "finalized": try: manifest = transition_manifest( manifest_path, "verified", reason="verify_run pass", ) check(checks, "status-transition", True, "status를 verified로 갱신했습니다.", "00_run.json") except InputError as exc: check(checks, "status-transition", False, str(exc), "00_run.json") verdict = "fail" elif verdict == "fail" and status not in {"hold_for_review", "failed", "incomplete"}: try: manifest = transition_manifest( manifest_path, "hold_for_review", reason="verify_run failed", ) check(checks, "status-transition", True, "status를 hold_for_review로 갱신했습니다.", "00_run.json") except InputError as exc: check(checks, "status-transition", False, str(exc), "00_run.json") post_transition_drift = [ *snapshot_drift(canonical_snapshot, canonical_paths), *snapshot_drift(runtime_snapshot, runtime_paths), *snapshot_drift(source_snapshot, current_source_paths), ] immutable_ok = immutable_manifest_fields(manifest) == initial_manifest_immutable if not immutable_ok: post_transition_drift.append("00_run.json:immutable-fields") check( checks, "verification-snapshot-post-transition", not post_transition_drift, "검증 snapshot이 최종 상태 기록 뒤에도 유지되었습니다." if not post_transition_drift else f"최종 상태 기록 중 변경된 파일: {sorted(set(post_transition_drift))}", ) if post_transition_drift and manifest.get("status") not in { "hold_for_review", "failed", "incomplete", }: try: manifest = transition_manifest( manifest_path, "hold_for_review", reason="verification snapshot changed during final transition", ) check( checks, "status-transition-after-drift", True, "동시 변경을 감지해 상태를 hold_for_review로 낮췄습니다.", "00_run.json", ) except InputError as exc: check( checks, "status-transition-after-drift", False, str(exc), "00_run.json", ) if pre_transition_drift or post_transition_drift: trusted_lint = None trusted_reviews = {} failures = [item for item in checks if item["status"] != "pass"] verdict = "fail" if failures else "pass" if manifest.get("mode") == "review" and verdict == "pass": document_verdict = review_mode_document_verdict(trusted_lint, trusted_reviews) elif manifest.get("mode") == "review": document_verdict = "not_evaluated" elif verdict == "pass": document_verdict = "pass" elif ( trusted_lint is not None and trusted_lint.get("verdict") == "fail" or any( trusted_reviews.get(name, {}).get("verdict") == "revise" for name in ("08_logic_review.json", "08_reader_review.json") ) ): document_verdict = "revise" else: document_verdict = "not_evaluated" report = { "schema_version": "1.0", "tool": "verify_run", "generated_at": utc_now(), "run_id": manifest.get("run_id", run_dir.name), "route": manifest.get("route_hint", "standard"), "mode": manifest.get("mode", "write"), "verdict": verdict, "document_verdict": document_verdict, "summary": { "passed": sum(item["status"] == "pass" for item in checks), "failed": len(failures), "required_artifacts": sorted(required), "omissions": manifest.get("omissions", []), "lint": lint_report_summary(trusted_lint), "reviews": review_report_summaries(trusted_reviews), "status": manifest.get("status"), }, "checks": checks, } validate_with_schema(report, "final-report.schema.json", "generated 09_final_report.json") return report, 0 if verdict == "pass" else 1, manifest def input_error_report(args: argparse.Namespace, message: str) -> dict[str, Any]: return { "schema_version": "1.0", "tool": "verify_run", "generated_at": utc_now(), "run_id": Path(args.run_dir).name, "route": "standard", "mode": "write", "verdict": "input_error", "document_verdict": "not_evaluated", "summary": { "passed": 0, "failed": 1, "required_artifacts": [], "omissions": [], "lint": None, "reviews": [], "status": None, }, "checks": [{"id": "input", "status": "error", "message": message, "artifact": None}], } def parser() -> argparse.ArgumentParser: value = argparse.ArgumentParser(description=__doc__) value.add_argument("--run-dir", required=True) value.add_argument( "--output", help="canonical {run-dir}/09_final_report.json 경로만 허용", ) value.add_argument("--contract", help="runtime-contract.json override") value.add_argument("--rules", help=argparse.SUPPRESS) return value def main(argv: list[str] | None = None) -> int: args = parser().parse_args(argv) try: run_dir = Path(args.run_dir).expanduser().resolve(strict=True) if not run_dir.is_dir(): raise InputError(f"run-dir은 디렉터리여야 합니다: {run_dir}") canonical_output = run_dir / "09_final_report.json" requested_output = Path(args.output) if args.output else canonical_output if not paths_alias(requested_output, canonical_output): raise InputError( "verify output은 canonical {run-dir}/09_final_report.json이어야 합니다." ) protected = [ run_dir / name for name in ( "00_run.json", "01_input.md", "01_sources.json", "02_reader_contract.json", "03_evidence_map.json", "04_logic_map.json", "05_term_ledger.json", "07_draft.md", "08_logic_review.json", "08_reader_review.json", "08_lint.json", "final.md", ) ] protected.extend( [ Path(args.contract) if args.contract else DEFAULT_CONTRACT_PATH, Path(args.rules) if args.rules else DEFAULT_RULES_PATH, ] ) output = prepare_report_output( requested_output, protected_paths=protected, expected_tool="verify_run", schema_name="final-report.schema.json", ) except (InputError, OSError, RuntimeError) as exc: print(f"input error: {exc}", file=sys.stderr) return 2 try: with run_lock(run_dir): try: report, exit_code, _ = verify(args) except (InputError, OSError, TypeError, ValueError, KeyError) as exc: report = input_error_report(args, str(exc)) exit_code = 2 output_path = publish_report_json(output, report) except (InputError, OSError, TypeError, ValueError, KeyError) as exc: print( f"input error: final report를 쓸 수 없습니다: {output.path}: {exc}", file=sys.stderr, ) return 2 print(f"{report['verdict']}: checks={len(report['checks'])} output={output_path}") return exit_code if __name__ == "__main__": raise SystemExit(main())