Add TOTEM workbook-manuscript bridge with strict JSON validation and dashboard wiring

app.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import hashlib
 import html
 import json
+import os
 import re
 from shutil import copy2
 from html import escape
@@ -28,6 +29,14 @@ from src.totem_workbook import (
 )
 
 from src.codex_extractor import process_upload, format_fingerprint_report
+from src.totem_bridge import (
+    BridgeError,
+    ManuscriptContext,
+    extract_manuscript_context,
+    extract_workbook_matrix,
+    recompute_gate,
+    run_totem_skill,
+)
 from smoke_signal_tab import smoke_signal_tab, SS_CSS
 
 ORIGINAL_WORKBOOK_PATH = "data/order69_macmillan_totem_rebuilt.xlsx"
@@ -2783,6 +2792,39 @@ def run_analysis(active_path: str):
     return render_dashboard(state), log_df, _score_summary(log_df)
 
 
+def _apply_llm_dashboard_state(
+    *,
+    active_path: str,
+    llm_state: dict,
+    notice: str = "",
+) -> tuple[str, pd.DataFrame, str, str]:
+    path = _validate_workbook_path(Path(active_path) if active_path else _preferred_workbook_path())
+    log_df = score_log(path)
+    state = _build_dashboard_state_from_workbook(path, "complete", notice)
+    state["metrics"]["overall_publishability"] = int(llm_state["overall_publishability"])
+    state["metrics"]["read_aloud_flow"] = int(llm_state["read_aloud_flow"])
+    state["metrics"]["emotional_truth"] = int(llm_state["emotional_truth"])
+    state["metrics"]["visual_strength"] = int(llm_state["visual_strength"])
+    state["metrics"]["commercial_viability"] = int(llm_state["commercial_visibility"])
+    state["revision_queue"] = list(llm_state.get("revision_priority_queue") or [])
+    state["risk_clusters"] = list(llm_state.get("risk_clusters") or [])
+    state["analysis_status"] = "complete"
+    if os.getenv("TOTEM_RECOMPUTE_GATES", "1").strip() not in {"0", "false", "False"}:
+        recomputed = recompute_gate(
+            {
+                "read_aloud_flow": state["metrics"]["read_aloud_flow"],
+                "emotional_truth": state["metrics"]["emotional_truth"],
+                "visual_strength": state["metrics"]["visual_strength"],
+                "commercial_visibility": state["metrics"]["commercial_viability"],
+            },
+            extract_workbook_matrix(path),
+        )
+        gate_summary = f"Gate={llm_state.get('gate','REVISE')} | Recomputed={recomputed}"
+    else:
+        gate_summary = f"Gate={llm_state.get('gate','REVISE')}"
+    return render_dashboard(state), log_df, _score_summary(log_df), gate_summary
+
+
 def analyze_manuscript_and_refresh(
     file_obj,
     author_name: str,
@@ -2826,11 +2868,69 @@ def analyze_manuscript_for_dashboard(file_obj, active_path: str):
         author_id=author_id,
         works_sampled=works_sampled,
     )
+    manuscript_ctx = extract_manuscript_context(file_path)
+    manuscript_state = {
+        "path": manuscript_ctx.path,
+        "cleaned_text": manuscript_ctx.cleaned_text,
+        "word_count": manuscript_ctx.word_count,
+    }
     dashboard_status = (
         f"{status} | Draft Mode active: extraction complete. "
         "No auto-scoring was run. Click 'Run TOTEM Analysis' to score."
     )
-    return report, json_out, dashboard_status
+    return report, json_out, dashboard_status, json.dumps(manuscript_state, ensure_ascii=False)
+
+
+def run_totem_analysis_from_context(active_path: str, manuscript_context_json: str):
+    """
+    Full bridge run:
+    - workbook matrix is source of truth
+    - manuscript context is evidence
+    - hidden OpenAI skill returns strict JSON
+    - validated JSON updates dashboard
+    """
+    if not manuscript_context_json:
+        dashboard_html, log_df, summary = run_analysis(active_path)
+        return dashboard_html, log_df, summary, "No manuscript context found. Ran workbook-only refresh."
+
+    try:
+        ctx_payload = json.loads(manuscript_context_json)
+        cleaned_text = str(ctx_payload.get("cleaned_text") or "").strip()
+        if not cleaned_text:
+            raise BridgeError("Manuscript cleaned text is empty.")
+        ctx_path = str(ctx_payload.get("path") or "").strip()
+        if ctx_path and Path(ctx_path).exists():
+            ctx = extract_manuscript_context(ctx_path)
+        else:
+            ctx = ManuscriptContext(
+                path=ctx_path or "(session)",
+                raw_text=cleaned_text,
+                cleaned_text=cleaned_text,
+                word_count=len(cleaned_text.split()),
+                page_trace=[],
+            )
+
+        path = _validate_workbook_path(Path(active_path) if active_path else _preferred_workbook_path())
+        rubric = extract_workbook_matrix(path)
+        llm_state, debug = run_totem_skill(rubric_matrix=rubric, manuscript=ctx)
+        dashboard_html, log_df, summary, gate_summary = _apply_llm_dashboard_state(
+            active_path=active_path,
+            llm_state=llm_state,
+            notice="TOTEM analysis complete (workbook + manuscript).",
+        )
+        debug_log = {
+            "status": "ok",
+            "gate_summary": gate_summary,
+            "model": debug.get("model"),
+            "manuscript_words": debug.get("word_count"),
+            "manuscript_path": ctx.path,
+            "rubric_metrics": len(rubric.get("metrics", [])),
+        }
+        return dashboard_html, log_df, summary, json.dumps(debug_log, indent=2)
+    except Exception as exc:
+        dashboard_html, log_df, summary = run_analysis(active_path)
+        debug_log = {"status": "error", "error": f"{type(exc).__name__}: {exc}"}
+        return dashboard_html, log_df, summary, json.dumps(debug_log, indent=2)
 
 
 def recalc_log(log_df, active_path: str):
@@ -2875,6 +2975,7 @@ def initial_dashboard_html() -> str:
 with gr.Blocks(title="TOTEM Studio") as demo:
     active_path = gr.State(str(_preferred_workbook_path()))
     log_state = gr.State(pd.DataFrame(columns=LOG_COLUMNS))
+    manuscript_context_state = gr.State("")
 
     with gr.Tabs():
 
@@ -2919,7 +3020,7 @@ with gr.Blocks(title="TOTEM Studio") as demo:
             with gr.Accordion("Manuscript Intake (score pipeline)", open=False, elem_id="codex-panel"):
                 with gr.Column(elem_classes=["codex-body"]):
                     gr.Markdown(
-                        "Upload manuscript here to run extraction and refresh workbook scoring context."
+                        "Upload manuscript here to build analysis context. Scoring updates only when you click Run TOTEM Analysis."
                     )
                     dashboard_manuscript_file = gr.File(
                         label="Manuscript (.txt, .docx, or .pdf)",
@@ -2946,6 +3047,13 @@ with gr.Blocks(title="TOTEM Studio") as demo:
                         interactive=False,
                         lines=10,
                     )
+                    bridge_diag_log = gr.Textbox(
+                        label="Bridge diagnostics (for bug reports)",
+                        interactive=False,
+                        lines=8,
+                        value='{"status":"idle","message":"No analysis run yet."}',
+                    )
+                    copy_bridge_log_btn = gr.Button("Copy Bridge Log", variant="secondary")
 
             with gr.Row(elem_id="path-panel"):
                 with gr.Column(elem_classes=["wrap"]):
@@ -3106,8 +3214,11 @@ with gr.Blocks(title="TOTEM Studio") as demo:
         inputs=[dashboard_workbook_file],
         outputs=[workbook_home_display, path_input, active_path, dashboard, log_state, score_status],
     )
-    run_button.click(run_analysis, inputs=[active_path],
-                     outputs=[dashboard, log_state, score_status])
+    run_button.click(
+        run_totem_analysis_from_context,
+        inputs=[active_path, manuscript_context_state],
+        outputs=[dashboard, log_state, score_status, bridge_diag_log],
+    )
     path_button.click(load_local_path, inputs=[path_input],
                       outputs=[active_path, dashboard, log_state, score_status])
     dashboard_manuscript_analyze.click(
@@ -3117,8 +3228,15 @@ with gr.Blocks(title="TOTEM Studio") as demo:
             dashboard_manuscript_report,
             dashboard_manuscript_json,
             dashboard_manuscript_status,
+            manuscript_context_state,
         ],
     )
+    copy_bridge_log_btn.click(
+        lambda v: v,
+        inputs=[bridge_diag_log],
+        outputs=[bridge_diag_log],
+        js="""(v) => { if (v) { navigator.clipboard.writeText(v); } return v; }""",
+    )
     recalc_button.click(recalc_log, inputs=[log_state, active_path],
                         outputs=[dashboard, log_state, score_status])
     export_button.click(export_log, inputs=[log_state, active_path], outputs=[exported_file])

requirements.txt

@@ -2,6 +2,7 @@ gradio>=4.44,<7
 huggingface_hub<1.0
 nltk>=3.8
 openpyxl>=3.1
+openai>=1.30.0
 pandas>=2.2
 pdfplumber>=0.10
 Pillow>=10.0.0

src/totem_bridge.py

@@ -0,0 +1,280 @@
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import pandas as pd
+
+from src.codex_extractor import clean_text, extract_text_from_file
+from src.totem_workbook import METRICS, protocol_table, protocol_weights
+
+
+class BridgeError(RuntimeError):
+    pass
+
+
+class BridgeValidationError(BridgeError):
+    pass
+
+
+@dataclass
+class ManuscriptContext:
+    path: str
+    raw_text: str
+    cleaned_text: str
+    word_count: int
+    page_trace: list[dict[str, Any]]
+
+
+def extract_workbook_matrix(path: Path) -> dict[str, Any]:
+    if not path.exists():
+        raise BridgeError(f"Workbook not found: {path}")
+    weights = protocol_weights(path)
+    protocol_df = protocol_table(path)
+
+    metric_rows: list[dict[str, Any]] = []
+    if protocol_df is not None and not protocol_df.empty:
+        for _, row in protocol_df.iterrows():
+            metric = str(row.get("Metric", "")).strip()
+            if metric not in METRICS:
+                continue
+            metric_rows.append(
+                {
+                    "metric": metric,
+                    "weight": float(weights.get(metric, 0.0)),
+                    "target": _safe_float(row.get("Target")),
+                    "min": _safe_float(row.get("Min")),
+                    "max": _safe_float(row.get("Max")),
+                    "gate_hint": str(row.get("Gate") or "").strip(),
+                }
+            )
+
+    if not metric_rows:
+        metric_rows = [{"metric": m, "weight": float(weights.get(m, 0.0))} for m in METRICS]
+
+    return {
+        "weights": {k: float(v) for k, v in weights.items()},
+        "metrics": metric_rows,
+        "gate_labels": ["HARD FAIL", "SOFT FAIL", "READ-ALOUD BLOCK", "COMMERCIAL CHECK", "GREENLIGHT", "REVISE"],
+    }
+
+
+def extract_manuscript_context(file_path: str | Path) -> ManuscriptContext:
+    path = Path(file_path)
+    story_text, raw_text, page_trace = extract_text_from_file(path)
+    cleaned = clean_text(story_text or raw_text or "")
+    return ManuscriptContext(
+        path=str(path),
+        raw_text=raw_text or "",
+        cleaned_text=cleaned,
+        word_count=len(cleaned.split()),
+        page_trace=page_trace or [],
+    )
+
+
+def run_totem_skill(
+    *,
+    rubric_matrix: dict[str, Any],
+    manuscript: ManuscriptContext,
+) -> tuple[dict[str, Any], dict[str, Any]]:
+    api_key = os.getenv("OPENAI_API_KEY")
+    if not api_key:
+        raise BridgeError("OPENAI_API_KEY is not configured.")
+
+    from openai import OpenAI
+
+    model = os.getenv("TOTEM_OPENAI_MODEL", "gpt-4.1-mini")
+    client = OpenAI(api_key=api_key)
+
+    system_prompt = (
+        "You are the hidden TOTEM Analysis skill. "
+        "Return strict JSON only with no markdown. "
+        "Use workbook matrix as scoring authority and manuscript text as evidence."
+    )
+    user_payload = {
+        "task": "score_manuscript_against_workbook_matrix",
+        "rubric_matrix": rubric_matrix,
+        "manuscript_cleaned_text": manuscript.cleaned_text,
+        "required_output_contract": {
+            "overall_publishability": "int 0..100",
+            "read_aloud_flow": "int 0..100",
+            "emotional_truth": "int 0..100",
+            "visual_strength": "int 0..100",
+            "commercial_visibility": "int 0..100",
+            "gate": "string",
+            "weakest_metric": "string",
+            "dashboard_message": "string",
+            "revision_priority_queue": [
+                {
+                    "block": "string",
+                    "weakest_dimension": "string",
+                    "gate": "string",
+                    "priority": "string",
+                    "recommended_action": "string",
+                }
+            ],
+            "risk_clusters": [
+                {
+                    "name": "string",
+                    "risk": "High Risk | Medium Risk | Low Risk",
+                    "summary": "string",
+                }
+            ],
+        },
+    }
+
+    resp = client.chat.completions.create(
+        model=model,
+        response_format={"type": "json_object"},
+        temperature=0.2,
+        messages=[
+            {"role": "system", "content": system_prompt},
+            {"role": "user", "content": json.dumps(user_payload)},
+        ],
+    )
+    raw = (resp.choices[0].message.content or "").strip()
+    try:
+        parsed = json.loads(raw)
+    except Exception as exc:
+        raise BridgeValidationError(f"Model returned invalid JSON: {exc}") from exc
+
+    validated = validate_dashboard_payload(parsed)
+    debug = {
+        "model": model,
+        "word_count": manuscript.word_count,
+        "raw_response_chars": len(raw),
+    }
+    return validated, debug
+
+
+def validate_dashboard_payload(payload: dict[str, Any]) -> dict[str, Any]:
+    required = [
+        "overall_publishability",
+        "read_aloud_flow",
+        "emotional_truth",
+        "visual_strength",
+        "commercial_visibility",
+        "gate",
+        "weakest_metric",
+        "dashboard_message",
+        "revision_priority_queue",
+        "risk_clusters",
+    ]
+    missing = [k for k in required if k not in payload]
+    if missing:
+        raise BridgeValidationError(f"Missing required output fields: {', '.join(missing)}")
+
+    out = {
+        "overall_publishability": _clamp_score(payload["overall_publishability"]),
+        "read_aloud_flow": _clamp_score(payload["read_aloud_flow"]),
+        "emotional_truth": _clamp_score(payload["emotional_truth"]),
+        "visual_strength": _clamp_score(payload["visual_strength"]),
+        "commercial_visibility": _clamp_score(payload["commercial_visibility"]),
+        "gate": str(payload.get("gate") or "REVISE").strip()[:48],
+        "weakest_metric": str(payload.get("weakest_metric") or "Read-aloud Flow").strip()[:80],
+        "dashboard_message": str(payload.get("dashboard_message") or "").strip()[:280],
+    }
+
+    rpq = payload.get("revision_priority_queue")
+    if not isinstance(rpq, list):
+        raise BridgeValidationError("revision_priority_queue must be a list.")
+    out["revision_priority_queue"] = [_normalize_queue_item(x) for x in rpq[:12]]
+
+    clusters = payload.get("risk_clusters")
+    if not isinstance(clusters, list):
+        raise BridgeValidationError("risk_clusters must be a list.")
+    out["risk_clusters"] = [_normalize_cluster_item(x) for x in clusters[:8]]
+    return out
+
+
+def recompute_gate(metrics: dict[str, int], rubric_matrix: dict[str, Any]) -> str:
+    weights = rubric_matrix.get("weights", {}) if isinstance(rubric_matrix, dict) else {}
+    if not isinstance(weights, dict) or not weights:
+        return "REVISE"
+
+    metric_key_map = {
+        "Clarity": None,
+        "Rhythm": None,
+        "Read-aloud Flow": "read_aloud_flow",
+        "Emotional Truth": "emotional_truth",
+        "Visual Strength": "visual_strength",
+        "Commercial Publishability": "commercial_visibility",
+    }
+    scores = []
+    for label, key in metric_key_map.items():
+        if key is None:
+            continue
+        score = float(metrics.get(key, 0))
+        w = float(weights.get(label, 0.0))
+        scores.append((label, score, w))
+    if not scores:
+        return "REVISE"
+
+    weighted = sum(s * w for _, s, w in scores)
+    low = min(s for _, s, _ in scores)
+    low_count = sum(1 for _, s, _ in scores if s <= 60)
+
+    if low <= 40:
+        return "HARD FAIL"
+    if low_count >= 2:
+        return "SOFT FAIL"
+    if weighted >= 80 and low >= 70:
+        return "GREENLIGHT"
+    return "REVISE"
+
+
+def _normalize_queue_item(item: Any) -> dict[str, str]:
+    if not isinstance(item, dict):
+        return {
+            "block": "—",
+            "weakest_dimension": "Read-aloud Flow",
+            "gate": "Soft Fail",
+            "priority": "Medium",
+            "recommended_action": "Review and revise.",
+        }
+    return {
+        "block": str(item.get("block") or "—").strip()[:48],
+        "weakest_dimension": str(item.get("weakest_dimension") or "Read-aloud Flow").strip()[:80],
+        "gate": str(item.get("gate") or "Soft Fail").strip()[:48],
+        "priority": str(item.get("priority") or "Medium").strip()[:24],
+        "recommended_action": str(item.get("recommended_action") or "Review and revise.").strip()[:280],
+    }
+
+
+def _normalize_cluster_item(item: Any) -> dict[str, Any]:
+    if not isinstance(item, dict):
+        return {
+            "name": "Rhythm",
+            "risk": "Medium Risk",
+            "description": "Risk signal detected.",
+            "sparkline": [8, 10, 9, 11, 12, 10, 9, 11],
+        }
+    risk = str(item.get("risk") or "Medium Risk").strip()
+    if risk not in {"High Risk", "Medium Risk", "Low Risk"}:
+        risk = "Medium Risk"
+    return {
+        "name": str(item.get("name") or "Risk").strip()[:60],
+        "risk": risk,
+        "description": str(item.get("summary") or item.get("description") or "Risk signal detected.").strip()[:220],
+        "sparkline": [8, 10, 9, 11, 12, 10, 9, 11],
+    }
+
+
+def _safe_float(value: Any) -> float | None:
+    try:
+        if value is None or value == "":
+            return None
+        return float(value)
+    except Exception:
+        return None
+
+
+def _clamp_score(value: Any) -> int:
+    try:
+        x = int(round(float(value)))
+    except Exception as exc:
+        raise BridgeValidationError(f"Invalid numeric score value: {value!r}") from exc
+    return max(0, min(100, x))