← Files VeraARCHIVED FILE
modules/new-client/scripts/review_server.py
42.9 KB · Oct 2, 2026 · 00:29 UTC
#!/usr/bin/env python3
"""Serve generated review-workbench widgets with local decision write-back."""
from __future__ import annotations
import argparse
import ipaddress
import json
import logging
import os
import re
import secrets
import shutil
import socket
import subprocess
import sys
import webbrowser
from dataclasses import dataclass
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
from urllib.parse import urlparse
__all__ = [
"LocalReviewWorkbench",
"build_session_payload",
"call_review_tool",
"create_review_http_server",
"main",
"render_review_html",
"serve_review",
]
ROOT = Path(__file__).resolve().parents[1]
MAX_ITEMS = 3000
MAX_POST_BYTES = 1_000_000
ALLOWED_ACTIONS = {
"accept",
"reject",
"edit",
"mark_unclear",
"request_more_documents",
"skip",
}
LOGGER = logging.getLogger(__name__)
NODE_OVERRIDE_ENV = "MPARANZA_REVIEW_NODE"
REVIEW_TOKEN_HEADER = "X-Mparanza-Review-Token"
REQUIRE_VERA_CUSTOMER_RUN = True
VERA_REVIEW_WORKFLOW_IDS = frozenset(
{
"archive-organization",
"open-item-reconciliation",
"check-entries",
"client-file-preparation",
"concordato-plan-review",
"deep-research-validator",
"financial-analysis",
"journal-bank-reconciliation",
"journal-sampling",
"new-client",
"previdenza-inps",
"prompt-optimizer",
"registro-imprese-sari",
"report-builder",
"sales-plan",
}
)
@dataclass(frozen=True)
class LocalReviewWorkbench:
"""Plugin/run paths for a local review-workbench session."""
plugin_dir: Path
output_dir: Path
@property
def plugin(self) -> str:
"""Return the plugin directory name."""
return self.plugin_dir.name
@property
def adapter_path(self) -> Path:
"""Return the workbench adapter path."""
return self.plugin_dir / "assets" / "review-workbench-adapter.json"
@property
def mcp_server_path(self) -> Path:
"""Return the available plugin review-tool server path."""
candidates = (
self.plugin_dir / "mcp" / "server.cjs",
self.plugin_dir / "scripts" / "review_mcp_server.cjs",
)
return next((path for path in candidates if path.is_file()), candidates[0])
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 _plugin_dir_from_args(plugin: str | None, plugin_dir: str | None) -> Path:
if plugin_dir:
directory = Path(plugin_dir).expanduser().resolve()
elif plugin:
repo_plugin_dir = (ROOT / "plugins" / plugin).resolve()
directory = repo_plugin_dir if repo_plugin_dir.exists() else ROOT.resolve()
else:
candidates = [
Path(__file__).resolve().parents[1],
Path.cwd().resolve(),
Path.cwd().resolve().parent,
]
directory = next(
(
candidate
for candidate in candidates
if (candidate / ".codex-plugin").exists()
),
Path(),
)
if not directory:
raise ValueError(
"pass --plugin or --plugin-dir when not running inside a plugin"
)
if not (directory / ".codex-plugin" / "plugin.json").exists():
raise ValueError(f"plugin directory is invalid: {directory}")
if plugin:
manifest = _read_json_object(directory / ".codex-plugin" / "plugin.json")
manifest_name = manifest.get("name")
if manifest_name != plugin:
raise ValueError(
f'plugin directory is for "{manifest_name}", not "{plugin}"'
)
if not (directory / "assets" / "review-workbench-adapter.json").exists():
raise ValueError(
f"plugin has no generated review workbench adapter: {directory}"
)
if not LocalReviewWorkbench(
plugin_dir=directory,
output_dir=directory,
).mcp_server_path.is_file():
raise ValueError(
f"plugin has no review-tool server for decision write-back: {directory}"
)
return directory
def _output_dir(path: str | Path) -> Path:
directory = Path(path).expanduser().resolve()
if not directory.is_dir():
raise ValueError(f"output folder does not exist: {directory}")
return directory
def _validate_vera_customer_run(
workbench: LocalReviewWorkbench,
*,
read_only: bool = False,
) -> dict[str, Any] | None:
"""Validate review ownership; only rendering may read a finalized run."""
if not REQUIRE_VERA_CUSTOMER_RUN:
return None
manifest = _read_json_object(
workbench.plugin_dir / ".codex-plugin" / "plugin.json",
required=True,
)
plugin = manifest.get("name")
if plugin not in VERA_REVIEW_WORKFLOW_IDS:
return None
module_roots = (
workbench.plugin_dir / "vendor" / "modules",
workbench.plugin_dir.parent / "_shared" / "vendor" / "modules",
)
module_root = next(
(root for root in module_roots if (root / "vera_assurance").is_dir()),
None,
)
if module_root is None:
raise ValueError("Vera customer-run validator is unavailable")
if str(module_root) not in sys.path:
sys.path.insert(0, str(module_root))
from vera_assurance import ( # noqa: PLC0415
load_client_engagement_context_file,
load_client_workflow_context_for_output,
)
if read_only:
output_root = next(
(
ancestor
for ancestor in (workbench.output_dir, *workbench.output_dir.parents)
if ancestor.name == "outputs"
and (ancestor.parent / "context.json").is_file()
),
None,
)
if output_root is None:
raise ValueError(
"workflow output is not inside a portable customer-folder run"
)
return load_client_engagement_context_file(
output_root.parent / "context.json",
expected_workflow_id=str(plugin),
output_dir=workbench.output_dir,
allowed_statuses=("running", "ready_for_review", "completed"),
)
return load_client_workflow_context_for_output(
workbench.output_dir,
expected_workflow_id=str(plugin),
)
def _with_vera_customer_context(
workbench: LocalReviewWorkbench,
args: dict[str, Any],
*,
read_only: bool = False,
) -> dict[str, Any]:
"""Add the live customer-run path to one server-owned MCP call."""
client_context = _validate_vera_customer_run(workbench, read_only=read_only)
if client_context is None:
return args
context_path = client_context.get("context_path")
if not isinstance(context_path, str) or not context_path:
raise ValueError("Vera customer-run context path is unavailable")
return {**args, "client_engagement": context_path}
def _validate_loopback_host(host: str) -> str:
normalized = host.strip()
if normalized.lower() == "localhost":
return normalized
try:
address = ipaddress.ip_address(normalized)
except ValueError as exc:
raise ValueError(
"review server host must be localhost or a loopback IP"
) from exc
if not address.is_loopback:
raise ValueError("review server host must be localhost or a loopback IP")
return normalized
def _review_url(host: str, port: int) -> str:
try:
address = ipaddress.ip_address(host)
except ValueError:
return f"http://{host}:{port}/review"
if address.version == 6:
return f"http://[{host}]:{port}/review"
return f"http://{host}:{port}/review"
def _server_class(host: str) -> type[ThreadingHTTPServer]:
try:
address = ipaddress.ip_address(host)
except ValueError:
return ThreadingHTTPServer
if address.version != 6:
return ThreadingHTTPServer
class IPv6ThreadingHTTPServer(ThreadingHTTPServer):
address_family = socket.AF_INET6
return IPv6ThreadingHTTPServer
def _adapter(workbench: LocalReviewWorkbench) -> dict[str, Any]:
return _read_json_object(workbench.adapter_path, required=True)
def _widget_path(workbench: LocalReviewWorkbench) -> Path:
html_files = sorted(
path
for path in (workbench.plugin_dir / "assets").glob("*.html")
if path.is_file()
)
if len(html_files) != 1:
raise ValueError(
f"expected exactly one review widget HTML asset for {workbench.plugin}"
)
return html_files[0]
def _validate_review_payload(
workbench: LocalReviewWorkbench,
adapter: dict[str, Any],
review_payload: Any,
) -> dict[str, Any]:
if not isinstance(review_payload, dict):
raise ValueError("review_payload must be a JSON object")
if review_payload.get("plugin") != workbench.plugin:
raise ValueError(f'review_payload.plugin must be "{workbench.plugin}"')
for field_name in ("schema_version", "workflow", "run_id"):
if (
not isinstance(review_payload.get(field_name), str)
or not review_payload[field_name].strip()
):
raise ValueError(f"review_payload.{field_name} must be a non-empty string")
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):
if not isinstance(item, dict):
raise ValueError(f"review_payload.items[{index}] must be an object")
for field_name in ("id", "item_type", "title"):
if (
not isinstance(item.get(field_name), str)
or not item[field_name].strip()
):
raise ValueError(
f"review_payload.items[{index}].{field_name} must be a non-empty string"
)
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}"
)
if not isinstance(adapter.get("saveTool"), str) or not adapter["saveTool"]:
raise ValueError("review-workbench adapter is missing saveTool")
if not isinstance(adapter.get("applyTool"), str) or not adapter["applyTool"]:
raise ValueError("review-workbench adapter is missing applyTool")
return review_payload
def _empty_ui_decisions(review_payload: dict[str, Any]) -> dict[str, Any]:
decisions = {
"schema_version": review_payload.get("schema_version", "1.0"),
"plugin": review_payload.get("plugin"),
"workflow": review_payload.get("workflow"),
"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",
}
content_sha256 = review_payload.get("content_sha256")
if isinstance(content_sha256, str) and content_sha256:
decisions["review_payload_content_sha256"] = content_sha256
return decisions
def _raw_session_payload(workbench: LocalReviewWorkbench) -> dict[str, Any]:
"""Load server-owned run artifacts without exposing them to the browser."""
adapter = _adapter(workbench)
run_intake = _read_json_object(
workbench.output_dir / "run_intake.json",
required=True,
)
review_payload = _validate_review_payload(
workbench,
adapter,
_read_json_object(workbench.output_dir / "review_payload.json", required=True),
)
ui_decisions = _read_json_object(workbench.output_dir / "ui_decisions.json")
applied_decisions = _read_json_object(
workbench.output_dir / "applied_decisions.json"
)
final_artifacts = _read_json_object(workbench.output_dir / "final_artifacts.json")
if run_intake.get("run_id") and run_intake["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") != workbench.plugin
):
raise ValueError("Applied decisions must belong to this review run")
return {
"widget_type": adapter.get("widgetType", f"{workbench.plugin}_review"),
"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": adapter["saveTool"],
"apply_tool": adapter["applyTool"],
"can_persist": True,
"fallback": "local_review_server",
},
}
def _render_tool_name(adapter: dict[str, Any]) -> str:
save_tool = adapter.get("saveTool")
if (
not isinstance(save_tool, str)
or not save_tool.startswith("save_")
or not save_tool.endswith("_decisions")
):
raise ValueError("review-workbench adapter saveTool cannot resolve render tool")
return f"render_{save_tool.removeprefix('save_').removesuffix('_decisions')}_review"
def _is_absolute_path(value: str) -> bool:
"""Return whether a string is an absolute POSIX or Windows path."""
return Path(value).is_absolute() or bool(re.match(r"^[A-Za-z]:[\\/]", value))
def _known_absolute_paths(value: Any) -> tuple[str, ...]:
"""Collect exact absolute paths present in server-owned review artifacts."""
paths: set[str] = set()
def visit(candidate: Any) -> None:
if isinstance(candidate, dict):
for item in candidate.values():
visit(item)
elif isinstance(candidate, list):
for item in candidate:
visit(item)
elif isinstance(candidate, str) and _is_absolute_path(candidate):
paths.add(candidate)
visit(value)
return tuple(sorted(paths, key=len, reverse=True))
def _sanitize_browser_payload(
value: Any,
*,
field: str | None = None,
redactions: tuple[str, ...] = (),
) -> Any:
"""Remove server-only paths and private QA text from any plugin render result."""
empty_path_lists = {"input_paths", "local_files_read"}
# Run-intake assumptions are server-owned context and may contain client names
# or other private case labels. They are not required to operate the browser
# review controls, so exclude the field rather than guessing which text is safe.
dropped_fields = {"assumptions", "output_dir", "required_text"}
if isinstance(value, dict):
sanitized: dict[str, Any] = {}
for key, item in value.items():
if key in dropped_fields:
continue
if key in empty_path_lists:
sanitized[key] = []
continue
sanitized[key] = _sanitize_browser_payload(
item,
field=key,
redactions=redactions,
)
return sanitized
if isinstance(value, list):
return [
_sanitize_browser_payload(item, field=field, redactions=redactions)
for item in value
]
if isinstance(value, str):
sanitized_text = value
for absolute_path in redactions:
sanitized_text = sanitized_text.replace(absolute_path, "<local-path>")
if _is_absolute_path(sanitized_text):
return "<local-path>"
return sanitized_text
return value
def build_session_payload(workbench: LocalReviewWorkbench) -> dict[str, Any]:
"""Return the plugin-rendered, browser-safe local review session."""
raw_session = _raw_session_payload(workbench)
read_only = False
render_args = {
"run_intake": raw_session["run_intake"],
"review_payload": raw_session["review_payload"],
"ui_decisions": raw_session["ui_decisions"],
"final_artifacts": raw_session["final_artifacts"],
}
if workbench.plugin == "journal-sampling":
render_args = {
"model_review_context": _read_json_object(
workbench.output_dir / "model_review_context.json", required=True
)
}
if (
raw_session["run_intake"].get("path_reference") == "run_root_relative"
or workbench.plugin == "archive-organization"
):
render_args = _with_vera_customer_context(
workbench, render_args, read_only=True
)
context_path = render_args.get("client_engagement")
if context_path:
run = _read_json_object(
Path(context_path).parent / "run.json", required=True
)
read_only = run.get("status") != "running"
rendered = _mcp_tool_result(
workbench,
_render_tool_name(_adapter(workbench)),
render_args,
browser_payload=True,
)
if rendered.get("ok") is False:
raise ValueError(str(rendered.get("error") or "plugin render tool failed"))
rendered_review = rendered.get("review_payload")
if not isinstance(rendered_review, dict):
raise ValueError("plugin render tool returned no review_payload")
expected_review_id = (
render_args["model_review_context"]["review_ref"]
if workbench.plugin == "journal-sampling"
else raw_session["review_payload"]["run_id"]
)
if (
rendered_review.get("plugin") != workbench.plugin
or rendered_review.get("run_id") != expected_review_id
):
raise ValueError("plugin render tool returned a mismatched review session")
if workbench.plugin == "journal-sampling":
# The MCP call has replayed and bound the current persisted review.
# Opaque aliases belong to model context; the local human-facing widget
# needs the verified filenames for traceability and declared downloads.
rendered = {**rendered, **raw_session}
redactions = _known_absolute_paths(
{
"session": raw_session,
"tool_args": render_args,
"workbench_output_dir": workbench.output_dir.resolve().as_posix(),
}
)
# Restore the persisted apply result through the same path redaction as the
# live tool response. A page reload must not reset its applied counter.
rendered["applied_decisions"] = raw_session["applied_decisions"]
if workbench.plugin == "new-client":
# The native model response omits free-text decisions. The validated
# local browser must still reopen the user's saved notes and requests.
rendered["ui_decisions"] = raw_session["ui_decisions"]
rendered["local_review_read_only"] = read_only
sanitized = _sanitize_browser_payload(rendered, redactions=redactions)
if not isinstance(sanitized, dict):
raise ValueError("sanitized plugin render result must remain an object")
return sanitized
def _script_json(value: Any) -> str:
"""Encode JSON for an executable script without allowing tag termination."""
return (
json.dumps(value, ensure_ascii=False, default=str)
.replace("&", "\\u0026")
.replace("<", "\\u003c")
.replace(">", "\\u003e")
.replace("\u2028", "\\u2028")
.replace("\u2029", "\\u2029")
)
def _bridge_html(workbench: LocalReviewWorkbench, session_token: str) -> str:
payload_json = _script_json(
build_session_payload(workbench),
)
plugin_json = _script_json(workbench.plugin)
token_json = _script_json(session_token)
return f"""<script>
(function () {{
const serverPayload = {payload_json};
const pluginName = {plugin_json};
const reviewToken = {token_json};
const stateKey = `${{pluginName}}:${{serverPayload.review_payload?.run_id || "run"}}`;
if (serverPayload.local_review_read_only) {{
document.addEventListener("DOMContentLoaded", () => {{
const messages = {{
it: "Risultato archiviato: sola lettura. Le decisioni si salvano prima di finalizzare l’esecuzione. Per ulteriori modifiche avvia una nuova esecuzione.",
en: "Archived result: read only. Save review decisions before finalizing the run. Start a new run for further changes.",
fr: "Résultat archivé : lecture seule. Enregistrez les décisions avant de finaliser l’exécution. Lancez une nouvelle exécution pour toute modification.",
de: "Archiviertes Ergebnis: schreibgeschützt. Prüfentscheidungen vor Abschluss speichern. Für Änderungen einen neuen Lauf starten.",
es: "Resultado archivado: solo lectura. Guarde las decisiones antes de finalizar la ejecución. Inicie una nueva ejecución para realizar cambios."
}};
const language = String(serverPayload.run_intake?.language || serverPayload.review_payload?.language || "en").slice(0,2);
const notice = document.createElement("p");
notice.id = "local-read-only-notice";
notice.setAttribute("role", "status");
notice.style.cssText = "padding:16px 24px;margin:0;background:#fff4d6;color:#493900";
notice.textContent = messages[language] || messages.en;
document.body.prepend(notice);
const protect = () => {{
for (const id of ["save-decisions", "apply-decisions", "use-recommended"]) {{
const button = document.getElementById(id);
if (button && !button.disabled) button.disabled = true;
}}
}};
protect();
new MutationObserver(protect).observe(document.body, {{subtree:true, childList:true, attributes:true, attributeFilter:["disabled"]}});
}}, {{once:true}});
}}
function readState() {{
try {{ return JSON.parse(window.sessionStorage.getItem(stateKey) || "null"); }}
catch {{ return null; }}
}}
window.localReviewDownloadOutput = async (path) => {{
const response = await fetch("/api/download-output", {{
method: "POST",
headers: {{"Content-Type": "application/json", "{REVIEW_TOKEN_HEADER}": reviewToken}},
body: JSON.stringify({{path}}),
}});
if (!response.ok) {{ const error = await response.json(); throw new Error(error.error || "Output unavailable"); }}
const url = URL.createObjectURL(await response.blob());
const link = document.createElement("a");
link.href = url; link.download = path.split(/[\\/]/).pop() || "output";
link.click(); setTimeout(() => URL.revokeObjectURL(url), 60000);
}};
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",
"{REVIEW_TOKEN_HEADER}": reviewToken,
}},
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>
"""
def render_review_html(
workbench: LocalReviewWorkbench,
*,
session_token: str | None = None,
) -> str:
"""Render the generated widget with a local ``window.openai`` bridge."""
html = _widget_path(workbench).read_text(encoding="utf-8")
needle = " <script>\n const CONFIG = "
if needle not in html:
raise ValueError("review widget script insertion point not found")
token = session_token or secrets.token_urlsafe(32)
return html.replace(needle, _bridge_html(workbench, token) + needle, 1)
def _server_tool_args(
workbench: LocalReviewWorkbench,
posted_args: dict[str, Any],
) -> dict[str, Any]:
session = _raw_session_payload(workbench)
decisions = posted_args.get("decisions")
if decisions is not None and not isinstance(decisions, list):
raise ValueError("decisions must be an array when provided")
args = {
"run_intake": session["run_intake"],
"review_payload": session["review_payload"],
"ui_decisions": session["ui_decisions"],
"final_artifacts": session["final_artifacts"],
"decision_source": "local_review_server",
}
if workbench.plugin == "journal-sampling":
args = {
"model_review_context": _read_json_object(
workbench.output_dir / "model_review_context.json", required=True
),
"decision_source": "local_review_server",
}
if decisions is not None:
args["decisions"] = decisions
reviewer = posted_args.get("reviewer")
if isinstance(reviewer, str) and reviewer.strip():
args["reviewer"] = reviewer.strip()
return args
def _codex_runtime_node_candidates() -> list[Path]:
"""Return bounded Node.js candidates supplied by local Codex runtimes."""
runtime_root = Path.home() / ".cache" / "codex-runtimes"
if not runtime_root.is_dir():
return []
return sorted(runtime_root.glob("*/dependencies/node/bin/node"), reverse=True)
def _node_executable() -> str:
override = os.environ.get(NODE_OVERRIDE_ENV)
if override:
candidate = Path(override).expanduser()
if candidate.is_file() and os.access(candidate, os.X_OK):
return candidate.resolve().as_posix()
raise ValueError(
f"{NODE_OVERRIDE_ENV} must point to an executable Node.js file"
)
node = shutil.which("node")
if node is not None:
return node
for candidate in _codex_runtime_node_candidates():
if candidate.is_file() and os.access(candidate, os.X_OK):
return candidate.resolve().as_posix()
raise ValueError(
"Node.js is required to call the plugin review-tool server; install Node.js "
f"or set {NODE_OVERRIDE_ENV} to its executable path"
)
def _mcp_tool_result(
workbench: LocalReviewWorkbench,
name: str,
args: dict[str, Any],
*,
browser_payload: bool = False,
) -> dict[str, Any]:
if browser_payload and name != _render_tool_name(_adapter(workbench)):
raise ValueError("component metadata is only available to browser rendering")
message = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {"name": name, "arguments": args},
}
command = [_node_executable(), workbench.mcp_server_path.as_posix(), "--stdio"]
if workbench.plugin_dir.name == "archive-organization":
# References belong to one server process. Validate the canonical local
# package and use its reference within the same bounded invocation.
# callTool retains the server's binding, digest and persistence checks.
command = [
_node_executable(),
"-e",
"const fs = require('node:fs');"
"const server = require(process.argv[1]);"
"const request = JSON.parse(fs.readFileSync(0, 'utf8'));"
"try { const args = request.params.arguments;"
"const validated = server.callTool(server.TOOL_NAMES.validateReview, args);"
"const result = server.callTool(request.params.name, "
"{...args, review_reference: validated.review_reference.reference});"
"process.stdout.write(JSON.stringify({id:request.id,result:{structuredContent:{ok:true,...result}}}));"
"} catch(error) { process.stdout.write(JSON.stringify({id:request.id,"
"error:{message:error.message}})); }",
workbench.mcp_server_path.as_posix(),
]
completed = subprocess.run(
command,
input=json.dumps(message) + "\n",
capture_output=True,
text=True,
check=False,
cwd=workbench.plugin_dir,
# The local server is already running in the selected environment.
# Its Node bridge defaults to that Python even when launched directly
# without shell activation. Preserve an explicit component override.
env={**os.environ, "PYTHON": os.environ.get("PYTHON") or sys.executable},
timeout=30,
)
if completed.returncode != 0:
raise ValueError((completed.stderr or completed.stdout).strip())
responses = [
json.loads(line) for line in completed.stdout.splitlines() if line.strip()
]
response = next((item for item in responses if item.get("id") == 1), None)
if response is None:
raise ValueError("MCP server returned no tools/call response")
if "error" in response:
raise ValueError(str(response["error"].get("message") or response["error"]))
result = response.get("result")
if not isinstance(result, dict):
raise ValueError("MCP tools/call result must be a JSON object")
structured = result.get("structuredContent")
if (
browser_payload
and isinstance(structured, dict)
and structured.get("ok") is True
):
metadata = result.get("_meta")
private = (
metadata.get("private_review_payload")
if isinstance(metadata, dict)
else None
)
if isinstance(private, dict):
# Component-only metadata stays in the local browser session. The
# normal MCP caller continues to receive only structuredContent.
return private
if isinstance(structured, dict):
if (
browser_payload
and name == _render_tool_name(_adapter(workbench))
and structured.get("ok") is not False
):
metadata = result.get("_meta")
private = (
metadata.get("private_review_payload")
if isinstance(metadata, dict)
else None
)
if isinstance(private, dict):
return private
return structured
return result
def call_review_tool(
workbench: LocalReviewWorkbench,
name: str,
posted_args: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Call the plugin MCP review tool using server-owned run artifacts."""
if not isinstance(name, str) or not name.strip():
raise ValueError("tool name must be a non-empty string")
args = _with_vera_customer_context(
workbench,
_server_tool_args(workbench, posted_args or {}),
)
adapter = _adapter(workbench)
allowed_tools = {adapter["saveTool"], adapter["applyTool"]}
if name not in allowed_tools:
raise ValueError(f"unsupported local review tool: {name}")
result = _mcp_tool_result(workbench, name, args)
if (
workbench.plugin in {"journal-sampling", "archive-organization", "new-client"}
and result.get("ok") is True
):
# A successful mutation returns a minimized model summary. Reload the
# actual persisted successor for browser counters and file downloads.
persisted = _raw_session_payload(workbench)
result = {
**result,
**{
key: persisted[key]
for key in ("ui_decisions", "applied_decisions", "final_artifacts")
},
}
redactions = _known_absolute_paths(
{
"tool_args": args,
"workbench_output_dir": workbench.output_dir.resolve().as_posix(),
}
)
sanitized = _sanitize_browser_payload(result, redactions=redactions)
if not isinstance(sanitized, dict):
raise ValueError("sanitized plugin tool result must remain an object")
return sanitized
def create_review_http_server(
workbench: LocalReviewWorkbench,
*,
host: str = "127.0.0.1",
port: int = 0,
) -> tuple[ThreadingHTTPServer, str]:
"""Return a configured local review server and its review URL."""
build_session_payload(workbench)
safe_host = _validate_loopback_host(host)
session_token = secrets.token_urlsafe(32)
httpd = _server_class(safe_host)(
(safe_host, port),
_handler(workbench, session_token=session_token),
)
actual_port = httpd.server_address[1]
return httpd, _review_url(safe_host, actual_port)
def _review_output_path(workbench: LocalReviewWorkbench, requested: str) -> Path:
"""Allow downloads only for declared regular output files within this run."""
session = _raw_session_payload(workbench)
records = session.get("final_artifacts", {}).get("outputs", [])
declared = [record.get("path") for record in records if isinstance(record, dict)]
declared += [item.get("output_path") for item in session["review_payload"]["items"]]
root = workbench.output_dir.resolve()
for raw in declared:
if not isinstance(raw, str) or not raw:
continue
candidate = Path(raw)
candidate = candidate if candidate.is_absolute() else root / candidate
try:
relative = candidate.relative_to(root)
except ValueError:
continue
if requested not in {raw, relative.as_posix()}:
continue
if ".." in relative.parts or any(
parent.is_symlink() for parent in (candidate, *candidate.parents)
):
raise ValueError(
"Output links and paths outside the run are not downloadable"
)
if not candidate.is_file():
raise ValueError("Declared output is not available")
if candidate.stat().st_size > 50_000_000:
raise ValueError("Open outputs larger than 50 MB from the run folder")
return candidate
raise ValueError("Output is not declared in this run")
def _handler(
workbench: LocalReviewWorkbench,
*,
session_token: str,
) -> type[BaseHTTPRequestHandler]:
class ReviewWorkbenchHandler(BaseHTTPRequestHandler):
server_version = "LocalReviewWorkbench/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(
render_review_html(
workbench,
session_token=session_token,
)
)
return
if route == "/api/session":
self._json_response(build_session_payload(workbench))
return
if route == "/api/health":
session = build_session_payload(workbench)
self._json_response(
{
"ok": True,
"plugin": workbench.plugin,
"run_id": session["review_payload"]["run_id"],
}
)
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 not in {"/api/call-tool", "/api/download-output"}:
self.send_error(HTTPStatus.NOT_FOUND.value, "Not found")
return
try:
supplied_token = self.headers.get(REVIEW_TOKEN_HEADER, "")
if not secrets.compare_digest(supplied_token, session_token):
self._json_response(
{"ok": False, "error": "invalid local review session token"},
status=HTTPStatus.FORBIDDEN,
)
return
content_type = self.headers.get("Content-Type", "")
media_type = content_type.partition(";")[0].strip().lower()
if media_type != "application/json":
self._json_response(
{
"ok": False,
"error": "Content-Type must be application/json",
},
status=HTTPStatus.UNSUPPORTED_MEDIA_TYPE,
)
return
length = int(self.headers.get("Content-Length", "0"))
if length < 0:
raise ValueError("request body length must be non-negative")
if length > MAX_POST_BYTES:
raise ValueError(f"request body exceeds {MAX_POST_BYTES} bytes")
payload = json.loads(self.rfile.read(length).decode("utf-8"))
if not isinstance(payload, dict):
raise ValueError("request body must be a JSON object")
if route == "/api/download-output":
output = _review_output_path(
workbench, str(payload.get("path", ""))
)
with output.open("rb") as stream:
body = stream.read(50_000_001)
if len(body) > 50_000_000:
raise ValueError("Output exceeds the download size limit")
self.send_response(HTTPStatus.OK.value)
self.send_header("Content-Type", "application/octet-stream")
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store")
self.send_header("X-Content-Type-Options", "nosniff")
self.send_header("Content-Security-Policy", "sandbox")
self.end_headers()
self.wfile.write(body)
return
name = payload.get("name")
args = (
payload.get("args") if isinstance(payload.get("args"), dict) else {}
)
self._json_response(call_review_tool(workbench, str(name or ""), args))
except json.JSONDecodeError as exc:
self._json_response(
{"ok": False, "error": f"invalid JSON request: {exc}"},
status=HTTPStatus.BAD_REQUEST,
)
except (OSError, subprocess.SubprocessError, TypeError, ValueError) as exc:
self._json_response(
{"ok": False, "error": str(exc)},
status=HTTPStatus.BAD_REQUEST,
)
return ReviewWorkbenchHandler
def serve_review(
workbench: LocalReviewWorkbench,
*,
host: str = "127.0.0.1",
port: int = 0,
open_browser: bool = True,
) -> None:
"""Serve the review UI on localhost and optionally open a browser."""
httpd, url = create_review_http_server(workbench, host=host, port=port)
LOGGER.info("%s review server: %s", workbench.plugin, url)
LOGGER.info("Output folder: %s", workbench.output_dir)
if open_browser:
webbrowser.open(url)
try:
httpd.serve_forever()
except KeyboardInterrupt:
LOGGER.info("Stopping %s review server", workbench.plugin)
finally:
httpd.server_close()
def _parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description=(
"Open a generated plugin review workbench 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(
"--plugin", help="Plugin name under the repository plugins folder"
)
parser.add_argument("--plugin-dir", help="Explicit plugin directory")
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:
workbench = LocalReviewWorkbench(
plugin_dir=_plugin_dir_from_args(args.plugin, args.plugin_dir),
output_dir=_output_dir(args.output_dir),
)
_validate_vera_customer_run(workbench, read_only=True)
serve_review(
workbench,
host=args.host,
port=args.port,
open_browser=not args.no_open,
)
except (OSError, TypeError, ValueError) as exc:
LOGGER.error("%s", exc)
return 2
return 0
if __name__ == "__main__":
raise SystemExit(main())
SHA-256: 02ad0fdfed1c005abf21e5235da1a4b260b2ac9f25bb71c71e4823d76d22330a