Add text-only deck critique endpoint

Wires /api/deck-critique to return a structured text-only critique before vision upload support lands.

- accepts deck_text/slides/startup context
- uses NVIDIA JSON mode when available
- falls back to local heuristics when the model is unavailable
- adds focused unittest coverage for validation, fallback, and model JSON paths

Verified with:
- python3 -m unittest tests/test_deck_critique.py -v
- python3 -m py_compile app.py core/*.py tests/test_deck_critique.py

app.py

@@ -13,7 +13,7 @@ from gradio import Server
 
 from core.api_handlers import (
     handle_chat_round,
-    handle_deck_critique_placeholder,
+    handle_deck_critique,
     handle_end_battle,
     handle_end_deal,
     handle_deal_round,
@@ -119,8 +119,8 @@ def api_end_deal(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
 
 
 @app.post("/api/deck-critique")
-def api_deck_critique(payload: dict[str, Any] = Body(default_factory=dict)) -> dict[str, str]:
-    return handle_deck_critique_placeholder(payload)
+def api_deck_critique(payload: dict[str, Any] = Body(default_factory=dict)) -> dict[str, Any]:
+    return handle_deck_critique(payload)
 
 
 # ---------------------------------------------------------------------------

core/api_handlers.py

@@ -1126,12 +1126,201 @@ def handle_deal_session_placeholder(_payload: dict[str, Any] | None = None) -> d
     }
 
 
-def handle_deck_critique_placeholder(_payload: dict[str, Any] | None = None) -> dict[str, str]:
-    """Reserved endpoint for pitch deck critique."""
+_DECK_TEXT_FIELDS = (
+    "deck_text",
+    "pitch_text",
+    "slides_text",
+    "content",
+)
+
+
+def _coerce_text_list(value: Any) -> list[str]:
+    if not isinstance(value, list):
+        return []
+    items: list[str] = []
+    for item in value:
+        if isinstance(item, str):
+            text = item.strip()
+            if text:
+                items.append(text)
+        elif isinstance(item, dict):
+            title = str(item.get("title") or item.get("heading") or "").strip()
+            body = str(item.get("body") or item.get("text") or item.get("content") or "").strip()
+            combined = ": ".join(part for part in (title, body) if part)
+            if combined:
+                items.append(combined)
+    return items
+
+
+def _deck_text_from_payload(payload: dict[str, Any]) -> str:
+    for field in _DECK_TEXT_FIELDS:
+        value = payload.get(field)
+        if isinstance(value, str) and value.strip():
+            return value.strip()
+
+    slide_items = _coerce_text_list(payload.get("slides"))
+    if slide_items:
+        return "\n".join(slide_items)
+
+    startup = payload.get("startup")
+    if isinstance(startup, dict):
+        lines = []
+        for field in _STARTUP_CONTEXT_FIELDS:
+            value = str(startup.get(field) or "").strip()
+            if value:
+                lines.append(f"{field.replace('_', ' ').title()}: {value}")
+        return "\n".join(lines).strip()
+
+    return ""
+
+
+def _as_string_list(value: Any, fallback: list[str]) -> list[str]:
+    if isinstance(value, list):
+        cleaned = [str(item).strip() for item in value if str(item).strip()]
+        if cleaned:
+            return cleaned[:5]
+    if isinstance(value, str) and value.strip():
+        return [value.strip()]
+    return fallback
+
+
+def _local_deck_critique(deck_text: str) -> dict[str, Any]:
+    text = re.sub(r"\s+", " ", deck_text).strip()
+    lower = text.lower()
+    has_number = bool(re.search(r"\d", text))
+    has_user = bool(re.search(r"\b(user|users|student|students|customer|customers|founder|founders)\b", lower))
+    has_problem = "problem" in lower or "pain" in lower or "miss" in lower
+    has_solution = "solution" in lower or "assistant" in lower or "platform" in lower or "app" in lower
+    has_business = bool(re.search(r"\b(revenue|pricing|pay|paid|charge|business model|buyer|customer)\b", lower))
+    has_competition = bool(re.search(r"\b(competitor|competitors|alternative|instead of|versus|vs\.?)\b", lower))
+
+    strengths: list[str] = []
+    if has_problem and has_user:
+        strengths.append("Names a clear user pain and target audience.")
+    if has_solution:
+        strengths.append("Explains the product direction in concrete terms.")
+    if has_number:
+        strengths.append("Includes at least one number that can become proof.")
+    if not strengths:
+        strengths.append("Gives enough raw material to start a pitch critique.")
+
+    risks: list[str] = []
+    if not has_business:
+        risks.append("The deck does not yet make the business model or buyer clear.")
+    if not has_competition:
+        risks.append("The competitive wedge is underdeveloped.")
+    if not has_number:
+        risks.append("The deck needs measurable proof, not only claims.")
+    if not risks:
+        risks.append("The story is promising; the main risk is whether the proof is strong enough under judge pressure.")
+
+    recommended_fixes: list[str] = []
+    if not has_business:
+        recommended_fixes.append("Add one slide that says who pays, why they pay, and the first realistic price point.")
+    if not has_competition:
+        recommended_fixes.append("Add a comparison slide showing why users choose this over the closest existing workflow.")
+    if not has_number:
+        recommended_fixes.append("Add one traction or validation number that a judge can remember.")
+    recommended_fixes.append("Rewrite the weakest claim as: claim, evidence, and why it matters now.")
+
+    summary = (
+        "The deck has a usable pitch foundation, but it needs sharper evidence "
+        "and a clearer business case before a serious judge round."
+    )
+    next_question = (
+        "What single metric proves this is a real user pain rather than a nice-to-have idea?"
+        if has_business
+        else "Who pays for this, and what evidence shows they will pay?"
+    )
+
     return {
-        "status": "not_implemented",
-        "message": (
-            "Deck critique endpoint is reserved and will be connected "
-            "after MiniCPM-V vision integration."
-        ),
+        "summary": summary,
+        "strengths": strengths[:5],
+        "risks": risks[:5],
+        "recommended_fixes": recommended_fixes[:5],
+        "next_pitch_question": next_question,
+    }
+
+
+def _build_deck_critique_messages(deck_text: str) -> list[dict[str, str]]:
+    return [
+        {
+            "role": "system",
+            "content": (
+                "You are PitchFight AI's pitch deck coach. Review text extracted from a founder's pitch deck. "
+                "Return ONLY valid JSON with this schema: "
+                '{"summary":"","strengths":[""],"risks":[""],"recommended_fixes":[""],"next_pitch_question":""}. '
+                "Be direct, specific, and useful for a student founder. Do not use markdown."
+            ),
+        },
+        {
+            "role": "user",
+            "content": (
+                "Critique this pitch deck text. Focus on clarity, proof, differentiation, business model, "
+                "and what a judge would challenge first.\n\n"
+                + deck_text[:6000]
+            ),
+        },
+    ]
+
+
+def handle_deck_critique(_payload: dict[str, Any] | None = None) -> dict[str, Any]:
+    """Return a text-only pitch deck critique.
+
+    This keeps the endpoint useful before MiniCPM-V vision upload support lands.
+    """
+    payload = _payload or {}
+    deck_text = _deck_text_from_payload(payload)
+    if len(deck_text.strip()) < 20:
+        return {
+            "ok": False,
+            "error": "deck_text, slides, or startup context is required for deck critique.",
+            "provider": "none",
+            "model_ok": False,
+        }
+
+    model_mode = str(payload.get("model_mode") or "premium_nvidia")
+    local = _local_deck_critique(deck_text)
+
+    model_error: str | None = None
+    try:
+        result = model_router.generate_deck_critique_response(
+            _build_deck_critique_messages(deck_text),
+            model_mode=model_mode,
+        )
+        if result.get("ok") and result.get("content"):
+            parsed, _repair_needed = parse_model_json(result["content"])
+            if isinstance(parsed, dict) and parsed:
+                return {
+                    "ok": True,
+                    "provider": result.get("provider", "nvidia"),
+                    "model_mode": result.get("model_mode", model_mode),
+                    "model_ok": True,
+                    "summary": str(parsed.get("summary") or local["summary"]).strip(),
+                    "strengths": _as_string_list(parsed.get("strengths"), local["strengths"]),
+                    "risks": _as_string_list(parsed.get("risks"), local["risks"]),
+                    "recommended_fixes": _as_string_list(
+                        parsed.get("recommended_fixes"), local["recommended_fixes"]
+                    ),
+                    "next_pitch_question": str(
+                        parsed.get("next_pitch_question") or local["next_pitch_question"]
+                    ).strip(),
+                }
+        model_error = result.get("error") or "Model returned empty deck critique."
+    except Exception as exc:
+        logger.warning("handle_deck_critique: model critique failed: %s", exc)
+        model_error = str(exc)
+
+    return {
+        "ok": True,
+        "provider": "local",
+        "model_mode": "local_fallback",
+        "model_ok": False,
+        **local,
+        **({"model_error": model_error} if model_error else {}),
     }
+
+
+def handle_deck_critique_placeholder(_payload: dict[str, Any] | None = None) -> dict[str, Any]:
+    """Backward-compatible wrapper for older imports."""
+    return handle_deck_critique(_payload)

core/model_router.py

@@ -606,6 +606,14 @@ def generate_structure_pitch_response(
     return _call_nvidia_json_mode(messages, "structure_pitch", model_mode, "structure pitch")
 
 
+def generate_deck_critique_response(
+    messages: list[dict[str, str]],
+    model_mode: str | None = None,
+) -> dict[str, Any]:
+    """Generate text-only pitch deck critique JSON."""
+    return _call_nvidia_json_mode(messages, "deck_critique", model_mode, "deck critique")
+
+
 def generate_structure_pitch_repair_response(
     raw_bad_content: str,
     model_mode: str | None = None,

core/nvidia_client.py

@@ -225,6 +225,13 @@ _TASK_DEFAULTS: dict[str, dict[str, Any]] = {
         "temperature": 0.0,
         "top_p": 0.95,
     },
+    "deck_critique": {
+        "enable_thinking": False,
+        "reasoning_budget": 0,
+        "max_tokens": 1800,
+        "temperature": 0.2,
+        "top_p": 0.95,
+    },
 }
 
 _VALID_AUDIO_FORMATS = frozenset({"webm", "wav", "mp3", "m4a", "ogg"})
@@ -258,6 +265,7 @@ _JSON_MODES: frozenset[str] = frozenset({
     "deal_scorecard_scoring_repair",
     "deal_scorecard_coaching",
     "deal_scorecard_repair",
+    "deck_critique",
 })
 
 

tests/test_deck_critique.py

@@ -0,0 +1,117 @@
+import json
+import types
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+import sys
+
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+# Keep the handler unit test lightweight. The app imports optional runtime
+# clients at module import time, but this test never calls external services.
+dotenv_module = types.ModuleType("dotenv")
+dotenv_module.load_dotenv = lambda *args, **kwargs: None
+sys.modules.setdefault("dotenv", dotenv_module)
+
+openai_module = types.ModuleType("openai")
+openai_module.OpenAI = object
+openai_module.APIConnectionError = type("APIConnectionError", (Exception,), {})
+openai_module.APIStatusError = type("APIStatusError", (Exception,), {})
+openai_module.APITimeoutError = type("APITimeoutError", (Exception,), {})
+sys.modules.setdefault("openai", openai_module)
+
+pymongo_module = types.ModuleType("pymongo")
+pymongo_module.MongoClient = object
+pymongo_errors_module = types.ModuleType("pymongo.errors")
+pymongo_collection_module = types.ModuleType("pymongo.collection")
+pymongo_collection_module.Collection = object
+pymongo_database_module = types.ModuleType("pymongo.database")
+pymongo_database_module.Database = object
+pymongo_errors_module.PyMongoError = type("PyMongoError", (Exception,), {})
+pymongo_errors_module.ServerSelectionTimeoutError = type(
+    "ServerSelectionTimeoutError", (Exception,), {}
+)
+sys.modules.setdefault("pymongo", pymongo_module)
+sys.modules.setdefault("pymongo.collection", pymongo_collection_module)
+sys.modules.setdefault("pymongo.database", pymongo_database_module)
+sys.modules.setdefault("pymongo.errors", pymongo_errors_module)
+
+from core.api_handlers import handle_deck_critique
+
+
+DECK_TEXT = """
+EventRadar AI
+Problem: students miss hackathons because events are scattered across WhatsApp, LinkedIn, Luma, and club pages.
+Solution: an AI event discovery assistant ranks events by skills, goals, location, and deadline urgency.
+Users: college students and student founders.
+Traction: tested with 80 students; 62 missed at least one relevant event last month.
+Ask: mentor feedback and pilot introductions.
+"""
+
+
+class DeckCritiqueTests(unittest.TestCase):
+    def test_deck_critique_returns_local_fallback_instead_of_placeholder(self):
+        with patch(
+            "core.api_handlers.model_router.generate_deck_critique_response",
+            return_value={
+                "ok": False,
+                "model_mode": "premium_nvidia",
+                "provider": "nvidia",
+                "content": "",
+                "error": "model unavailable",
+            },
+        ):
+            result = handle_deck_critique({"deck_text": DECK_TEXT})
+
+        self.assertTrue(result["ok"])
+        self.assertNotEqual(result.get("status"), "not_implemented")
+        self.assertEqual(result["provider"], "local")
+        self.assertFalse(result["model_ok"])
+        self.assertTrue(result["summary"])
+        self.assertGreaterEqual(len(result["strengths"]), 1)
+        self.assertGreaterEqual(len(result["risks"]), 1)
+        self.assertGreaterEqual(len(result["recommended_fixes"]), 1)
+        self.assertTrue(result["next_pitch_question"].endswith("?"))
+
+    def test_deck_critique_requires_text_or_startup_context(self):
+        result = handle_deck_critique({})
+
+        self.assertFalse(result["ok"])
+        self.assertIn("deck_text", result["error"])
+
+    def test_deck_critique_uses_model_json_when_available(self):
+        model_payload = {
+            "summary": "The deck has a focused student-founder wedge but needs sharper proof.",
+            "strengths": ["Clear student pain", "Specific initial traction"],
+            "risks": ["Business model is still vague"],
+            "recommended_fixes": ["Add one slide on who pays and why this is urgent"],
+            "next_pitch_question": "Who pays for this, and what evidence shows they will pay?",
+        }
+
+        with patch(
+            "core.api_handlers.model_router.generate_deck_critique_response",
+            return_value={
+                "ok": True,
+                "model_mode": "premium_nvidia",
+                "provider": "nvidia",
+                "content": json.dumps(model_payload),
+                "error": None,
+            },
+            create=True,
+        ):
+            result = handle_deck_critique(
+                {"deck_text": DECK_TEXT, "model_mode": "premium_nvidia"}
+            )
+
+        self.assertTrue(result["ok"])
+        self.assertTrue(result["model_ok"])
+        self.assertEqual(result["provider"], "nvidia")
+        self.assertEqual(result["summary"], model_payload["summary"])
+        self.assertEqual(result["recommended_fixes"], model_payload["recommended_fixes"])
+
+
+if __name__ == "__main__":
+    unittest.main()