#!/usr/bin/env python3
"""Build a prompt/choice boundary review for the Scene/Event VM goal.
The map1_01a trace already contains useful text-flow candidates, but the raw
trace mixes prompt text, choice labels, wait/render opcodes, and branch calls in
one long table. This builder separates those layers and deliberately keeps the
unproven pieces blocked instead of promoting the whole sequence as executed
story flow.
"""
from __future__ import annotations
import argparse
import html
import json
from collections import Counter
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "out"
WEB = ROOT / "web"
def load_json(path: Path, default: Any) -> Any:
try:
return json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError:
return default
def write_json(path: Path, payload: Any) -> None:
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
def write_text(path: Path, text: str) -> None:
path.write_text(text, encoding="utf-8")
def h(value: Any) -> str:
return html.escape("" if value is None else str(value))
def short(value: Any, limit: int = 180) -> str:
if isinstance(value, (dict, list)):
value = json.dumps(value, ensure_ascii=False, sort_keys=True)
text = " ".join((str(value) if value is not None else "").split())
return text if len(text) <= limit else text[: limit - 1] + "..."
def rel(path: Path) -> str:
try:
return path.resolve().relative_to(ROOT).as_posix()
except ValueError:
return path.as_posix()
def prompt_text(prompt: dict[str, Any], limit: int = 180) -> str:
return short((prompt.get("displayText") or "").replace("\n", " / "), limit)
def command_boundary_rows(command_rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
focused = {"0x00", "0x02", "0x06", "0x09", "0x0b", "0x18", "0x35"}
rows = []
for row in command_rows:
if row.get("opcodeHex") not in focused:
continue
rows.append(
{
"promptOrder": row.get("promptOrder"),
"commandOrderInPrompt": row.get("commandOrderInPrompt"),
"vaHex": row.get("vaHex"),
"opcodeHex": row.get("opcodeHex"),
"opcodeEvidenceLevel": row.get("opcodeEvidenceLevel"),
"label": row.get("label"),
"roles": row.get("roles") or [],
"branchTargetVaHex": row.get("branchTargetVaHex"),
"rawBytes": row.get("rawBytes"),
}
)
return rows
def choice_boundaries(prompts: list[dict[str, Any]]) -> list[dict[str, Any]]:
choices = [row for row in prompts if row.get("status") == "choice-marker-delimited"]
result = []
for choice in choices:
order = int(choice.get("order") or 0)
prev_prompt = next((row for row in reversed(prompts) if int(row.get("order") or 0) < order), None)
next_choice_order = next(
(
int(row.get("order") or 0)
for row in prompts
if int(row.get("order") or 0) > order and row.get("status") == "choice-marker-delimited"
),
10**9,
)
following = [
row
for row in prompts
if order < int(row.get("order") or 0) < next_choice_order
]
trace_choice_prompts = [
row for row in following if "0x18" in (row.get("traceOpcodeSet") or [])
]
branch_calls = []
for row in trace_choice_prompts:
for trace_row in row.get("trace") or []:
if trace_row.get("opcodeHex") == "0x09":
branch_calls.append(
{
"promptOrder": row.get("order"),
"vaHex": trace_row.get("vaHex"),
"branchTargetVaHex": trace_row.get("branchTargetVaHex"),
"rawBytes": trace_row.get("rawBytes"),
}
)
choice_markers = []
for row in trace_choice_prompts:
for trace_row in row.get("trace") or []:
if trace_row.get("opcodeHex") == "0x18":
choice_markers.append(
{
"promptOrder": row.get("order"),
"vaHex": trace_row.get("vaHex"),
"rawBytes": trace_row.get("rawBytes"),
"roles": trace_row.get("roles") or [],
}
)
result.append(
{
"choicePromptOrder": choice.get("order"),
"choicePromptId": choice.get("promptId"),
"choiceStartVaHex": choice.get("startVaHex"),
"choiceText": prompt_text(choice, 240),
"previousPromptOrder": (prev_prompt or {}).get("order"),
"previousPromptText": prompt_text(prev_prompt or {}, 180),
"followingCandidateCount": len(following),
"followingCandidates": [
{
"order": row.get("order"),
"promptId": row.get("promptId"),
"status": row.get("status"),
"startVaHex": row.get("startVaHex"),
"hasChoiceOpcode": "0x18" in (row.get("traceOpcodeSet") or []),
"text": prompt_text(row, 180),
}
for row in following[:8]
],
"tracePromptOrdersWithChoiceOpcode": [row.get("order") for row in trace_choice_prompts],
"choiceOpcodeRows": choice_markers,
"branchCallRows": branch_calls,
"selectionToNextPromptProofFound": False,
"promotion": "partial" if choice_markers or branch_calls else "candidate",
"remainingGap": (
"choice marker and branch-call rows are visible, but selected option -> next prompt target is not proven"
if choice_markers or branch_calls
else "choice text is identified, but the opcode/control rows for this displayed slice are not shown"
),
}
)
return result
def prompt_rows(prompts: list[dict[str, Any]]) -> list[dict[str, Any]]:
rows = []
for row in prompts:
rows.append(
{
"order": row.get("order"),
"promptId": row.get("promptId"),
"matchedStoryPromptId": row.get("matchedStoryPromptId"),
"status": row.get("status"),
"evidenceLevel": row.get("evidenceLevel"),
"startVaHex": row.get("startVaHex"),
"renderVaHex": row.get("renderVaHex"),
"waitVaHex": row.get("waitVaHex"),
"traceOpcodeSet": row.get("traceOpcodeSet") or [],
"text": prompt_text(row, 220),
}
)
return rows
def build(args: argparse.Namespace) -> dict[str, Any]:
trace_payload = load_json(args.map1_trace, {})
trace = trace_payload.get("map1_01aTrace") or {}
prompts = trace.get("prompts") or []
command_rows = trace.get("commandTraceRows") or []
boundaries = command_boundary_rows(command_rows)
choices = choice_boundaries(prompts)
opcode_counts = Counter(row.get("opcodeHex") for row in boundaries)
choice_rows_shown = [row for row in prompts if row.get("status") == "choice-marker-delimited"]
render_rows = [row for row in boundaries if row.get("opcodeHex") == "0x0b"]
wait_rows = [row for row in boundaries if row.get("opcodeHex") == "0x06"]
branch_rows = [row for row in boundaries if row.get("opcodeHex") == "0x09"]
choice_opcode_rows = [row for row in boundaries if row.get("opcodeHex") == "0x18"]
end_rows = [row for row in boundaries if row.get("opcodeHex") == "0x00"]
choice_marker_rows = [
marker
for choice in choices
for marker in (choice.get("choiceOpcodeRows") or [])
]
choice_branch_rows = [
branch
for choice in choices
for branch in (choice.get("branchCallRows") or [])
]
summary = {
"map": trace.get("map"),
"selectedGroupId": trace.get("selectedGroupId"),
"selectedSequenceId": trace.get("selectedSequenceId"),
"selectedRootVaHex": trace.get("selectedRootVaHex"),
"selectedRootEndVaHex": trace.get("selectedRootEndVaHex"),
"selectedEvidenceStatus": trace.get("selectedEvidenceStatus"),
"promotionStatus": "prompt-sequence-partial-choice-branch-blocked",
"promptSequenceProofFound": False,
"promptRootCandidatePartial": bool(trace.get("selectedRootVaHex")),
"directScenePromptBindingFound": False,
"choiceBranchProofFound": False,
"selectionToNextPromptProofFound": False,
"promptCountInSequence": trace.get("promptCountInSequence"),
"promptsShown": trace.get("promptsShown"),
"choiceCountInSequence": trace.get("choiceCountInSequence"),
"choiceRowsShown": len(choice_rows_shown),
"commandTraceRowsShown": trace.get("commandTraceRowsShown"),
"boundaryRowCount": len(boundaries),
"renderOpcodeRowCount": len(render_rows),
"waitOpcodeRowCount": len(wait_rows),
"choiceOpcodeRowCount": len(choice_opcode_rows),
"branchCallRowCount": len(branch_rows),
"choiceBoundaryMarkerRowCount": len(choice_marker_rows),
"choiceBoundaryBranchCallRowCount": len(choice_branch_rows),
"endClearRowCount": len(end_rows),
"opcodeBoundaryCounts": dict(sorted(opcode_counts.items())),
}
decisions = [
{
"item": "selected text-root prompt range",
"promotion": "partial",
"evidence": f"{summary['selectedGroupId']} / {summary['selectedSequenceId']} root {summary['selectedRootVaHex']}..{summary['selectedRootEndVaHex']}",
"remainingGap": "direct normal scene record -> executed prompt root binding is still missing",
},
{
"item": "prompt render/wait boundary",
"promotion": "grounded-in-trace",
"evidence": f"0x0b render rows={summary['renderOpcodeRowCount']}, 0x06 wait rows={summary['waitOpcodeRowCount']}",
"remainingGap": "rows are still selector-root-range trace evidence rather than full field-scene runtime proof",
},
{
"item": "choice marker boundary",
"promotion": "partial",
"evidence": f"choice text rows shown={summary['choiceRowsShown']}, choice-boundary opcode 0x18 rows={summary['choiceBoundaryMarkerRowCount']}",
"remainingGap": "choice text rows and opcode 0x18 rows are not always in the same prompt row, so it remains a boundary candidate",
},
{
"item": "choice branch call",
"promotion": "partial-blocked",
"evidence": f"choice-boundary opcode 0x09 branch-call rows={summary['choiceBoundaryBranchCallRowCount']}",
"remainingGap": "selected option value and next prompt target are not proven",
},
{
"item": "prompt sequence as story flow",
"promotion": "blocked",
"evidence": "early Ataho cave text matches the observed text sample",
"remainingGap": "sequence contains candidate alternatives around choices; it must not be treated as a single linear script",
},
]
return {
"scope": "scene/event VM prompt sequence and choice boundary review",
"promotionStatus": summary["promotionStatus"],
"summary": summary,
"decisions": decisions,
"choiceBoundaries": choices,
"promptRows": prompt_rows(prompts),
"commandBoundaryRows": boundaries,
"remainingProofs": [
"direct map1_01a scene record -> selected prompt root binding",
"selected choice value -> next prompt target edge",
"branch target 0x004bfdb4 role for each choice path",
"normal field-scene runtime proof that this candidate sequence is the executed sequence",
"full unshown prompt/choice coverage beyond the first 32 prompt rows",
],
"sourceArtifacts": {
"map1Trace": rel(args.map1_trace),
"sceneVmReference": "docs/SCENE_EVENT_VM_REFERENCE.md",
"sceneVmGoal": "docs/SCENE_EVENT_VM_GOAL.md",
},
}
def html_doc(payload: dict[str, Any]) -> str:
s = payload["summary"]
def tag(status: Any) -> str:
text = str(status)
cls = "good" if text in {"grounded", "confirmed", "grounded-in-trace"} else "warn" if "partial" in text or "grounded" in text else "bad"
return f'{h(text)}'
metrics = [
("sequence", f"{s['selectedGroupId']} / {s['selectedSequenceId']}"),
("root", f"{s['selectedRootVaHex']}..{s['selectedRootEndVaHex']}"),
("prompts", f"{s['promptsShown']} shown / {s['promptCountInSequence']} sequence"),
("choices", f"{s['choiceRowsShown']} shown / {s['choiceCountInSequence']} sequence"),
("choice boundary markers", s["choiceBoundaryMarkerRowCount"]),
("choice boundary branch calls", s["choiceBoundaryBranchCallRowCount"]),
("branch proof", s["choiceBranchProofFound"]),
]
metric_html = "\n".join(f"
{h(k)}{h(v)}
" for k, v in metrics)
decision_rows = "\n".join(
f"
{h(row['item'])}
{tag(row['promotion'])}
{h(row['evidence'])}
{h(row['remainingGap'])}
"
for row in payload["decisions"]
)
choice_rows = []
for row in payload["choiceBoundaries"]:
candidates = " ".join(
f"#{h(item['order'])} {h(item['promptId'])} {h(item['text'])}"
for item in row["followingCandidates"]
)
branch_calls = " ".join(
f"{h(item.get('vaHex'))} -> {h(item.get('branchTargetVaHex'))}"
for item in row["branchCallRows"]
)
marker_raws = " ".join(
f"{h(item.get('vaHex'))}{h(item.get('rawBytes'))}"
for item in row["choiceOpcodeRows"]
)
choice_rows.append(
f"
{h(row['choicePromptOrder'])}
{h(row['choiceStartVaHex'])}
{h(row['choiceText'])}
{h(row['tracePromptOrdersWithChoiceOpcode'])}
{marker_raws or '-'}
{branch_calls or '-'}
{candidates}
{h(row['remainingGap'])}
"
)
prompt_rows_html = "\n".join(
f"
{h(row['order'])}
{tag(row['evidenceLevel'])}
{h(row['status'])}
{h(row['startVaHex'])}
{h(row['promptId'])}
{h(row['text'])}
{h(', '.join(row['traceOpcodeSet']))}
"
for row in payload["promptRows"]
)
boundary_rows = "\n".join(
f"
{h(row['promptOrder'])}
{h(row['commandOrderInPrompt'])}
{h(row['vaHex'])}
{h(row['opcodeHex'])}
{h(row['label'])}
{h(', '.join(row['roles']))}
{h(row['branchTargetVaHex'] or '-')}
{h(row['rawBytes'])}
"
for row in payload["commandBoundaryRows"]
)
proofs = "\n".join(f"
{h(row)}
" for row in payload["remainingProofs"])
artifacts = "\n".join(f"