Wire command center operator actions

app.py

@@ -44,6 +44,15 @@ DEFAULT_PROMPT = (
 MODEL_RELAY = WeaverModelRelay()
 
 
+def _default_operator_state() -> dict[str, Any]:
+    return {
+        "provider_state": "idle",
+        "checkpoint": "pending",
+        "export": "pending",
+        "message": "No operator action yet.",
+    }
+
+
 def _zero_gpu_entrypoint(fn: Any) -> Any:
     """Expose one callback to ZeroGPU without making local development depend on Spaces."""
     gpu_decorator = getattr(spaces, "GPU", None) if spaces is not None else None
@@ -73,6 +82,7 @@ def _dashboard_regions(
     adult_mode: bool = False,
     scan: dict[str, Any] | None = None,
     active_section: str = "Forge",
+    operator_state: dict[str, Any] | None = None,
 ) -> dict[str, str]:
     return render_dashboard_regions(
         run=run,
@@ -80,6 +90,7 @@ def _dashboard_regions(
         scan=scan,
         relay_status=_relay_snapshot(adult_mode),
         active_section=active_section,
+        operator_state=operator_state,
     )
 
 
@@ -91,7 +102,25 @@ def run_weave(
     adult_mode: bool,
     upload: Any,
     active_section: str,
-) -> tuple[str, str, str, str, str, str, str, str, str, str, dict[str, Any], dict[str, Any], dict[str, Any]]:
+) -> tuple[
+    str,
+    str,
+    str,
+    str,
+    str,
+    str,
+    str,
+    str,
+    str,
+    str,
+    dict[str, Any],
+    dict[str, Any],
+    dict[str, Any],
+    Any,
+    dict[str, Any],
+    dict[str, Any],
+    Any,
+]:
     prompt = prompt.strip() or DEFAULT_PROMPT
     run = build_command_center_run(
         prompt=prompt,
@@ -100,7 +129,19 @@ def run_weave(
         adult_mode=adult_mode,
     )
     scan = scan_file(_file_path(upload))
-    regions = _dashboard_regions(run=run, adult_mode=adult_mode, scan=scan, active_section=active_section)
+    operator_state = {
+        "provider_state": "checkpointed",
+        "checkpoint": "pending_review",
+        "export": scan.get("export_gate", "pending"),
+        "message": "Run packet generated. Human checkpoint required before provider promotion or export.",
+    }
+    regions = _dashboard_regions(
+        run=run,
+        adult_mode=adult_mode,
+        scan=scan,
+        active_section=active_section,
+        operator_state=operator_state,
+    )
     catalog = render_catalog_table(adult_mode=adult_mode)
     return (
         regions["topbar"],
@@ -116,6 +157,10 @@ def run_weave(
         run.to_dict(),
         catalog_summary(adult_mode),
         scan,
+        run,
+        scan,
+        operator_state,
+        gr.update(interactive=True),
     )
 
 
@@ -123,9 +168,13 @@ def toggle_adult_visibility(
     adult_mode: bool,
     active_section: str,
     upload: Any,
-) -> tuple[str, str, str, str, str, str, str, dict[str, Any], dict[str, Any]]:
+) -> tuple[str, str, str, str, str, str, str, dict[str, Any], dict[str, Any], dict[str, Any]]:
     scan = scan_file(_file_path(upload))
-    regions = _dashboard_regions(adult_mode=adult_mode, scan=scan, active_section=active_section)
+    operator_state = {
+        **_default_operator_state(),
+        "message": "Adult catalog visibility changed. ST3GG, consent, and export gates remain active.",
+    }
+    regions = _dashboard_regions(adult_mode=adult_mode, scan=scan, active_section=active_section, operator_state=operator_state)
     return (
         regions["topbar"],
         regions["command_rail"],
@@ -136,22 +185,172 @@ def toggle_adult_visibility(
         render_catalog_table(adult_mode=adult_mode),
         catalog_summary(adult_mode),
         scan,
+        operator_state,
     )
 
 
 def refresh_section(
     active_section: str,
     adult_mode: bool,
-    upload: Any,
+    run: Any | None,
+    scan: dict[str, Any] | None,
+    operator_state: dict[str, Any] | None,
 ) -> tuple[str, str, str, str, str, dict[str, Any]]:
-    scan = scan_file(_file_path(upload))
-    regions = _dashboard_regions(adult_mode=adult_mode, scan=scan, active_section=active_section)
+    scan = scan or scan_file(None)
+    regions = _dashboard_regions(
+        run=run,
+        adult_mode=adult_mode,
+        scan=scan,
+        active_section=active_section,
+        operator_state=operator_state or _default_operator_state(),
+    )
     return regions["command_rail"], regions["operations"], regions["inspector"], regions["artifacts"], regions["providers"], scan
 
 
-initial_regions = _dashboard_regions(scan=scan_file(None))
+def _render_stateful(
+    run: Any | None,
+    adult_mode: bool,
+    scan: dict[str, Any] | None,
+    active_section: str,
+    operator_state: dict[str, Any],
+) -> tuple[str, str, str, str, str, str, str, str, str, str, dict[str, Any], dict[str, Any], dict[str, Any], dict[str, Any], Any]:
+    scan = scan or scan_file(None)
+    regions = _dashboard_regions(
+        run=run,
+        adult_mode=adult_mode,
+        scan=scan,
+        active_section=active_section,
+        operator_state=operator_state,
+    )
+    return (
+        regions["topbar"],
+        regions["command_rail"],
+        regions["workflow"],
+        regions["operations"],
+        regions["inspector"],
+        regions["drawer"],
+        regions["status"],
+        regions["artifacts"],
+        regions["providers"],
+        render_catalog_table(adult_mode=adult_mode),
+        run.to_dict() if hasattr(run, "to_dict") else {},
+        catalog_summary(adult_mode),
+        scan,
+        operator_state,
+        gr.update(interactive=run is not None and operator_state.get("provider_state") not in {"idle", "stopped", "exported"}),
+    )
+
+
+def scan_reference(
+    run: Any | None,
+    adult_mode: bool,
+    upload: Any,
+    active_section: str,
+    operator_state: dict[str, Any] | None,
+) -> tuple[str, str, str, str, str, str, str, str, str, str, dict[str, Any], dict[str, Any], dict[str, Any], dict[str, Any], Any, dict[str, Any]]:
+    scan = scan_file(_file_path(upload))
+    next_state = {
+        **(operator_state or _default_operator_state()),
+        "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.",
+    }
+    rendered = _render_stateful(run, adult_mode, scan, active_section, next_state)
+    return (*rendered, scan)
+
+
+def approve_checkpoint(
+    run: Any | None,
+    adult_mode: bool,
+    scan: dict[str, Any] | None,
+    active_section: str,
+    operator_state: dict[str, Any] | None,
+) -> tuple[str, str, str, str, str, str, str, str, str, str, dict[str, Any], dict[str, Any], dict[str, Any], dict[str, Any], Any]:
+    scan = scan or scan_file(None)
+    if run is None:
+        next_state = {**_default_operator_state(), "provider_state": "blocked", "message": "No run exists yet. Run Active Weave first."}
+    else:
+        export_state = scan.get("export_gate", "pending")
+        next_state = {
+            **(operator_state or _default_operator_state()),
+            "provider_state": "export_ready" if export_state == "clear" else "checkpointed",
+            "checkpoint": "approved",
+            "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.",
+        }
+    return _render_stateful(run, adult_mode, scan, active_section, next_state)
+
+
+def export_packet(
+    run: Any | None,
+    adult_mode: bool,
+    scan: dict[str, Any] | None,
+    active_section: str,
+    operator_state: dict[str, Any] | None,
+) -> tuple[str, str, str, str, str, str, str, str, str, str, dict[str, Any], dict[str, Any], dict[str, Any], dict[str, Any], Any]:
+    scan = scan or scan_file(None)
+    state = operator_state or _default_operator_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 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:
+        next_state = {**state, "provider_state": "exported", "export": "clear", "message": "Governed export packet prepared with run, catalog, and ST3GG evidence."}
+    return _render_stateful(run, adult_mode, scan, active_section, next_state)
+
+
+def stop_provider_job(
+    run: Any | None,
+    adult_mode: bool,
+    scan: dict[str, Any] | None,
+    active_section: str,
+    operator_state: dict[str, Any] | None,
+) -> tuple[str, str, str, str, str, str, str, str, str, str, dict[str, Any], dict[str, Any], dict[str, Any], dict[str, Any], Any]:
+    scan = scan or scan_file(None)
+    next_state = {
+        **(operator_state or _default_operator_state()),
+        "provider_state": "stopped",
+        "message": "Provider handoff stopped. Local dry-run packet and evidence remain available.",
+    }
+    return _render_stateful(run, adult_mode, scan, active_section, next_state)
+
+
+def reset_demo(
+    adult_mode: bool,
+    active_section: str,
+) -> tuple[str, str, str, str, str, str, str, str, str, str, dict[str, Any], dict[str, Any], dict[str, Any], None, dict[str, Any], dict[str, Any], Any]:
+    scan = scan_file(None)
+    operator_state = _default_operator_state()
+    regions = _dashboard_regions(adult_mode=adult_mode, scan=scan, active_section=active_section, operator_state=operator_state)
+    return (
+        regions["topbar"],
+        regions["command_rail"],
+        regions["workflow"],
+        regions["operations"],
+        regions["inspector"],
+        regions["drawer"],
+        regions["status"],
+        regions["artifacts"],
+        regions["providers"],
+        render_catalog_table(adult_mode=adult_mode),
+        {},
+        catalog_summary(adult_mode),
+        scan,
+        None,
+        scan,
+        operator_state,
+        gr.update(interactive=False),
+    )
+
+
+initial_operator_state = _default_operator_state()
+initial_regions = _dashboard_regions(scan=scan_file(None), operator_state=initial_operator_state)
 
 with gr.Blocks(title="NEXUS Visual Weaver") as demo:
+    active_run_state = gr.State(None)
+    scan_state = gr.State(scan_file(None))
+    operator_state = gr.State(initial_operator_state)
     topbar_html = gr.HTML(initial_regions["topbar"], container=False)
 
     with gr.Group(elem_id="nw-inputs", elem_classes=["nw-control-panel"]):
@@ -190,6 +389,11 @@ with gr.Blocks(title="NEXUS Visual Weaver") as demo:
             )
             run_btn = gr.Button("Run Active Weave", variant="primary", scale=1)
             stop_btn = gr.Button("Stop Provider Job", variant="stop", interactive=False, scale=1)
+        with gr.Row(elem_id="nw-operator-actions", elem_classes=["nw-operator-actions"]):
+            scan_btn = gr.Button("Scan Reference", scale=1)
+            checkpoint_btn = gr.Button("Approve Checkpoint", scale=1)
+            export_btn = gr.Button("Prepare Export Packet", scale=1)
+            reset_btn = gr.Button("Reset Demo State", scale=1)
 
     with gr.Row(elem_id="nw-workspace", elem_classes=["nw-workspace"]):
         with gr.Column(scale=1, min_width=150, elem_id="nw-native-rail"):
@@ -218,44 +422,36 @@ with gr.Blocks(title="NEXUS Visual Weaver") as demo:
             catalog_json = gr.JSON(label="Catalog Summary")
             scan_json = gr.JSON(label="ST3GG Scan")
 
+    dashboard_outputs = [
+        topbar_html,
+        command_rail_html,
+        workflow_html,
+        operations_html,
+        inspector_html,
+        drawer_html,
+        status_html,
+        artifact_html,
+        provider_html,
+        catalog_html,
+        run_json,
+        catalog_json,
+        scan_json,
+    ]
+
+    stateful_outputs = dashboard_outputs + [active_run_state, scan_state, operator_state, stop_btn]
+
+    operator_outputs = dashboard_outputs + [operator_state, stop_btn]
+
     run_btn.click(
         fn=run_weave,
         inputs=[prompt, reasoning_mode, video_preset, adult_mode, upload, section_nav],
-        outputs=[
-            topbar_html,
-            command_rail_html,
-            workflow_html,
-            operations_html,
-            inspector_html,
-            drawer_html,
-            status_html,
-            artifact_html,
-            provider_html,
-            catalog_html,
-            run_json,
-            catalog_json,
-            scan_json,
-        ],
+        outputs=stateful_outputs,
         api_name="run_active_weave",
     )
     prompt.submit(
         fn=run_weave,
         inputs=[prompt, reasoning_mode, video_preset, adult_mode, upload, section_nav],
-        outputs=[
-            topbar_html,
-            command_rail_html,
-            workflow_html,
-            operations_html,
-            inspector_html,
-            drawer_html,
-            status_html,
-            artifact_html,
-            provider_html,
-            catalog_html,
-            run_json,
-            catalog_json,
-            scan_json,
-        ],
+        outputs=stateful_outputs,
         api_name=False,
     )
     adult_mode.change(
@@ -271,18 +467,49 @@ with gr.Blocks(title="NEXUS Visual Weaver") as demo:
             catalog_html,
             catalog_json,
             scan_json,
+            operator_state,
         ],
         api_name="toggle_adult_catalog",
     )
     section_nav.change(
         fn=refresh_section,
-        inputs=[section_nav, adult_mode, upload],
+        inputs=[section_nav, adult_mode, active_run_state, scan_state, operator_state],
         outputs=[command_rail_html, operations_html, inspector_html, artifact_html, provider_html, scan_json],
         api_name=False,
     )
+    scan_btn.click(
+        fn=scan_reference,
+        inputs=[active_run_state, adult_mode, upload, section_nav, operator_state],
+        outputs=dashboard_outputs + [operator_state, stop_btn, scan_state],
+        api_name="scan_reference",
+    )
+    checkpoint_btn.click(
+        fn=approve_checkpoint,
+        inputs=[active_run_state, adult_mode, scan_state, section_nav, operator_state],
+        outputs=operator_outputs,
+        api_name="approve_checkpoint",
+    )
+    export_btn.click(
+        fn=export_packet,
+        inputs=[active_run_state, adult_mode, scan_state, section_nav, operator_state],
+        outputs=operator_outputs,
+        api_name="prepare_export_packet",
+    )
+    stop_btn.click(
+        fn=stop_provider_job,
+        inputs=[active_run_state, adult_mode, scan_state, section_nav, operator_state],
+        outputs=operator_outputs,
+        api_name="stop_provider_job",
+    )
+    reset_btn.click(
+        fn=reset_demo,
+        inputs=[adult_mode, section_nav],
+        outputs=stateful_outputs,
+        api_name="reset_demo_state",
+    )
     demo.load(
-        fn=lambda: (render_catalog_table(False), catalog_summary(False), scan_file(None)),
-        outputs=[catalog_html, catalog_json, scan_json],
+        fn=lambda: (render_catalog_table(False), catalog_summary(False), scan_file(None), scan_file(None), _default_operator_state()),
+        outputs=[catalog_html, catalog_json, scan_json, scan_state, operator_state],
         api_name=False,
     )
 

src/nexus_visual_weaver/render.py

@@ -137,13 +137,20 @@ def render_command_rail(active_section: str = "Forge") -> str:
     """
 
 
-def render_workflow(run: GenerationRun | None = None) -> str:
+def render_workflow(run: GenerationRun | None = None, operator_state: dict | None = None) -> str:
+    operator_state = operator_state or {}
     workflow = WorkflowState.default()
     score = run.checkpoint.trust_score if run else 0.82
     checkpoint_id = run.checkpoint.checkpoint_id if run else "nw-dry-run"
-    recommendation = run.checkpoint.recommendation.replace("_", " ").title() if run else "Awaiting Run"
+    checkpoint_status = str(operator_state.get("checkpoint", "pending_review" if run else "pending"))
+    provider_state = str(operator_state.get("provider_state", "dry-run" if run else "idle"))
+    recommendation = operator_state.get("message") or (run.checkpoint.recommendation.replace("_", " ").title() if run else "Awaiting Run")
     required_actions = run.checkpoint.required_actions if run else ["Review candidate thumbnails before promotion"]
     action_label = required_actions[0] if required_actions else "No action pending"
+    if checkpoint_status == "approved":
+        action_label = "Checkpoint approved; export packet may be prepared after ST3GG gate."
+    elif provider_state == "stopped":
+        action_label = "Provider handoff stopped; dry-run evidence remains available."
     model_label = _short_repo(run.model_stack[0].repo_id) if run and run.model_stack else "FLUX.2"
     locate_label = run.inspection.locate_model.split("/")[-1] if run else "LocateAnything-3B"
     nodes = {
@@ -210,13 +217,13 @@ def render_workflow(run: GenerationRun | None = None) -> str:
         {"".join(cards)}
       </svg>
       <div class="nw-legend">
-        {badge("Text Flow", "accent")} {badge("Refine Loop", "violet")} {badge("Policy Gate", "blue")} {badge("Media Flow", "cyan")} {badge("Human Gate", "warn")}
+        {badge("Text Flow", "accent")} {badge("Refine Loop", "violet")} {badge("Policy Gate", "blue")} {badge("Media Flow", "cyan")} {badge("Human Gate", "warn")} {badge(provider_state.replace("_", " ").upper(), "pass" if provider_state == "exported" else "warn" if provider_state in {"blocked", "stopped"} else "muted")}
       </div>
       <div class="nw-weave-console">
         <div class="nw-console-card nw-console-primary">
           <small>Selected Node</small>
           <strong>Human Checkpoint</strong>
-          <span>{escape(recommendation)} / {escape(checkpoint_id)}</span>
+          <span>{escape(str(recommendation))} / {escape(checkpoint_id)}</span>
         </div>
         <div class="nw-console-card">
           <small>Next Operator Action</small>
@@ -238,20 +245,31 @@ def render_workflow(run: GenerationRun | None = None) -> str:
     """
 
 
-def render_artifact_lane(run: GenerationRun | None = None, scan: dict | None = None) -> str:
+def render_artifact_lane(run: GenerationRun | None = None, scan: dict | None = None, operator_state: dict | None = None) -> str:
     scan = scan or {"status": "idle", "export_gate": "pending"}
+    operator_state = operator_state or {}
     prompt_label = "Prompt proof"
     outfit_label = "Outfit map"
     locate_label = "Grounding overlay"
     video_label = run.video.preset if run else "Video path"
     active_prompt = run.refined_prompt.refined[:150] if run else "Awaiting first weave. The preview stage shows dry-run handoff packets until provider output exists."
-    checkpoint = getattr(run.checkpoint, "recommendation", "pending") if run else "pending"
+    checkpoint = operator_state.get("checkpoint", getattr(run.checkpoint, "recommendation", "pending") if run else "pending")
+    provider_state = str(operator_state.get("provider_state", "dry-run" if run else "idle"))
+    preview_mode = {
+        "idle": "Idle",
+        "dry-run": "Dry Run",
+        "checkpointed": "Checkpointed",
+        "export_ready": "Export Ready",
+        "exported": "Exported",
+        "blocked": "Blocked",
+        "stopped": "Stopped",
+    }.get(provider_state, provider_state.replace("_", " ").title())
     demo_seed = (run.checkpoint.checkpoint_id[-4:] if run else "0000").upper()
     artifacts = [
         (prompt_label, "Taste-refined brief", "dry-run", "material-0", "01"),
         (outfit_label, "Wardrobe slots and locks", "checkpointed", "material-1", "02"),
         (locate_label, "LocateAnything region plan", "configured", "material-4", "03"),
-        (video_label, "Checkpointed storyboard", "blocked" if scan.get("export_gate") == "blocked" else "ready", "story-2", "04"),
+        (video_label, "Checkpointed storyboard", "blocked" if scan.get("export_gate") == "blocked" or provider_state == "blocked" else "ready", "story-2", "04"),
     ]
     cards = "".join(
         f"""
@@ -285,13 +303,13 @@ def render_artifact_lane(run: GenerationRun | None = None, scan: dict | None = N
         <div class="nw-preview-meta">
           <div><small>checkpoint</small><strong>{escape(str(checkpoint).replace("_", " ").title())}</strong></div>
           <div><small>export gate</small><strong>{escape(export_gate)}</strong></div>
-          <div><small>preview mode</small><strong>Dry Run</strong></div>
+          <div><small>preview mode</small><strong>{escape(preview_mode)}</strong></div>
         </div>
       </div>
       <div class="nw-preview-ribbon">
         <span>{icon("security")} ST3GG before export</span>
         <span>{icon("wardrobe")} continuity: {escape(continuity)}</span>
-        <span>{icon("models")} provider call remains checkpointed; state: dry-run / configured / blocked / failed</span>
+        <span>{icon("models")} provider call remains checkpointed; state: dry-run / configured / blocked / failed / exported; current: {escape(provider_state)}</span>
       </div>
       <div class="nw-artifact-grid">{cards}</div>
     </section>
@@ -305,9 +323,11 @@ def render_operations_panel(
     relay_status: dict | None = None,
     *,
     adult_mode: bool = False,
+    operator_state: dict | None = None,
 ) -> str:
     scan = scan or {"status": "idle", "export_gate": "pending", "findings": []}
     relay_status = relay_status or {}
+    operator_state = operator_state or {}
     section = active_section if active_section in {"Forge", "Wardrobe", "Lore", "Models", "Security", "Runs"} else "Forge"
     checkpoint = getattr(run, "checkpoint", None) if run else None
     outfit = getattr(run, "outfit", None) if run else None
@@ -317,6 +337,8 @@ def render_operations_panel(
     lore_count = len(getattr(lore, "beats", []) or [])
     scan_status = str(scan.get("status", "idle")).upper()
     export_gate = str(scan.get("export_gate", "pending")).upper()
+    provider_state = str(operator_state.get("provider_state", "idle")).replace("_", " ").upper()
+    operator_message = str(operator_state.get("message", "No operator action yet."))
     decisions = relay_status.get("decisions", [])
     first_decision = decisions[0] if decisions else {}
     first_primary = (first_decision.get("primary") or {}) if first_decision else {}
@@ -325,7 +347,7 @@ def render_operations_panel(
         "Forge": [
             ("Prompt contract", "Taste-refined prompt, material locks, negative purge, and checkpoint requirements."),
             ("Active run", f"{run_id} / checkpoint remains human-reviewed before video promotion."),
-            ("Provider posture", "Dry-run packets are visible before paid or gated calls."),
+            ("Provider posture", f"{provider_state}: {operator_message}"),
         ],
         "Wardrobe": [
             ("Slot coverage", f"{outfit_count or 9} garment/accessory regions tracked with locks and edit priority."),
@@ -353,7 +375,7 @@ def render_operations_panel(
         ],
         "Runs": [
             ("Current checkpoint", run_id),
-            ("Ledger mode", "Run JSON, catalog summary, and ST3GG evidence remain in the evidence accordion."),
+            ("Ledger mode", f"Operator state: {provider_state}. Run JSON, catalog summary, and ST3GG evidence remain in the evidence accordion."),
             ("Rollback path", "Feature branches and draft PRs carry implementation checkpoints."),
         ],
     }
@@ -593,8 +615,11 @@ def render_drawer(run: GenerationRun | None = None) -> str:
     """
 
 
-def render_status_bar() -> str:
+def render_status_bar(operator_state: dict | None = None) -> str:
     space = _space_runtime_status()
+    operator_state = operator_state or {}
+    provider_state = str(operator_state.get("provider_state", "idle")).replace("_", " ").title()
+    stop_class = "nw-stop-idle" if provider_state == "Idle" else "nw-stop-active"
     return f"""
     <footer class="nw-statusbar">
       {_metric("Runs", "112")}
@@ -604,7 +629,7 @@ def render_status_bar() -> str:
       {_metric("Temp", "62 C")}
       {_metric("HF Space", space["hardware"])}
       <div class="nw-autosave"><span class="nw-live-dot"></span><strong>Auto-save</strong><small>On</small></div>
-      <div class="nw-stop nw-stop-idle">No live provider job</div>
+      <div class="nw-stop {stop_class}">{escape(provider_state)}</div>
     </footer>
     """
 
@@ -637,8 +662,9 @@ def render_dashboard(
     scan: dict | None = None,
     relay_status: dict | None = None,
     active_section: str = "Forge",
+    operator_state: dict | None = None,
 ) -> str:
-    regions = render_dashboard_regions(run, adult_mode, scan, relay_status, active_section)
+    regions = render_dashboard_regions(run, adult_mode, scan, relay_status, active_section, operator_state)
     return f"""
     <div class="nw-app">
       {regions["topbar"]}
@@ -666,16 +692,17 @@ def render_dashboard_regions(
     scan: dict | None = None,
     relay_status: dict | None = None,
     active_section: str = "Forge",
+    operator_state: dict | None = None,
 ) -> dict[str, str]:
     return {
         "topbar": render_topbar(adult_mode, relay_status),
         "rail": render_left_rail(active_section),
         "command_rail": render_command_rail(active_section),
-        "workflow": render_workflow(run),
-        "operations": render_operations_panel(active_section, run, scan, relay_status, adult_mode=adult_mode),
+        "workflow": render_workflow(run, operator_state),
+        "operations": render_operations_panel(active_section, run, scan, relay_status, adult_mode=adult_mode, operator_state=operator_state),
         "inspector": render_inspector(run, scan, relay_status),
         "drawer": render_drawer(run),
-        "status": render_status_bar(),
-        "artifacts": render_artifact_lane(run, scan),
+        "status": render_status_bar(operator_state),
+        "artifacts": render_artifact_lane(run, scan, operator_state),
         "providers": render_provider_cards(relay_status, adult_mode),
     }

src/nexus_visual_weaver/styles.py

@@ -482,6 +482,28 @@ footer { display: none !important; }
   background: linear-gradient(180deg, #ff3b62, #c51a3b) !important;
   border-color: rgba(255,54,95,.58) !important;
 }
+#nw-operator-actions {
+  margin-top: 10px;
+  padding-top: 10px;
+  border-top: 1px solid rgba(255,255,255,.06);
+  gap: 8px !important;
+}
+#nw-operator-actions button {
+  min-height: 34px !important;
+  text-transform: uppercase;
+  letter-spacing: 0 !important;
+  font-size: 11px !important;
+  font-weight: 750 !important;
+}
+#nw-operator-actions button:nth-child(2) {
+  border-color: rgba(16,185,129,.42) !important;
+}
+#nw-operator-actions button:nth-child(3) {
+  border-color: rgba(245,158,11,.46) !important;
+}
+#nw-operator-actions button:nth-child(4) {
+  border-color: rgba(148,163,184,.34) !important;
+}
 .nw-control-panel {
   margin: 8px 8px 10px 8px;
   padding: 12px;
@@ -852,6 +874,10 @@ footer { display: none !important; }
   background: rgba(255,255,255,.025);
   color: var(--nw-muted);
 }
+.nw-stop-active {
+  display: grid;
+  place-items: center;
+}
 @media (max-width: 1100px) {
   .nw-topbar { grid-template-columns: 1fr; }
   .nw-shell { grid-template-columns: 1fr; grid-template-rows: auto; }

tests/test_app_callbacks.py

@@ -0,0 +1,59 @@
+import app
+
+
+def test_run_weave_returns_stateful_dashboard_packet() -> None:
+    result = app.run_weave(
+        "gothic patent leather platform boots, crimson hardware",
+        "Strict",
+        "Wan2.2 I2V",
+        False,
+        None,
+        "Forge",
+    )
+
+    assert len(result) == 17
+    assert result[13].checkpoint.checkpoint_id.startswith("nw-")
+    assert result[15]["provider_state"] == "checkpointed"
+    assert result[16]["interactive"] is True
+
+
+def test_operator_actions_transition_checkpoint_export_and_stop() -> None:
+    result = app.run_weave(
+        "gothic patent leather platform boots, crimson hardware",
+        "Strict",
+        "Wan2.2 I2V",
+        False,
+        None,
+        "Forge",
+    )
+    run = result[13]
+    operator_state = result[15]
+    clean_scan = {"status": "pass", "export_gate": "clear", "findings": [], "purification_actions": []}
+
+    approved = app.approve_checkpoint(run, False, clean_scan, "Forge", operator_state)
+    assert approved[13]["checkpoint"] == "approved"
+    assert approved[13]["provider_state"] == "export_ready"
+
+    exported = app.export_packet(run, False, clean_scan, "Forge", approved[13])
+    assert exported[13]["provider_state"] == "exported"
+    assert exported[13]["export"] == "clear"
+
+    stopped = app.stop_provider_job(run, False, clean_scan, "Forge", operator_state)
+    assert stopped[13]["provider_state"] == "stopped"
+
+
+def test_export_blocks_without_checkpoint() -> None:
+    result = app.run_weave(
+        "gothic patent leather platform boots, crimson hardware",
+        "Strict",
+        "Wan2.2 I2V",
+        False,
+        None,
+        "Forge",
+    )
+    clean_scan = {"status": "pass", "export_gate": "clear", "findings": [], "purification_actions": []}
+
+    blocked = app.export_packet(result[13], False, clean_scan, "Forge", result[15])
+
+    assert blocked[13]["provider_state"] == "blocked"
+    assert "checkpoint" in blocked[13]["message"].lower()