path: root/debug/risc-v-gsp-rm-vm/harness/state_seed.py

#!/usr/bin/env python3
"""RM-object state seeding: declarative state for rehost/valprobe runs.

Why this exists
---------------
``harness/rehost.py`` runs one firmware function with caller-supplied bytes and
``harness/valprobe.py`` runs one validation branch with crafted registers and
memory.  Both hit the same wall on RM handlers: the function early-outs (or
faults on a NULL callback) before the branch of interest because the RM object
state it walks -- clients, channels, graphics contexts, memdesc/context-buffer
objects -- does not exist.  This harness adds the missing layer: a declarative
**state seed** (``greeneye-state-seed/1``) that places symbolic objects at
chosen addresses, fills their fields and pointers, sets registers/entry, and
runs the function under a bounded budget while recording which seeded state the
firmware actually reads and where it first stops.

Every run reports the ordered first-touch state reads and a ``gate``: the
address and the state the returning/faulting path checked.  A clean early-out
with the gate identified is the success criterion; a fault is the first missing
state observed, never a product defect.

Honesty rules:

* Not a boot and not a model: only the seeded state exists.  The report lists
  the layout, every field it wrote, and every field marked ``unknown``.
* The object field map below is partial and cites its evidence per field; a
  field without evidence is marked ``unknown`` and must not be invented.
* No timing/cycle claim (L0/L1 fidelity, docs/06).

Offline only: never touches a GPU, kernel module, mdev, guest or network.
"""
from __future__ import annotations

import argparse
import json
import os
import sys
import time
from typing import Any

TOOL_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if TOOL_ROOT not in sys.path:
    sys.path.insert(0, TOOL_ROOT)

from greeneye.elf import load_elf  # noqa: E402
from greeneye.errors import GreenEyeError  # noqa: E402
from greeneye.sim_backend import launch_native  # noqa: E402
from sim.devices import load_profile, registry_from_profile  # noqa: E402
from sim.gsp_boot import _apply_boot_shim  # noqa: E402
from sim.libos import LibosSyscalls  # noqa: E402
from sim.machine import Fault, R_BP, R_BUDGET, R_ECALL, R_FAULT, R_WATCH  # noqa: E402

from harness.rehost import (  # noqa: E402
    DEFAULT_IMAGE,
    DEFAULT_PROFILE,
    EBREAK,
    RUN_CHUNK,
    _parse_u64,
    _registers,
    _sha256_file,
)
from harness.valprobe import (  # noqa: E402
    BUFFER_ID,
    INSTANCE_GETTER,
    OUTCOMES,
    PROMOTE_VALIDATION,
    decode_memory_spec,
    encode_return_stub,
    reg_index,
)

#: Schema tag of a state seed.
SEED_FORMAT = "greeneye-state-seed/1"
#: Schema tag of a single seeded run.
REPORT_FORMAT = "greeneye-state-seed-report/1"
#: Schema tag of the bundled demo.
DEMO_FORMAT = "greeneye-state-seed-demo/1"

DEFAULT_BUDGET = 50_000
DEFAULT_TIMEOUT = 20.0
DEFAULT_ALLOC_BASE = 0x10000000
DEFAULT_ALLOC_ALIGN = 16

#: Object types that carry a documented field map (evidence per field below).
_CHANNEL_FIELDS = {
    0x0D28: {"name": "instance_index", "width": 4, "status": "known",
             "role": "graphics-instance index (u32); selects client+0x37c+idx",
             "evidence": "promote-validation predicate (docs/47)"},
    0x0D01: {"name": "flag_d01", "width": 1, "status": "unknown",
             "role": "u8 flag on the failed-lookup path (0 takes the panic path)"},
    0x0CD1: {"name": "flag_cd1", "width": 1, "status": "unknown",
             "role": "u8 flag before the promote calls"},
    0x0648: {"name": "callback_648", "width": 8, "status": "unknown",
             "role": "indirect callback on the invalid-state path"},
    0x0898: {"name": "callback_898", "width": 8, "status": "unknown",
             "role": "indirect callback first reached after the +0x600 gate"},
    0x03B8: {"name": "marker_3b8", "width": 8, "status": "unknown",
             "role": "identity marker compared across objects"},
    0x00EE: {"name": "obj_ee", "width": 8, "status": "unknown",
             "role": "sub-object; +0xe8 read as the gfid/index"},
}

_GRAPHICS_INSTANCE_FIELDS = {
    0x0190: {"name": "memdesc_table", "width": 8, "status": "known",
             "role": "pointer to the per-buffer memDesc slot table",
             "evidence": "promote-validation predicate (docs/47)"},
    0x0161: {"name": "flag_161", "width": 1, "status": "unknown",
             "role": "u8 flag gating the 0x529bd44 call"},
    0x03B8: {"name": "marker_3b8", "width": 8, "status": "unknown",
             "role": "identity marker compared across objects"},
}

FIELD_MAP: dict[str, dict[int, dict]] = {
    "client": {
        0x1BE0: {"name": "instance_table", "width": 8, "status": "known",
                 "role": "graphics-instance pointer array base (index 0x37c + idx)",
                 "evidence": "promote-validation predicate (docs/47)"},
        0x1AA8: {"name": "obj_1aa8", "width": 8, "status": "unknown",
                 "role": "sub-object consumed by the construct path (+0x24c callback)"},
        0x1CC8: {"name": "obj_1cc8", "width": 8, "status": "unknown",
                 "role": "sub-object consumed by the construct path (+0x150, +0x1d0)"},
        0x1DD8: {"name": "engine_ctx", "width": 8, "status": "known",
                 "role": "engine-context pointer; 0 selects the no-engine path",
                 "evidence": "construct-path disassembly (docs/52)"},
        0x1DE8: {"name": "subdevice", "width": 8, "status": "known",
                 "role": "sub-device object whose +0x600 callback is called",
                 "evidence": "construct-path disassembly (docs/52)"},
        0x1E48: {"name": "obj_1e48", "width": 8, "status": "unknown",
                 "role": "sub-object consumed on the no-engine path (+0x110 callback)"},
    },
    "channel": _CHANNEL_FIELDS,
    "graphics_context": _CHANNEL_FIELDS,
    "graphics_instance": _GRAPHICS_INSTANCE_FIELDS,
    "memdesc": {
        0x0000: {"name": "unknown_0", "width": 8, "status": "unknown",
                 "role": "slot value semantics outside the promote predicate"},
        0x0008: {"name": "unknown_8", "width": 8, "status": "unknown",
                 "role": "unknown"},
    },
    "ctx_buffer": {
        0x0000: {"name": "unknown_0", "width": 8, "status": "unknown",
                 "role": "context-buffer (memdesc) object; field semantics unknown"},
    },
    "rm_object": {},
}

#: Object types accepted in a seed.  ``rm_object`` carries no field map.
KNOWN_TYPES = tuple(FIELD_MAP)
MISSING_MARKER = "unknown"


class StateSeedError(GreenEyeError):
    """A state seed, image or scenario is invalid."""


def _require_u64(value: Any, what: str = "value") -> int:
    try:
        return _parse_u64(value)
    except GreenEyeError as exc:
        raise StateSeedError(f"{what}: {exc}") from exc


def _field_width(obj_type: str, offset: int) -> int:
    spec = FIELD_MAP.get(obj_type, {}).get(offset)
    return int(spec["width"]) if spec else 8


def _field_spec(obj_type: str, offset: int) -> dict | None:
    return FIELD_MAP.get(obj_type, {}).get(offset)


def _parse_offset(value: Any) -> int:
    text = str(value).strip()
    if text.startswith("+"):
        text = text[1:]
    try:
        return int(text, 0)
    except ValueError as exc:
        raise StateSeedError(f"bad field offset {value!r}") from exc


def _encode_int(value: int, width: int) -> bytes:
    if width not in (1, 2, 4, 8):
        raise StateSeedError(f"unsupported field width {width}")
    return (value & ((1 << (8 * width)) - 1)).to_bytes(width, "little")


def _is_symbol(value: Any) -> bool:
    return isinstance(value, str) and value.strip().startswith("@")


def _symbol_expr(value: str) -> tuple[str, int]:
    text = value.strip()[1:]
    name, sep, offset = text.partition("+")
    if not name:
        raise StateSeedError(f"bad pointer expression {value!r}")
    if sep:
        try:
            delta = int(offset, 0)
        except ValueError as exc:
            raise StateSeedError(f"bad pointer expression {value!r}") from exc
    else:
        delta = 0
    return name, delta


class _Symbols:
    """Name -> address table for symbolic object/stub pointers."""

    def __init__(self) -> None:
        self._by_name: dict[str, int] = {}

    def add(self, name: str, addr: int, owner: str) -> None:
        if name in self._by_name:
            raise StateSeedError(f"duplicate symbol {name!r} ({owner})")
        self._by_name[name] = addr

    def resolve(self, value: str, *, owner: str) -> int:
        name, delta = _symbol_expr(value)
        if name not in self._by_name:
            raise StateSeedError(f"{owner}: unknown symbol {name!r}")
        return (self._by_name[name] + delta) & 0xFFFF_FFFF_FFFF_FFFF


def load_seed(seed: dict) -> dict:
    """Validate a ``greeneye-state-seed/1`` seed and compile an application plan."""
    if not isinstance(seed, dict):
        raise StateSeedError(f"seed must be an object, got {type(seed).__name__}")
    fmt = seed.get("format", SEED_FORMAT)
    if fmt != SEED_FORMAT:
        raise StateSeedError(f"unsupported seed format {fmt!r} (want {SEED_FORMAT})")
    name = str(seed.get("name") or "state-seed")
    seeded_registers = seed.get("registers") or {}
    if not isinstance(seeded_registers, dict):
        raise StateSeedError("registers must be an object")
    entry_value = seed.get("entry", seeded_registers.get("pc"))
    if entry_value is None:
        raise StateSeedError("seed needs an entry address (or registers.pc)")
    entry = _require_u64(entry_value)

    allocator = seed.get("allocator") or {}
    cursor = _require_u64(allocator.get("base", DEFAULT_ALLOC_BASE))
    default_align = _require_u64(allocator.get("align", DEFAULT_ALLOC_ALIGN))
    if default_align <= 0 or default_align & (default_align - 1):
        raise StateSeedError("allocator align must be a positive power of two")

    objects = seed.get("objects") or []
    if not isinstance(objects, list):
        raise StateSeedError("objects must be a list")
    layout: list[dict] = []
    symbols = _Symbols()
    for obj in objects:
        if not isinstance(obj, dict):
            raise StateSeedError(f"object must be an object, got {obj!r}")
        obj_type = str(obj.get("type") or "")
        if obj_type not in KNOWN_TYPES:
            raise StateSeedError(f"unknown object type {obj_type!r} (known: {', '.join(KNOWN_TYPES)})")
        obj_name = str(obj.get("name") or "")
        if not obj_name:
            raise StateSeedError(f"{obj_type} object needs a name")
        align = _require_u64(obj.get("align", default_align))
        if align <= 0 or align & (align - 1):
            raise StateSeedError(f"{obj_name}: align must be a positive power of two")
        fields = obj.get("fields") or {}
        if not isinstance(fields, dict):
            raise StateSeedError(f"{obj_name}: fields must be an object")
        extent = 0
        for key in fields:
            offset = _parse_offset(key)
            extent = max(extent, offset + _field_width(obj_type, offset))
        size = _require_u64(obj.get("size", max(extent, default_align)))
        if size <= 0:
            raise StateSeedError(f"{obj_name}: size must be positive")
        if "pa" in obj and "va" in obj:
            raise StateSeedError(f"{obj_name}: use pa or va, not both")
        explicit = obj.get("pa", obj.get("va"))
        if explicit is not None:
            pa = _require_u64(explicit)
        else:
            cursor = (cursor + align - 1) & ~(align - 1)
            pa = cursor
            cursor += (size + align - 1) & ~(align - 1)
        record = {"name": obj_name, "type": obj_type, "pa": pa, "size": size,
                  "fields": {}, "unknown": []}
        layout.append(record)
        symbols.add(obj_name, pa, obj_name)
        for key, value in fields.items():
            record["fields"][_parse_offset(key)] = value

    for i, first in enumerate(layout):
        for second in layout[i + 1:]:
            if first["pa"] < second["pa"] + second["size"] and second["pa"] < first["pa"] + first["size"]:
                raise StateSeedError(
                    f"objects {first['name']!r} and {second['name']!r} overlap"
                )

    writes: list[dict] = []
    unknown_fields: list[dict] = []

    stubs = seed.get("stubs") or []
    if not isinstance(stubs, list):
        raise StateSeedError("stubs must be a list")
    stub_records: list[dict] = []
    for stub in stubs:
        if not isinstance(stub, dict) or "return" not in stub:
            raise StateSeedError(f"stub needs 'return': {stub!r}")
        ret = _require_u64(stub["return"])
        if "addr" in stub:
            addr = _require_u64(stub["addr"])
        else:
            cursor = (cursor + default_align - 1) & ~(default_align - 1)
            addr = cursor
            cursor += default_align
        stub_name = stub.get("name")
        if stub_name:
            symbols.add(str(stub_name), addr, f"stub {stub_name}")
        code = encode_return_stub(ret)
        stub_records.append({"name": stub_name, "addr": addr, "return": ret,
                             "code_hex": code.hex()})
        writes.append({"pa": addr, "hex": code.hex(), "label": f"stub:{stub_name or hex(addr)}",
                       "kind": "stub"})

    def resolve_field(owner: str, obj_type: str, offset: int, value: Any) -> tuple[int | None, int, str]:
        width = _field_width(obj_type, offset)
        if value == MISSING_MARKER or (isinstance(value, dict) and value.get("unknown")):
            return None, width, MISSING_MARKER
        if isinstance(value, dict):
            if "unknown" in value and value["unknown"]:
                return None, width, MISSING_MARKER
            if "ptr" in value:
                value = value["ptr"]
            elif "value" in value:
                raw = value["value"]
                if raw == MISSING_MARKER:
                    return None, _require_u64(value.get("size", width)), MISSING_MARKER
                width = _require_u64(value.get("size", width))
                return _require_u64(raw), width, "value"
            else:
                raise StateSeedError(f"{owner}+{offset:#x}: field dict needs value/ptr/unknown")
        if _is_symbol(value):
            return symbols.resolve(str(value), owner=f"{owner}+{offset:#x}"), width, "ptr"
        return _parse_u64(value), width, "value"

    for record in layout:
        pa = record["pa"]
        for offset in sorted(record["fields"]):
            value = record["fields"][offset]
            resolved, width, kind = resolve_field(record["name"], record["type"], offset, value)
            spec = _field_spec(record["type"], offset)
            label = f"{record['name']}+{offset:#x}"
            if resolved is None:
                unknown_fields.append({
                    "object": record["name"], "type": record["type"], "off": f"{offset:#x}",
                    "name": spec["name"] if spec else None,
                    "role": spec["role"] if spec else "not documented",
                    "reason": "seed marks this field unknown; left at zero",
                })
                continue
            writes.append({"pa": pa + offset, "hex": _encode_int(resolved, width).hex(),
                           "label": label, "kind": kind})

    memory = seed.get("memory") or []
    if not isinstance(memory, list):
        raise StateSeedError("memory must be a list")
    for item in memory:
        if not isinstance(item, dict):
            raise StateSeedError(f"memory item must be an object: {item!r}")
        normalized = dict(item)
        if "pa" in normalized and "addr" in normalized:
            raise StateSeedError(f"memory item: use pa or addr, not both: {item!r}")
        if "pa" in normalized:
            normalized["addr"] = normalized.pop("pa")
        try:
            addr, data = decode_memory_spec(normalized)
        except GreenEyeError as exc:
            raise StateSeedError(str(exc)) from exc
        writes.append({"pa": addr, "hex": data.hex(),
                       "label": item.get("label") or f"mem+{addr:#x}", "kind": "memory"})

    registers: dict[str, int] = {}
    for reg_name, value in seeded_registers.items():
        if str(reg_name).strip().lower() == "pc":
            continue
        try:
            reg_index(str(reg_name))
        except GreenEyeError as exc:
            raise StateSeedError(str(exc)) from exc
        if _is_symbol(value):
            registers[str(reg_name)] = symbols.resolve(str(value), owner=f"register {reg_name}")
        else:
            registers[str(reg_name)] = _require_u64(value)

    stack = seed.get("stack")
    outcomes = {_require_u64(a): str(label)
                for a, label in (seed.get("outcomes") or {}).items()}
    reads = []
    for item in seed.get("reads") or ():
        if "pa" in item and "addr" not in item:
            item = dict(item)
            item["addr"] = item.pop("pa")
        if "addr" not in item:
            raise StateSeedError(f"read item needs addr/pa: {item!r}")
        reads.append(item)
    watch: list[dict] = []
    seen_labels: dict[str, int] = {}

    def _add_watch(addr: int, size: int, label: str) -> None:
        count = seen_labels.get(label, 0) + 1
        seen_labels[label] = count
        if count > 1:
            label = f"{label}~{count}"
        watch.append({"addr": addr, "size": size, "label": label})

    for item in seed.get("watch") or ():
        if "pa" in item and "addr" not in item:
            item = dict(item)
            item["addr"] = item.pop("pa")
        if "addr" not in item:
            raise StateSeedError(f"watch item needs addr/pa: {item!r}")
        _add_watch(_require_u64(item["addr"]), _require_u64(item.get("size", 8)),
                   str(item.get("label") or f"{_require_u64(item['addr']):#x}"))
    for record in layout:
        for offset in record["fields"]:
            _add_watch(record["pa"] + offset, _field_width(record["type"], offset),
                       f"{record['name']}+{offset:#x}")

    return {
        "format": SEED_FORMAT,
        "name": name,
        "entry": entry,
        "stack": _require_u64(stack) if stack is not None else None,
        "registers": registers,
        "writes": writes,
        "layout": layout,
        "unknown_fields": unknown_fields,
        "stubs": stub_records,
        "outcomes": outcomes,
        "reads": reads,
        "watch": watch,
        "provenance": seed.get("provenance") or {},
        "notes": seed.get("notes") or [],
    }


def apply_seed(machine, plan: dict) -> dict:
    """Apply a compiled plan to a machine; return the same plan (chainable)."""
    for item in plan["writes"]:
        machine.write_mem(item["pa"], bytes.fromhex(item["hex"]))
    for reg_name, value in plan["registers"].items():
        machine.x[reg_index(reg_name)] = value
    return plan


def _read_after(machine, reads) -> list[dict]:
    out = []
    for item in reads or ():
        addr = _require_u64(item["addr"])
        size = _require_u64(item.get("size", 8))
        data = bytes(machine.read_mem(addr, size))
        record = {"addr": f"{addr:#x}", "size": size, "hex": data.hex()}
        if item.get("label"):
            record["label"] = item["label"]
        record["u64"] = int.from_bytes(data.ljust(8, b"\x00"), "little")
        out.append(record)
    return out


def _gate_from_run(status: str, stop_pc: int, fault: dict | None,
                   state_reads: list[dict]) -> dict | None:
    if not state_reads:
        return None
    last = state_reads[-1]
    if status == "fault" and stop_pc == 0 and last.get("value") == 0:
        return {
            "kind": "null_indirect_call",
            "pc": last["pc"],
            "addr": last["addr"],
            "label": last["label"],
            "value": 0,
            "detail": (
                f"indirect call target {last['label']} is NULL; execution faulted at "
                f"pc 0x0 (fault={fault})"
            ),
        }
    if status == "fault":
        return {
            "kind": "fault",
            "pc": f"{stop_pc:#x}",
            "addr": last["addr"],
            "label": last["label"],
            "value": last.get("value"),
            "detail": f"fault before the next seeded-state read (fault={fault})",
        }
    if status == "outcome":
        return None
    return {
        "kind": f"{status}_before_gate",
        "pc": f"{stop_pc:#x}",
        "addr": last["addr"],
        "label": last["label"],
        "value": last.get("value"),
        "detail": f"run stopped as {status}; last seeded state read shown",
    }


def run_seed(
    seed: dict,
    *,
    image: str = DEFAULT_IMAGE,
    profile_path: str | None = DEFAULT_PROFILE,
    entry: int | str | None = None,
    out: str | None = None,
    record: bool | None = None,
    budget: int = DEFAULT_BUDGET,
    timeout: float = DEFAULT_TIMEOUT,
    name: str = "state-seed",
) -> dict:
    """Load a seed into the native machine, run it bounded, return a report."""
    if not os.path.isfile(image):
        raise StateSeedError(f"image not found: {image}")
    if budget <= 0:
        raise StateSeedError(f"budget must be positive, got {budget}")
    image_obj = load_elf(image)
    if not image_obj.is_riscv:
        raise StateSeedError(f"{image}: not a RISC-V ELF image")
    plan = load_seed(seed)
    entry_addr = _require_u64(entry) if entry is not None else plan["entry"]

    profile = load_profile(profile_path) if profile_path else None
    load = (profile or {}).get("load", {})
    ram_base = _require_u64(load.get("ram_base", 0))
    memory_mb = int(load.get("memory_mb", 256))
    ram_top = ram_base + memory_mb * 1024 * 1024

    if record is None:
        record = bool(out)
    session = launch_native(
        kernel=image,
        memory_mb=memory_mb,
        ram_base=ram_base,
        record=record,
        name=f"{name}-{plan['name']}"[:80],
        device_map=None,
        unknown_mmio="lenient",
        bias=0,
    )
    machine = session.target.machine
    try:
        if profile is not None:
            machine.devices = registry_from_profile(profile, machine=machine)
            machine.translator = getattr(machine.devices, "dma", None)
            machine.syscall_handler = LibosSyscalls().handle
            shim = dict(profile.get("boot_shim", {}))
            shim["_image_path"] = image
            _apply_boot_shim(machine, shim, image_obj, {"enabled": True, "assumptions": []})

        sentinel = (ram_top - 0x100) & ~0xF
        default_sp = (ram_top - 0x8000) & ~0xF
        machine.write_mem(sentinel, EBREAK.to_bytes(4, "little"))

        apply_seed(machine, plan)
        machine.x[1] = sentinel
        if machine.x[2] == 0:
            machine.x[2] = plan["stack"] if plan["stack"] is not None else default_sp
        machine.pc = entry_addr

        breakpoints = set(plan["outcomes"]) | {sentinel}
        watchers = []
        for item in plan["watch"]:
            addr = _require_u64(item["addr"])
            size = _require_u64(item.get("size", 8))
            label = str(item.get("label") or f"{addr:#x}")
            watchers.append((addr, addr + size, label, 3))

        entered = {f"x{i}": machine.x[i] for i in range(32) if i and machine.x[i]}
        entered["pc"] = entry_addr
        state_reads: list[dict] = []
        started = time.monotonic()
        status = "budget"
        stop_pc = entry_addr
        fault: dict | None = None
        outcome: str | None = None
        remaining = budget
        while remaining > 0:
            if time.monotonic() - started > timeout:
                status = "timeout"
                break
            chunk = min(RUN_CHUNK, remaining)
            reason, stop_pc, a, b = machine.run(
                chunk, breakpoints=breakpoints, watchers=tuple(watchers)
            )
            if reason == R_WATCH:
                width = 8
                for watcher in watchers:
                    if watcher[2] == b:
                        width = watcher[1] - watcher[0]
                        break
                try:
                    value = int.from_bytes(bytes(machine.read_mem(a, width)), "little")
                except (GreenEyeError, Fault):
                    value = None
                state_reads.append({"pc": f"{stop_pc:#x}", "addr": f"{a:#x}",
                                    "label": b, "value": value, "width": width})
                watchers = [watcher for watcher in watchers if watcher[2] != b]
                continue
            remaining -= chunk
            if reason == R_BP:
                if stop_pc == sentinel:
                    status = "return"
                    break
                outcome = plan["outcomes"].get(stop_pc, "unlabelled")
                status = "outcome"
                break
            if reason == R_FAULT:
                status = "fault"
                fault = {"cause": a, "tval": f"{b:#x}"}
                break
            if reason == R_ECALL:
                handled, result = (machine.syscall_handler(machine, stop_pc)
                                   if machine.syscall_handler else (False, None))
                if not handled:
                    status = "ecall"
                    break
                if result is not None:
                    machine.x[10] = result & 0xFFFF_FFFF_FFFF_FFFF
                continue
            if reason == R_BUDGET:
                status = "budget"

        gate = _gate_from_run(status, stop_pc, fault, state_reads)
        report: dict[str, Any] = {
            "format": REPORT_FORMAT,
            "seed": plan["name"],
            "seed_format": plan["format"],
            "seed_provenance": plan["provenance"],
            "image": os.path.abspath(image),
            "image_sha256": _sha256_file(image),
            "profile": os.path.basename(profile_path) if profile_path else None,
            "entry": f"{entry_addr:#x}",
            "status": status,
            "outcome": outcome,
            "stop_pc": f"{stop_pc:#x}",
            "fault": fault,
            "retval": machine.x[10],
            "instructions": machine.insns,
            "wall_seconds": round(time.monotonic() - started, 4),
            "objects": [
                {"name": item["name"], "type": item["type"], "pa": f"{item['pa']:#x}",
                 "size": item["size"]}
                for item in plan["layout"]
            ],
            "unknown_fields": plan["unknown_fields"],
            "stubs": plan["stubs"],
            "registers_entered": {k: f"{v:#x}" if isinstance(v, int) else v
                                  for k, v in entered.items()},
            "registers_exited": {k: f"{v:#x}" if isinstance(v, int) else v
                                 for k, v in _registers(machine).items()},
            "state_reads": state_reads,
            "gate": gate,
            "reads": _read_after(machine, plan["reads"]),
            "limits": [
                "Seeded run, not a boot: only the supplied objects/memory exist.",
                "A fault is the first missing state observed, not a product defect.",
                "No timing/cycle claim (L0/L1 fidelity, docs/06).",
                "A field marked unknown was left at zero and reported.",
            ],
        }
        if getattr(session.recorder, "dir", None) is not None:
            session.note("state-seed", seed=plan["name"], status=status,
                         outcome=outcome, stop_pc=f"{stop_pc:#x}",
                         gate=(gate or {}).get("kind"))
            report["session"] = session.recorder.dir
        if out:
            os.makedirs(out, exist_ok=True)
            path = os.path.join(out, "state-seed-report.json")
            with open(path, "w", encoding="utf-8") as fh:
                json.dump(report, fh, indent=2, sort_keys=True)
                fh.write("\n")
            report["report_json"] = os.path.abspath(path)
        return report
    finally:
        session.close()


# ---------------------------------------------------------------------------
# Bundled seeds.
#
# 1. fecs_seed: the promote-validation predicate seeded through this loader
#    (the mechanism proof -- it works without RM objects).  It reproduces the
#    valprobe variant memory/registers/stub exactly, so a slot non-NULL seed
#    must refuse 0x1f and a slot NULL seed must take the commit path.
# 2. target_seed: the construct target 0x52a70ec with a minimal crafted
#    client/channel/graphics-context/memdesc set.  The first gate is reported
#    (see docs/52); with the +0x600 callback stubbed it advances to the next.
# ---------------------------------------------------------------------------


def _fecs_seed(*, slot: int, label: str) -> dict:
    scratch = 0x10000000
    layout = {
        "scratch": scratch,
        "sp": scratch + 0x4000,
        "s0": scratch + 0x5000,
        "pgpu": scratch + 0x1000,
        "getter_out": scratch + 0x1800,
        "instance": scratch + 0x3800,
        "entry": scratch + 0x6000,
    }
    slot_addr = layout["instance"] + 0x300 + 6 * 8
    memory = [
        {"pa": f"{layout['entry'] + 0x30:#x}", "u64": 0x1F4150000},
        {"pa": f"{layout['entry'] + 0x38:#x}", "u64": 0x204608000},
        {"pa": f"{layout['entry'] + 0x40:#x}", "u64": 0x10000},
        {"pa": f"{layout['entry'] + 0x48:#x}", "u32": 4},
        {"pa": f"{layout['entry'] + 0x4c:#x}", "u16": BUFFER_ID["FECS_EVENT"]},
        {"pa": f"{layout['entry'] + 0x4e:#x}", "u8": 1},
        {"pa": f"{layout['entry'] + 0x4f:#x}", "u8": 0},
        {"pa": f"{layout['pgpu'] + 0x1be0:#x}", "u64": layout["instance"]},
        {"pa": f"{layout['instance'] + 0x190:#x}", "u64": layout["instance"] + 0x300},
        {"pa": f"{slot_addr:#x}", "u64": slot, "label": "fecs-slot"},
    ]
    return {
        "format": SEED_FORMAT,
        "name": label,
        "entry": f"{PROMOTE_VALIDATION:#x}",
        "outcomes": dict(OUTCOMES),
        "registers": {
            "s0": f"{layout['s0']:#x}",
            "s2": f"{layout['pgpu']:#x}",
            "s5": f"{layout['entry']:#x}",
            "sp": f"{layout['sp']:#x}",
            "a3": BUFFER_ID["FECS_EVENT"],
            "t1": 0xC,
        },
        "stubs": [{"addr": f"{INSTANCE_GETTER:#x}", "return": f"{layout['getter_out']:#x}",
                   "name": "instance_getter"}],
        "memory": memory,
        "watch": [{"pa": f"{slot_addr:#x}", "size": 8, "label": "fecs-slot"}],
        "reads": [{"pa": f"{slot_addr:#x}", "size": 8, "label": "fecs-slot-value"}],
        "provenance": {
            "source": "promote-validation predicate state (docs/47), rebuilt as a seed",
            "slot": slot,
        },
        "notes": [
            "Mechanism proof: a path that needs no RM objects, seeded end to end.",
        ],
    }


def fecs_seed(*, slot: int = 1) -> dict:
    """Build the FECS promote-validation seed (slot 1 refuses, slot 0 accepts)."""
    label = "fecs-promote-slot-nonnull" if slot else "fecs-promote-slot-null"
    return _fecs_seed(slot=slot, label=label)


def target_seed(*, advance: bool = False) -> dict:
    """Build the minimal client/channel/context seed for the 0x52a70ec target.

    ``advance=False`` leaves the +0x600 callback unknown (the first gate).
    ``advance=True`` points it at a return-0 stub so the next gate is reached.
    """
    callback = "@engine_stub" if advance else MISSING_MARKER
    name = "construct-gate-advanced" if advance else "construct-gate-minimal"
    return {
        "format": SEED_FORMAT,
        "name": name,
        "entry": "0x52a70ec",
        "allocator": {"base": "0x10010000", "align": "0x1000"},
        "stubs": [{"name": "engine_stub", "return": 0}] if advance else [],
        "registers": {
            "a0": "@client",
            "a1": "@channel",
            "a2": "0x1003f000",
        },
        "stack": "0x1003e000",
        "objects": [
            {"name": "client", "type": "client", "size": "0x2000", "fields": {
                "0x1be0": "@graphics_instance",
                "0x1aa8": "@obj_1aa8",
                "0x1cc8": "@obj_1cc8",
                "0x1dd8": {"value": 0, "size": 8},
                "0x1de8": "@engine_node",
                "0x1e48": "@obj_1e48",
            }},
            {"name": "channel", "type": "channel", "size": "0x1000", "fields": {
                "0xd28": {"value": 0, "size": 4},
                "0xd01": {"value": 1, "size": 1},
                "0xcd1": {"value": 0, "size": 1},
                "0x898": MISSING_MARKER,
            }},
            {"name": "graphics_instance", "type": "graphics_instance", "size": "0x1000",
             "fields": {
                 "0x190": "@memdesc_table",
                 "0x161": {"value": 0, "size": 1},
             }},
            {"name": "engine_node", "type": "rm_object", "size": "0x1000", "fields": {
                "0x600": callback,
            }},
            {"name": "obj_1cc8", "type": "rm_object", "size": "0x400", "fields": {
                "0x150": MISSING_MARKER,
                "0x1d0": {"value": 0x20, "size": 4},
            }},
            {"name": "obj_1aa8", "type": "rm_object", "size": "0x400", "fields": {
                "0x24c": MISSING_MARKER,
            }},
            {"name": "obj_1e48", "type": "rm_object", "size": "0x400", "fields": {
                "0x110": MISSING_MARKER,
            }},
            {"name": "memdesc_table", "type": "rm_object", "size": "0x100", "fields": {}},
            {"name": "fecs_memdesc", "type": "memdesc", "size": "0x100", "fields": {
                "0x0": MISSING_MARKER,
                "0x8": MISSING_MARKER,
            }},
        ],
        "reads": [{"pa": "0x1003f000", "size": 8, "label": "a2-out-scratch"}],
        "provenance": {
            "source": "construct-path disassembly (docs/52, this wave)",
            "callback": callback,
        },
        "notes": [
            "Only the fields the known firmware path reads are seeded; a field marked",
            "unknown is left zero and reported so the gate is not mistaken for a seed bug.",
        ],
    }


def run_demo(out: str | None = None) -> dict:
    """Prove the seeder on the FECS predicate, then attempt the construct target."""
    if not os.path.isfile(DEFAULT_IMAGE):
        raise StateSeedError(f"firmware image absent: {DEFAULT_IMAGE}")
    profile = DEFAULT_PROFILE

    fecs_nonnull = run_seed(fecs_seed(slot=1), profile_path=profile, record=False,
                            name="state-seed-fecs")
    fecs_nonnull_again = run_seed(fecs_seed(slot=1), profile_path=profile, record=False,
                                  name="state-seed-fecs")
    fecs_null = run_seed(fecs_seed(slot=0), profile_path=profile, record=False,
                         name="state-seed-fecs")
    target = run_seed(target_seed(advance=False), profile_path=profile, record=False,
                      name="state-seed-target")
    advanced = run_seed(target_seed(advance=True), profile_path=profile, record=False,
                        name="state-seed-target")

    verdicts = [
        {"label": "FECS init slot non-NULL", "outcome": fecs_nonnull["outcome"],
         "expect": "refuse-fecs-slot-exists-0x1f",
         "pass": fecs_nonnull["outcome"] == "refuse-fecs-slot-exists-0x1f"},
        {"label": "FECS init slot NULL", "outcome": fecs_null["outcome"],
         "expect": "accept-commit-path",
         "pass": fecs_null["outcome"] == "accept-commit-path"},
    ]
    deterministic = (
        fecs_nonnull["outcome"] == fecs_nonnull_again["outcome"]
        and fecs_nonnull["gate"] == fecs_nonnull_again["gate"]
    )
    result = {
        "format": DEMO_FORMAT,
        "ok": all(row["pass"] for row in verdicts) and deterministic,
        "deterministic": deterministic,
        "mechanism": {
            "seed": "fecs-promote",
            "verdicts": verdicts,
            "wall_seconds": round(fecs_nonnull["wall_seconds"] + fecs_null["wall_seconds"], 4),
        },
        "target": {
            "entry": "0x52a70ec",
            "status": target["status"],
            "stop_pc": target["stop_pc"],
            "gate": target["gate"],
            "state_reads": target["state_reads"],
            "unknown_fields": target["unknown_fields"],
        },
        "target_advanced": {
            "entry": "0x52a70ec",
            "status": advanced["status"],
            "stop_pc": advanced["stop_pc"],
            "gate": advanced["gate"],
            "state_reads": advanced["state_reads"],
        },
    }
    if out:
        os.makedirs(out, exist_ok=True)
        path = os.path.join(out, "state-seed-demo.json")
        with open(path, "w", encoding="utf-8") as fh:
            json.dump(result, fh, indent=2, sort_keys=True)
            fh.write("\n")
        result["artifacts"] = {"demo_json": os.path.abspath(path)}
    return result


def _load_seed_file(path: str) -> dict:
    if not os.path.isfile(path):
        raise StateSeedError(f"seed not found: {path}")
    with open(path, "r", encoding="utf-8") as fh:
        return json.load(fh)


def _emit_seed(kind: str) -> dict:
    if kind == "fecs":
        return fecs_seed()
    if kind == "target-minimal":
        return target_seed(advance=False)
    if kind == "target-advanced":
        return target_seed(advance=True)
    raise StateSeedError(f"unknown seed kind {kind!r}")


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="state_seed.py",
        description=(
            "Seed declarative RM object state into a bounded firmware run "
            "(offline; never touches hardware)."
        ),
    )
    sub = parser.add_subparsers(dest="command")
    run_cmd = sub.add_parser("run", help="run one seed file")
    run_cmd.add_argument("--seed", required=True, help="greeneye-state-seed/1 JSON")
    run_cmd.add_argument("--entry", default=None, help="override the seed entry (hex)")
    run_cmd.add_argument("--image", default=DEFAULT_IMAGE, help="firmware ELF")
    run_cmd.add_argument("--profile", default=DEFAULT_PROFILE, help="device profile JSON")
    run_cmd.add_argument("--budget", type=int, default=DEFAULT_BUDGET)
    run_cmd.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT)
    run_cmd.add_argument("--out", help="output directory (session + report JSON)")
    run_cmd.add_argument("--no-record", action="store_true", help="skip session recording")
    validate = sub.add_parser("validate", help="validate a seed and print its layout")
    validate.add_argument("--seed", required=True)
    emit = sub.add_parser("emit", help="write a bundled seed")
    emit.add_argument("--kind", required=True,
                      choices=["fecs", "target-minimal", "target-advanced"])
    emit.add_argument("--out", required=True)
    demo = sub.add_parser("demo", help="FECS mechanism proof + construct-gate attempt")
    demo.add_argument("--out", help="output directory")
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        if args.command == "run":
            seed = _load_seed_file(args.seed)
            report = run_seed(seed, image=args.image, profile_path=args.profile or None,
                              entry=args.entry, budget=args.budget, timeout=args.timeout,
                              out=args.out, record=False if args.no_record else None)
            summary = {k: report[k] for k in
                       ("format", "seed", "entry", "status", "outcome", "stop_pc",
                        "fault", "gate", "instructions", "wall_seconds",
                        "report_json", "session") if k in report}
            print(json.dumps(summary, indent=2, sort_keys=True))
            return 0
        if args.command == "validate":
            seed = _load_seed_file(args.seed)
            plan = load_seed(seed)
            print(json.dumps({
                "format": plan["format"],
                "name": plan["name"],
                "entry": f"{plan['entry']:#x}",
                "objects": [{"name": item["name"], "type": item["type"],
                             "pa": f"{item['pa']:#x}", "size": item["size"]}
                            for item in plan["layout"]],
                "unknown_fields": plan["unknown_fields"],
                "writes": len(plan["writes"]),
                "stubs": plan["stubs"],
            }, indent=2, sort_keys=True))
            return 0
        if args.command == "emit":
            seed = _emit_seed(args.kind)
            os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True)
            with open(args.out, "w", encoding="utf-8") as fh:
                json.dump(seed, fh, indent=2, sort_keys=True)
                fh.write("\n")
            print(os.path.abspath(args.out))
            return 0
        if args.command == "demo":
            result = run_demo(args.out)
            print(json.dumps({k: result[k] for k in
                              ("format", "ok", "deterministic", "mechanism",
                               "target", "target_advanced", "artifacts")
                              if k in result}, indent=2, sort_keys=True))
            return 0 if result["ok"] else 3
        parser.print_help()
        return 2
    except StateSeedError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())