Deploy v4 review hardening

app.py

@@ -629,7 +629,11 @@ def approve_checkpoint(
             "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.",
+            "message": (
+                "Checkpoint approved. Export is ready after clear ST3GG scan."
+                if export_state == "clear"
+                else "Checkpoint approved. ST3GG is not clear; add an override reason and click Prepare Export Packet to write an audit packet."
+            ),
         }
     return _render_stateful(run, adult_mode, scan, active_section, next_state)
 
@@ -675,9 +679,10 @@ def export_packet(
         next_state = {**state, "provider_state": "blocked", "export": scan.get("export_gate", "blocked"), "message": "Export blocked: ST3GG gate is not clear. Add an explicit override reason to write an audit packet."}
     else:
         export_state = "clear" if scan.get("export_gate") == "clear" else "override"
+        override_applies = scan.get("export_gate") != "clear" and bool(override_reason)
         export_operator_state = {
             **state,
-            **({"st3gg_override_reason": override_reason} if override_reason else {}),
+            **({"st3gg_override_reason": override_reason} if override_applies else {}),
             "export": export_state,
         }
         export = write_export_packet(run=run, scan=scan, operator_state=export_operator_state, adult_mode=adult_mode)
@@ -839,7 +844,7 @@ with gr.Blocks(title="NEXUS Visual Weaver") as demo:
             reset_btn = gr.Button("Reset Demo State", scale=1)
         override_reason = gr.Textbox(
             label="ST3GG Override Reason",
-            placeholder="Required only when exporting an audit packet for a reviewed/blocked artifact.",
+            placeholder="Required when ST3GG is review/blocked; explain why this audit packet may be written.",
             lines=2,
             max_lines=3,
         )

docs/HACKATHON_EVALUATION.md

@@ -13,8 +13,8 @@ NEXUS Visual Weaver should open as a working command center, not a landing page.
 ## Current Strengths
 
 - Gradio-compatible app shape with `mcp_server=True`.
-- Pinned model governance is visible: FLUX.2 Klein 4B, LocateAnything-3B, and ST3GG.
-- Real FLUX.2 Klein 4B generation is wired for HF Space and falls back to an honest dry-run state outside Space.
+- Pinned model governance is visible: FLUX.2 Klein 9B, LocateAnything-3B, and ST3GG, with FLUX.2 Klein 4B kept as a sidecar fallback.
+- Real FLUX.2 Klein 9B generation is wired for HF Space, falls back to the 4B sidecar when needed, and reports an honest dry-run state outside Space.
 - Generated artifacts are scanned by ST3GG before checkpoint/export.
 - Above-fold trust strip makes ST3GG verdict, export gate, fixture evidence, and adult-mode safety boundaries visible immediately.
 - OpenBMB MiniCPM-V 4.6 and NVIDIA Nemotron evidence lanes are represented as real optional provider adapters with missing-secret/failed/success states.
@@ -48,5 +48,5 @@ NEXUS Visual Weaver should open as a working command center, not a landing page.
 | OpenBMB | Claim only after MiniCPM-V returns `success` in export evidence. |
 | NVIDIA | Claim only after Nemotron returns `success` in export evidence. |
 | OpenAI Codex | GitHub branch/PR provides Codex development trail. |
-| Tiny Titan | Public demo stack is eligible: active public models are each <=4B. |
+| Tiny Titan | Sidecar-only: claim only from an explicit 4B export packet where every active model is <=4B. |
 | Modal | Not claimed unless a real Modal job runs. |

docs/HANDOFF_FINAL_HACKATHON.md

@@ -8,13 +8,13 @@
 - Public Space URL: `https://build-small-hackathon-nexus-visual-weaver-a107340.hf.space/`
 - HF rollback SHA: `410a467c55d11e7308249198bd5fe0b2c190aec6`.
 - Branch discipline: use only `main` and `codex/specimba/ui-polish-command-center`; no extra recovery branches.
-- Primary goal: finish a countable Build Small submission with real FLUX.2 4B generation, ST3GG scan, optional OpenBMB MiniCPM-V judge evidence, optional NVIDIA Nemotron evidence, checkpointed export packet, README prize mapping, demo video, and social post.
+- Primary goal: finish a countable Build Small submission with real FLUX.2 Klein 9B generation, 4B sidecar fallback, ST3GG scan, optional OpenBMB MiniCPM-V judge evidence, optional NVIDIA Nemotron evidence, checkpointed export packet, README prize mapping, demo video, and social post.
 
 ## Secrets Needed
 
 Do not paste these into chat, commits, logs, or export packets.
 
-- `HF_TOKEN`: optional for public FLUX.2 Klein 4B access and required only if private/gated research lanes are enabled.
+- `HF_TOKEN`: required when the FLUX.2 Klein 9B Space runtime needs authenticated access; the 4B sidecar remains the public-safe fallback/Tiny Titan lane.
 - `MINICPM_BASE_URL`: OpenBMB OpenAI-compatible endpoint base URL.
 - `MINICPM_API_KEY`: OpenBMB bearer token.
 - `MINICPM_MODEL`: default `MiniCPM-V-4.6`.
@@ -39,7 +39,7 @@ Avoid pytest `--basetemp=C:\tmp` in this Windows sandbox if `tmp_path` fixtures
 ## Runtime Flow
 
 1. `run_active_weave` builds the Raven Chronicle run packet.
-2. FLUX.2 Klein 4B generates the image on Space when HF runtime is enabled.
+2. FLUX.2 Klein 9B generates the image on Space when HF runtime is enabled, with FLUX.2 Klein 4B as the fallback/sidecar lane.
 3. Generated artifact is scanned by ST3GG.
 4. MiniCPM-V judge runs when OpenBMB secrets are present.
 5. Nemotron evidence runs when Nemotron/NVIDIA endpoint secrets are present.
@@ -51,14 +51,14 @@ 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 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.
+- Tiny Titan can be claimed only from an explicit 4B sidecar export packet where every active model is <=4B.
+- Raven Quality Stack uses FLUX.2 Klein 9B by default; OFFELLIA/Gemma remain private research options behind Adult Mode/catalog partitioning.
 - Modal is not claimed unless a real Modal job runs and is documented.
 
 ## Known Risks
 
 - GitHub CLI may fail behind proxy `127.0.0.1:9`; use local git status and HF verification when blocked.
-- Real FLUX generation depends on Space GPU availability and the public 4B runtime loading successfully.
+- Real FLUX generation depends on Space GPU availability and the 9B runtime loading successfully, with a 4B sidecar fallback when configured.
 - OpenBMB and Nemotron endpoints are optional and must show `missing secret` rather than fake success when not configured.
 - Demo video and social post links must be added before final submission.
 

src/nexus_visual_weaver/exporter.py

@@ -64,7 +64,18 @@ def _artifact_name(output_path: Any) -> str | None:
     """
     if not output_path:
         return None
-    return Path(str(output_path)).name
+    text = str(output_path)
+    if "\\" in text or re.match(r"^[A-Za-z]:[\\/]", text):
+        return PureWindowsPath(text).name
+    return Path(text).name
+
+
+def _safe_run_id(value: Any) -> str:
+    text = str(value or f"nw-{int(time.time())}")
+    safe = re.sub(r"[^A-Za-z0-9_.-]+", "-", text).strip(".-")
+    while ".." in safe:
+        safe = safe.replace("..", ".")
+    return safe or f"nw-{int(time.time())}"
 
 
 SENSITIVE_KEY_RE = re.compile(r"(token|secret|api[_-]?key|authorization|payload_excerpt|raw|base64|bytes)", re.IGNORECASE)
@@ -245,7 +256,8 @@ def write_export_packet(
         Dictionary with "path" (str) pointing to the written JSON file and "packet" (dict)
         containing the constructed export packet.
     """
-    run_id = getattr(getattr(run, "checkpoint", None), "checkpoint_id", f"nw-{int(time.time())}")
+    raw_run_id = getattr(getattr(run, "checkpoint", None), "checkpoint_id", f"nw-{int(time.time())}")
+    run_id = _safe_run_id(raw_run_id)
     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)
@@ -260,11 +272,12 @@ def write_export_packet(
     st3gg_scan = _safe_scan(scan)
     minicpm_judge = _safe_provider(operator_state.get("minicpm_judge") or {})
     nemotron_evidence = _safe_provider(operator_state.get("nemotron_evidence") or {})
+    operator_provider_state = _safe_dict(operator_state.get("provider_state"))
     provider_states = {
         "generation": generation.get("provider_state"),
         "minicpm": minicpm_judge.get("status"),
         "nemotron": nemotron_evidence.get("status"),
-        "operator": operator_state.get("provider_state"),
+        "operator": operator_provider_state,
     }
     override_reason = _safe_dict(operator_state.get("st3gg_override_reason", ""))
     packet = {
@@ -272,8 +285,8 @@ def write_export_packet(
         "run_id": run_id,
         "created_at_epoch": int(time.time()),
         "adult_mode": run_adult_mode,
-        "prompt": getattr(getattr(run, "request", None), "prompt", ""),
-        "refined_prompt": getattr(getattr(run, "refined_prompt", None), "refined", ""),
+        "prompt": _safe_dict(getattr(getattr(run, "request", None), "prompt", "")),
+        "refined_prompt": _safe_dict(getattr(getattr(run, "refined_prompt", None), "refined", "")),
         "artifact": artifact,
         "image_basename": artifact,
         "creator_controls": creator_controls,
@@ -296,13 +309,13 @@ def write_export_packet(
         "minicpm_judge": minicpm_judge,
         "nemotron_evidence": nemotron_evidence,
         "checkpoint": {
-            "status": operator_state.get("checkpoint"),
-            "message": operator_state.get("message"),
-            "recommendation": getattr(getattr(run, "checkpoint", None), "recommendation", None),
+            "status": _safe_dict(operator_state.get("checkpoint")),
+            "message": _safe_dict(operator_state.get("message")),
+            "recommendation": _safe_dict(getattr(getattr(run, "checkpoint", None), "recommendation", None)),
             "trust_score": getattr(getattr(run, "checkpoint", None), "trust_score", None),
-            "required_actions": getattr(getattr(run, "checkpoint", None), "required_actions", []),
+            "required_actions": _safe_dict(getattr(getattr(run, "checkpoint", None), "required_actions", [])),
         },
-        "provider_state": operator_state.get("provider_state"),
+        "provider_state": operator_provider_state,
         "provider_states": provider_states,
         "model_stack": [
             {
@@ -324,6 +337,9 @@ def write_export_packet(
             "st3gg_export_gate": st3gg_scan.get("export_gate"),
         },
     }
-    target = export_root() / f"{run_id}.json"
+    root = export_root()
+    target = (root / f"{run_id}.json").resolve(strict=False)
+    if not (target == root or _is_within(target, root)):
+        raise ValueError("Unsafe export target path.")
     target.write_text(json.dumps(packet, indent=2, ensure_ascii=True), encoding="utf-8")
     return {"path": str(target), "packet": packet}

src/nexus_visual_weaver/hf_runtime.py

@@ -8,6 +8,7 @@ import time
 from dataclasses import dataclass, asdict
 from pathlib import Path
 from typing import Any
+from uuid import uuid4
 
 from .catalog import ADAPTER_CATALOG
 from .lora_adapter import load_and_apply, unload_all
@@ -274,7 +275,7 @@ def generate_flux_image(
                 ).images[0]
             finally:
                 unload_all(pipe)
-            output_path = _output_dir() / f"nexus_flux_{int(time.time())}_{seed}.png"
+            output_path = _output_dir() / f"nexus_flux_{time.time_ns()}_{seed}_{uuid4().hex[:8]}.png"
             image.save(output_path)
             return HFGenerationResult(
                 status="success",

src/nexus_visual_weaver/lora_adapter.py

@@ -116,6 +116,7 @@ def load_and_apply(
             pipe.set_adapters([adapter_name], adapter_weights=[recipe.weight])
         return _status("loaded", recipe, message="Adapter loaded and applied for this generation.", adapter_name=adapter_name)
     except Exception as exc:
+        unload_all(pipe)
         return _status("failed", recipe, message=_short_error(exc), adapter_name=adapter_name)
 
 

src/nexus_visual_weaver/model_relay.py

@@ -196,7 +196,8 @@ class WeaverModelRelay:
         lane_records = [record for record in self.records.values() if record.lane == lane]
         if lane in PINNED_LANES:
             primary = next((record for record in lane_records if record.pinned), None)
-            fallbacks = [self.records[model_id] for model_id in (primary.fallback_chain if primary else ()) if model_id in self.records]
+            fallback_records = [self.records[model_id] for model_id in (primary.fallback_chain if primary else ()) if model_id in self.records]
+            fallbacks, skipped = self._eligible_records(fallback_records, budget_b, public_demo, strategy, now)
             return LaneDecision(
                 lane=lane,
                 strategy="pinned",
@@ -206,7 +207,7 @@ class WeaverModelRelay:
                 expected_cost_hint=primary.cost_hint if primary else "unavailable",
                 quota_impact=self._quota_impact(primary, now) if primary else {},
                 context_packet=context,
-                skipped=[] if primary else [f"{lane}: no pinned model registered"],
+                skipped=skipped if primary else [f"{lane}: no pinned model registered"],
             )
 
         candidates, skipped = self._eligible_records(lane_records, budget_b, public_demo, strategy, now)

src/nexus_visual_weaver/planner.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from copy import deepcopy
 from uuid import uuid4
 
 from .catalog import ADAPTER_CATALOG, active_stack
@@ -34,12 +35,19 @@ def build_command_center_run(
     Returns:
         A GenerationRun containing the refined prompt, generated outfit, inspection results, model stack, adapters, video plan, lore components, and a human checkpoint with approval recommendation based on a computed trust score.
     """
-    controls = creator_controls or {}
-    references = reference_metadata or []
+    controls = deepcopy(creator_controls or {})
+    references = deepcopy(reference_metadata or [])
+    reference_ids: list[str] = []
+    for item in references:
+        if not isinstance(item, dict):
+            continue
+        value = item.get("id") or item.get("basename") or item.get("url_hash")
+        if value is not None and str(value):
+            reference_ids.append(str(value))
     request = CreativeRequest(
         prompt=prompt,
         adult_mode=adult_mode,
-        references=[str(item.get("id") or item.get("basename") or item.get("url_hash")) for item in references],
+        references=reference_ids,
         creator_controls=controls,
         reference_metadata=references,
     )

src/nexus_visual_weaver/provider_runtime.py

@@ -17,6 +17,14 @@ from typing import Any
 OPENBMB_REPO_ID = "openbmb/MiniCPM-V-4.6"
 NEMOTRON_PARSE_REPO_ID = "nvidia/NVIDIA-Nemotron-Parse-v1.2"
 NEMOTRON_NANO_REPO_ID = "nvidia/NVIDIA-Nemotron-3-Nano-4B-GGUF"
+MAX_PROVIDER_IMAGE_BYTES = 10 * 1024 * 1024
+IMAGE_MIME_BY_SUFFIX = {
+    ".png": "image/png",
+    ".jpg": "image/jpeg",
+    ".jpeg": "image/jpeg",
+    ".webp": "image/webp",
+}
+SENSITIVE_PROVIDER_KEYS = ("token", "secret", "api_key", "authorization", "payload", "raw", "base64", "bytes")
 
 
 @dataclass(frozen=True)
@@ -67,13 +75,36 @@ def _image_data_url(path: str | None) -> str | None:
         if not target.exists() or not target.is_file():
             return None
         suffix = target.suffix.lower()
-        mime = "image/png" if suffix == ".png" else "image/jpeg" if suffix in {".jpg", ".jpeg"} else "image/webp"
+        mime = IMAGE_MIME_BY_SUFFIX.get(suffix)
+        if mime is None or target.stat().st_size > MAX_PROVIDER_IMAGE_BYTES:
+            return None
         data = base64.b64encode(target.read_bytes()).decode("ascii")
         return f"data:{mime};base64,{data}"
     except OSError:
         return None
 
 
+def _is_loopback_host(hostname: str | None) -> bool:
+    return hostname in {"localhost", "127.0.0.1", "::1"}
+
+
+def _safe_provider_payload(value: Any) -> Any:
+    if isinstance(value, dict):
+        redacted: dict[str, Any] = {}
+        for key, item in value.items():
+            normalized = str(key).lower()
+            if any(marker in normalized for marker in SENSITIVE_PROVIDER_KEYS):
+                redacted[str(key)] = "[redacted]"
+            else:
+                redacted[str(key)] = _safe_provider_payload(item)
+        return redacted
+    if isinstance(value, list):
+        return [_safe_provider_payload(item) for item in value]
+    if isinstance(value, tuple):
+        return [_safe_provider_payload(item) for item in value]
+    return value
+
+
 def _post_json(url: str, token: str, payload: dict[str, Any], timeout: float) -> dict[str, Any]:
     """
     Send an authenticated JSON POST request and return the parsed response.
@@ -91,6 +122,8 @@ def _post_json(url: str, token: str, payload: dict[str, Any], timeout: float) ->
     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}.")
+    if parsed.scheme == "http" and not _is_loopback_host(parsed.hostname):
+        raise ValueError("Provider URLs must use HTTPS unless targeting loopback development hosts.")
     body = json.dumps(payload).encode("utf-8")
     request = urllib.request.Request(
         url,
@@ -287,12 +320,14 @@ def judge_with_nemotron(
         "Return strict JSON only with keys: sponsor_model_used, structured_parse, "
         "risk_notes, parameter_budget_notes, final_claim_status. Parse this visual creation run."
     )
+    safe_run_packet = _safe_provider_payload(run_packet)
+    safe_minicpm_result = _safe_provider_payload(minicpm_result or {})
     payload = {
         "model": model,
         "messages": [
             {
                 "role": "user",
-                "content": f"{instruction}\nPrompt: {prompt}\nRun: {json.dumps(run_packet, ensure_ascii=True)[:6000]}\nMiniCPM: {json.dumps(minicpm_result or {}, ensure_ascii=True)[:2500]}",
+                "content": f"{instruction}\nPrompt: {prompt}\nRun: {json.dumps(safe_run_packet, ensure_ascii=True)[:6000]}\nMiniCPM: {json.dumps(safe_minicpm_result, ensure_ascii=True)[:2500]}",
             }
         ],
         "temperature": 0.1,

src/nexus_visual_weaver/wardrobe.py

@@ -17,6 +17,10 @@ SLOT_BLUEPRINTS: list[tuple[str, str, str, str, str]] = [
 ]
 
 
+def _material_token(value: object) -> str:
+    return str(value).strip().lower().replace("-", "_").replace(" ", "_")
+
+
 def build_outfit_graph(prompt: str, adult_mode: bool = False, controls: dict | None = None) -> OutfitGraph:
     """
     Constructs an outfit graph with wardrobe slots configured by prompt analysis and optional controls.
@@ -42,7 +46,7 @@ def build_outfit_graph(prompt: str, adult_mode: bool = False, controls: dict | N
         if name == "footwear" and controls.get("footwear"):
             description = str(controls["footwear"])
         if name == "jewelry" and controls.get("hardware"):
-            material = str(controls["hardware"])
+            material = _material_token(controls["hardware"])
         if controls.get("palette"):
             palette = str(controls["palette"])
         locked = any(token in lowered for token in name.split("_")) or material.replace("_", " ") in lowered

tests/test_app_callbacks.py

@@ -159,6 +159,34 @@ def test_operator_actions_transition_checkpoint_export_and_stop() -> None:
     assert stopped[13]["provider_state"] == "stopped"
 
 
+def test_clear_export_ignores_stale_override_reason(monkeypatch) -> None:
+    monkeypatch.setenv("NEXUS_EXPORT_DIR", "outputs/test-exports")
+    result = app.run_weave(
+        "gothic patent leather platform boots, crimson hardware",
+        "Strict",
+        "Wan2.2 I2V",
+        False,
+        None,
+        "Forge",
+    )
+    run = result[13]
+    artifact_path = app.ROOT / "outputs" / "test-clear-override-artifact.png"
+    artifact_path.parent.mkdir(parents=True, exist_ok=True)
+    artifact_path.write_bytes(b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIEND\xaeB`\x82")
+    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)
+    exported = app.export_packet(run, False, clean_scan, "Forge", approved[13], "stale text from prior blocked run")
+
+    assert exported[13]["provider_state"] == "exported"
+    assert exported[13]["export"] == "clear"
+    assert "st3gg_override_reason" not in exported[13]
+
+
 def test_export_blocks_without_checkpoint() -> None:
     result = app.run_weave(
         "gothic patent leather platform boots, crimson hardware",

tests/test_command_center.py

@@ -94,6 +94,23 @@ def test_command_center_run_is_checkpointed() -> None:
     assert any(model.repo_id == "nvidia/LocateAnything-3B" for model in run.model_stack)
 
 
+def test_command_center_run_snapshots_controls_and_skips_empty_reference_ids() -> None:
+    controls = {"wardrobe": {"footwear": "platform boots"}}
+    references = [{"source": "upload"}, {"basename": "reference.png"}]
+
+    run = build_command_center_run(
+        "gothic couture archivist",
+        creator_controls=controls,
+        reference_metadata=references,
+    )
+    controls["wardrobe"]["footwear"] = "mutated"
+    references[1]["basename"] = "mutated.png"
+
+    assert run.request.references == ["reference.png"]
+    assert run.request.creator_controls["wardrobe"]["footwear"] == "platform boots"
+    assert run.request.reference_metadata[1]["basename"] == "reference.png"
+
+
 def test_security_scan_does_not_return_payload_excerpt() -> None:
     sample = Path(__file__).parent / "fixtures" / "sample.png"
     scan = scan_file(str(sample))
@@ -746,7 +763,7 @@ def test_build_outfit_graph_controls_override_hardware() -> None:
     jewelry_slot = next((s for s in outfit.slots if s.name == "jewelry"), None)
 
     assert jewelry_slot is not None
-    assert jewelry_slot.material == "silver occult buckles"
+    assert jewelry_slot.material == "silver_occult_buckles"
 
 
 def test_build_outfit_graph_locked_slots_control_locks_named_slots() -> None:

tests/test_exporter.py

@@ -1,4 +1,5 @@
 import json
+from dataclasses import replace
 from pathlib import Path
 
 from nexus_visual_weaver.exporter import export_root, write_export_packet
@@ -73,6 +74,52 @@ def test_write_export_packet_records_evidence_without_secrets(monkeypatch) -> No
     assert "/data/" not in json.dumps(payload)
 
 
+def test_export_packet_sanitizes_run_id_before_writing(monkeypatch) -> None:
+    monkeypatch.setenv("NEXUS_EXPORT_DIR", "outputs/test-exports")
+    run = build_command_center_run("path traversal export brief")
+    run = replace(run, checkpoint=replace(run.checkpoint, checkpoint_id="../unsafe/path\\with:chars"))
+    scan = {"status": "pass", "export_gate": "clear"}
+
+    result = write_export_packet(run=run, scan=scan, operator_state=_make_base_state(), adult_mode=False)
+    path = Path(result["path"])
+    payload = json.loads(path.read_text(encoding="utf-8"))
+
+    assert path.parent == export_root()
+    assert ".." not in path.name
+    assert "\\" not in path.name
+    assert "/" not in path.name
+    assert payload["run_id"] == "unsafe-path-with-chars"
+
+
+def test_export_packet_sanitizes_prompt_checkpoint_and_provider_state(monkeypatch) -> None:
+    monkeypatch.setenv("NEXUS_EXPORT_DIR", "outputs/test-exports")
+    run = build_command_center_run("prompt contains MINICPM_API_KEY and C:/Users/speci.000/secret.png")
+    run = replace(
+        run,
+        refined_prompt=replace(run.refined_prompt, refined="refined path /data/nexus_visual_weaver/secret.png"),
+        checkpoint=replace(
+            run.checkpoint,
+            recommendation="set NEMOTRON_API_KEY",
+            required_actions=["review C:/Users/speci.000/Downloads/raw.png"],
+        ),
+    )
+    scan = {"status": "pass", "export_gate": "clear"}
+    state = _make_base_state(
+        provider_state="Bearer " + "hidden-provider-token",
+        message="operator message with HF_TOKEN",
+    )
+
+    result = write_export_packet(run=run, scan=scan, operator_state=state, adult_mode=False)
+    serialized = json.dumps(json.loads(Path(result["path"]).read_text(encoding="utf-8")))
+
+    assert "MINICPM_API_KEY" not in serialized
+    assert "NEMOTRON_API_KEY" not in serialized
+    assert "HF_TOKEN" not in serialized
+    assert "C:/Users/speci.000" not in serialized
+    assert "/data/" not in serialized
+    assert "Bearer " + "hidden-provider-token" not in serialized
+
+
 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")

tests/test_hf_runtime.py

@@ -1,9 +1,13 @@
 from PIL import Image
+from pathlib import Path
+import types
 
+import nexus_visual_weaver.hf_runtime as hf_runtime
 from nexus_visual_weaver.hf_runtime import (
     FLUX_REPO_ID,
     PRIVATE_RESEARCH_FLUX_REPO_ID,
     TINY_TITAN_FLUX_REPO_ID,
+    _PIPELINE_CACHE,
     _adapter_recipe,
     _repo_candidates,
     active_flux_repo_id,
@@ -14,6 +18,9 @@ from nexus_visual_weaver.hf_runtime import (
 from nexus_visual_weaver.render import render_artifact_lane
 
 
+RUNTIME_FIXTURE_DIR = Path("tests/fixtures/runtime")
+
+
 def test_hf_runtime_is_disabled_locally_by_default(monkeypatch) -> None:
     monkeypatch.delenv("SPACE_ID", raising=False)
     monkeypatch.delenv("HF_SPACE_ID", raising=False)
@@ -209,3 +216,67 @@ def test_default_lora_repo_id_excludes_requires_image_adapters() -> None:
     recipe_result = _adapter_recipe(result)
     assert recipe_result is not None
     assert recipe_result.requires_image is False
+
+
+def test_generate_flux_image_reports_sidecar_fallback(monkeypatch) -> None:
+    class FakeCuda:
+        @staticmethod
+        def is_available() -> bool:
+            return True
+
+    class FakeGenerator:
+        def __init__(self, device):
+            self.device = device
+
+        def manual_seed(self, seed):
+            self.seed = seed
+            return self
+
+    fake_torch = types.SimpleNamespace(cuda=FakeCuda(), bfloat16="bfloat16", Generator=FakeGenerator)
+
+    class FakePipeline:
+        @classmethod
+        def from_pretrained(cls, repo_id, torch_dtype=None, token=None):
+            if repo_id == FLUX_REPO_ID:
+                raise RuntimeError("primary denied")
+            return cls(repo_id)
+
+        def __init__(self, repo_id):
+            self.repo_id = repo_id
+
+        def enable_model_cpu_offload(self):
+            return None
+
+        def set_progress_bar_config(self, disable):
+            self.progress_disabled = disable
+
+        def __call__(self, **kwargs):
+            return types.SimpleNamespace(images=[Image.new("RGB", (8, 8), color=(2, 4, 6))])
+
+    fake_diffusers = types.SimpleNamespace(Flux2KleinPipeline=FakePipeline)
+    monkeypatch.setitem(__import__("sys").modules, "torch", fake_torch)
+    monkeypatch.setitem(__import__("sys").modules, "diffusers", fake_diffusers)
+    monkeypatch.setenv("NEXUS_ENABLE_REAL_HF", "1")
+    monkeypatch.delenv("NEXUS_DISABLE_REAL_HF", raising=False)
+    monkeypatch.delenv("NEXUS_DISABLE_TINY_TITAN_FALLBACK", raising=False)
+    output_dir = RUNTIME_FIXTURE_DIR / "hf-output"
+    output_dir.mkdir(parents=True, exist_ok=True)
+    monkeypatch.setenv("NEXUS_OUTPUT_DIR", str(output_dir))
+    monkeypatch.setattr(hf_runtime, "load_and_apply", lambda pipe, recipe, repo_id, adult_mode=False: {"status": "disabled", "repo_id": None, "message": "fake"})
+    monkeypatch.setattr(hf_runtime, "unload_all", lambda pipe: None)
+    _PIPELINE_CACHE.clear()
+
+    try:
+        result = generate_flux_image("prompt", seed=7)
+    finally:
+        for artifact in output_dir.glob("nexus_flux_*_7_*.png"):
+            artifact.unlink(missing_ok=True)
+
+    assert result.status == "success"
+    assert result.repo_id == TINY_TITAN_FLUX_REPO_ID
+    assert result.fallback_used is True
+    assert result.primary_error is not None
+    assert "primary denied" in result.primary_error
+    assert result.output_path is not None
+    assert Path(result.output_path).name.startswith("nexus_flux_")
+    assert "_7_" in Path(result.output_path).name

tests/test_lora_adapter.py

@@ -36,6 +36,11 @@ class FailingLoraPipe(FakeLoraPipe):
         raise RuntimeError("adapter load failed")
 
 
+class FailingSetAdapterPipe(FakeLoraPipe):
+    def set_adapters(self, adapter_names: list[str], adapter_weights: list[float]) -> None:
+        raise RuntimeError("adapter apply failed")
+
+
 class UnsupportedPipe:
     pass
 
@@ -95,6 +100,21 @@ def test_load_and_apply_reports_failed_without_raising() -> None:
     assert "RuntimeError" in result["message"]
 
 
+def test_load_and_apply_unloads_when_apply_fails() -> None:
+    recipe = AdapterRecipe(
+        repo_id="example/style-lora",
+        adapter_for="black-forest-labs/FLUX.2-klein-9B",
+        task="style",
+    )
+    pipe = FailingSetAdapterPipe()
+
+    result = load_and_apply(pipe, recipe, "black-forest-labs/FLUX.2-klein-9B")
+
+    assert result["status"] == "failed"
+    assert pipe.loaded == [("example/style-lora", {"adapter_name": "nexus_style"})]
+    assert pipe.unloaded is True
+
+
 def test_unload_all_uses_pipeline_unload_hook() -> None:
     pipe = FakeLoraPipe()
 

tests/test_model_relay.py

@@ -201,6 +201,19 @@ def test_flux2_klein_9b_is_pinned_with_4b_sidecar_fallback() -> None:
     assert fallback.pinned is False
 
 
+def test_pinned_lane_fallbacks_respect_governance_filters() -> None:
+    relay = WeaverModelRelay()
+    fallback = relay.records["flux2-klein-4b-sidecar"]
+    fallback.health = "unhealthy"
+
+    decision = relay.select_lane("image_generation")
+
+    assert decision.primary is not None
+    assert decision.primary.model_id == "flux2-klein-9b-quality"
+    assert decision.fallbacks == []
+    assert any("flux2-klein-4b-sidecar: health=unhealthy" in reason for reason in decision.skipped)
+
+
 def test_flux2_klein_9b_is_not_pinned_and_in_private_research() -> None:
     relay = WeaverModelRelay()
 

tests/test_provider_runtime.py

@@ -1,4 +1,5 @@
 from PIL import Image
+from pathlib import Path
 
 from nexus_visual_weaver.provider_runtime import (
     NEMOTRON_NANO_REPO_ID,
@@ -9,12 +10,21 @@ from nexus_visual_weaver.provider_runtime import (
     _image_data_url,
     _post_json,
     _safe_json_from_text,
+    _safe_provider_payload,
     _short_error,
     judge_with_minicpm,
     judge_with_nemotron,
 )
 
 
+RUNTIME_FIXTURE_DIR = Path("tests/fixtures/runtime")
+
+
+def _runtime_fixture_path(name: str) -> Path:
+    RUNTIME_FIXTURE_DIR.mkdir(parents=True, exist_ok=True)
+    return RUNTIME_FIXTURE_DIR / name
+
+
 def test_minicpm_reports_missing_secret(monkeypatch) -> None:
     monkeypatch.delenv("MINICPM_BASE_URL", raising=False)
     monkeypatch.delenv("MINICPM_API_KEY", raising=False)
@@ -48,7 +58,7 @@ def test_minicpm_blocks_when_artifact_missing(monkeypatch) -> None:
 
 
 def test_minicpm_success_with_mocked_post(monkeypatch) -> None:
-    image = "outputs/test-provider-artifact.png"
+    image = _runtime_fixture_path("test-provider-artifact.png")
     Image.new("RGB", (8, 8), color=(12, 16, 20)).save(image)
     monkeypatch.setenv("MINICPM_BASE_URL", "http://example.test")
     monkeypatch.setenv("MINICPM_API_KEY", "test-token")
@@ -61,12 +71,15 @@ def test_minicpm_success_with_mocked_post(monkeypatch) -> None:
 
     monkeypatch.setattr("nexus_visual_weaver.provider_runtime._post_json", fake_post)
 
-    result = judge_with_minicpm(
-        prompt="gothic couture",
-        image_path=image,
-        scan={"export_gate": "clear"},
-        wardrobe_summary="boots and lace",
-    )
+    try:
+        result = judge_with_minicpm(
+            prompt="gothic couture",
+            image_path=str(image),
+            scan={"export_gate": "clear"},
+            wardrobe_summary="boots and lace",
+        )
+    finally:
+        image.unlink(missing_ok=True)
 
     assert result.status == "success"
     assert result.evidence["overall_status"] == "pass"
@@ -119,25 +132,52 @@ def test_image_data_url_returns_none_for_missing_file() -> None:
 
 
 def test_image_data_url_encodes_png_as_base64() -> None:
-    image_path = "outputs/test-img-data-url.png"
+    image_path = _runtime_fixture_path("test-img-data-url.png")
     Image.new("RGB", (4, 4), color=(0, 0, 0)).save(image_path)
 
-    result = _image_data_url(image_path)
+    try:
+        result = _image_data_url(str(image_path))
+    finally:
+        image_path.unlink(missing_ok=True)
 
     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_path = _runtime_fixture_path("test-img-data-url.jpg")
     Image.new("RGB", (4, 4), color=(0, 0, 0)).save(image_path)
 
-    result = _image_data_url(image_path)
+    try:
+        result = _image_data_url(str(image_path))
+    finally:
+        image_path.unlink(missing_ok=True)
 
     assert result is not None
     assert result.startswith("data:image/jpeg;base64,")
 
 
+def test_image_data_url_rejects_unknown_suffix() -> None:
+    image_path = _runtime_fixture_path("not-an-image.txt")
+    image_path.write_text("not an image", encoding="utf-8")
+
+    try:
+        assert _image_data_url(str(image_path)) is None
+    finally:
+        image_path.unlink(missing_ok=True)
+
+
+def test_image_data_url_rejects_oversized_file(monkeypatch) -> None:
+    image_path = _runtime_fixture_path("large.png")
+    image_path.write_bytes(b"not actually decoded because size fails")
+    monkeypatch.setattr("nexus_visual_weaver.provider_runtime.MAX_PROVIDER_IMAGE_BYTES", 4)
+
+    try:
+        assert _image_data_url(str(image_path)) is None
+    finally:
+        image_path.unlink(missing_ok=True)
+
+
 # --- _extract_content tests ---
 
 def test_extract_content_returns_string_from_choices() -> None:
@@ -263,7 +303,7 @@ def test_nemotron_failed_api_call_returns_failed_status(monkeypatch) -> None:
 def test_minicpm_failed_api_call_returns_failed_status(monkeypatch) -> None:
     import urllib.error
 
-    image_path = "outputs/test-minicpm-fail.png"
+    image_path = _runtime_fixture_path("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")
@@ -273,12 +313,15 @@ def test_minicpm_failed_api_call_returns_failed_status(monkeypatch) -> None:
 
     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",
-    )
+    try:
+        result = judge_with_minicpm(
+            prompt="gothic couture",
+            image_path=str(image_path),
+            scan={"export_gate": "clear"},
+            wardrobe_summary="platform boots",
+        )
+    finally:
+        image_path.unlink(missing_ok=True)
 
     assert result.status == "failed"
     assert result.provider_state == "failed"
@@ -321,7 +364,7 @@ def test_minicpm_uses_openbmb_api_key_as_fallback(monkeypatch) -> None:
     monkeypatch.delenv("MINICPM_API_KEY", raising=False)
     monkeypatch.setenv("OPENBMB_API_KEY", "openbmb-fallback-token")
 
-    image_path = "outputs/test-openbmb-key.png"
+    image_path = _runtime_fixture_path("test-openbmb-key.png")
     Image.new("RGB", (4, 4), color=(0, 0, 0)).save(image_path)
 
     captured = {}
@@ -332,12 +375,15 @@ def test_minicpm_uses_openbmb_api_key_as_fallback(monkeypatch) -> None:
 
     monkeypatch.setattr("nexus_visual_weaver.provider_runtime._post_json", fake_post)
 
-    result = judge_with_minicpm(
-        prompt="test",
-        image_path=image_path,
-        scan={},
-        wardrobe_summary="",
-    )
+    try:
+        result = judge_with_minicpm(
+            prompt="test",
+            image_path=str(image_path),
+            scan={},
+            wardrobe_summary="",
+        )
+    finally:
+        image_path.unlink(missing_ok=True)
 
     assert result.status == "success"
     assert captured["token"] == "openbmb-fallback-token"
@@ -362,6 +408,47 @@ def test_nemotron_uses_nvidia_api_key_as_fallback(monkeypatch) -> None:
     assert captured["token"] == "nvidia-fallback-token"
 
 
+def test_safe_provider_payload_redacts_sensitive_nested_keys() -> None:
+    result = _safe_provider_payload(
+        {
+            "checkpoint": "ok",
+            "HF_TOKEN": "should-not-leak",
+            "nested": {"raw_payload": "hidden", "items": [{"base64_image": "hidden"}, {"safe": "visible"}]},
+        }
+    )
+
+    assert result["checkpoint"] == "ok"
+    assert result["HF_TOKEN"] == "[redacted]"
+    assert result["nested"]["raw_payload"] == "[redacted]"
+    assert result["nested"]["items"][0]["base64_image"] == "[redacted]"
+    assert result["nested"]["items"][1]["safe"] == "visible"
+
+
+def test_nemotron_redacts_run_packet_before_provider_call(monkeypatch) -> None:
+    monkeypatch.setenv("NEMOTRON_BASE_URL", "http://localhost:8001")
+    monkeypatch.setenv("NEMOTRON_API_KEY", "nvidia-token")
+    captured = {}
+
+    def fake_post(url, token, payload, timeout):
+        captured["content"] = payload["messages"][0]["content"]
+        return {"choices": [{"message": {"content": '{"final_claim_status":"pass"}'}}]}
+
+    monkeypatch.setattr("nexus_visual_weaver.provider_runtime._post_json", fake_post)
+
+    result = judge_with_nemotron(
+        prompt="brief",
+        run_packet={"safe": "visible", "api_key": "hidden-key", "nested": {"payload_bytes": "hidden-bytes"}},
+        minicpm_result={"status": "success", "authorization": "hidden-auth"},
+    )
+
+    assert result.status == "success"
+    assert "visible" in captured["content"]
+    assert "hidden-key" not in captured["content"]
+    assert "hidden-bytes" not in captured["content"]
+    assert "hidden-auth" not in captured["content"]
+    assert "[redacted]" in captured["content"]
+
+
 def test_post_json_rejects_unsupported_url_schemes_before_urlopen(monkeypatch) -> None:
     called = False
 
@@ -381,3 +468,44 @@ def test_post_json_rejects_unsupported_url_schemes_before_urlopen(monkeypatch) -
             raise AssertionError(f"{url} should have been rejected")
 
     assert called is False
+
+
+def test_post_json_rejects_plain_http_non_loopback_before_urlopen(monkeypatch) -> None:
+    called = False
+
+    def fake_urlopen(*args, **kwargs):
+        nonlocal called
+        called = True
+        raise AssertionError("urlopen should not be called for plaintext remote provider URLs")
+
+    monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
+
+    try:
+        _post_json("http://example.test/v1/chat/completions", "token", {"ok": True}, 1.0)
+    except ValueError as exc:
+        assert "HTTPS" in str(exc)
+        assert "loopback" in str(exc)
+    else:
+        raise AssertionError("remote http provider URL should have been rejected")
+
+    assert called is False
+
+
+def test_post_json_allows_loopback_http_for_local_tests(monkeypatch) -> None:
+    class FakeResponse:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+        def read(self):
+            return b'{"ok": true}'
+
+    def fake_urlopen(request, timeout):
+        assert request.full_url == "http://127.0.0.1:8000/v1/chat/completions"
+        return FakeResponse()
+
+    monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
+
+    assert _post_json("http://127.0.0.1:8000/v1/chat/completions", "token", {"ok": True}, 1.0) == {"ok": True}