from __future__ import annotations import json import os import time 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.token_meter import ( append_token_audit, compact_json, estimate_components, estimate_text_tokens, extract_response_usage, new_transaction_id, stable_hash, ) from src.totem_workbook import METRICS, protocol_table, protocol_weights APP_ROOT = Path(__file__).resolve().parents[1] DEFAULT_SKILL_PATH = APP_ROOT / "data" / "skills" / "TOTEM_Manuscript_Scoring_Skill.md" SCORE_FIELDS = ( "clarity", "rhythm", "read_aloud_flow", "emotional_truth", "visual_strength", "commercial_viability", ) SCORE_LABELS = { "clarity": "Clarity", "rhythm": "Rhythm", "read_aloud_flow": "Read-aloud Flow", "emotional_truth": "Emotional Truth", "visual_strength": "Visual Strength", "commercial_viability": "Commercial Viability", } VALID_GATES = {"HARD FAIL", "SOFT FAIL", "READ-ALOUD BLOCK", "COMMERCIAL CHECK", "GREENLIGHT", "REVISE"} VALID_RISKS = {"High Risk", "Medium Risk", "Low Risk"} 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]] @dataclass class LLMProviderConfig: provider: str model: str api_key: str api_key_env: str base_url: str | None = None 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": sorted(VALID_GATES), } 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 load_totem_skill(skill_path: str | Path | None = None) -> tuple[str, str]: path = Path(skill_path or os.getenv("TOTEM_SKILL_PATH") or DEFAULT_SKILL_PATH) if not path.exists(): raise BridgeError(f"TOTEM scoring skill not found: {path}") text = path.read_text(encoding="utf-8").strip() if not text: raise BridgeError(f"TOTEM scoring skill is empty: {path}") return text, str(path) def dashboard_output_contract() -> dict[str, Any]: return { "scores": {field: "integer 0..100" for field in SCORE_FIELDS}, "gate": "HARD FAIL | SOFT FAIL | READ-ALOUD BLOCK | COMMERCIAL CHECK | GREENLIGHT | REVISE", "weakest_metric": "one of: Clarity, Rhythm, Read-aloud Flow, Emotional Truth, Visual Strength, Commercial Viability", "dashboard_message": "short dashboard status message, max 220 characters", "revision_priority_queue": [ { "block": "short block/page/section identifier", "weakest_dimension": "scoring dimension name", "gate": "gate label", "priority": "High | Medium | Low", "recommended_action": "specific rewrite action, max 240 characters", } ], "risk_clusters": [ { "name": "risk cluster name", "risk": "High Risk | Medium Risk | Low Risk", "summary": "short evidence-based risk summary", } ], "evidence_summary": "brief evidence summary, max 500 characters", } def run_totem_skill( *, rubric_matrix: dict[str, Any], manuscript: ManuscriptContext, ) -> tuple[dict[str, Any], dict[str, Any]]: skill_text, skill_path = load_totem_skill() transaction_id = new_transaction_id("totem") contract = dashboard_output_contract() system_prompt = f""" You are the hidden TOTEM Analysis skill inside TOTEM Studio. Use the skill instructions below as the scoring method, but do not create a DOCX report for this dashboard run. Return strict JSON only. No prose. No markdown. No code fences. Use the workbook matrix as scoring authority and manuscript text as evidence. --- TOTEM SKILL INSTRUCTIONS --- {skill_text} --- END TOTEM SKILL INSTRUCTIONS --- """.strip() user_payload = { "task": "score_manuscript_against_workbook_matrix_for_dashboard", "rubric_matrix": rubric_matrix, "manuscript_cleaned_text": manuscript.cleaned_text, "scoring_dimensions": SCORE_LABELS, "required_output_contract": contract, "instruction": ( "Return exactly one JSON object matching required_output_contract. " "Scores must be 0..100 dashboard values for the six dimensions. " "Do not write a report and do not include manuscript excerpts." ), } user_payload_json = json.dumps(user_payload, ensure_ascii=False) if _env_truthy("TOTEM_FAKE_API"): if os.getenv("SPACE_ID") and not _env_truthy("TOTEM_ALLOW_FAKE_API_IN_SPACE"): raise BridgeError("TOTEM_FAKE_API is disabled on Hugging Face Spaces.") raw_payload = _fake_dashboard_payload(manuscript) validated = validate_dashboard_payload(raw_payload) raw_payload_json = json.dumps(raw_payload, ensure_ascii=False) audit_path = _write_token_audit( transaction_id=transaction_id, status="ok", provider="fake", model="fake-audit", api_key_env=None, skill_path=skill_path, skill_text=skill_text, rubric_matrix=rubric_matrix, manuscript=manuscript, contract=contract, system_prompt=system_prompt, user_payload_json=user_payload_json, actual_tokens=None, raw_response=raw_payload_json, latency_ms=0, error=None, ) return validated, { "provider": "fake", "model": "fake-audit", "word_count": manuscript.word_count, "raw_response_chars": len(raw_payload_json), "skill_path": skill_path, "skill_loaded": True, "schema_pass": True, "token_audit_path": audit_path, "token_audit_transaction_id": transaction_id, "raw_model_payload": raw_payload, "validated_payload": validated, } provider_config = _select_llm_provider() from openai import OpenAI client_kwargs: dict[str, Any] = {"api_key": provider_config.api_key} if provider_config.base_url: client_kwargs["base_url"] = provider_config.base_url client = OpenAI(**client_kwargs) started = time.perf_counter() resp = None raw = "" try: resp = client.chat.completions.create( model=provider_config.model, response_format={"type": "json_object"}, temperature=0.2, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_payload_json}, ], ) except Exception as exc: latency_ms = int(round((time.perf_counter() - started) * 1000)) audit_path = _write_token_audit( transaction_id=transaction_id, status="api_error", provider=provider_config.provider, model=provider_config.model, api_key_env=provider_config.api_key_env, skill_path=skill_path, skill_text=skill_text, rubric_matrix=rubric_matrix, manuscript=manuscript, contract=contract, system_prompt=system_prompt, user_payload_json=user_payload_json, actual_tokens=None, raw_response="", latency_ms=latency_ms, error=f"{type(exc).__name__}: {exc}", ) raise BridgeError( f"{provider_config.provider} API call failed. Token audit: {audit_path}. {type(exc).__name__}: {exc}" ) from exc latency_ms = int(round((time.perf_counter() - started) * 1000)) actual_tokens = extract_response_usage(resp) raw = (resp.choices[0].message.content or "").strip() try: parsed = json.loads(raw) except Exception as exc: audit_path = _write_token_audit( transaction_id=transaction_id, status="invalid_json", provider=provider_config.provider, model=provider_config.model, api_key_env=provider_config.api_key_env, skill_path=skill_path, skill_text=skill_text, rubric_matrix=rubric_matrix, manuscript=manuscript, contract=contract, system_prompt=system_prompt, user_payload_json=user_payload_json, actual_tokens=actual_tokens, raw_response=raw, latency_ms=latency_ms, error=f"{type(exc).__name__}: {exc}", ) raise BridgeValidationError(f"Model returned invalid JSON: {exc}") from exc try: validated = validate_dashboard_payload(parsed) except BridgeValidationError as exc: audit_path = _write_token_audit( transaction_id=transaction_id, status="schema_error", provider=provider_config.provider, model=provider_config.model, api_key_env=provider_config.api_key_env, skill_path=skill_path, skill_text=skill_text, rubric_matrix=rubric_matrix, manuscript=manuscript, contract=contract, system_prompt=system_prompt, user_payload_json=user_payload_json, actual_tokens=actual_tokens, raw_response=raw, latency_ms=latency_ms, error=f"{type(exc).__name__}: {exc}", ) raise BridgeValidationError(f"{exc}. Token audit: {audit_path}") from exc audit_path = _write_token_audit( transaction_id=transaction_id, status="ok", provider=provider_config.provider, model=provider_config.model, api_key_env=provider_config.api_key_env, skill_path=skill_path, skill_text=skill_text, rubric_matrix=rubric_matrix, manuscript=manuscript, contract=contract, system_prompt=system_prompt, user_payload_json=user_payload_json, actual_tokens=actual_tokens, raw_response=raw, latency_ms=latency_ms, error=None, ) debug = { "provider": provider_config.provider, "model": provider_config.model, "api_key_env": provider_config.api_key_env, "word_count": manuscript.word_count, "raw_response_chars": len(raw), "skill_path": skill_path, "skill_loaded": True, "schema_pass": True, "actual_tokens": actual_tokens, "token_audit_path": audit_path, "token_audit_transaction_id": transaction_id, "raw_model_payload": parsed, "validated_payload": validated, } return validated, debug def _select_llm_provider() -> LLMProviderConfig: # Hugging Face exposes secrets exactly as named. Accept Jamal's early # `grok_key` spelling, but normalize internally to the standard Groq name. if os.environ.get("grok_key") and not os.environ.get("GROQ_API_KEY"): os.environ["GROQ_API_KEY"] = os.environ["grok_key"] requested = str(os.getenv("TOTEM_LLM_PROVIDER") or "").strip().lower() if requested == "grok": requested = "groq" groq_key, groq_env = _first_env_value( ( "GROQ_API_KEY", "groq_key", "GROQ_KEY", "GROK_API_KEY", "GROK_KEY", ) ) openai_key, openai_env = _first_env_value(("OPENAI_API_KEY",)) if not requested: requested = "groq" if groq_key else "openai" if requested == "groq": if not groq_key: raise BridgeError("Groq is selected but no GROQ_API_KEY/groq_key secret is configured.") return LLMProviderConfig( provider="groq", model=os.getenv("TOTEM_GROQ_MODEL") or os.getenv("GROQ_MODEL") or "llama-3.3-70b-versatile", api_key=groq_key, api_key_env=groq_env or "GROQ_API_KEY", base_url=os.getenv("GROQ_BASE_URL") or "https://api.groq.com/openai/v1", ) if requested == "openai": if not openai_key and groq_key: return LLMProviderConfig( provider="groq", model=os.getenv("TOTEM_GROQ_MODEL") or os.getenv("GROQ_MODEL") or "llama-3.3-70b-versatile", api_key=groq_key, api_key_env=groq_env or "GROQ_API_KEY", base_url=os.getenv("GROQ_BASE_URL") or "https://api.groq.com/openai/v1", ) if not openai_key: raise BridgeError("OpenAI is selected but OPENAI_API_KEY is not configured.") return LLMProviderConfig( provider="openai", model=os.getenv("TOTEM_OPENAI_MODEL", "gpt-4.1-mini"), api_key=openai_key, api_key_env=openai_env or "OPENAI_API_KEY", base_url=os.getenv("OPENAI_BASE_URL") or None, ) raise BridgeError("TOTEM_LLM_PROVIDER must be 'groq' or 'openai'.") def _write_token_audit( *, transaction_id: str, status: str, provider: str, model: str, api_key_env: str | None, skill_path: str, skill_text: str, rubric_matrix: dict[str, Any], manuscript: ManuscriptContext, contract: dict[str, Any], system_prompt: str, user_payload_json: str, actual_tokens: dict[str, int | None] | None, raw_response: str, latency_ms: int, error: str | None, ) -> str: rubric_json = compact_json(rubric_matrix) contract_json = compact_json(contract) user_wrapper_json = compact_json( { "task": "score_manuscript_against_workbook_matrix_for_dashboard", "scoring_dimensions": SCORE_LABELS, "instruction": "Return exactly one JSON object matching required_output_contract.", } ) system_wrapper = """ You are the hidden TOTEM Analysis skill inside TOTEM Studio. Use the skill instructions as the scoring method, but do not create a DOCX report for this dashboard run. Return strict JSON only. No prose. No markdown. No code fences. Use the workbook matrix as scoring authority and manuscript text as evidence. """.strip() component_estimates = estimate_components( { "system_wrapper": system_wrapper, "skill_md": skill_text, "workbook_rubric_json": rubric_json, "manuscript_cleaned_text": manuscript.cleaned_text, "output_contract_json": contract_json, "user_wrapper_json": user_wrapper_json, }, model=model, ) system_wire = estimate_text_tokens(system_prompt, model=model) user_wire = estimate_text_tokens(user_payload_json, model=model) record = { "transaction_id": transaction_id, "created_at_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "status": status, "provider": provider, "model": model, "api_key_env": api_key_env, "latency_ms": latency_ms, "skill": { "path": skill_path, "sha256": stable_hash(skill_text), "chars": len(skill_text), }, "workbook": { "sha256": stable_hash(rubric_matrix), "metric_count": len(rubric_matrix.get("metrics", [])) if isinstance(rubric_matrix, dict) else 0, }, "manuscript": { "path": manuscript.path, "sha256": stable_hash(manuscript.cleaned_text), "chars": len(manuscript.cleaned_text), "word_count": manuscript.word_count, }, "estimated_tokens": { **component_estimates, "wire_prompt": { "system_prompt": system_wire, "user_message": user_wire, "estimated_prompt_total": system_wire["tokens"] + user_wire["tokens"], }, }, "actual_tokens": actual_tokens or { "prompt_tokens": None, "completion_tokens": None, "total_tokens": None, }, "response": { "sha256": stable_hash(raw_response) if raw_response else None, "chars": len(raw_response or ""), }, "error": error, } return append_token_audit(record) def _first_env_value(names: tuple[str, ...]) -> tuple[str | None, str | None]: for name in names: value = os.getenv(name) if value: return value, name return None, None def validate_dashboard_payload(payload: dict[str, Any]) -> dict[str, Any]: if not isinstance(payload, dict): raise BridgeValidationError("Dashboard payload must be a JSON object.") scores = payload.get("scores") if not isinstance(scores, dict): # Backward-compatible normalization for older flat model responses. scores = { "clarity": payload.get("clarity"), "rhythm": payload.get("rhythm"), "read_aloud_flow": payload.get("read_aloud_flow"), "emotional_truth": payload.get("emotional_truth"), "visual_strength": payload.get("visual_strength"), "commercial_viability": payload.get("commercial_viability", payload.get("commercial_visibility")), } required = [ "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)}") missing_scores = [field for field in SCORE_FIELDS if field not in scores or scores.get(field) in (None, "")] if missing_scores: raise BridgeValidationError(f"Missing required score fields: {', '.join(missing_scores)}") gate = str(payload.get("gate") or "REVISE").strip().upper() if gate not in VALID_GATES: gate = "REVISE" out = { "scores": {field: _clamp_score(scores[field]) for field in SCORE_FIELDS}, "gate": gate, "weakest_metric": str(payload.get("weakest_metric") or "Read-aloud Flow").strip()[:80], "dashboard_message": str(payload.get("dashboard_message") or "").strip()[:280], "evidence_summary": str(payload.get("evidence_summary") or "").strip()[:500], } out.update(out["scores"]) 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": "clarity", "Rhythm": "rhythm", "Read-aloud Flow": "read_aloud_flow", "Emotional Truth": "emotional_truth", "Visual Strength": "visual_strength", "Commercial Publishability": "commercial_viability", } scores = [] for label, key in metric_key_map.items(): 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) rhythm = float(metrics.get("rhythm", 0)) flow = float(metrics.get("read_aloud_flow", 0)) commercial = float(metrics.get("commercial_viability", 0)) if low <= 40: return "HARD FAIL" if low_count >= 2: return "SOFT FAIL" if rhythm < 70 or flow < 70: return "READ-ALOUD BLOCK" if commercial < 70: return "COMMERCIAL CHECK" 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": "REVISE", "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 "REVISE").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 VALID_RISKS: 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 _fake_dashboard_payload(manuscript: ManuscriptContext) -> dict[str, Any]: word_count = max(1, manuscript.word_count) base = max(45, min(88, 62 + (word_count // 160))) long_sentence_penalty = 8 if word_count > 850 else 3 return { "scores": { "clarity": max(0, min(100, base - long_sentence_penalty)), "rhythm": max(0, min(100, base - 11)), "read_aloud_flow": max(0, min(100, base - 7)), "emotional_truth": max(0, min(100, base + 5)), "visual_strength": max(0, min(100, base + 8)), "commercial_viability": max(0, min(100, base + 1)), }, "gate": "REVISE", "weakest_metric": "Rhythm", "dashboard_message": "TOTEM analysis complete using fake audit response.", "revision_priority_queue": [ { "block": "Opening third", "weakest_dimension": "Rhythm", "gate": "REVISE", "priority": "High", "recommended_action": "Run a read-aloud pass and cut any line that stalls the beat.", }, { "block": "Middle turn", "weakest_dimension": "Clarity", "gate": "REVISE", "priority": "Medium", "recommended_action": "Clarify the character action before adding extra comic business.", }, ], "risk_clusters": [ {"name": "Rhythm", "risk": "High Risk", "summary": "Read-aloud pressure likely clusters around longer lines."}, {"name": "Clarity", "risk": "Medium Risk", "summary": "Some manuscript beats may need cleaner cause and effect."}, ], "evidence_summary": f"Fake audit used {word_count} manuscript words.", } def _env_truthy(name: str) -> bool: return str(os.getenv(name, "")).strip().lower() in {"1", "true", "yes", "on"} 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))