Deploy PR5 scan gate hardening 1cca102

README.md

@@ -92,17 +92,18 @@ Public demo mode excludes private, commercial-uncleared, and research-only helpe
 | <=32B models | Public stack is 11.42B: FLUX.2 Klein 4B + LocateAnything 3.83B + MiniCPM-V 1.30B + Nemotron Parse 0.94B + MiniCPM5 1.08B + FunctionGemma 0.27B. |
 | Off Brand | Custom command-center UI, dense inspector, workflow graph, wardrobe/lore drawer, and provider cards. |
 | Best Agent | Multi-step prompt, generation, scan, judge, checkpoint, export workflow. |
-| OpenBMB | Claimed only when MiniCPM-V returns judge evidence in an export packet. |
-| NVIDIA | Claimed only when Nemotron returns evidence in an export packet. LocateAnything remains visible but is not the Nemotron claim by itself. |
+| OpenBMB | Claimed only when MiniCPM-V returns success status in an export packet. |
+| NVIDIA | Claimed only when Nemotron returns success status in an export packet. LocateAnything remains visible but is not the Nemotron claim by itself. |
 | OpenAI Codex | Development branch and PR include Codex-authored implementation commits. |
 | Demo / social | Add final links here before submission: `DEMO_VIDEO_URL` and `SOCIAL_POST_URL`. |
 
-Tiny Titan can be claimed for the public demo stack because the active public models are each <=4B. The stronger FLUX.2 Klein 9B and OFFELLIA/Gemma stack remains private research only.
+Tiny Titan can be claimed only from a successful public-demo export packet because the active public models are each <=4B. The stronger FLUX.2 Klein 9B and OFFELLIA/Gemma stack remains private research only.
 
 ## Local Setup
 
 ```powershell
 python -m pip install -r requirements.txt
+$env:NEXUS_DISABLE_REAL_HF='1'
 python app.py
 ```
 
@@ -112,6 +113,7 @@ The app reads `NEXUS_PORT` or `PORT` when present, otherwise it launches on `786
 
 ```powershell
 python -m compileall app.py src tests
+$env:NEXUS_DISABLE_REAL_HF='1'
 $env:PYTEST_DISABLE_PLUGIN_AUTOLOAD='1'
 python -m pytest -q tests -p no:cacheprovider --basetemp=C:\tmp\pytest-nvw-full
 ```

app.py

@@ -77,6 +77,20 @@ def _file_path(uploaded: Any) -> str | None:
     return str(path) if path else None
 
 
+def _generated_output_path(operator_state: dict[str, Any] | None) -> str | None:
+    generation = (operator_state or {}).get("generation") or {}
+    output_path = generation.get("output_path")
+    return str(output_path) if output_path else None
+
+
+def _authoritative_generated_scan(operator_state: dict[str, Any] | None) -> dict[str, Any]:
+    output_path = _generated_output_path(operator_state)
+    if output_path:
+        return scan_file(output_path)
+    stored_scan = (operator_state or {}).get("generated_scan")
+    return stored_scan if isinstance(stored_scan, dict) else scan_file(None)
+
+
 def _checkpoint_seed(checkpoint_id: str) -> int:
     suffix = "".join(char for char in checkpoint_id[-8:] if char in "0123456789abcdefABCDEF")
     if not suffix:
@@ -133,7 +147,7 @@ def run_weave(
     )
     reference_scan = scan_file(_file_path(upload))
     generation = generate_flux_image(run.refined_prompt.refined, seed=_checkpoint_seed(run.checkpoint.checkpoint_id))
-    generated_scan = scan_file(generation.output_path) if generation.output_path else reference_scan
+    generated_scan = scan_file(generation.output_path) if generation.output_path else scan_file(None)
     minicpm = judge_with_minicpm(
         prompt=run.refined_prompt.refined,
         image_path=generation.output_path,
@@ -278,25 +292,32 @@ def scan_reference(
     active_section: str,
     operator_state: dict[str, Any] | None,
 ) -> tuple[Any, ...]:
-    scan = scan_file(_file_path(upload))
+    state = operator_state or _default_operator_state()
+    reference_path = _file_path(upload)
+    reference_scan = scan_file(reference_path)
+    generated_scan = _authoritative_generated_scan(state)
     minicpm = None
-    if run is not None:
-        generated_path = ((operator_state or {}).get("generation") or {}).get("output_path")
+    if run is not None and reference_path:
         minicpm = judge_with_minicpm(
             prompt=getattr(getattr(run, "refined_prompt", None), "refined", DEFAULT_PROMPT),
-            image_path=generated_path or _file_path(upload),
-            scan=scan,
+            image_path=reference_path,
+            scan=reference_scan,
             wardrobe_summary=_wardrobe_summary(run),
         )
     next_state = {
-        **(operator_state or _default_operator_state()),
+        **state,
         **({"reference_judge": minicpm.to_dict()} if minicpm else {}),
-        "reference_scan": scan,
-        "export": scan.get("export_gate", "pending"),
-        "message": "Reference scan complete. Export gate is clear." if scan.get("export_gate") == "clear" else "Reference scan requires review before export.",
+        "reference_scan": reference_scan,
+        "reference_export_gate": reference_scan.get("export_gate", "pending"),
+        "export": state.get("export", generated_scan.get("export_gate", "pending")),
+        "message": (
+            "Reference scan complete. Generated artifact export gate is unchanged."
+            if reference_scan.get("export_gate") == "clear"
+            else "Reference scan requires review. Generated artifact export gate is unchanged."
+        ),
     }
-    rendered = _render_stateful(run, adult_mode, scan, active_section, next_state)
-    return (*rendered, scan)
+    rendered = _render_stateful(run, adult_mode, generated_scan, active_section, next_state)
+    return (*rendered, generated_scan)
 
 
 def approve_checkpoint(
@@ -306,12 +327,13 @@ def approve_checkpoint(
     active_section: str,
     operator_state: dict[str, Any] | None,
 ) -> tuple[Any, ...]:
-    scan = scan or scan_file(None)
+    state = operator_state or _default_operator_state()
+    scan = _authoritative_generated_scan(state)
     if run is None:
         next_state = {**_default_operator_state(), "provider_state": "blocked", "message": "No run exists yet. Run Active Weave first."}
-    elif not ((operator_state or {}).get("generation") or {}).get("output_path"):
+    elif not _generated_output_path(state):
         next_state = {
-            **(operator_state or _default_operator_state()),
+            **state,
             "provider_state": "blocked",
             "checkpoint": "pending",
             "message": "Checkpoint blocked: no generated artifact exists yet.",
@@ -319,9 +341,10 @@ def approve_checkpoint(
     else:
         export_state = scan.get("export_gate", "pending")
         next_state = {
-            **(operator_state or _default_operator_state()),
+            **state,
             "provider_state": "export_ready" if export_state == "clear" else "checkpointed",
             "checkpoint": "approved",
+            "generated_scan": scan,
             "export": export_state,
             "message": "Checkpoint approved. Export is ready after clear ST3GG scan." if export_state == "clear" else "Checkpoint approved, but export still waits on ST3GG review.",
         }
@@ -335,12 +358,14 @@ def export_packet(
     active_section: str,
     operator_state: dict[str, Any] | None,
 ) -> tuple[Any, ...]:
-    scan = scan or scan_file(None)
     state = operator_state or _default_operator_state()
+    scan = _authoritative_generated_scan(state)
     if run is None:
         next_state = {**state, "provider_state": "blocked", "export": "blocked", "message": "Export blocked: no active run packet exists."}
     elif state.get("checkpoint") != "approved":
         next_state = {**state, "provider_state": "blocked", "export": "blocked", "message": "Export blocked: human checkpoint has not been approved."}
+    elif not _generated_output_path(state):
+        next_state = {**state, "provider_state": "blocked", "export": "blocked", "message": "Export blocked: no generated artifact exists."}
     elif scan.get("export_gate") != "clear":
         next_state = {**state, "provider_state": "blocked", "export": scan.get("export_gate", "blocked"), "message": "Export blocked: ST3GG gate is not clear."}
     else:

docs/HANDOFF_FINAL_HACKATHON.md

@@ -26,8 +26,8 @@ Do not paste these into chat, commits, logs, or export packets.
 
 ```powershell
 python -m compileall app.py src tests
-python -c "import app; print('app import ok')"
 $env:NEXUS_DISABLE_REAL_HF='1'
+python -c "import app; print('app import ok')"
 $env:PYTEST_DISABLE_PLUGIN_AUTOLOAD='1'
 python -m pytest -q tests -p no:cacheprovider
 hf auth whoami --format json
@@ -51,7 +51,7 @@ Avoid pytest `--basetemp=C:\tmp` in this Windows sandbox if `tmp_path` fixtures
 - OpenBMB prize claim requires `minicpm_judge.status == "success"` in an export packet.
 - NVIDIA prize claim requires `nemotron_evidence.status == "success"` in an export packet.
 - LocateAnything supports the grounding story but does not replace Nemotron for the NVIDIA prize.
-- Tiny Titan can be claimed for the public demo stack because each active public model is <=4B.
+- Tiny Titan can be claimed only from a successful public-demo export packet because each active public model is <=4B.
 - FLUX.2 Klein 9B and OFFELLIA/Gemma remain private research options only.
 - Modal is not claimed unless a real Modal job runs and is documented.
 

src/nexus_visual_weaver/exporter.py

@@ -10,6 +10,43 @@ from typing import Any
 
 from .catalog import active_stack, parameter_budget
 
+REPO_ROOT = Path(__file__).resolve().parents[2]
+ALLOWED_LOCAL_EXPORT_ROOTS = (
+    REPO_ROOT / "outputs",
+    REPO_ROOT / ".pi",
+    REPO_ROOT / ".codex",
+)
+
+
+def _is_within(path: Path, root: Path) -> bool:
+    try:
+        path.relative_to(root)
+        return True
+    except ValueError:
+        return False
+
+
+def _safe_export_candidate(candidate: Path) -> Path | None:
+    if not candidate.is_absolute():
+        candidate = REPO_ROOT / candidate
+    resolved = candidate.resolve(strict=False)
+    src_root = (REPO_ROOT / "src").resolve(strict=False)
+    if resolved == src_root or _is_within(resolved, src_root):
+        return None
+
+    allowed_roots = [root.resolve(strict=False) for root in ALLOWED_LOCAL_EXPORT_ROOTS]
+    if Path("/data").exists():
+        allowed_roots.append(Path("/data/nexus_visual_weaver").resolve(strict=False))
+    if any(resolved == root or _is_within(resolved, root) for root in allowed_roots):
+        return resolved
+    return None
+
+
+def _artifact_name(output_path: Any) -> str | None:
+    if not output_path:
+        return None
+    return Path(str(output_path)).name
+
 
 def export_root() -> Path:
     requested = os.environ.get("NEXUS_EXPORT_DIR")
@@ -18,14 +55,15 @@ def export_root() -> Path:
         candidates.append(Path("/data/nexus_visual_weaver/exports"))
     candidates.append(Path("outputs/exports"))
     for candidate in candidates:
-        if candidate is None:
+        safe_candidate = _safe_export_candidate(candidate)
+        if safe_candidate is None:
             continue
         try:
-            candidate.mkdir(parents=True, exist_ok=True)
-            return candidate
+            safe_candidate.mkdir(parents=True, exist_ok=True)
+            return safe_candidate
         except OSError:
             continue
-    fallback = Path("outputs/exports")
+    fallback = (REPO_ROOT / "outputs/exports").resolve(strict=False)
     fallback.mkdir(parents=True, exist_ok=True)
     return fallback
 
@@ -38,17 +76,22 @@ def write_export_packet(
     adult_mode: bool,
 ) -> dict[str, Any]:
     run_id = getattr(getattr(run, "checkpoint", None), "checkpoint_id", f"nw-{int(time.time())}")
-    stack = active_stack(adult_mode)
+    run_adult_mode = bool(getattr(getattr(run, "request", None), "adult_mode", adult_mode))
+    stack = list(getattr(run, "model_stack", None) or active_stack(run_adult_mode))
+    budget = parameter_budget(stack)
     generation = dict(operator_state.get("generation") or {})
     generation.pop("hf_token_present", None)
+    artifact = _artifact_name(generation.get("output_path"))
+    if "output_path" in generation:
+        generation["output_path"] = artifact
     packet = {
         "schema": "nexus_visual_weaver.export_packet.v1",
         "run_id": run_id,
         "created_at_epoch": int(time.time()),
-        "adult_mode": bool(adult_mode),
+        "adult_mode": run_adult_mode,
         "prompt": getattr(getattr(run, "request", None), "prompt", ""),
         "refined_prompt": getattr(getattr(run, "refined_prompt", None), "refined", ""),
-        "artifact": (operator_state.get("generation") or {}).get("output_path"),
+        "artifact": artifact,
         "generation": generation,
         "st3gg_scan": scan,
         "minicpm_judge": operator_state.get("minicpm_judge") or {},
@@ -68,9 +111,9 @@ def write_export_packet(
             }
             for model in stack
         ],
-        "parameter_budget": parameter_budget(stack),
+        "parameter_budget": budget,
         "hackathon_claims": {
-            "build_small_32b": parameter_budget(stack)["status"] == "pass",
+            "build_small_32b": budget["status"] == "pass",
             "gradio_space": True,
             "off_brand_custom_ui": True,
             "openbmb_lane": (operator_state.get("minicpm_judge") or {}).get("status") == "success",

src/nexus_visual_weaver/provider_runtime.py

@@ -7,6 +7,7 @@ import json
 import os
 import time
 import urllib.error
+import urllib.parse
 import urllib.request
 from dataclasses import asdict, dataclass
 from pathlib import Path
@@ -56,6 +57,9 @@ def _image_data_url(path: str | None) -> str | None:
 
 
 def _post_json(url: str, token: str, payload: dict[str, Any], timeout: float) -> dict[str, Any]:
+    parsed = urllib.parse.urlparse(url)
+    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+        raise ValueError(f"Invalid URL: expected http(s) URL with host, got {url!r}.")
     body = json.dumps(payload).encode("utf-8")
     request = urllib.request.Request(
         url,

src/nexus_visual_weaver/render.py

@@ -48,6 +48,31 @@ def _env_configured(*names: str) -> bool:
     return any(bool(os.environ.get(name)) for name in names)
 
 
+def _provider_configured(base_url_name: str, *key_names: str) -> bool:
+    return bool(os.environ.get(base_url_name)) and _env_configured(*key_names)
+
+
+_SCAN_REDACTION_TERMS = (
+    "payload",
+    "payload_excerpt",
+    "hidden content",
+    "recovered",
+    "raw byte",
+    "base64",
+    "hex dump",
+)
+
+
+def _redact_scan_text(value: object) -> str:
+    text = str(value)
+    lowered = text.lower()
+    if any(term in lowered for term in _SCAN_REDACTION_TERMS):
+        return "Redacted scan detail; review the gated ST3GG evidence packet."
+    if len(text) > 180:
+        return text[:177] + "..."
+    return text
+
+
 def _space_runtime_status() -> dict[str, str]:
     space_id = os.environ.get("SPACE_ID") or os.environ.get("HF_SPACE_ID") or "local-preview"
     hardware = os.environ.get("SPACE_HARDWARE") or os.environ.get("NEXUS_SPACE_HARDWARE") or "ZeroGPU"
@@ -55,9 +80,9 @@ def _space_runtime_status() -> dict[str, str]:
     configured = []
     if _env_configured("HF_TOKEN", "HUGGING_FACE_HUB_TOKEN"):
         configured.append("HF")
-    if _env_configured("MINICPM_API_KEY", "OPENBMB_API_KEY"):
+    if _provider_configured("MINICPM_BASE_URL", "MINICPM_API_KEY", "OPENBMB_API_KEY"):
         configured.append("MiniCPM")
-    if _env_configured("NEMOTRON_API_KEY", "NVIDIA_API_KEY"):
+    if _provider_configured("NEMOTRON_BASE_URL", "NEMOTRON_API_KEY", "NVIDIA_API_KEY"):
         configured.append("Nemotron")
     if _env_configured("FAL_KEY", "NETLIFY_AUTH_TOKEN", "OPENAI_API_KEY", "MODAL_TOKEN_ID"):
         configured.append("optional")
@@ -106,12 +131,12 @@ def render_trust_strip(scan: dict | None = None, operator_state: dict | None = N
     status = str(scan.get("status", "idle")).upper()
     export_gate = str(scan.get("export_gate", "pending")).upper()
     checkpoint = str(operator_state.get("checkpoint", "pending")).replace("_", " ").title()
-    findings = scan.get("findings") or ["No upload selected. Always-on scanner ready."]
-    actions = scan.get("purification_actions") or [
+    findings = [_redact_scan_text(item) for item in (scan.get("findings") or ["No upload selected. Always-on scanner ready."])]
+    actions = [_redact_scan_text(item) for item in (scan.get("purification_actions") or [
         "strip metadata before export",
         "truncate PNG after IEND when needed",
         "run LSB statistical review",
-    ]
+    ])]
     tone = _scan_status_tone(str(scan.get("status", "idle")))
     return f"""
     <section class="nw-trust-strip">
@@ -460,7 +485,7 @@ def render_operations_panel(
             ("ST3GG state", f"{scan_status} / export {export_gate}"),
             (
                 "Findings",
-                "; ".join(str(item) for item in (scan.get("findings") or [])[:2])
+                "; ".join(_redact_scan_text(item) for item in (scan.get("findings") or [])[:2])
                 or ("No findings." if scan_status != "IDLE" else "No upload selected."),
             ),
             ("Public export", "Consent, provenance, metadata, age, dataset, and payload gates stay active."),
@@ -563,8 +588,8 @@ def render_provider_cards(relay_status: dict | None = None, adult_mode: bool = F
     relay_status = relay_status or {}
     decisions = relay_status.get("decisions", [])
     optional_statuses = {
-        "openbmb": "configured" if _env_configured("MINICPM_API_KEY", "OPENBMB_API_KEY") else "missing secret",
-        "nvidia": "configured" if _env_configured("NEMOTRON_API_KEY", "NVIDIA_API_KEY") else "missing secret",
+        "openbmb": "configured" if _provider_configured("MINICPM_BASE_URL", "MINICPM_API_KEY", "OPENBMB_API_KEY") else "missing secret",
+        "nvidia": "configured" if _provider_configured("NEMOTRON_BASE_URL", "NEMOTRON_API_KEY", "NVIDIA_API_KEY") else "missing secret",
         "fal": "configured" if _env_configured("FAL_KEY") else "blocked",
         "netlify": "configured" if _env_configured("NETLIFY_AUTH_TOKEN", "NETLIFY_SITE_ID", "OPENAI_BASE_URL") else "blocked",
         "cloudflare": "configured" if _env_configured("CLOUDFLARE_API_TOKEN", "CF_ACCOUNT_ID") else "blocked",
@@ -664,10 +689,10 @@ def render_inspector(
             ("NVIDIA Nemotron", "nvidia/NVIDIA-Nemotron-Parse-v1.2", nemotron),
         ]
     )
-    findings = scan.get("findings") or []
-    actions = scan.get("purification_actions") or ["metadata strip ready", "IEND truncation ready", "LSB review ready"]
-    finding_rows = "".join(f"<li>{escape(str(item))}</li>" for item in findings[:4]) or "<li>No upload selected. Scanner ready.</li>"
-    action_rows = "".join(f"<li>{escape(str(item))}</li>" for item in actions[:4])
+    findings = [_redact_scan_text(item) for item in (scan.get("findings") or [])]
+    actions = [_redact_scan_text(item) for item in (scan.get("purification_actions") or ["metadata strip ready", "IEND truncation ready", "LSB review ready"])]
+    finding_rows = "".join(f"<li>{escape(item)}</li>" for item in findings[:4]) or "<li>No upload selected. Scanner ready.</li>"
+    action_rows = "".join(f"<li>{escape(item)}</li>" for item in actions[:4])
     export_gate = str(scan.get("export_gate", "pending")).upper()
     return f"""
     <aside class="nw-panel nw-inspector">

tests/test_app_callbacks.py

@@ -1,4 +1,86 @@
 import app
+from nexus_visual_weaver.planner import build_command_center_run
+
+
+# --- _checkpoint_seed tests ---
+
+def test_checkpoint_seed_parses_valid_hex_suffix() -> None:
+    # "nw-" prefix + "abcdef12" -> int("abcdef12", 16) % 1_000_000
+    checkpoint_id = "nw-abcdef12"
+    result = app._checkpoint_seed(checkpoint_id)
+    assert result == int("abcdef12", 16) % 1_000_000
+
+
+def test_checkpoint_seed_handles_empty_suffix() -> None:
+    assert app._checkpoint_seed("") == 0
+    assert app._checkpoint_seed("nw-") == 0
+
+
+def test_checkpoint_seed_handles_non_hex_suffix() -> None:
+    # No hex chars -> return 0
+    result = app._checkpoint_seed("nw-zzzzzzzz")
+    assert result == 0
+
+
+def test_checkpoint_seed_ignores_non_hex_chars_in_last_8() -> None:
+    # Strips non-hex: "abc-xyz" -> only "abcef" are hex in the last 8 chars ("-xyz" strips to "abc")
+    # "0000000g" -> only "0000000" are valid hex
+    result = app._checkpoint_seed("0000000g")
+    assert result == int("0000000", 16) % 1_000_000
+    assert result == 0
+
+
+def test_checkpoint_seed_returns_value_within_range() -> None:
+    checkpoint_id = "nw-deadbeef"
+    result = app._checkpoint_seed(checkpoint_id)
+    assert 0 <= result < 1_000_000
+
+
+def test_checkpoint_seed_uses_only_last_8_chars() -> None:
+    # Long checkpoint id: only last 8 chars considered
+    # "nw-aabbccddee11223344" -> last 8 is "11223344"
+    result = app._checkpoint_seed("nw-aabbccddee11223344")
+    assert result == int("11223344", 16) % 1_000_000
+
+
+# --- _wardrobe_summary tests ---
+
+def test_wardrobe_summary_returns_string_with_slot_fields() -> None:
+    run = build_command_center_run("gothic patent leather platform boots, crimson hardware")
+    summary = app._wardrobe_summary(run)
+
+    # Summary is a semicolon-delimited string of slot info
+    assert isinstance(summary, str)
+    assert len(summary) > 0
+    # Should contain material= and palette= from slot info
+    assert "material=" in summary
+    assert "palette=" in summary
+    assert "locked=" in summary
+
+
+def test_wardrobe_summary_handles_run_with_no_outfit() -> None:
+    class FakeRun:
+        outfit = None
+
+    summary = app._wardrobe_summary(FakeRun())
+    assert summary == ""
+
+
+def test_wardrobe_summary_handles_none_run() -> None:
+    # Pass something with no outfit attribute
+    class Empty:
+        pass
+
+    summary = app._wardrobe_summary(Empty())
+    assert summary == ""
+
+
+def test_wardrobe_summary_includes_slot_names() -> None:
+    run = build_command_center_run("dark couture archivist, patent leather, platform boots")
+    summary = app._wardrobe_summary(run)
+
+    # Slot names like footwear, outerwear, upper_body should appear
+    assert any(slot_name in summary for slot_name in ["footwear", "outerwear", "upper_body", "jewelry"])
 
 
 def test_run_weave_returns_stateful_dashboard_packet() -> None:
@@ -18,6 +100,8 @@ def test_run_weave_returns_stateful_dashboard_packet() -> None:
 
 
 def test_operator_actions_transition_checkpoint_export_and_stop() -> None:
+    from pathlib import Path
+
     result = app.run_weave(
         "gothic patent leather platform boots, crimson hardware",
         "Strict",
@@ -27,7 +111,10 @@ def test_operator_actions_transition_checkpoint_export_and_stop() -> None:
         "Forge",
     )
     run = result[13]
-    operator_state = {**result[15], "generation": {**result[15]["generation"], "output_path": "outputs/test-generated-artifact.png"}}
+    artifact_path = Path("outputs/test-generated-artifact.png")
+    artifact_path.parent.mkdir(parents=True, exist_ok=True)
+    artifact_path.write_bytes(b"\x89PNG\r\n\x1a\n" + b"\x00" * 100)
+    operator_state = {**result[15], "generation": {**result[15]["generation"], "output_path": str(artifact_path)}}
     clean_scan = {"status": "pass", "export_gate": "clear", "findings": [], "purification_actions": []}
 
     approved = app.approve_checkpoint(run, False, clean_scan, "Forge", operator_state)
@@ -59,6 +146,66 @@ def test_export_blocks_without_checkpoint() -> None:
     assert "checkpoint" in blocked[13]["message"].lower()
 
 
+def test_reference_scan_cannot_clear_blocked_generated_artifact() -> None:
+    base = app.ROOT / "outputs" / "test-app-callbacks"
+    base.mkdir(parents=True, exist_ok=True)
+    blocked_artifact = base / "blocked-generated.png"
+    blocked_artifact.write_bytes(
+        b"\x89PNG\r\n\x1a\n"
+        b"\x00\x00\x00\rIEND\xaeB`\x82"
+        b"NEXUS_TRAILING_PAYLOAD"
+    )
+    clean_reference = base / "clean-reference.png"
+    clean_reference.write_bytes(b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIEND\xaeB`\x82")
+    run = build_command_center_run("gothic patent leather platform boots")
+    state = {
+        "provider_state": "checkpointed",
+        "checkpoint": "pending_review",
+        "export": "blocked",
+        "generation": {"status": "success", "output_path": str(blocked_artifact)},
+        "generated_scan": app.scan_file(str(blocked_artifact)),
+    }
+
+    scanned = app.scan_reference(run, False, str(clean_reference), "Forge", state)
+    assert scanned[13]["reference_scan"]["export_gate"] == "clear"
+    assert scanned[13]["export"] == "blocked"
+    assert scanned[15]["export_gate"] == "blocked"
+
+    approved = app.approve_checkpoint(run, False, scanned[15], "Forge", scanned[13])
+    assert approved[13]["provider_state"] == "checkpointed"
+    assert approved[13]["export"] == "blocked"
+
+
+def test_blocked_reference_scan_does_not_block_clear_generated_artifact() -> None:
+    base = app.ROOT / "outputs" / "test-app-callbacks"
+    base.mkdir(parents=True, exist_ok=True)
+    generated = base / "clear-generated.png"
+    generated.write_bytes(b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIEND\xaeB`\x82")
+    blocked_reference = base / "blocked-reference.png"
+    blocked_reference.write_bytes(
+        b"\x89PNG\r\n\x1a\n"
+        b"\x00\x00\x00\rIEND\xaeB`\x82"
+        b"NEXUS_TRAILING_PAYLOAD"
+    )
+    run = build_command_center_run("gothic patent leather platform boots")
+    state = {
+        "provider_state": "checkpointed",
+        "checkpoint": "pending_review",
+        "export": "clear",
+        "generation": {"status": "success", "output_path": str(generated)},
+        "generated_scan": app.scan_file(str(generated)),
+    }
+
+    scanned = app.scan_reference(run, False, str(blocked_reference), "Forge", state)
+    assert scanned[13]["reference_scan"]["export_gate"] == "blocked"
+    assert scanned[13]["export"] == "clear"
+    assert scanned[15]["export_gate"] == "clear"
+
+    approved = app.approve_checkpoint(run, False, scanned[15], "Forge", scanned[13])
+    assert approved[13]["provider_state"] == "export_ready"
+    assert approved[13]["export"] == "clear"
+
+
 def test_checkpoint_blocks_without_generated_artifact() -> None:
     result = app.run_weave(
         "gothic patent leather platform boots, crimson hardware",

tests/test_command_center.py

@@ -1,10 +1,26 @@
 from pathlib import Path
 
-from nexus_visual_weaver.catalog import active_stack, catalog_summary, filter_catalog, parameter_budget
+from nexus_visual_weaver.catalog import (
+    DEFAULT_ACTIVE_STACK,
+    PRIVATE_RESEARCH_STACK,
+    active_stack,
+    catalog_summary,
+    filter_catalog,
+    parameter_budget,
+)
 from nexus_visual_weaver.grounding import inspect_outfit
 from nexus_visual_weaver.model_relay import WeaverModelRelay
 from nexus_visual_weaver.planner import build_command_center_run
-from nexus_visual_weaver.render import render_command_header, render_dashboard_regions, render_operations_panel
+from nexus_visual_weaver.render import (
+    render_command_header,
+    render_dashboard_regions,
+    render_inspector,
+    render_operations_panel,
+    render_provider_cards,
+    render_topbar,
+    render_trust_strip,
+)
+from nexus_visual_weaver.schema import ModelCandidate
 from nexus_visual_weaver.security import scan_file
 from nexus_visual_weaver.taste import refine_prompt, score_prompt
 from nexus_visual_weaver.wardrobe import build_outfit_graph
@@ -218,3 +234,283 @@ def test_catalog_summary_reflects_adult_scope() -> None:
     assert default_summary["adult_catalog"] == "hidden"
     assert adult_summary["adult_catalog"] == "enabled"
     assert adult_summary["models_visible"] > default_summary["models_visible"]
+
+
+# --- render_trust_strip tests ---
+
+def test_render_trust_strip_defaults_to_idle_state() -> None:
+    html = render_trust_strip()
+
+    assert "TRUST MODEL" in html
+    assert "Generation is not export." in html
+    assert "ST3GG IDLE" in html
+    assert "EXPORT PENDING" in html
+    assert "Clean PNG -> pass." in html
+    assert "FIXTURE EVIDENCE" in html
+
+
+def test_render_trust_strip_pass_scan_shows_clear_export() -> None:
+    scan = {"status": "pass", "export_gate": "clear", "findings": ["all checks passed"], "purification_actions": ["none needed"]}
+    html = render_trust_strip(scan=scan)
+
+    assert "ST3GG PASS" in html
+    assert "EXPORT CLEAR" in html
+    assert "all checks passed" in html
+
+
+def test_render_trust_strip_review_scan_shows_blocked_export() -> None:
+    scan = {"status": "review", "export_gate": "blocked", "findings": ["trailing data after IEND"], "purification_actions": ["truncate PNG"]}
+    html = render_trust_strip(scan=scan)
+
+    assert "ST3GG REVIEW" in html
+    assert "EXPORT BLOCKED" in html
+    assert "trailing data after IEND" in html
+
+
+def test_render_trust_strip_redacts_payload_details() -> None:
+    scan = {
+        "status": "review",
+        "export_gate": "blocked",
+        "findings": ["payload excerpt: deadbeef hidden content recovered"],
+        "purification_actions": ["raw byte hex dump should stay quarantined"],
+    }
+    html = render_trust_strip(scan=scan)
+
+    assert "Redacted scan detail" in html
+    assert "deadbeef" not in html
+    assert "hex dump" not in html
+
+
+def test_render_trust_strip_approved_checkpoint_shows_pass() -> None:
+    operator_state = {"checkpoint": "approved", "provider_state": "export_ready"}
+    html = render_trust_strip(operator_state=operator_state)
+
+    assert "CHECKPOINT APPROVED" in html
+
+
+def test_render_trust_strip_pending_review_checkpoint_label() -> None:
+    operator_state = {"checkpoint": "pending_review"}
+    html = render_trust_strip(operator_state=operator_state)
+
+    assert "CHECKPOINT PENDING REVIEW" in html
+
+
+# --- render_topbar tests ---
+
+def test_render_topbar_includes_trust_strip() -> None:
+    html = render_topbar()
+
+    assert "TRUST MODEL" in html
+    assert "nw-trust-strip" in html
+
+
+def test_render_topbar_with_scan_passes_to_trust_strip() -> None:
+    scan = {"status": "pass", "export_gate": "clear", "findings": ["no issues"], "purification_actions": []}
+    html = render_topbar(scan=scan)
+
+    assert "ST3GG PASS" in html
+    assert "EXPORT CLEAR" in html
+
+
+def test_render_topbar_with_operator_state_shows_checkpoint() -> None:
+    operator_state = {"checkpoint": "approved"}
+    html = render_topbar(operator_state=operator_state)
+
+    assert "CHECKPOINT APPROVED" in html
+
+
+def test_render_topbar_default_scan_shows_fixture_evidence() -> None:
+    html = render_topbar()
+
+    assert "Clean PNG -> pass." in html
+    assert "PNG trailing bytes -> blocked." in html
+
+
+# --- render_inspector sponsor evidence tests ---
+
+def test_render_inspector_shows_sponsor_evidence_section() -> None:
+    html = render_inspector()
+
+    assert "Sponsor Evidence" in html
+    assert "OpenBMB MiniCPM" in html
+    assert "NVIDIA Nemotron" in html
+
+
+def test_render_inspector_with_missing_secret_shows_pending() -> None:
+    operator_state = {
+        "minicpm_judge": {"status": "missing_secret", "repo_id": "openbmb/MiniCPM-V-4.6"},
+        "nemotron_evidence": {"status": "missing_secret", "repo_id": "nvidia/NVIDIA-Nemotron-Parse-v1.2"},
+    }
+    html = render_inspector(operator_state=operator_state)
+
+    assert "MISSING_SECRET" in html
+
+
+def test_render_inspector_with_success_judge_shows_success_status() -> None:
+    operator_state = {
+        "minicpm_judge": {"status": "success", "repo_id": "openbmb/MiniCPM-V-4.6"},
+        "nemotron_evidence": {"status": "success", "repo_id": "nvidia/NVIDIA-Nemotron-Parse-v1.2"},
+    }
+    html = render_inspector(operator_state=operator_state)
+
+    assert html.count("SUCCESS") >= 2
+
+
+def test_render_inspector_shows_default_stack_label_without_run() -> None:
+    html = render_inspector()
+
+    assert "FLUX.2 4B / MiniCPM / LocateAnything" in html
+
+
+# --- render_provider_cards sponsor lane tests ---
+
+def test_render_provider_cards_shows_openbmb_and_nvidia_entries() -> None:
+    html = render_provider_cards()
+
+    assert "Openbmb" in html or "openbmb" in html.lower()
+    assert "Nvidia" in html or "nvidia" in html.lower()
+
+
+def test_render_provider_cards_shows_sponsor_lane_badge_for_openbmb(monkeypatch) -> None:
+    monkeypatch.setenv("MINICPM_API_KEY", "test-key")
+    monkeypatch.setenv("MINICPM_BASE_URL", "https://minicpm.example.test")
+    html = render_provider_cards()
+
+    assert "SPONSOR LANE" in html
+
+
+def test_render_provider_cards_shows_missing_secret_for_unconfigured_sponsor(monkeypatch) -> None:
+    monkeypatch.delenv("MINICPM_API_KEY", raising=False)
+    monkeypatch.delenv("OPENBMB_API_KEY", raising=False)
+    monkeypatch.delenv("NEMOTRON_API_KEY", raising=False)
+    monkeypatch.delenv("NVIDIA_API_KEY", raising=False)
+    html = render_provider_cards()
+
+    assert "MISSING SECRET" in html
+
+
+def test_render_provider_cards_configured_openbmb_shows_configured(monkeypatch) -> None:
+    monkeypatch.setenv("MINICPM_API_KEY", "test-key")
+    monkeypatch.delenv("MINICPM_BASE_URL", raising=False)
+    monkeypatch.delenv("NEMOTRON_API_KEY", raising=False)
+    monkeypatch.delenv("NVIDIA_API_KEY", raising=False)
+    monkeypatch.delenv("NEMOTRON_BASE_URL", raising=False)
+    monkeypatch.delenv("FAL_KEY", raising=False)
+    monkeypatch.delenv("NETLIFY_AUTH_TOKEN", raising=False)
+    monkeypatch.delenv("OPENAI_BASE_URL", raising=False)
+    monkeypatch.delenv("CLOUDFLARE_API_TOKEN", raising=False)
+    html = render_provider_cards()
+
+    assert "CONFIGURED" not in html
+    assert "MISSING SECRET" in html
+
+
+def test_render_provider_cards_configured_openbmb_requires_key_and_base_url(monkeypatch) -> None:
+    monkeypatch.setenv("MINICPM_API_KEY", "test-key")
+    monkeypatch.setenv("MINICPM_BASE_URL", "https://minicpm.example.test")
+    html = render_provider_cards()
+
+    assert "CONFIGURED" in html
+
+
+def test_render_operations_and_inspector_redact_payload_details() -> None:
+    scan = {
+        "status": "review",
+        "export_gate": "blocked",
+        "findings": ["payload excerpt: recovered hidden content"],
+        "purification_actions": ["base64 raw bytes quarantined"],
+    }
+
+    operations = render_operations_panel(active_section="Security", scan=scan)
+    inspector = render_inspector(scan=scan)
+
+    assert "Redacted scan detail" in operations
+    assert "Redacted scan detail" in inspector
+    assert "recovered hidden content" not in operations
+    assert "base64 raw bytes" not in inspector
+
+
+# --- catalog public_demo field tests ---
+
+def test_filter_catalog_excludes_flux_9b_in_public_mode() -> None:
+    models, _ = filter_catalog(False)
+    repo_ids = {model.repo_id for model in models}
+
+    assert "black-forest-labs/FLUX.2-klein-9B" not in repo_ids
+
+
+def test_filter_catalog_includes_flux_9b_in_adult_mode() -> None:
+    models, _ = filter_catalog(True)
+    repo_ids = {model.repo_id for model in models}
+
+    assert "black-forest-labs/FLUX.2-klein-9B" in repo_ids
+
+
+def test_filter_catalog_excludes_offellia_in_public_mode() -> None:
+    models, _ = filter_catalog(False)
+    repo_ids = {model.repo_id for model in models}
+
+    assert not any("OFFELLIA" in repo_id for repo_id in repo_ids)
+
+
+def test_active_stack_uses_private_research_stack_in_adult_mode() -> None:
+    stack = active_stack(True)
+    repo_ids = {model.repo_id for model in stack}
+
+    assert "black-forest-labs/FLUX.2-klein-9B" in repo_ids
+    assert "black-forest-labs/FLUX.2-klein-4B" not in repo_ids
+
+
+def test_private_research_stack_constant_contains_9b_and_offellia() -> None:
+    assert "black-forest-labs/FLUX.2-klein-9B" in PRIVATE_RESEARCH_STACK
+    assert "Brunobkr/OFFELLIA_Q4_0_gemma-4-12B-it.gguf" in PRIVATE_RESEARCH_STACK
+
+
+def test_default_active_stack_constant_uses_4b_and_sponsor_models() -> None:
+    assert "black-forest-labs/FLUX.2-klein-4B" in DEFAULT_ACTIVE_STACK
+    assert "openbmb/MiniCPM-V-4.6" in DEFAULT_ACTIVE_STACK
+    assert "nvidia/NVIDIA-Nemotron-Parse-v1.2" in DEFAULT_ACTIVE_STACK
+    assert "black-forest-labs/FLUX.2-klein-9B" not in DEFAULT_ACTIVE_STACK
+
+
+# --- schema ModelCandidate public_demo tests ---
+
+def test_model_candidate_public_demo_defaults_to_true() -> None:
+    candidate = ModelCandidate(
+        repo_id="test/model",
+        role="test_role",
+        task="text-to-image",
+        params_b=1.0,
+        runtime="local",
+        license="apache-2.0",
+    )
+    assert candidate.public_demo is True
+
+
+def test_model_candidate_public_demo_can_be_set_false() -> None:
+    candidate = ModelCandidate(
+        repo_id="test/private-model",
+        role="private_role",
+        task="text-to-image",
+        params_b=9.0,
+        runtime="gated provider",
+        license="other",
+        public_demo=False,
+    )
+    assert candidate.public_demo is False
+
+
+def test_public_demo_false_models_are_excluded_from_public_filter() -> None:
+    private = ModelCandidate(
+        repo_id="test/hidden-model",
+        role="private",
+        task="text-to-image",
+        params_b=2.0,
+        runtime="local",
+        license="other",
+        public_demo=False,
+    )
+    # public_demo=False should mean filter_catalog(False) excludes it
+    # The catalog-level test: verify FLUX 9B (public_demo=False) is absent
+    models_public, _ = filter_catalog(False)
+    assert all(m.public_demo for m in models_public)

tests/test_exporter.py

@@ -1,15 +1,11 @@
 import json
 from pathlib import Path
 
-from nexus_visual_weaver.exporter import write_export_packet
+from nexus_visual_weaver.exporter import export_root, write_export_packet
 from nexus_visual_weaver.planner import build_command_center_run
 
 
-def test_write_export_packet_records_evidence_without_secrets(monkeypatch) -> None:
-    export_dir = Path("outputs/test-exports")
-    monkeypatch.setenv("NEXUS_EXPORT_DIR", str(export_dir))
-    run = build_command_center_run("gothic couture archivist, platform boots")
-    scan = {"status": "pass", "export_gate": "clear", "findings": [], "purification_actions": []}
+def _make_base_state(**overrides):
     state = {
         "provider_state": "export_ready",
         "checkpoint": "approved",
@@ -18,6 +14,16 @@ def test_write_export_packet_records_evidence_without_secrets(monkeypatch) -> No
         "minicpm_judge": {"status": "success", "repo_id": "openbmb/MiniCPM-V-4.6"},
         "nemotron_evidence": {"status": "missing_secret", "repo_id": "nvidia/NVIDIA-Nemotron-Parse-v1.2"},
     }
+    state.update(overrides)
+    return state
+
+
+def test_write_export_packet_records_evidence_without_secrets(monkeypatch) -> None:
+    export_dir = Path("outputs/test-exports")
+    monkeypatch.setenv("NEXUS_EXPORT_DIR", str(export_dir))
+    run = build_command_center_run("gothic couture archivist, platform boots")
+    scan = {"status": "pass", "export_gate": "clear", "findings": [], "purification_actions": []}
+    state = _make_base_state()
 
     result = write_export_packet(run=run, scan=scan, operator_state=state, adult_mode=False)
     payload = json.loads(Path(result["path"]).read_text(encoding="utf-8"))
@@ -27,3 +33,154 @@ def test_write_export_packet_records_evidence_without_secrets(monkeypatch) -> No
     assert payload["hackathon_claims"]["nvidia_nemotron_lane"] is False
     assert payload["parameter_budget"]["status"] == "pass"
     assert "token" not in json.dumps(payload).lower()
+    assert payload["artifact"] == "artifact.png"
+    assert payload["generation"]["output_path"] == "artifact.png"
+    assert "/data/" not in json.dumps(payload)
+
+
+def test_export_packet_has_correct_schema_version(monkeypatch) -> None:
+    monkeypatch.setenv("NEXUS_EXPORT_DIR", "outputs/test-exports")
+    run = build_command_center_run("dark couture brief")
+    scan = {"status": "pass", "export_gate": "clear"}
+    state = _make_base_state()
+
+    result = write_export_packet(run=run, scan=scan, operator_state=state, adult_mode=False)
+    payload = json.loads(Path(result["path"]).read_text(encoding="utf-8"))
+
+    assert payload["schema"] == "nexus_visual_weaver.export_packet.v1"
+
+
+def test_export_packet_stores_adult_mode_flag(monkeypatch) -> None:
+    monkeypatch.setenv("NEXUS_EXPORT_DIR", "outputs/test-exports")
+    run_public = build_command_center_run("dark couture brief", adult_mode=False)
+    run_private = build_command_center_run("dark couture brief", adult_mode=True)
+    scan = {"status": "pass", "export_gate": "clear"}
+
+    result_public = write_export_packet(run=run_public, scan=scan, operator_state=_make_base_state(), adult_mode=True)
+    result_private = write_export_packet(run=run_private, scan=scan, operator_state=_make_base_state(), adult_mode=False)
+
+    public_payload = json.loads(Path(result_public["path"]).read_text(encoding="utf-8"))
+    private_payload = json.loads(Path(result_private["path"]).read_text(encoding="utf-8"))
+
+    assert public_payload["adult_mode"] is False
+    assert private_payload["adult_mode"] is True
+    assert public_payload["model_stack"][0]["repo_id"] == run_public.model_stack[0].repo_id
+    assert private_payload["model_stack"][0]["repo_id"] == run_private.model_stack[0].repo_id
+
+
+def test_export_packet_removes_hf_token_from_generation(monkeypatch) -> None:
+    monkeypatch.setenv("NEXUS_EXPORT_DIR", "outputs/test-exports")
+    run = build_command_center_run("gothic platform boots brief")
+    scan = {"status": "pass", "export_gate": "clear"}
+    state = _make_base_state(
+        generation={"status": "success", "output_path": "/data/artifact.png", "hf_token_present": True},
+    )
+
+    result = write_export_packet(run=run, scan=scan, operator_state=state, adult_mode=False)
+    payload = json.loads(Path(result["path"]).read_text(encoding="utf-8"))
+
+    assert "hf_token_present" not in payload["generation"]
+    assert "token" not in json.dumps(payload).lower()
+
+
+def test_export_packet_hackathon_claims_both_success(monkeypatch) -> None:
+    monkeypatch.setenv("NEXUS_EXPORT_DIR", "outputs/test-exports")
+    run = build_command_center_run("couture archivist brief")
+    scan = {"status": "pass", "export_gate": "clear"}
+    state = _make_base_state(
+        minicpm_judge={"status": "success"},
+        nemotron_evidence={"status": "success"},
+    )
+
+    result = write_export_packet(run=run, scan=scan, operator_state=state, adult_mode=False)
+    payload = json.loads(Path(result["path"]).read_text(encoding="utf-8"))
+
+    assert payload["hackathon_claims"]["openbmb_lane"] is True
+    assert payload["hackathon_claims"]["nvidia_nemotron_lane"] is True
+    assert payload["hackathon_claims"]["gradio_space"] is True
+    assert payload["hackathon_claims"]["off_brand_custom_ui"] is True
+
+
+def test_export_packet_hackathon_claims_st3gg_export_gate(monkeypatch) -> None:
+    monkeypatch.setenv("NEXUS_EXPORT_DIR", "outputs/test-exports")
+    run_clear = build_command_center_run("scan gate test brief clear")
+    run_blocked = build_command_center_run("scan gate test brief blocked")
+    scan_clear = {"status": "pass", "export_gate": "clear"}
+    scan_blocked = {"status": "review", "export_gate": "blocked"}
+
+    result_clear = write_export_packet(run=run_clear, scan=scan_clear, operator_state=_make_base_state(), adult_mode=False)
+    result_blocked = write_export_packet(run=run_blocked, scan=scan_blocked, operator_state=_make_base_state(), adult_mode=False)
+
+    payload_clear = json.loads(Path(result_clear["path"]).read_text(encoding="utf-8"))
+    payload_blocked = json.loads(Path(result_blocked["path"]).read_text(encoding="utf-8"))
+
+    assert payload_clear["hackathon_claims"]["st3gg_export_gate"] == "clear"
+    assert payload_blocked["hackathon_claims"]["st3gg_export_gate"] == "blocked"
+
+
+def test_export_packet_includes_model_stack_and_prompts(monkeypatch) -> None:
+    monkeypatch.setenv("NEXUS_EXPORT_DIR", "outputs/test-exports")
+    run = build_command_center_run("raven archivist couture brief")
+    scan = {"status": "pass", "export_gate": "clear"}
+
+    result = write_export_packet(run=run, scan=scan, operator_state=_make_base_state(), adult_mode=False)
+    payload = json.loads(Path(result["path"]).read_text(encoding="utf-8"))
+
+    assert isinstance(payload["model_stack"], list)
+    assert len(payload["model_stack"]) > 0
+    assert all("repo_id" in entry and "params_b" in entry for entry in payload["model_stack"])
+    assert isinstance(payload["prompt"], str)
+    assert isinstance(payload["refined_prompt"], str)
+    assert isinstance(payload["created_at_epoch"], int)
+
+
+def test_export_root_uses_nexus_export_dir_env(monkeypatch) -> None:
+    custom_dir = Path("outputs/test-exports/custom_export")
+    monkeypatch.setenv("NEXUS_EXPORT_DIR", str(custom_dir))
+
+    root = export_root()
+
+    assert root == custom_dir.resolve()
+    assert root.is_dir()
+
+
+def test_export_root_rejects_src_export_dir(monkeypatch) -> None:
+    unsafe_dir = Path("src/nexus_visual_weaver/export-leak")
+    monkeypatch.setenv("NEXUS_EXPORT_DIR", str(unsafe_dir))
+
+    root = export_root()
+
+    src_root = (Path.cwd() / "src").resolve()
+    assert root != src_root
+    assert src_root not in root.parents
+    assert not unsafe_dir.exists()
+
+
+def test_export_root_falls_back_to_outputs_when_no_env(monkeypatch) -> None:
+    monkeypatch.delenv("NEXUS_EXPORT_DIR", raising=False)
+    # Patch /data so it is not seen as existing
+    import nexus_visual_weaver.exporter as exporter_mod
+    original_exists = Path.exists
+
+    def patched_exists(self):
+        if str(self) == "/data":
+            return False
+        return original_exists(self)
+
+    monkeypatch.setattr(Path, "exists", patched_exists)
+
+    root = export_root()
+    assert "outputs" in str(root) or "exports" in str(root)
+
+
+def test_export_packet_returns_both_path_and_packet(monkeypatch) -> None:
+    monkeypatch.setenv("NEXUS_EXPORT_DIR", "outputs/test-exports")
+    run = build_command_center_run("return structure brief")
+    scan = {"status": "pass", "export_gate": "clear"}
+
+    result = write_export_packet(run=run, scan=scan, operator_state=_make_base_state(), adult_mode=False)
+
+    assert "path" in result
+    assert "packet" in result
+    assert isinstance(result["packet"], dict)
+    assert result["packet"]["schema"] == "nexus_visual_weaver.export_packet.v1"

tests/test_model_relay.py

@@ -138,3 +138,79 @@ def test_optional_external_gateways_are_registered_but_excluded_by_default() ->
     assert relay.records["netflix-void-modal"].health == "healthy"
     assert relay.records["netlify-ai-gateway-helper"].health == "excluded"
     assert relay.records["fal-media-adapter"].health == "excluded"
+
+
+def test_minicpm_v46_is_registered_in_taste_judge_lane() -> None:
+    relay = WeaverModelRelay()
+
+    record = relay.records["minicpm-v46-visual-judge"]
+
+    assert record.lane == "taste_judge"
+    assert record.repo_id == "openbmb/MiniCPM-V-4.6"
+    assert record.provider == "openbmb"
+    assert record.params_b == 1.30
+    assert record.license_gate == "apache-2.0"
+    assert record.health == "healthy"
+
+
+def test_nemotron_parse_is_registered_in_taste_judge_lane() -> None:
+    relay = WeaverModelRelay()
+
+    record = relay.records["nemotron-parse-v12-evidence"]
+
+    assert record.lane == "taste_judge"
+    assert record.repo_id == "nvidia/NVIDIA-Nemotron-Parse-v1.2"
+    assert record.provider == "hf_nvidia"
+    assert record.params_b == 0.94
+    assert record.health == "healthy"
+
+
+def test_nemotron_nano_is_registered_as_fallback_for_parse() -> None:
+    relay = WeaverModelRelay()
+
+    parse_record = relay.records["nemotron-parse-v12-evidence"]
+    nano_record = relay.records["nemotron-nano-4b-gguf-evidence"]
+
+    assert "nemotron-nano-4b-gguf-evidence" in (parse_record.fallback_chain or ())
+    assert nano_record.repo_id == "nvidia/NVIDIA-Nemotron-3-Nano-4B-GGUF"
+    assert nano_record.lane == "taste_judge"
+    assert nano_record.params_b == 3.97
+
+
+def test_private_image_research_lane_is_rotatable() -> None:
+    from nexus_visual_weaver.model_relay import PINNED_LANES, ROTATABLE_LANES
+
+    assert "private_image_research" not in PINNED_LANES
+    assert "private_image_research" in ROTATABLE_LANES
+
+
+def test_flux2_klein_4b_is_pinned_with_apache_license() -> None:
+    relay = WeaverModelRelay()
+
+    record = relay.records["flux2-klein-4b-public"]
+
+    assert record.lane == "image_generation"
+    assert record.pinned is True
+    assert record.params_b == 4.0
+    assert record.license_gate == "apache-2.0"
+    assert record.repo_id == "black-forest-labs/FLUX.2-klein-4B"
+
+
+def test_flux2_klein_9b_is_not_pinned_and_in_private_research() -> None:
+    relay = WeaverModelRelay()
+
+    record = relay.records["flux2-klein-9b-private"]
+
+    assert record.lane == "private_image_research"
+    assert record.pinned is False
+    assert record.params_b == 9.0
+    assert record.license_gate == "review_required"
+
+
+def test_minicpm_has_fallback_chain_configured() -> None:
+    relay = WeaverModelRelay()
+
+    record = relay.records["minicpm-v46-visual-judge"]
+
+    assert record.fallback_chain is not None
+    assert len(record.fallback_chain) > 0

tests/test_provider_runtime.py

@@ -1,6 +1,18 @@
 from PIL import Image
 
-from nexus_visual_weaver.provider_runtime import judge_with_minicpm, judge_with_nemotron
+from nexus_visual_weaver.provider_runtime import (
+    NEMOTRON_NANO_REPO_ID,
+    NEMOTRON_PARSE_REPO_ID,
+    OPENBMB_REPO_ID,
+    ProviderJudgeResult,
+    _extract_content,
+    _image_data_url,
+    _post_json,
+    _safe_json_from_text,
+    _short_error,
+    judge_with_minicpm,
+    judge_with_nemotron,
+)
 
 
 def test_minicpm_reports_missing_secret(monkeypatch) -> None:
@@ -70,3 +82,302 @@ def test_nemotron_reports_missing_secret(monkeypatch) -> None:
     assert result.status == "missing_secret"
     assert result.provider == "NVIDIA"
     assert "Nemotron" in result.message
+
+
+# --- _short_error tests ---
+
+def test_short_error_prefixes_class_name() -> None:
+    exc = ValueError("something went wrong")
+    result = _short_error(exc)
+    assert result.startswith("ValueError:")
+    assert "something went wrong" in result
+
+
+def test_short_error_truncates_long_messages() -> None:
+    long_message = "x" * 500
+    exc = RuntimeError(long_message)
+    result = _short_error(exc)
+    assert len(result) <= len("RuntimeError: ") + 360
+    assert result.endswith("...")
+
+
+def test_short_error_collapses_newlines() -> None:
+    exc = OSError("line one\nline two\nline three")
+    result = _short_error(exc)
+    assert "\n" not in result
+    assert "line one" in result
+
+
+# --- _image_data_url tests ---
+
+def test_image_data_url_returns_none_for_none_path() -> None:
+    assert _image_data_url(None) is None
+
+
+def test_image_data_url_returns_none_for_missing_file() -> None:
+    assert _image_data_url("/nonexistent/path/to/image.png") is None
+
+
+def test_image_data_url_encodes_png_as_base64() -> None:
+    image_path = "outputs/test-img-data-url.png"
+    Image.new("RGB", (4, 4), color=(0, 0, 0)).save(image_path)
+
+    result = _image_data_url(image_path)
+
+    assert result is not None
+    assert result.startswith("data:image/png;base64,")
+
+
+def test_image_data_url_uses_jpeg_mime_for_jpg() -> None:
+    image_path = "outputs/test-img-data-url.jpg"
+    Image.new("RGB", (4, 4), color=(0, 0, 0)).save(image_path)
+
+    result = _image_data_url(image_path)
+
+    assert result is not None
+    assert result.startswith("data:image/jpeg;base64,")
+
+
+# --- _extract_content tests ---
+
+def test_extract_content_returns_string_from_choices() -> None:
+    response = {"choices": [{"message": {"content": "hello world"}}]}
+    assert _extract_content(response) == "hello world"
+
+
+def test_extract_content_returns_empty_for_no_choices() -> None:
+    assert _extract_content({"choices": []}) == ""
+    assert _extract_content({}) == ""
+
+
+def test_extract_content_serializes_non_string_content() -> None:
+    response = {"choices": [{"message": {"content": {"key": "value"}}}]}
+    result = _extract_content(response)
+    assert "key" in result
+    assert "value" in result
+
+
+# --- _safe_json_from_text tests ---
+
+def test_safe_json_from_text_parses_valid_json() -> None:
+    text = '{"status": "pass", "score": 0.9}'
+    result = _safe_json_from_text(text)
+    assert result == {"status": "pass", "score": 0.9}
+
+
+def test_safe_json_from_text_returns_empty_dict_for_empty_string() -> None:
+    assert _safe_json_from_text("") == {}
+    assert _safe_json_from_text("   ") == {}
+
+
+def test_safe_json_from_text_falls_back_on_invalid_json() -> None:
+    text = "this is not valid json"
+    result = _safe_json_from_text(text)
+    assert "raw_summary" in result
+    assert "this is not valid json" in result["raw_summary"]
+
+
+def test_safe_json_from_text_extracts_embedded_json() -> None:
+    text = 'Some prefix text {"result": "ok"} trailing text'
+    result = _safe_json_from_text(text)
+    assert result == {"result": "ok"}
+
+
+def test_safe_json_from_text_truncates_fallback_to_1200_chars() -> None:
+    long_text = "not json " + "x" * 2000
+    result = _safe_json_from_text(long_text)
+    assert "raw_summary" in result
+    assert len(result["raw_summary"]) <= 1200
+
+
+# --- ProviderJudgeResult.to_dict tests ---
+
+def test_provider_judge_result_to_dict_contains_all_fields() -> None:
+    result = ProviderJudgeResult(
+        status="success",
+        provider_state="configured",
+        provider="OpenBMB",
+        repo_id=OPENBMB_REPO_ID,
+        model="MiniCPM-V-4.6",
+        message="judge returned evidence",
+        evidence={"overall_status": "pass"},
+        latency_seconds=1.23,
+    )
+    d = result.to_dict()
+
+    assert d["status"] == "success"
+    assert d["provider_state"] == "configured"
+    assert d["provider"] == "OpenBMB"
+    assert d["repo_id"] == OPENBMB_REPO_ID
+    assert d["model"] == "MiniCPM-V-4.6"
+    assert d["latency_seconds"] == 1.23
+    assert d["evidence"]["overall_status"] == "pass"
+
+
+# --- nemotron success/failure tests ---
+
+def test_nemotron_success_with_mocked_post(monkeypatch) -> None:
+    monkeypatch.setenv("NEMOTRON_BASE_URL", "http://nemotron.test")
+    monkeypatch.setenv("NEMOTRON_API_KEY", "nvidia-token")
+
+    def fake_post(url, token, payload, timeout):
+        assert "Nemotron" in payload["model"] or "nemotron" in payload["model"]
+        return {"choices": [{"message": {"content": '{"final_claim_status":"pass","structured_parse":"ok"}'}}]}
+
+    monkeypatch.setattr("nexus_visual_weaver.provider_runtime._post_json", fake_post)
+
+    result = judge_with_nemotron(
+        prompt="gothic couture brief",
+        run_packet={"checkpoint": {"checkpoint_id": "nw-test-123"}},
+        minicpm_result={"status": "success"},
+    )
+
+    assert result.status == "success"
+    assert result.provider == "NVIDIA"
+    assert result.evidence["final_claim_status"] == "pass"
+    assert result.latency_seconds is not None
+
+
+def test_nemotron_failed_api_call_returns_failed_status(monkeypatch) -> None:
+    import urllib.error
+
+    monkeypatch.setenv("NEMOTRON_BASE_URL", "http://nemotron.test")
+    monkeypatch.setenv("NEMOTRON_API_KEY", "nvidia-token")
+
+    def fake_post(url, token, payload, timeout):
+        raise urllib.error.URLError("connection refused")
+
+    monkeypatch.setattr("nexus_visual_weaver.provider_runtime._post_json", fake_post)
+
+    result = judge_with_nemotron(
+        prompt="brief",
+        run_packet={"id": "nw-fail"},
+    )
+
+    assert result.status == "failed"
+    assert result.provider_state == "failed"
+    assert "URLError" in result.message or "connection" in result.message.lower()
+    assert result.evidence.get("configured") is True
+
+
+def test_minicpm_failed_api_call_returns_failed_status(monkeypatch) -> None:
+    import urllib.error
+
+    image_path = "outputs/test-minicpm-fail.png"
+    Image.new("RGB", (4, 4), color=(0, 0, 0)).save(image_path)
+    monkeypatch.setenv("MINICPM_BASE_URL", "http://minicpm.test")
+    monkeypatch.setenv("MINICPM_API_KEY", "test-token")
+
+    def fake_post(url, token, payload, timeout):
+        raise urllib.error.URLError("network unreachable")
+
+    monkeypatch.setattr("nexus_visual_weaver.provider_runtime._post_json", fake_post)
+
+    result = judge_with_minicpm(
+        prompt="gothic couture",
+        image_path=image_path,
+        scan={"export_gate": "clear"},
+        wardrobe_summary="platform boots",
+    )
+
+    assert result.status == "failed"
+    assert result.provider_state == "failed"
+    assert result.provider == "OpenBMB"
+    assert result.evidence.get("configured") is True
+
+
+def test_nemotron_uses_parse_repo_id_for_parse_model(monkeypatch) -> None:
+    monkeypatch.setenv("NEMOTRON_BASE_URL", "http://nemotron.test")
+    monkeypatch.setenv("NEMOTRON_API_KEY", "nvidia-token")
+    monkeypatch.setenv("NEMOTRON_MODEL", "nvidia/NVIDIA-Nemotron-Parse-v1.2")
+
+    def fake_post(url, token, payload, timeout):
+        return {"choices": [{"message": {"content": "{}"}}]}
+
+    monkeypatch.setattr("nexus_visual_weaver.provider_runtime._post_json", fake_post)
+
+    result = judge_with_nemotron(prompt="brief", run_packet={})
+
+    assert result.repo_id == NEMOTRON_PARSE_REPO_ID
+
+
+def test_nemotron_uses_nano_repo_id_for_non_parse_model(monkeypatch) -> None:
+    monkeypatch.setenv("NEMOTRON_BASE_URL", "http://nemotron.test")
+    monkeypatch.setenv("NEMOTRON_API_KEY", "nvidia-token")
+    monkeypatch.setenv("NEMOTRON_MODEL", "nvidia/NVIDIA-Nemotron-3-Nano-4B-GGUF")
+
+    def fake_post(url, token, payload, timeout):
+        return {"choices": [{"message": {"content": "{}"}}]}
+
+    monkeypatch.setattr("nexus_visual_weaver.provider_runtime._post_json", fake_post)
+
+    result = judge_with_nemotron(prompt="brief", run_packet={})
+
+    assert result.repo_id == NEMOTRON_NANO_REPO_ID
+
+
+def test_minicpm_uses_openbmb_api_key_as_fallback(monkeypatch) -> None:
+    monkeypatch.setenv("MINICPM_BASE_URL", "http://minicpm.test")
+    monkeypatch.delenv("MINICPM_API_KEY", raising=False)
+    monkeypatch.setenv("OPENBMB_API_KEY", "openbmb-fallback-token")
+
+    image_path = "outputs/test-openbmb-key.png"
+    Image.new("RGB", (4, 4), color=(0, 0, 0)).save(image_path)
+
+    captured = {}
+
+    def fake_post(url, token, payload, timeout):
+        captured["token"] = token
+        return {"choices": [{"message": {"content": '{"status": "ok"}'}}]}
+
+    monkeypatch.setattr("nexus_visual_weaver.provider_runtime._post_json", fake_post)
+
+    result = judge_with_minicpm(
+        prompt="test",
+        image_path=image_path,
+        scan={},
+        wardrobe_summary="",
+    )
+
+    assert result.status == "success"
+    assert captured["token"] == "openbmb-fallback-token"
+
+
+def test_nemotron_uses_nvidia_api_key_as_fallback(monkeypatch) -> None:
+    monkeypatch.setenv("NEMOTRON_BASE_URL", "http://nemotron.test")
+    monkeypatch.delenv("NEMOTRON_API_KEY", raising=False)
+    monkeypatch.setenv("NVIDIA_API_KEY", "nvidia-fallback-token")
+
+    captured = {}
+
+    def fake_post(url, token, payload, timeout):
+        captured["token"] = token
+        return {"choices": [{"message": {"content": "{}"}}]}
+
+    monkeypatch.setattr("nexus_visual_weaver.provider_runtime._post_json", fake_post)
+
+    result = judge_with_nemotron(prompt="brief", run_packet={})
+
+    assert result.status == "success"
+    assert captured["token"] == "nvidia-fallback-token"
+
+
+def test_post_json_rejects_unsupported_url_schemes_before_urlopen(monkeypatch) -> None:
+    called = False
+
+    def fake_urlopen(*args, **kwargs):
+        nonlocal called
+        called = True
+        raise AssertionError("urlopen should not be called for invalid schemes")
+
+    monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
+
+    for url in ["file:///tmp/payload.json", "ftp://example.test/api", "http:///missing-host"]:
+        try:
+            _post_json(url, "token", {"ok": True}, 1.0)
+        except ValueError as exc:
+            assert "Invalid URL" in str(exc)
+        else:
+            raise AssertionError(f"{url} should have been rejected")
+
+    assert called is False