← Files VeraARCHIVED FILE
modules/open-item-reconciliation/scripts/review_server.py
57.2 KB · Oct 2, 2026 · 00:29 UTC
from __future__ import annotations
import sys as _bootstrap_sys
_bootstrap_sys.dont_write_bytecode = True
_bootstrap_sys.pycache_prefix = (
r"Z:\__open_item_reconciliation_no_bytecode__"
if _bootstrap_sys.platform == "win32"
else "/dev/null/open-item-reconciliation"
)
import os as _bootstrap_os
_BOOTSTRAP_PATH = _bootstrap_os.path.join(
_bootstrap_os.path.dirname(_bootstrap_os.path.abspath(__file__)),
"implementation_bootstrap.py",
)
_BOOTSTRAP_NAMESPACE = {
"__file__": _BOOTSTRAP_PATH,
"__name__": "_open_item_reconciliation_implementation_bootstrap",
}
_bootstrap_stat = _bootstrap_os.lstat(_BOOTSTRAP_PATH)
if _bootstrap_stat.st_mode & 0o170000 != 0o100000 or _bootstrap_stat.st_nlink != 1:
raise RuntimeError(
"implementation bootstrap must be an ordinary single-link regular file"
)
_bootstrap_descriptor = _bootstrap_os.open(
_BOOTSTRAP_PATH,
_bootstrap_os.O_RDONLY | getattr(_bootstrap_os, "O_NOFOLLOW", 0),
)
try:
_bootstrap_open_stat = _bootstrap_os.fstat(_bootstrap_descriptor)
_bootstrap_identity = (
_bootstrap_stat.st_dev,
_bootstrap_stat.st_ino,
_bootstrap_stat.st_size,
_bootstrap_stat.st_mtime_ns,
_bootstrap_stat.st_nlink,
)
if _bootstrap_identity != (
_bootstrap_open_stat.st_dev,
_bootstrap_open_stat.st_ino,
_bootstrap_open_stat.st_size,
_bootstrap_open_stat.st_mtime_ns,
_bootstrap_open_stat.st_nlink,
):
raise RuntimeError("implementation bootstrap changed before it was read")
with _bootstrap_os.fdopen(
_bootstrap_descriptor,
"rb",
closefd=False,
) as _bootstrap_handle:
_bootstrap_source = _bootstrap_handle.read()
_bootstrap_after_stat = _bootstrap_os.fstat(_bootstrap_descriptor)
if (
_bootstrap_identity
!= (
_bootstrap_after_stat.st_dev,
_bootstrap_after_stat.st_ino,
_bootstrap_after_stat.st_size,
_bootstrap_after_stat.st_mtime_ns,
_bootstrap_after_stat.st_nlink,
)
or len(_bootstrap_source) != _bootstrap_after_stat.st_size
):
raise RuntimeError("implementation bootstrap changed while it was read")
finally:
_bootstrap_os.close(_bootstrap_descriptor)
# Execute only the pre-opened single-link bootstrap source.
exec( # nosec B102
compile(_bootstrap_source, _BOOTSTRAP_PATH, "exec"),
_BOOTSTRAP_NAMESPACE,
)
_BOOTSTRAP_NAMESPACE["activate_implementation_boundary"](
("locale_support", "reconciliation_helpers", "review_session")
)
_SCRIPTS_DIR = _bootstrap_os.path.dirname(_bootstrap_os.path.abspath(__file__))
if _SCRIPTS_DIR not in _bootstrap_sys.path:
_bootstrap_sys.path.insert(0, _SCRIPTS_DIR)
import argparse
import ipaddress
import json
import logging
import re
import shutil
import tempfile
import webbrowser
from datetime import datetime, timezone
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
from urllib.parse import urlparse
try:
from .audit_assurance import (
build_applied_review_authority,
build_review_payload_mapping,
capture_review_transition_predecessor,
retain_review_transition,
review_decision_fingerprint,
run_review_output_transaction,
validate_assurance_run,
)
except ImportError: # pragma: no cover - direct script/importlib support
from audit_assurance import ( # type: ignore
build_applied_review_authority,
build_review_payload_mapping,
capture_review_transition_predecessor,
retain_review_transition,
review_decision_fingerprint,
run_review_output_transaction,
validate_assurance_run,
)
from vera_assurance import (
load_client_engagement_context_file,
load_client_workflow_context_for_output,
)
__all__ = [
"apply_decisions",
"build_session_payload",
"main",
"render_review_html",
"save_decisions",
"serve_review",
]
SCHEMA_VERSION = "1.0"
PLUGIN_NAME = "open-item-reconciliation"
WORKFLOW_NAME = "open-item-reconciliation"
TOOL_VALIDATE = "validate_open_item_reconciliation_review"
TOOL_RENDER = "render_open_item_reconciliation_review"
TOOL_SAVE = "save_open_item_reconciliation_decisions"
TOOL_APPLY = "apply_open_item_reconciliation_decisions"
MAX_ITEMS = 2500
MAX_DECISION_TEXT_LENGTH = 10_000
ALLOWED_ACTIONS = {
"accept",
"reject",
"edit",
"mark_unclear",
"request_more_documents",
"skip",
}
ACTION_STATUSES = {
"accept": "accepted",
"reject": "rejected",
"edit": "edited",
"mark_unclear": "needs_evidence",
"request_more_documents": "needs_evidence",
"skip": "skipped",
}
BLOCKING_ACTIONS = {"reject", "mark_unclear", "request_more_documents"}
REQUIRED_REVIEW_ITEM_TYPES = {
"closure_evidence_review",
"probable_payment_review",
"manual_review",
"review_exception",
"check_exception",
}
LOGGER = logging.getLogger(__name__)
def _validate_loopback_host(host: str) -> str:
"""Enforce the fixed local-only security boundary before socket binding."""
normalized = host.strip()
if normalized.lower() == "localhost":
return "127.0.0.1"
try:
address = ipaddress.ip_address(normalized)
except ValueError as exc:
raise ValueError(
"review server host must be localhost or an IPv4 loopback address"
) from exc
if not address.is_loopback or address.version != 4:
raise ValueError(
"review server host must be localhost or an IPv4 loopback address"
)
return normalized
def _utc_now() -> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
def _read_json_object(path: Path, *, required: bool = False) -> dict[str, Any]:
if not path.exists():
if required:
raise ValueError(f"{path.name} is required in the output folder")
return {}
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
raise ValueError(f"{path.name} is not readable JSON") from exc
if not isinstance(payload, dict):
raise ValueError(f"{path.name} must contain a JSON object")
return payload
def _write_json(path: Path, payload: dict[str, Any]) -> None:
path.write_text(
json.dumps(payload, ensure_ascii=False, indent=2, default=str) + "\n",
encoding="utf-8",
)
def _output_dir(path: str | Path) -> Path:
requested = Path(path).expanduser()
if requested.is_symlink():
raise ValueError("output folder must be a real directory")
output_dir = requested.resolve()
if not output_dir.is_dir():
raise ValueError(f"output folder does not exist: {output_dir}")
return output_dir
def _bounded_optional_string(value: Any, field_path: str) -> str:
if value is None:
return ""
if not isinstance(value, str):
raise ValueError(f"{field_path} must be a string when provided")
if len(value) > MAX_DECISION_TEXT_LENGTH:
raise ValueError(f"{field_path} exceeds {MAX_DECISION_TEXT_LENGTH} characters")
return value.strip()
def _require_string(value: Any, field_path: str) -> str:
text = _bounded_optional_string(value, field_path)
if not text:
raise ValueError(f"{field_path} must be a non-empty string")
return text
def _validate_item(item: Any, index: int) -> dict[str, Any]:
if not isinstance(item, dict):
raise ValueError(f"review_payload.items[{index}] must be an object")
_require_string(item.get("id"), f"review_payload.items[{index}].id")
_require_string(item.get("item_type"), f"review_payload.items[{index}].item_type")
_require_string(item.get("title"), f"review_payload.items[{index}].title")
allowed_actions = item.get("allowed_actions")
if not isinstance(allowed_actions, list) or not allowed_actions:
raise ValueError(
f"review_payload.items[{index}].allowed_actions must be a non-empty array"
)
for action in allowed_actions:
if action not in ALLOWED_ACTIONS:
raise ValueError(
"review_payload.items"
f"[{index}].allowed_actions contains unsupported action: {action}"
)
recommended_action = item.get("recommended_action")
if recommended_action is not None and recommended_action not in ALLOWED_ACTIONS:
raise ValueError(
f"review_payload.items[{index}].recommended_action is not supported"
)
return item
def _validate_review_payload(review_payload: Any) -> dict[str, Any]:
if not isinstance(review_payload, dict):
raise ValueError("review_payload must be an object")
_require_string(
review_payload.get("schema_version"), "review_payload.schema_version"
)
if review_payload.get("plugin") != PLUGIN_NAME:
raise ValueError(f'review_payload.plugin must be "{PLUGIN_NAME}"')
_require_string(review_payload.get("workflow"), "review_payload.workflow")
_require_string(review_payload.get("run_id"), "review_payload.run_id")
items = review_payload.get("items")
if not isinstance(items, list):
raise ValueError("review_payload.items must be an array")
if len(items) > MAX_ITEMS:
raise ValueError(f"review_payload.items exceeds {MAX_ITEMS} items")
if review_payload.get("item_count") != len(items):
raise ValueError(
"review_payload.item_count must equal review_payload.items.length"
)
for index, item in enumerate(items):
_validate_item(item, index)
return review_payload
def _readable_client_context(directory: Path) -> dict[str, Any]:
output_root = next(
(
ancestor
for ancestor in (directory, *directory.parents)
if ancestor.name == "outputs"
and (ancestor.parent / "context.json").is_file()
),
None,
)
if output_root is None:
raise ValueError("Review is not inside a portable customer-folder run")
return load_client_engagement_context_file(
output_root.parent / "context.json",
expected_workflow_id="open-item-reconciliation",
output_dir=directory,
allowed_statuses=("running", "ready_for_review", "completed"),
)
def build_session_payload(
output_dir: str | Path, *, for_display: bool = False
) -> dict[str, Any]:
"""Load the review payload served to the local browser page."""
directory = _output_dir(output_dir)
run_intake = _read_json_object(directory / "run_intake.json", required=True)
read_only = False
if for_display and run_intake.get("path_reference") == "run_root_relative":
context = _readable_client_context(directory)
run = _read_json_object(Path(context["run_root"]) / "run.json", required=True)
read_only = run.get("status") != "running"
review_payload = _validate_review_payload(
_read_json_object(directory / "review_payload.json", required=True)
)
ui_decisions = _read_json_object(directory / "ui_decisions.json")
applied_decisions = _read_json_object(directory / "applied_decisions.json")
final_artifacts = _read_json_object(directory / "final_artifacts.json")
if (
run_intake.get("run_id")
and run_intake.get("run_id") != review_payload["run_id"]
):
raise ValueError("run_intake.run_id must match review_payload.run_id")
if applied_decisions and (
applied_decisions.get("run_id") != review_payload["run_id"]
or applied_decisions.get("plugin") != PLUGIN_NAME
):
raise ValueError("applied_decisions must belong to the current review run")
# Native regeneration recreates an uncollected decision template. The
# separately persisted applied decisions remain the saved review baseline.
# Do not replace an explicit later Save, including an empty one.
if (
applied_decisions
and ui_decisions.get("decision_source") == "not_collected"
and ui_decisions.get("decided_at") is None
and not ui_decisions.get("decisions")
):
decisions = applied_decisions.get("decisions", [])
ui_decisions = {
**ui_decisions,
"decisions": decisions,
"decision_count": len(decisions),
"decided_at": applied_decisions.get("applied_at"),
"decision_source": "applied_decisions",
}
return {
"widget_type": "open_item_reconciliation_review",
"local_review_read_only": read_only,
"run_intake": run_intake,
"review_payload": review_payload,
"ui_decisions": ui_decisions or _empty_ui_decisions(review_payload),
"applied_decisions": applied_decisions or None,
"final_artifacts": final_artifacts or None,
"decision_policy": {
"save_tool": TOOL_SAVE,
"apply_tool": TOOL_APPLY,
"can_persist": True,
"fallback": "local_review_server",
},
}
def _empty_ui_decisions(review_payload: dict[str, Any]) -> dict[str, Any]:
return {
"schema_version": review_payload.get("schema_version", SCHEMA_VERSION),
"plugin": review_payload.get("plugin", PLUGIN_NAME),
"workflow": review_payload.get("workflow", WORKFLOW_NAME),
"run_id": review_payload["run_id"],
"decided_at": None,
"decision_source": "not_collected",
"review_payload_path": "review_payload.json",
"decisions": [],
"decision_count": 0,
"item_count": review_payload["item_count"],
"status": "pending_review",
}
def _requested_documents(value: Any, field_path: str) -> list[str]:
if value is None:
return []
if not isinstance(value, list):
raise ValueError(f"{field_path} must be an array when provided")
documents: list[str] = []
for index, entry in enumerate(value):
document = _bounded_optional_string(entry, f"{field_path}[{index}]")
if not document:
raise ValueError(f"{field_path}[{index}] must be a non-empty string")
documents.append(document)
return documents
def _normalize_decision(
decision: Any,
*,
item_by_id: dict[str, dict[str, Any]],
seen_ids: set[str],
decided_at: str,
index: int,
) -> dict[str, Any]:
if not isinstance(decision, dict):
raise ValueError(f"decisions[{index}] must be an object")
item_id = _bounded_optional_string(
decision.get("item_id", decision.get("id")),
f"decisions[{index}].item_id",
)
if not item_id:
raise ValueError(f"decisions[{index}].item_id must be a non-empty string")
if item_id in seen_ids:
raise ValueError(f"decisions contains duplicate item_id: {item_id}")
seen_ids.add(item_id)
item = item_by_id.get(item_id)
if item is None:
raise ValueError(
f"decisions[{index}].item_id is not in review_payload.items: {item_id}"
)
action = _bounded_optional_string(
decision.get("action"),
f"decisions[{index}].action",
)
if action not in ALLOWED_ACTIONS:
raise ValueError(f"decisions[{index}].action is not supported: {action}")
if action not in item["allowed_actions"]:
raise ValueError(
f"decisions[{index}].action is not allowed for item {item_id}: {action}"
)
reviewer_note = _bounded_optional_string(
decision.get("reviewer_note", decision.get("note")),
f"decisions[{index}].reviewer_note",
)
edit_value = _bounded_optional_string(
decision.get("edit_value", decision.get("user_text")),
f"decisions[{index}].edit_value",
)
if action == "edit" and not edit_value:
raise ValueError(
f"decisions[{index}].edit_value is required when action is edit"
)
normalized = {
"item_id": item_id,
"item_type": item["item_type"],
"title": item["title"],
"action": action,
"status": ACTION_STATUSES[action],
"decided_at": decided_at,
}
requested_documents = _requested_documents(
decision.get("requested_documents"),
f"decisions[{index}].requested_documents",
)
if reviewer_note:
normalized["reviewer_note"] = reviewer_note
if edit_value:
normalized["edit_value"] = edit_value
if requested_documents:
normalized["requested_documents"] = requested_documents
return normalized
def _build_ui_decisions(
output_dir: str | Path,
input_args: dict[str, Any],
) -> dict[str, Any]:
session = build_session_payload(output_dir)
review_payload = session["review_payload"]
if not isinstance(input_args.get("decisions"), list):
raise ValueError("decisions must be an array")
raw_decisions = input_args["decisions"]
if len(raw_decisions) > review_payload["item_count"]:
raise ValueError("decisions cannot exceed review_payload.items.length")
decided_at = _utc_now()
item_by_id = {item["id"]: item for item in review_payload["items"]}
seen_ids: set[str] = set()
decisions = [
_normalize_decision(
decision,
item_by_id=item_by_id,
seen_ids=seen_ids,
decided_at=decided_at,
index=index,
)
for index, decision in enumerate(raw_decisions)
]
reviewer = _bounded_optional_string(input_args.get("reviewer"), "reviewer")
status = (
"pending_review"
if not decisions
else (
"reviewed"
if len(decisions) == review_payload["item_count"]
else "partial_review"
)
)
ui_decisions = {
"schema_version": review_payload["schema_version"],
"plugin": review_payload["plugin"],
"workflow": review_payload["workflow"],
"run_id": review_payload["run_id"],
"decided_at": decided_at if decisions else None,
"decision_source": "local_review_server",
"review_payload_path": "review_payload.json",
"decisions": decisions,
"decision_count": len(decisions),
"item_count": review_payload["item_count"],
"status": status,
}
if reviewer:
ui_decisions["reviewer"] = reviewer
return ui_decisions
def _save_decisions_in_place(
output_dir: str | Path,
input_args: dict[str, Any],
) -> dict[str, Any]:
"""Validate and persist browser review decisions to ui_decisions.json."""
directory = _output_dir(output_dir)
ui_decisions = _build_ui_decisions(directory, input_args)
decision_output_path = directory / "ui_decisions.json"
_write_json(decision_output_path, ui_decisions)
return {
"ok": True,
"validation_type": "open_item_reconciliation_decisions",
"run_id": ui_decisions["run_id"],
"decision_count": ui_decisions["decision_count"],
"item_count": ui_decisions["item_count"],
"status": ui_decisions["status"],
"persisted": True,
"ui_decisions_path": decision_output_path.as_posix(),
"message": (
f"Saved {ui_decisions['decision_count']} Open-item Reconciliation decisions."
),
"ui_decisions": ui_decisions,
}
def save_decisions(
output_dir: str | Path,
input_args: dict[str, Any],
) -> dict[str, Any]:
"""Persist browser decisions under an exact whole-tree transaction."""
directory = _output_dir(output_dir)
if (
_read_json_object(directory / "run_intake.json").get("path_reference")
== "run_root_relative"
):
load_client_workflow_context_for_output(
directory, expected_workflow_id=PLUGIN_NAME
)
return run_review_output_transaction(
directory,
lambda working_dir: _save_decisions_in_place(working_dir, input_args),
)
def _short_string(value: Any) -> str:
return value.strip() if isinstance(value, str) else ""
def _application_effect(
decision: dict[str, Any],
item: dict[str, Any],
applied_at: str,
) -> dict[str, Any]:
data = item.get("data") if isinstance(item.get("data"), dict) else {}
target_artifact = (
_short_string(data.get("target_artifact"))
or _short_string(item.get("output_path"))
or _short_string(data.get("path"))
)
target_path = (
_short_string(data.get("target_path"))
or _short_string(data.get("field_path"))
or _short_string(data.get("field"))
)
target_id_field = _short_string(data.get("target_id_field")) or _short_string(
data.get("record_id_field")
)
target_record_id = _short_string(data.get("target_record_id")) or _short_string(
data.get("record_id")
)
target_field = _short_string(data.get("target_field")) or _short_string(
data.get("edit_field")
)
target_records_key = _short_string(data.get("target_records_key")) or _short_string(
data.get("records_key")
)
requires_followup = decision["action"] in BLOCKING_ACTIONS
effect = {
"item_id": decision["item_id"],
"item_type": decision["item_type"],
"title": decision["title"],
"action": decision["action"],
"status": decision["status"],
"applied_at": applied_at,
"applied": True,
"requires_followup": requires_followup,
"target_artifact": target_artifact or None,
"target_path": target_path or None,
"target_id_field": target_id_field or None,
"target_record_id": target_record_id or None,
"target_field": target_field or None,
"target_records_key": target_records_key or None,
"source_path": _short_string(item.get("source_path")) or None,
"artifact_update": (
"revision_artifact_pending"
if decision["action"] == "edit"
else ("decision_manifest_only" if target_artifact else "review_record_only")
),
}
for field in ("reviewer_note", "edit_value", "requested_documents"):
if field in decision:
effect[field] = decision[field]
return effect
def _safe_path_segment(value: Any, fallback: str) -> str:
cleaned = re.sub(r"[^a-zA-Z0-9._-]+", "-", _short_string(value)).strip("-._")
return cleaned or fallback
def _resolve_safe_output_path(output_dir: Path, value: Any) -> tuple[str, Path] | None:
raw_path = _short_string(value)
if not raw_path or Path(raw_path).is_absolute():
return None
candidate = (output_dir / raw_path).resolve()
try:
relative = candidate.relative_to(output_dir.resolve()).as_posix()
except ValueError:
return None
return relative, candidate
def _structured_update_spec(effect: dict[str, Any]) -> dict[str, str] | None:
if effect.get("action") != "edit" or not _short_string(effect.get("edit_value")):
return None
if not (
effect.get("target_artifact")
and effect.get("target_id_field")
and effect.get("target_record_id")
and effect.get("target_field")
):
return None
return {
"id_field": _short_string(effect.get("target_id_field")),
"record_id": _short_string(effect.get("target_record_id")),
"target_field": _short_string(effect.get("target_field")),
"records_key": _short_string(effect.get("target_records_key")),
}
def _update_matching_record(
records: list[Any],
spec: dict[str, str],
edit_value: str,
) -> int:
updated_rows = 0
for record in records:
if not isinstance(record, dict):
continue
if str(record.get(spec["id_field"]) or "") != spec["record_id"]:
continue
record[spec["target_field"]] = edit_value
updated_rows += 1
if updated_rows != 1:
raise ValueError(
"structured edit expected exactly one row for "
f"{spec['id_field']}={spec['record_id']}, found {updated_rows}"
)
return updated_rows
def _update_json_artifact(
path: Path,
effect: dict[str, Any],
spec: dict[str, str],
) -> tuple[int, int]:
payload = json.loads(path.read_text(encoding="utf-8"))
edit_value = _short_string(effect.get("edit_value"))
if isinstance(payload, list):
updated_rows = _update_matching_record(payload, spec, edit_value)
path.write_text(
json.dumps(payload, ensure_ascii=False, indent=2, default=str) + "\n",
encoding="utf-8",
)
return updated_rows, len(payload)
if (
isinstance(payload, dict)
and spec["records_key"]
and isinstance(payload.get(spec["records_key"]), list)
):
records = payload[spec["records_key"]]
updated_rows = _update_matching_record(records, spec, edit_value)
path.write_text(
json.dumps(payload, ensure_ascii=False, indent=2, default=str) + "\n",
encoding="utf-8",
)
return updated_rows, len(records)
if (
isinstance(payload, dict)
and str(payload.get(spec["id_field"]) or "") == spec["record_id"]
):
payload[spec["target_field"]] = edit_value
path.write_text(
json.dumps(payload, ensure_ascii=False, indent=2, default=str) + "\n",
encoding="utf-8",
)
return 1, 1
raise ValueError("JSON structured edit requires an object, array, or records key")
def _original_backup_path(effect: dict[str, Any], target_relative_path: str) -> str:
target = Path(target_relative_path)
suffix = target.suffix or ".json"
stem = target.stem or "artifact"
item_id = _safe_path_segment(effect.get("item_id"), "item")
return (Path("revisions") / "originals" / f"{stem}__{item_id}{suffix}").as_posix()
def _write_structured_artifact_updates(
output_dir: Path,
effects: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
target_outputs: list[dict[str, Any]] = []
backup_outputs: list[dict[str, Any]] = []
for effect in effects:
spec = _structured_update_spec(effect)
if spec is None:
continue
resolved = _resolve_safe_output_path(output_dir, effect.get("target_artifact"))
if resolved is None:
continue
target_relative_path, target_path = resolved
if target_path.suffix.lower() != ".json" or not target_path.exists():
continue
backup_relative_path = _original_backup_path(effect, target_relative_path)
backup_path = output_dir / backup_relative_path
backup_path.parent.mkdir(parents=True, exist_ok=True)
if not backup_path.exists():
shutil.copy2(target_path, backup_path)
updated_rows, row_count = _update_json_artifact(target_path, effect, spec)
effect["target_artifact"] = target_relative_path
effect["original_artifact_backup"] = backup_relative_path
effect["artifact_update"] = "structured_artifact_updated"
effect["structured_update"] = {
"id_field": spec["id_field"],
"record_id": spec["record_id"],
"target_field": spec["target_field"],
"records_key": spec["records_key"],
"updated_rows": updated_rows,
}
target_outputs.append(
{
"path": target_relative_path,
"kind": "json",
"status": "updated_from_review",
"item_id": effect["item_id"],
"row_count": row_count,
"required_columns": [spec["id_field"], spec["target_field"]],
}
)
backup_outputs.append(
{
"path": backup_relative_path,
"kind": "json",
"status": "backup_original",
"source_artifact": target_relative_path,
"item_id": effect["item_id"],
}
)
return target_outputs, backup_outputs
def _review_decision_fingerprint(
review_payload: dict[str, Any],
effects: list[dict[str, Any]],
) -> str:
"""Identify the semantic decision set independently of application time."""
return review_decision_fingerprint(
build_review_payload_mapping(review_payload),
effects,
)
def _professional_review_from_effects(
*,
output_dir: Path,
assurance: dict[str, Any],
review_payload: dict[str, Any],
effects: list[dict[str, Any]],
reviewer: str | None,
decision_fingerprint: str,
) -> dict[str, Any]:
"""Bind accepted review actions one-to-one to the sealed source-row set."""
reconciliation_result = _read_json_object(
output_dir
/ str(assurance["final_output_inventory"]["boundary_root"])
/ "reconciliation_results.json",
required=True,
)
authority = build_applied_review_authority(
predecessor_assurance=assurance,
predecessor_professional_review=assurance["professional_review_authority"],
predecessor_reconciliation=reconciliation_result,
review_payload_mapping=build_review_payload_mapping(review_payload),
effects=effects,
reviewer_ref=reviewer,
)
if authority["decision_fingerprint"] != decision_fingerprint:
raise ValueError("professional review decision fingerprint is stale")
return authority
def _completion_blockers(
review_payload: dict[str, Any],
effects: list[dict[str, Any]],
output_dir: Path | None = None,
decision_fingerprint: str | None = None,
expected_predecessor_checkpoint: str | None = None,
) -> list[dict[str, str]]:
"""Return unresolved run gates that reviewer clicks cannot override."""
blockers: list[dict[str, str]] = []
summary = (
review_payload.get("summary")
if isinstance(review_payload.get("summary"), dict)
else {}
)
if summary.get("checks_pass") is False or int(
summary.get("failed_check_count") or 0
):
blockers.append(
{
"kind": "failed_deterministic_checks",
"detail": "Failed deterministic checks require correction and rerun.",
}
)
effect_by_id = {effect["item_id"]: effect for effect in effects}
for item in review_payload.get("items", []):
if (
not isinstance(item, dict)
or item.get("item_type") not in REQUIRED_REVIEW_ITEM_TYPES
):
continue
effect = effect_by_id.get(item.get("id"))
if effect is None:
blockers.append(
{
"kind": "pending_required_review",
"detail": f"Required review item remains pending: {item.get('id')}",
}
)
elif effect.get("action") == "skip":
blockers.append(
{
"kind": "skipped_required_review",
"detail": f"Required review item was skipped: {item.get('id')}",
}
)
assurance_path = (
output_dir / "assurance_gates.json" if output_dir is not None else None
)
successor_replayed = False
if assurance_path is not None and assurance_path.is_file():
try:
assurance = validate_assurance_run(
output_dir,
expected_predecessor_checkpoint=expected_predecessor_checkpoint,
)
except ValueError:
blockers.append(
{
"kind": "assurance_replay_failed",
"detail": (
"The latest assurance seal did not replay; native outputs "
"must be regenerated before final readiness."
),
}
)
else:
register = assurance.get("gate_register")
gates = register.get("gates") if isinstance(register, dict) else None
semantic = gates.get("semantic_review") if isinstance(gates, dict) else None
reporting = gates.get("reporting") if isinstance(gates, dict) else None
if (
not isinstance(semantic, dict)
or semantic.get("status") not in {"passed", "not_applicable"}
or not isinstance(reporting, dict)
or reporting.get("status") != "passed"
):
blockers.append(
{
"kind": "assurance_replay_failed",
"detail": (
"The prior assurance gates were not report-ready; "
"native outputs must be regenerated."
),
}
)
authority = assurance.get("professional_review_authority")
successor_replayed = bool(
isinstance(authority, dict)
and authority.get("origin") == "applied_decisions"
and decision_fingerprint
and authority.get("decision_fingerprint") == decision_fingerprint
and assurance.get("gate_register", {}).get("report_ready") is True
)
if not successor_replayed:
blockers.append(
{
"kind": "assurance_replay_required",
"detail": (
"Reviewed decisions require native regeneration and a fresh "
"successor assurance receipt replay before final readiness."
),
}
)
return blockers
def _application_status(
effects: list[dict[str, Any]],
item_count: int,
completion_blockers: list[dict[str, str]],
) -> str:
if not effects:
return "pending_review"
if any(effect["requires_followup"] for effect in effects):
return "blocked"
if completion_blockers:
return "blocked"
if len(effects) < item_count:
return "partial_review_applied"
return "final_ready"
def _upsert_output(
outputs: list[dict[str, Any]],
record: dict[str, Any],
) -> list[dict[str, Any]]:
for index, output in enumerate(outputs):
if isinstance(output, dict) and output.get("path") == record["path"]:
outputs[index] = {**output, **record}
return outputs
outputs.append(record)
return outputs
def _final_artifacts_with_application(
output_dir: Path,
applied_decisions: dict[str, Any],
target_outputs: list[dict[str, Any]] | None = None,
backup_outputs: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
current = _read_json_object(output_dir / "final_artifacts.json")
outputs = list(
current.get("outputs") if isinstance(current.get("outputs"), list) else []
)
_upsert_output(
outputs,
{"path": "ui_decisions.json", "kind": "json", "status": "written_reviewed"},
)
_upsert_output(
outputs,
{
"path": "applied_decisions.json",
"kind": "json",
"status": applied_decisions["application_status"],
},
)
professional_review = applied_decisions.get("professional_review")
if isinstance(professional_review, dict):
_upsert_output(
outputs,
{
"path": "professional_review.json",
"kind": "json",
"status": (
"successor_assurance_replayed"
if professional_review.get("successor_assurance_replayed")
else "written_pending_successor_reseal"
),
},
)
for output in target_outputs or []:
_upsert_output(outputs, output)
for output in backup_outputs or []:
_upsert_output(outputs, output)
final_artifacts = {
**current,
"schema_version": current.get("schema_version", SCHEMA_VERSION),
"plugin": current.get("plugin", PLUGIN_NAME),
"workflow": current.get("workflow", WORKFLOW_NAME),
"run_id": current.get("run_id", applied_decisions["run_id"]),
"outputs": outputs,
"blockers": applied_decisions.get("completion_blockers", []),
"status": applied_decisions["application_status"],
"review_application": {
"applied_at": applied_decisions["applied_at"],
"application_status": applied_decisions["application_status"],
"decision_count": applied_decisions["decision_count"],
"item_count": applied_decisions["item_count"],
"blocker_count": applied_decisions["blocker_count"],
"completion_blockers": applied_decisions.get(
"completion_blockers",
[],
),
"target_update_count": applied_decisions.get("target_update_count", 0),
"target_update_paths": applied_decisions.get("target_update_paths", []),
"structured_update_count": applied_decisions.get(
"structured_update_count", 0
),
"structured_update_paths": applied_decisions.get(
"structured_update_paths", []
),
"original_backup_paths": applied_decisions.get("original_backup_paths", []),
"applied_decisions_path": "applied_decisions.json",
"professional_review_path": (
"professional_review.json"
if isinstance(professional_review, dict)
else None
),
"decision_fingerprint": (
professional_review.get("decision_fingerprint")
if isinstance(professional_review, dict)
else None
),
"successor_assurance_replayed": (
professional_review.get("successor_assurance_replayed") is True
if isinstance(professional_review, dict)
else False
),
},
}
return final_artifacts
def _unique_strings(values: list[Any]) -> list[str]:
seen: set[str] = set()
unique: list[str] = []
for value in values:
if not isinstance(value, str):
continue
text = value.strip()
if not text or text in seen:
continue
seen.add(text)
unique.append(text)
return unique
def _review_application_trace_outputs(
applied_decisions: dict[str, Any],
final_artifacts: dict[str, Any],
) -> list[str]:
outputs: list[Any] = [
"ui_decisions.json",
"applied_decisions.json",
"final_artifacts.json",
]
if isinstance(applied_decisions.get("professional_review"), dict):
outputs.append("professional_review.json")
review_application = final_artifacts.get("review_application")
if not isinstance(review_application, dict):
review_application = {}
for field_name in [
"applied_decisions_path",
"revision_paths",
"target_update_paths",
"structured_update_paths",
"native_regeneration_paths",
"native_regenerated_paths",
"downstream_regenerated_paths",
"original_backup_paths",
]:
value = review_application.get(field_name, applied_decisions.get(field_name))
if isinstance(value, list):
outputs.extend(value)
else:
outputs.append(value)
return _unique_strings(outputs)
def _append_review_application_trace(
output_dir: Path,
applied_decisions: dict[str, Any],
final_artifacts: dict[str, Any],
) -> Path | None:
"""Record the deterministic local decision-application step for auditability."""
run_intake_path = output_dir / "run_intake.json"
run_intake = _read_json_object(run_intake_path)
if not run_intake:
return None
trace = run_intake.get("execution_trace")
if not isinstance(trace, list):
trace = []
applied_at = str(applied_decisions.get("applied_at") or _utc_now())
step_id_suffix = re.sub(r"[^A-Za-z0-9]+", "_", applied_at).strip("_")
trace.append(
{
"step_id": f"{WORKFLOW_NAME}_review_apply_{step_id_suffix}",
"kind": "deterministic_review_apply",
"status": "passed",
"execution_location": "local_codex_workspace",
"command": ["open-item-reconciliation-review-server", "apply_decisions"],
"inputs": _unique_strings(
[
applied_decisions.get("review_payload", {}).get(
"path", "review_payload.json"
),
"ui_decisions.json",
"final_artifacts.json",
]
),
"outputs": _review_application_trace_outputs(
applied_decisions,
final_artifacts,
),
}
)
run_intake["execution_trace"] = trace
_write_json(run_intake_path, run_intake)
return run_intake_path
def _apply_decisions_in_place(
output_dir: str | Path,
input_args: dict[str, Any],
) -> dict[str, Any]:
"""Persist browser review decisions and write application manifests."""
directory = _output_dir(output_dir)
ui_decisions = _build_ui_decisions(directory, input_args)
review_payload = build_session_payload(directory)["review_payload"]
item_by_id = {item["id"]: item for item in review_payload["items"]}
applied_at = _utc_now()
effects = [
_application_effect(decision, item_by_id[decision["item_id"]], applied_at)
for decision in ui_decisions["decisions"]
]
decision_fingerprint = _review_decision_fingerprint(review_payload, effects)
expected_predecessor_checkpoint = input_args.get("expected_predecessor_checkpoint")
assurance_path = directory / "assurance_receipts.json"
assurance = (
validate_assurance_run(
directory,
expected_predecessor_checkpoint=(
expected_predecessor_checkpoint
if isinstance(expected_predecessor_checkpoint, str)
else None
),
)
if assurance_path.exists() or assurance_path.is_symlink()
else None
)
current_authority = (
assurance.get("professional_review_authority")
if isinstance(assurance, dict)
else None
)
successor_replayed = bool(
isinstance(current_authority, dict)
and current_authority.get("origin") == "applied_decisions"
and current_authority.get("decision_fingerprint") == decision_fingerprint
and isinstance(assurance, dict)
and assurance.get("gate_register", {}).get("report_ready") is True
)
professional_review = (
current_authority
if successor_replayed
else (
_professional_review_from_effects(
output_dir=directory,
assurance=assurance,
review_payload=review_payload,
effects=effects,
reviewer=_short_string(ui_decisions.get("reviewer")) or None,
decision_fingerprint=decision_fingerprint,
)
if isinstance(assurance, dict)
else None
)
)
transition_capture: Path | None = None
if isinstance(assurance, dict) and not successor_replayed:
transition_capture = Path(
tempfile.mkdtemp(
prefix=".audit-review-transition-capture-",
dir=directory.parent,
)
)
capture_review_transition_predecessor(
directory,
transition_capture,
expected_predecessor_checkpoint=(
expected_predecessor_checkpoint
if isinstance(expected_predecessor_checkpoint, str)
else None
),
)
target_outputs, backup_outputs = _write_structured_artifact_updates(
directory,
effects,
)
structured_update_paths = [
effect["target_artifact"]
for effect in effects
if effect.get("artifact_update") == "structured_artifact_updated"
]
completion_blockers = _completion_blockers(
review_payload,
effects,
directory,
decision_fingerprint,
(
expected_predecessor_checkpoint
if isinstance(expected_predecessor_checkpoint, str)
else None
),
)
blocker_count = sum(1 for effect in effects if effect["requires_followup"]) + len(
completion_blockers
)
application_status = _application_status(
effects,
review_payload["item_count"],
completion_blockers,
)
applied_decisions = {
"schema_version": review_payload["schema_version"],
"plugin": review_payload["plugin"],
"workflow": review_payload["workflow"],
"run_id": review_payload["run_id"],
"applied_at": applied_at,
"decision_source": "local_review_server",
"review_payload": {
"path": ui_decisions["review_payload_path"],
"item_count": review_payload["item_count"],
"review_type": review_payload.get("review_type"),
},
"decisions": ui_decisions["decisions"],
"effects": effects,
"decision_count": ui_decisions["decision_count"],
"item_count": review_payload["item_count"],
"blocker_count": blocker_count,
"completion_blockers": completion_blockers,
"target_update_count": len(target_outputs),
"target_update_paths": [output["path"] for output in target_outputs],
"structured_update_count": len(structured_update_paths),
"structured_update_paths": structured_update_paths,
"original_backup_paths": [output["path"] for output in backup_outputs],
"application_status": application_status,
}
if professional_review is not None:
applied_decisions["professional_review"] = {
"path": "professional_review.json",
"decision_fingerprint": decision_fingerprint,
"successor_assurance_replayed": successor_replayed,
}
if "reviewer" in ui_decisions:
applied_decisions["reviewer"] = ui_decisions["reviewer"]
final_artifacts = _final_artifacts_with_application(
directory,
applied_decisions,
target_outputs,
backup_outputs,
)
ui_decisions_path = directory / "ui_decisions.json"
applied_decisions_path = directory / "applied_decisions.json"
final_artifacts_path = directory / "final_artifacts.json"
professional_review_path = directory / "professional_review.json"
_write_json(ui_decisions_path, ui_decisions)
_write_json(applied_decisions_path, applied_decisions)
if professional_review is not None and not successor_replayed:
_write_json(professional_review_path, professional_review)
if transition_capture is not None:
retain_review_transition(
directory,
transition_capture,
expected_predecessor_checkpoint=(
expected_predecessor_checkpoint
if isinstance(expected_predecessor_checkpoint, str)
else None
),
)
shutil.rmtree(transition_capture)
_write_json(final_artifacts_path, final_artifacts)
run_intake_path = _append_review_application_trace(
directory,
applied_decisions,
final_artifacts,
)
return {
"ok": True,
"validation_type": "open_item_reconciliation_application",
"run_id": applied_decisions["run_id"],
"decision_count": applied_decisions["decision_count"],
"item_count": applied_decisions["item_count"],
"blocker_count": blocker_count,
"target_update_count": applied_decisions["target_update_count"],
"structured_update_count": applied_decisions["structured_update_count"],
"application_status": application_status,
"persisted": True,
"ui_decisions_path": ui_decisions_path.as_posix(),
"applied_decisions_path": applied_decisions_path.as_posix(),
"final_artifacts_path": final_artifacts_path.as_posix(),
"professional_review_path": (
professional_review_path.as_posix()
if professional_review is not None
else None
),
"run_intake_path": run_intake_path.as_posix() if run_intake_path else None,
"message": (
f"Applied {applied_decisions['decision_count']} "
"Open-item Reconciliation decisions."
),
"applied_decisions": applied_decisions,
"final_artifacts": final_artifacts,
}
def apply_decisions(
output_dir: str | Path,
input_args: dict[str, Any],
) -> dict[str, Any]:
"""Apply browser decisions under an exact whole-tree transaction."""
directory = _output_dir(output_dir)
if (
_read_json_object(directory / "run_intake.json").get("path_reference")
== "run_root_relative"
):
load_client_workflow_context_for_output(
directory, expected_workflow_id=PLUGIN_NAME
)
return run_review_output_transaction(
directory,
lambda working_dir: _apply_decisions_in_place(working_dir, input_args),
expected_predecessor_checkpoint=(
input_args.get("expected_predecessor_checkpoint")
if isinstance(input_args.get("expected_predecessor_checkpoint"), str)
else None
),
)
def _widget_html(output_dir: Path) -> str:
widget_path = (
Path(__file__).resolve().parents[1]
/ "assets"
/ "open-item-reconciliation-review-widget.html"
)
html = widget_path.read_text(encoding="utf-8")
payload_json = json.dumps(
build_session_payload(output_dir, for_display=True),
ensure_ascii=False,
default=str,
)
bridge = f"""<script>
(function () {{
const serverPayload = {payload_json};
const stateKey = `open-item-reconciliation:${{serverPayload.review_payload?.run_id || "run"}}`;
function readState() {{
try {{ return JSON.parse(window.sessionStorage.getItem(stateKey) || "null"); }}
catch {{ return null; }}
}}
window.openai = {{
toolOutput: serverPayload,
widgetState: readState(),
setWidgetState(value) {{
try {{ window.sessionStorage.setItem(stateKey, JSON.stringify(value || null)); }}
catch {{ }}
}},
async callTool(name, args) {{
const response = await fetch("/api/call-tool", {{
method: "POST",
headers: {{ "Content-Type": "application/json" }},
body: JSON.stringify({{ name, args: args || {{}} }}),
}});
const result = await response.json();
if (!response.ok || result.ok === false) {{
throw new Error(result.error || `Local review server call failed: ${{name}}`);
}}
if (result.ui_decisions) serverPayload.ui_decisions = result.ui_decisions;
if (result.final_artifacts) serverPayload.final_artifacts = result.final_artifacts;
if (result.applied_decisions) serverPayload.applied_decisions = result.applied_decisions;
return result;
}},
}};
}}());
</script>
"""
needle = " <script>\n const CONFIG = "
if needle not in html:
raise ValueError(
"open-item reconciliation widget script insertion point not found"
)
return html.replace(needle, bridge + needle, 1)
def render_review_html(output_dir: str | Path) -> str:
"""Render the browser review page HTML with the local persistence bridge."""
return _widget_html(_output_dir(output_dir))
def _tool_result(output_dir: Path, payload: dict[str, Any]) -> dict[str, Any]:
name = _bounded_optional_string(payload.get("name"), "name")
args = payload.get("args") if isinstance(payload.get("args"), dict) else {}
if name == TOOL_VALIDATE:
session = build_session_payload(output_dir, for_display=True)
return {
"ok": True,
"validation_type": "open_item_reconciliation_review",
"run_id": session["review_payload"]["run_id"],
"item_count": session["review_payload"]["item_count"],
"review_type": session["review_payload"].get("review_type"),
"message": "Open-item Reconciliation review payload is valid.",
"review_payload": session["review_payload"],
}
if name == TOOL_RENDER:
return build_session_payload(output_dir, for_display=True)
if name == TOOL_SAVE:
return save_decisions(output_dir, args)
if name == TOOL_APPLY:
return apply_decisions(output_dir, args)
raise ValueError(f"unknown Open-item Reconciliation widget tool: {name}")
def _handler(output_dir: Path) -> type[BaseHTTPRequestHandler]:
class ReviewServerHandler(BaseHTTPRequestHandler):
server_version = "AuditReconciliationReviewServer/1.0"
def log_message(self, format_string: str, *args: object) -> None:
LOGGER.info("%s - %s", self.client_address[0], format_string % args)
def _json_response(
self,
payload: dict[str, Any],
*,
status: HTTPStatus = HTTPStatus.OK,
) -> None:
body = json.dumps(payload, ensure_ascii=False, default=str).encode("utf-8")
self.send_response(status.value)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(body)
def _html_response(self, html: str) -> None:
body = html.encode("utf-8")
self.send_response(HTTPStatus.OK.value)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(body)
def do_GET(self) -> None:
route = urlparse(self.path).path
try:
if route in {"/", "/review", "/review_ui.html"}:
self._html_response(_widget_html(output_dir))
return
if route == "/api/session":
self._json_response(
build_session_payload(output_dir, for_display=True)
)
return
if route == "/api/health":
session = build_session_payload(output_dir, for_display=True)
self._json_response(
{
"ok": True,
"plugin": PLUGIN_NAME,
"run_id": session["review_payload"]["run_id"],
"output_dir": output_dir.as_posix(),
}
)
return
self.send_error(HTTPStatus.NOT_FOUND.value, "Not found")
except (OSError, TypeError, ValueError) as exc:
self._json_response(
{"ok": False, "error": str(exc)},
status=HTTPStatus.INTERNAL_SERVER_ERROR,
)
def do_POST(self) -> None:
route = urlparse(self.path).path
if route != "/api/call-tool":
self.send_error(HTTPStatus.NOT_FOUND.value, "Not found")
return
try:
length = int(self.headers.get("Content-Length", "0"))
payload = json.loads(self.rfile.read(length).decode("utf-8"))
if not isinstance(payload, dict):
raise ValueError("request body must be a JSON object")
self._json_response(_tool_result(output_dir, payload))
except json.JSONDecodeError as exc:
self._json_response(
{"ok": False, "error": f"invalid JSON request: {exc}"},
status=HTTPStatus.BAD_REQUEST,
)
except (OSError, TypeError, ValueError) as exc:
self._json_response(
{"ok": False, "error": str(exc)},
status=HTTPStatus.BAD_REQUEST,
)
return ReviewServerHandler
def serve_review(
output_dir: str | Path,
*,
host: str = "127.0.0.1",
port: int = 0,
open_browser: bool = True,
) -> None:
"""Serve the review UI on localhost and optionally open the system browser."""
directory = _output_dir(output_dir)
safe_host = _validate_loopback_host(host)
_readable_client_context(directory)
build_session_payload(directory, for_display=True)
httpd = ThreadingHTTPServer((safe_host, port), _handler(directory))
actual_port = httpd.server_address[1]
url = f"http://{safe_host}:{actual_port}/review"
LOGGER.info("Open-item Reconciliation review server: %s", url)
LOGGER.info("Output folder: %s", directory)
if open_browser:
webbrowser.open(url)
try:
httpd.serve_forever()
except KeyboardInterrupt:
LOGGER.info("Stopping Open-item Reconciliation review server")
finally:
httpd.server_close()
def _parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description=(
"Open the Open-item Reconciliation review UI in a local browser and "
"persist decisions into the run output folder."
)
)
parser.add_argument("output_dir", help="Run output folder with review_payload.json")
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=0)
parser.add_argument(
"--no-open",
action="store_true",
help="Start the server without opening the browser automatically.",
)
return parser
def main(argv: list[str] | None = None) -> int:
logging.basicConfig(level=logging.INFO, format="%(message)s")
args = _parser().parse_args(argv)
try:
serve_review(
args.output_dir,
host=args.host,
port=args.port,
open_browser=not args.no_open,
)
except (OSError, ValueError) as exc:
LOGGER.error("%s", exc)
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())
SHA-256: dcecb645b0be020893741b99b64099a70953ac17880dc05bbbb11cbe0c38d091