stage 2: dashboard state contract

Scope: add dashboard state contract, gauge formula, escaping helper, and boundary adapter without changing existing scoring logic or callback return formats.

Validation: py_compile + contract smoke checks passed (0/35/weighted signal behavior verified); stage acceptance gate passed.

app.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import hashlib
+import html
 import json
 import re
 from html import escape
@@ -350,6 +351,173 @@ REVISION_ACTIONS = {
     "Commercial Publishability": "Tighten hook, age fit, and list-readiness.",
 }
 
+DASHBOARD_STATE_KEYS = (
+    "project_name",
+    "workbook_loaded",
+    "analysis_status",
+    "last_analysis_at",
+    "totem_signal",
+    "metrics",
+    "metric_history",
+    "revision_queue",
+    "risk_clusters",
+)
+
+
+def esc(value) -> str:
+    """HTML-escape UI text payloads safely."""
+    return html.escape(str(value or ""))
+
+
+def _state_bool(value) -> bool:
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        low = value.strip().lower()
+        if low in {"1", "true", "yes", "y", "on"}:
+            return True
+        if low in {"0", "false", "no", "n", "off", "", "none", "null"}:
+            return False
+    return bool(value)
+
+
+def _state_float(value, default: float = 0.0) -> float:
+    try:
+        return float(value)
+    except Exception:
+        return float(default)
+
+
+def compute_totem_signal(
+    metrics: dict,
+    workbook_loaded: bool,
+    analysis_timestamp: str | None,
+) -> int:
+    """
+    Hero gauge contract from the design manual:
+    - 0 when no workbook is loaded.
+    - 35 when workbook is loaded but analysis has not run yet.
+    - Otherwise weighted metric blend, clamped 0..100.
+    """
+    if not _state_bool(workbook_loaded):
+        return 0
+    if not analysis_timestamp:
+        return 35
+
+    weights = {
+        "overall_publishability": 0.30,
+        "read_aloud_flow": 0.15,
+        "emotional_truth": 0.20,
+        "visual_strength": 0.20,
+        "commercial_viability": 0.15,
+    }
+
+    total = 0.0
+    metric_map = metrics or {}
+    for key, weight in weights.items():
+        total += _state_float(metric_map.get(key, 0.0), 0.0) * weight
+    return max(0, min(100, int(round(total))))
+
+
+def get_initial_dashboard_state() -> dict:
+    """
+    Stage 2 state contract.
+    Placeholder values are deliberate before first analysis run.
+    """
+    metrics = {
+        "overall_publishability": 0,
+        "read_aloud_flow": 0,
+        "emotional_truth": 0,
+        "visual_strength": 0,
+        "commercial_viability": 0,
+    }
+    return {
+        "project_name": "Editorial Workspace",
+        "workbook_loaded": False,
+        "analysis_status": "idle",  # idle | ready | running | complete | error
+        "last_analysis_at": None,
+        "totem_signal": compute_totem_signal(metrics, workbook_loaded=False, analysis_timestamp=None),
+        "metrics": metrics,
+        "metric_history": {
+            "overall_publishability": [],
+            "read_aloud_flow": [],
+            "emotional_truth": [],
+            "visual_strength": [],
+            "commercial_viability": [],
+        },
+        "revision_queue": [],
+        "risk_clusters": [],
+    }
+
+
+def normalize_dashboard_state(raw_existing_outputs) -> dict:
+    """
+    Boundary adapter that normalizes scattered callback outputs into the Stage 2 state contract.
+    This adapter is UI-boundary only and does not alter extractor/scoring logic internals.
+    """
+    state = get_initial_dashboard_state()
+    if raw_existing_outputs is None:
+        return state
+    if not isinstance(raw_existing_outputs, dict):
+        return state
+
+    state["project_name"] = str(raw_existing_outputs.get("project_name") or state["project_name"])
+    state["workbook_loaded"] = _state_bool(raw_existing_outputs.get("workbook_loaded", state["workbook_loaded"]))
+
+    status = str(raw_existing_outputs.get("analysis_status") or state["analysis_status"]).strip().lower()
+    if status not in {"idle", "ready", "running", "complete", "error"}:
+        status = state["analysis_status"]
+    state["analysis_status"] = status
+
+    ts_value = raw_existing_outputs.get("last_analysis_at")
+    state["last_analysis_at"] = str(ts_value).strip() if ts_value else None
+
+    incoming_metrics = raw_existing_outputs.get("metrics")
+    if isinstance(incoming_metrics, dict):
+        for key in state["metrics"]:
+            state["metrics"][key] = int(round(_state_float(incoming_metrics.get(key, state["metrics"][key]))))
+            state["metrics"][key] = max(0, min(100, state["metrics"][key]))
+
+    incoming_history = raw_existing_outputs.get("metric_history")
+    if isinstance(incoming_history, dict):
+        normalized_history = {}
+        for key in state["metric_history"].keys():
+            values = incoming_history.get(key, [])
+            if isinstance(values, list):
+                normalized_history[key] = [
+                    max(0, min(100, int(round(_state_float(v)))))
+                    for v in values
+                ]
+            else:
+                normalized_history[key] = []
+        state["metric_history"] = normalized_history
+
+    revision_queue = raw_existing_outputs.get("revision_queue")
+    if isinstance(revision_queue, list):
+        state["revision_queue"] = revision_queue
+
+    risk_clusters = raw_existing_outputs.get("risk_clusters")
+    if isinstance(risk_clusters, list):
+        state["risk_clusters"] = risk_clusters
+
+    if "totem_signal" in raw_existing_outputs:
+        explicit = int(round(_state_float(raw_existing_outputs.get("totem_signal"), state["totem_signal"])))
+        state["totem_signal"] = max(0, min(100, explicit))
+    else:
+        state["totem_signal"] = compute_totem_signal(
+            state["metrics"],
+            workbook_loaded=state["workbook_loaded"],
+            analysis_timestamp=state["last_analysis_at"],
+        )
+
+    return state
+
+
+def _dashboard_state_contract_smoke_test() -> tuple[bool, list[str]]:
+    state = get_initial_dashboard_state()
+    missing = [key for key in DASHBOARD_STATE_KEYS if key not in state]
+    return len(missing) == 0, missing
+
 
 def _clean_path(uploaded_file) -> Path:
     return workbook_path(uploaded_file)
@@ -1439,4 +1607,9 @@ with gr.Blocks(title="TOTEM Studio") as demo:
 
 
 if __name__ == "__main__":
+    state_ok, missing_keys = _dashboard_state_contract_smoke_test()
+    if state_ok:
+        print("[stage2] dashboard_state contract OK")
+    else:
+        print(f"[stage2] dashboard_state missing keys: {missing_keys}")
     demo.launch(ssr_mode=False, css=CSS + TOTEM_CSS + SS_CSS, head=HEAD)