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]] fingerprint: dict[str, Any] | None = None fingerprint_report: str = "" @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}, "score_evidence": { field: { "score": "integer 0..100 matching scores.", "fingerprint_metrics_used": ["VM metric ids used, e.g. VM-001, VM-010, VM-025"], "fingerprint_metric_values": {"VM-010": "numeric/string value used from manuscript_fingerprint"}, "workbook_rule": "target/threshold/rubric rule used from workbook matrix", "manuscript_evidence": "short paraphrased manuscript evidence; no long excerpts", "reason": "why this score follows from the workbook rule and fingerprint metrics", } 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, manuscript fingerprint VM metrics as measured evidence, and manuscript text only as supporting 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_fingerprint": manuscript.fingerprint or {}, "manuscript_fingerprint_report": manuscript.fingerprint_report[:6000] if manuscript.fingerprint_report else "", "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. " "Every score_evidence entry must cite at least one VM metric id from manuscript_fingerprint " "with its observed value in fingerprint_metric_values, plus one workbook rule/target from rubric_matrix. " "If the fingerprint is missing or insufficient, return gate REVISE and explain the missing evidence in dashboard_message. " "Hard caps for picture-book scoring: if VM-010 sentence length mean is above 30 or VM-025 reading age " "is above 10, clarity and read_aloud_flow must not exceed 75. If VM-025 reading age is above 10, " "commercial_viability must not exceed 75. If VM-002 syllable variance is above 5, rhythm must not exceed 75. " "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, manuscript.fingerprint) 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, "fingerprint_supplied": bool(manuscript.fingerprint), "fingerprint_metric_count": _fingerprint_metric_count(manuscript.fingerprint), "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, manuscript.fingerprint) 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, "fingerprint_supplied": bool(manuscript.fingerprint), "fingerprint_metric_count": _fingerprint_metric_count(manuscript.fingerprint), "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) fingerprint_json = compact_json(manuscript.fingerprint or {}) 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, manuscript fingerprint VM metrics as measured evidence, and manuscript text only as supporting evidence. """.strip() component_estimates = estimate_components( { "system_wrapper": system_wrapper, "skill_md": skill_text, "workbook_rubric_json": rubric_json, "manuscript_fingerprint_json": fingerprint_json, "manuscript_fingerprint_report": manuscript.fingerprint_report or "", "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, }, "fingerprint": { "supplied": bool(manuscript.fingerprint), "sha256": stable_hash(manuscript.fingerprint or {}), "metric_count": _fingerprint_metric_count(manuscript.fingerprint), "report_chars": len(manuscript.fingerprint_report or ""), }, "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], fingerprint: dict[str, Any] | None = None) -> 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"]) score_evidence = payload.get("score_evidence") if not isinstance(score_evidence, dict): raise BridgeValidationError("score_evidence must be a JSON object keyed by scoring dimension.") out["score_evidence"] = _normalize_score_evidence(score_evidence, out["scores"], fingerprint) _validate_fingerprint_score_alignment(out, fingerprint) 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 _normalize_score_evidence( evidence: dict[str, Any], scores: dict[str, int], fingerprint: dict[str, Any] | None = None, ) -> dict[str, dict[str, Any]]: normalized: dict[str, dict[str, Any]] = {} missing: list[str] = [] weak: list[str] = [] for field in SCORE_FIELDS: item = evidence.get(field) if not isinstance(item, dict): missing.append(field) continue metrics = item.get("fingerprint_metrics_used") or item.get("vm_metrics_used") or [] if isinstance(metrics, str): metrics = [metrics] metric_ids = [str(x).strip() for x in metrics if str(x).strip()] if not metric_ids or not any(m.upper().startswith("VM-") for m in metric_ids): weak.append(field) required_metric = _required_bottleneck_metric(field, fingerprint) if required_metric and required_metric not in {m.upper().split("_", 1)[0] for m in metric_ids}: weak.append(field) workbook_rule = str(item.get("workbook_rule") or "").strip()[:260] manuscript_evidence = str(item.get("manuscript_evidence") or "").strip()[:260] reason = str(item.get("reason") or "").strip()[:320] if not workbook_rule or not reason: weak.append(field) normalized[field] = { "score": scores[field], "fingerprint_metrics_used": metric_ids[:8], "fingerprint_metric_values": { metric_id.upper().split("_", 1)[0]: _fingerprint_value(fingerprint, metric_id) for metric_id in metric_ids[:8] }, "workbook_rule": workbook_rule, "manuscript_evidence": manuscript_evidence, "reason": reason, } if missing: raise BridgeValidationError(f"Missing score_evidence fields: {', '.join(missing)}") if weak: deduped = ", ".join(dict.fromkeys(weak)) raise BridgeValidationError(f"score_evidence lacks VM metric/workbook grounding for: {deduped}") return normalized def _required_bottleneck_metric(field: str, fingerprint: dict[str, Any] | None) -> str | None: sentence_mean = _fingerprint_float(fingerprint, "VM-010") reading_age = _fingerprint_float(fingerprint, "VM-025") syllable_variance = _fingerprint_float(fingerprint, "VM-002") if field == "rhythm" and syllable_variance is not None and syllable_variance > 5: return "VM-002" if field in {"clarity", "read_aloud_flow"}: if sentence_mean is not None and sentence_mean > 30: return "VM-010" if reading_age is not None and reading_age > 10: return "VM-025" if field == "commercial_viability" and reading_age is not None and reading_age > 10: return "VM-025" return None def _validate_fingerprint_score_alignment(out: dict[str, Any], fingerprint: dict[str, Any] | None) -> None: if not isinstance(fingerprint, dict) or not fingerprint: return sentence_mean = _fingerprint_float(fingerprint, "VM-010") reading_age = _fingerprint_float(fingerprint, "VM-025") syllable_variance = _fingerprint_float(fingerprint, "VM-002") problems: list[str] = [] if sentence_mean is not None and sentence_mean > 30: if out["scores"].get("clarity", 0) > 75: problems.append(f"clarity={out['scores']['clarity']} exceeds cap 75 while VM-010 sentence mean is {sentence_mean}") if out["scores"].get("read_aloud_flow", 0) > 75: problems.append( f"read_aloud_flow={out['scores']['read_aloud_flow']} exceeds cap 75 while VM-010 sentence mean is {sentence_mean}" ) if reading_age is not None and reading_age > 10: if out["scores"].get("read_aloud_flow", 0) > 75: problems.append( f"read_aloud_flow={out['scores']['read_aloud_flow']} exceeds cap 75 while VM-025 reading age is {reading_age}" ) if out["scores"].get("commercial_viability", 0) > 75: problems.append( f"commercial_viability={out['scores']['commercial_viability']} exceeds cap 75 while VM-025 reading age is {reading_age}" ) if syllable_variance is not None and syllable_variance > 5 and out["scores"].get("rhythm", 0) > 75: problems.append(f"rhythm={out['scores']['rhythm']} exceeds cap 75 while VM-002 syllable variance is {syllable_variance}") if problems: raise BridgeValidationError("Fingerprint/score contradiction: " + "; ".join(problems)) 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 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)), } return { "scores": scores, "score_evidence": { field: { "score": score, "fingerprint_metrics_used": ["VM-024"], "workbook_rule": "Fake provider test rule; real runs must cite workbook targets.", "manuscript_evidence": "Fake provider uses word-count-only smoke evidence.", "reason": f"Fake audit score for {field} derived from {word_count} words.", } for field, score in scores.items() }, "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)) def _fingerprint_metric_count(fingerprint: dict[str, Any] | None) -> int: if not isinstance(fingerprint, dict): return 0 return sum(1 for key in fingerprint if str(key).upper().startswith("VM-")) def _fingerprint_value(fingerprint: dict[str, Any] | None, metric_id: str) -> Any: if not isinstance(fingerprint, dict): return None wanted = str(metric_id).upper().split("_", 1)[0] for key, value in fingerprint.items(): if str(key).upper().startswith(wanted): return value return None def _fingerprint_float(fingerprint: dict[str, Any] | None, metric_id: str) -> float | None: value = _fingerprint_value(fingerprint, metric_id) try: if value is None or value == "": return None return float(value) except Exception: return None