Files
Cesarus85--mjolnir-vr/tools/semantic_progress.py
T

293 lines
12 KiB
Python

"""Conservative progress corrections for proven objdiff false negatives.
Entries in config/semantic_matches.json are never trusted on their own. Each
entry is re-verified against the current target and rebuilt COFF objects before
its function is credited. This keeps the ordinary objdiff report authoritative
except where a stricter semantic relocation comparison proves exact equality.
"""
import json
from pathlib import Path
from typing import Any, Dict, List
from .coff_compare import (
CoffError,
load,
section_info,
section_info_resolved,
section_infos_equal,
)
class SemanticProgressError(RuntimeError):
pass
def _percent(numerator: int, denominator: int) -> float:
return 100.0 * numerator / denominator if denominator else 0.0
def _credit(measures: Dict[str, Any], code_bytes: int) -> None:
measures["matched_code"] = int(measures.get("matched_code", 0)) + code_bytes
measures["matched_functions"] = int(measures.get("matched_functions", 0)) + 1
measures["matched_code_percent"] = _percent(
measures["matched_code"], int(measures.get("total_code", 0))
)
measures["matched_functions_percent"] = _percent(
measures["matched_functions"], int(measures.get("total_functions", 0))
)
def _debit(measures: Dict[str, Any], code_bytes: int) -> None:
measures["matched_code"] = int(measures.get("matched_code", 0)) - code_bytes
measures["matched_functions"] = int(measures.get("matched_functions", 0)) - 1
if measures["matched_code"] < 0 or measures["matched_functions"] < 0:
raise SemanticProgressError("semantic rejection would make progress negative")
measures["matched_code_percent"] = _percent(
measures["matched_code"], int(measures.get("total_code", 0))
)
measures["matched_functions_percent"] = _percent(
measures["matched_functions"], int(measures.get("total_functions", 0))
)
def _credit_data(measures: Dict[str, Any], data_bytes: int) -> None:
measures["matched_data"] = int(measures.get("matched_data", 0)) + data_bytes
total_data = int(measures.get("total_data", 0))
if measures["matched_data"] > total_data:
raise SemanticProgressError("semantic data credit exceeds total data")
measures["matched_data_percent"] = _percent(
measures["matched_data"], total_data)
def apply_semantic_rejections(
report: Dict[str, Any],
semantic_report_path: Path,
) -> List[str]:
"""Remove objdiff credits rejected by the stricter COFF-shape audit."""
if not semantic_report_path.is_file():
return []
semantic_report = json.loads(semantic_report_path.read_text(encoding="utf-8"))
report_units = {unit["name"]: unit for unit in report.get("units", [])}
categories = {item["id"]: item for item in report.get("categories", [])}
rejected = []
for entry in semantic_report.get("ordinary_rejected", []):
unit_name = entry["unit"]
function_name = entry["function"]
report_unit = report_units.get(unit_name)
if report_unit is None:
raise SemanticProgressError(f"semantic rejection unit not found: {unit_name}")
functions = [
function for function in report_unit.get("functions", [])
if function.get("name") == function_name
]
if len(functions) != 1 or functions[0].get("fuzzy_match_percent") != 100.0:
raise SemanticProgressError(
f"semantic rejection is not an objdiff exact function: "
f"{unit_name}:{function_name}"
)
code_bytes = int(functions[0]["size"])
_debit(report["measures"], code_bytes)
_debit(report_unit["measures"], code_bytes)
progress_categories = report_unit.get("metadata", {}).get(
"progress_categories", []
)
if isinstance(progress_categories, str):
progress_categories = [progress_categories]
for category_id in progress_categories:
if category_id not in categories:
raise SemanticProgressError(f"progress category not found: {category_id}")
_debit(categories[category_id]["measures"], code_bytes)
rejected.append(
f"{unit_name}:{function_name} (-{code_bytes} code bytes, -1 function)"
)
return rejected
def apply_semantic_matches(
report: Dict[str, Any],
project_root: Path,
manifest_path: Path,
objdiff_config_path: Path,
) -> List[str]:
"""Verify and credit manifest entries that objdiff did not count exactly.
Returns human-readable notes for credited entries. Missing, ambiguous, or
non-equal evidence raises instead of silently inflating progress.
"""
if not manifest_path.is_file():
return []
entries = json.loads(manifest_path.read_text(encoding="utf-8"))
objdiff = json.loads(objdiff_config_path.read_text(encoding="utf-8"))
report_units = {unit["name"]: unit for unit in report.get("units", [])}
config_units = {unit["name"]: unit for unit in objdiff.get("units", [])}
categories = {item["id"]: item for item in report.get("categories", [])}
credited = []
for entry in entries:
unit_name = entry["unit"]
function_name = entry["function"]
if unit_name not in report_units or unit_name not in config_units:
raise SemanticProgressError(f"semantic match unit not found: {unit_name}")
report_unit = report_units[unit_name]
functions = [
function for function in report_unit.get("functions", [])
if function.get("name") == function_name
]
if len(functions) != 1:
raise SemanticProgressError(
f"expected one report function {unit_name}:{function_name}, found {len(functions)}"
)
function = functions[0]
if function.get("fuzzy_match_percent") == 100.0:
continue
config_unit = config_units[unit_name]
try:
target = load(project_root / config_unit["target_path"])
base = load(project_root / config_unit["base_path"])
target_info = section_info(target, function_name)
base_info = section_info(base, function_name)
except (CoffError, KeyError, OSError) as error:
raise SemanticProgressError(
f"cannot verify semantic match {unit_name}:{function_name}: {error}"
) from error
if not section_infos_equal(target_info, base_info):
raise SemanticProgressError(
f"semantic match is no longer exact: {unit_name}:{function_name}"
)
code_bytes = int(function["size"])
_credit(report["measures"], code_bytes)
_credit(report_unit["measures"], code_bytes)
progress_categories = report_unit.get("metadata", {}).get("progress_categories", [])
if isinstance(progress_categories, str):
progress_categories = [progress_categories]
for category_id in progress_categories:
if category_id not in categories:
raise SemanticProgressError(f"progress category not found: {category_id}")
_credit(categories[category_id]["measures"], code_bytes)
credited.append(
f"{unit_name}:{function_name} (+{code_bytes} code bytes, +1 function)"
)
return credited
def apply_semantic_data_matches(
report: Dict[str, Any],
project_root: Path,
manifest_path: Path,
objdiff_config_path: Path,
symbol_manifest_path: Path,
) -> List[str]:
"""Verify and credit executable-split data relocation aliases.
A manifest entry is accepted only when the target and rebuilt data
sections have identical normalized bytes, relocation locations/types,
and independently resolved final image destinations. The credited
section must account for the unit's entire remaining unmatched data, and
the unit must already be explicitly marked complete in objdiff config.
"""
if not manifest_path.is_file():
return []
entries = json.loads(manifest_path.read_text(encoding="utf-8"))
objdiff = json.loads(objdiff_config_path.read_text(encoding="utf-8"))
symbol_entries = json.loads(symbol_manifest_path.read_text(encoding="utf-8"))
symbol_addresses = {}
for item in symbol_entries:
name = item["name"]
address = int(item["file_offset"])
if name in symbol_addresses:
raise SemanticProgressError(
f"duplicate image symbol address for {name}")
symbol_addresses[name] = address
report_units = {unit["name"]: unit for unit in report.get("units", [])}
config_units = {unit["name"]: unit for unit in objdiff.get("units", [])}
categories = {item["id"]: item for item in report.get("categories", [])}
credited = []
for entry in entries:
unit_name = entry["unit"]
section_symbol = entry["symbol"]
if unit_name not in report_units or unit_name not in config_units:
raise SemanticProgressError(
f"semantic data match unit not found: {unit_name}")
report_unit = report_units[unit_name]
config_unit = config_units[unit_name]
if not config_unit.get("metadata", {}).get("complete"):
raise SemanticProgressError(
f"semantic data unit is not marked complete: {unit_name}")
try:
target_info = section_info_resolved(
load(project_root / config_unit["target_path"]),
section_symbol,
symbol_addresses,
)
base_info = section_info_resolved(
load(project_root / config_unit["base_path"]),
section_symbol,
symbol_addresses,
)
except (CoffError, KeyError, OSError) as error:
raise SemanticProgressError(
f"cannot verify semantic data match {unit_name}:{section_symbol}: "
f"{error}") from error
if target_info != base_info:
raise SemanticProgressError(
f"semantic data match is no longer exact: "
f"{unit_name}:{section_symbol}")
expected = entry.get("measurements", {})
snapshot = {
key: target_info[key]
for key in ("size", "relocation_count", "normalized_sha256")
}
if expected != snapshot:
raise SemanticProgressError(
f"semantic data target snapshot changed: "
f"{unit_name}:{section_symbol}")
unit_measures = report_unit["measures"]
unmatched_data = (
int(unit_measures.get("total_data", 0))
- int(unit_measures.get("matched_data", 0))
)
if unmatched_data == 0:
continue
if unmatched_data != target_info["size"]:
raise SemanticProgressError(
f"semantic data section does not cover all unmatched data: "
f"{unit_name}:{section_symbol} covers {target_info['size']}, "
f"remaining {unmatched_data}")
_credit_data(report["measures"], unmatched_data)
_credit_data(unit_measures, unmatched_data)
progress_categories = report_unit.get("metadata", {}).get(
"progress_categories", [])
if isinstance(progress_categories, str):
progress_categories = [progress_categories]
for category_id in progress_categories:
if category_id not in categories:
raise SemanticProgressError(
f"semantic data category not found: {category_id}")
_credit_data(categories[category_id]["measures"], unmatched_data)
credited.append(
f"{unit_name}:{section_symbol} (+{unmatched_data} data bytes)")
return credited