feat: add small model assisted analysis

Co-authored-by: Codex <noreply@openai.com>

README.md

@@ -18,9 +18,11 @@ telemetry by default and analyzes only the agent's visible narrative messages:
 what it planned, where it got stuck, how it detoured, how it recovered, and how
 it claimed completion.
 
-Built for the Build Small Hackathon as a Gradio app. The stable MVP uses a
+Built for the Build Small Hackathon as a Gradio app. The default engine uses a
 verified deterministic codebook analyzer so the Space can always start and
-produce a report; the analysis schema is model-ready for small-model assistance.
+produce a report. The app also exposes explicit small-model assist modes for
+`nvidia/NVIDIA-Nemotron-3-Nano-30B-A3B-BF16` and `Qwen/Qwen3.5-9B` through
+Hugging Face Inference Providers when the runtime has provider access.
 
 ## Run Locally
 
@@ -37,6 +39,18 @@ python app.py
 python3.11 -m unittest discover -s tests
 ```
 
+## Analysis Engines
+
+- `Deterministic field notes`: default, local, no model dependency.
+- `Small-model assist: NVIDIA Nemotron 3 Nano 30B-A3B`: uses the hackathon-sized
+  30B total-parameter Nemotron model when Hugging Face Inference Providers can
+  serve it.
+- `Quick small-model assist: Qwen3.5 9B`: optional lower-latency model-assisted
+  memo.
+
+If a selected model is unavailable, the report records the error in model notes
+and returns the deterministic analysis instead of failing the whole Space.
+
 ## Agent Session Locations
 
 ```bash

analyzer.py

@@ -8,6 +8,7 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Iterable
 
+from model_runtime import MODEL_CHOICES, run_model_assist
 from parser import parse_trace
 from redaction import redact_text
 from schemas import AnalysisResult, DifficultyEpisode, MessageSpan, NarrativeMessage
@@ -117,6 +118,7 @@ def analyze_trace_file(
     redact_secrets: bool = True,
     ignore_tool_calls: bool = True,
     report_style: str = "field_notes",
+    analysis_engine: str = "deterministic",
 ) -> tuple[AnalysisResult, str]:
     """Parse, optionally redact, and analyze an uploaded trace file."""
 
@@ -178,6 +180,29 @@ def analyze_trace_file(
         engine="deterministic-codebook",
     )
     narrative_text = render_redacted_narrative(messages)
+
+    if analysis_engine != "deterministic":
+        if analysis_engine not in MODEL_CHOICES:
+            result.model_notes.append(
+                f"Unknown analysis engine {analysis_engine!r}; deterministic analysis was returned."
+            )
+        else:
+            try:
+                assist = run_model_assist(
+                    engine=analysis_engine,
+                    result=result,
+                    narrative_text=narrative_text,
+                )
+            except Exception as exc:
+                result.model_notes.append(
+                    "Small-model assist was requested but unavailable: "
+                    f"{type(exc).__name__}: {exc}. Deterministic analysis was returned."
+                )
+            else:
+                result.engine = f"deterministic-codebook + {assist.model_id}"
+                result.model_memo = assist.memo
+                result.model_notes.append(assist.note)
+
     return result, narrative_text
 
 

app.py

@@ -10,6 +10,7 @@ from typing import Any
 import gradio as gr
 
 from analyzer import analyze_trace_file
+from model_runtime import MODEL_CHOICES
 from parser import TraceParseError
 from report_renderer import render_report
 
@@ -99,6 +100,7 @@ def analyze_trace(
     redact_secrets: bool = True,
     ignore_tool_calls: bool = True,
     report_style: str = "field_notes",
+    analysis_engine: str = "deterministic",
 ) -> tuple[str, dict[str, Any], str, str, str]:
     """Gradio-callable analysis endpoint."""
 
@@ -113,6 +115,7 @@ def analyze_trace(
             redact_secrets=redact_secrets,
             ignore_tool_calls=ignore_tool_calls,
             report_style=report_style,
+            analysis_engine=analysis_engine,
         )
     except TraceParseError as exc:
         raise gr.Error(str(exc)) from exc
@@ -194,6 +197,14 @@ with gr.Blocks(
                 label="Report style",
                 interactive=False,
             )
+            analysis_engine = gr.Radio(
+                choices=[
+                    (str(choice["label"]), key)
+                    for key, choice in MODEL_CHOICES.items()
+                ],
+                value="deterministic",
+                label="Analysis engine",
+            )
             analyze_button = gr.Button("Analyze My Trace", variant="primary")
         with gr.Column(scale=2):
             gr.Markdown(SESSION_PATHS_MD)
@@ -208,8 +219,24 @@ with gr.Blocks(
         )
 
     gr.Examples(
-        examples=[["examples/sample_trace_redacted.jsonl", True, True, True, "field_notes"]],
-        inputs=[trace_input, include_user_context, redact_secrets, ignore_tool_calls, report_style],
+        examples=[
+            [
+                "examples/sample_trace_redacted.jsonl",
+                True,
+                True,
+                True,
+                "field_notes",
+                "deterministic",
+            ]
+        ],
+        inputs=[
+            trace_input,
+            include_user_context,
+            redact_secrets,
+            ignore_tool_calls,
+            report_style,
+            analysis_engine,
+        ],
         label="Try a redacted sample trace",
     )
 
@@ -229,6 +256,7 @@ with gr.Blocks(
             redact_secrets,
             ignore_tool_calls,
             report_style,
+            analysis_engine,
         ],
         outputs=[
             report_output,

model_runtime.py

@@ -0,0 +1,153 @@
+"""Optional small-model assistance through Hugging Face Inference Providers."""
+
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass
+from typing import Any, Protocol
+
+from schemas import AnalysisResult
+
+
+PRIMARY_MODEL_ID = "nvidia/NVIDIA-Nemotron-3-Nano-30B-A3B-BF16"
+QUICK_MODEL_ID = "Qwen/Qwen3.5-9B"
+
+MODEL_CHOICES = {
+    "deterministic": {
+        "label": "Deterministic field notes",
+        "model_id": None,
+    },
+    "nemotron": {
+        "label": "Small-model assist: NVIDIA Nemotron 3 Nano 30B-A3B",
+        "model_id": PRIMARY_MODEL_ID,
+    },
+    "qwen": {
+        "label": "Quick small-model assist: Qwen3.5 9B",
+        "model_id": QUICK_MODEL_ID,
+    },
+}
+
+
+class ChatClient(Protocol):
+    def chat_completion(self, *args: Any, **kwargs: Any) -> Any:
+        ...
+
+
+@dataclass(slots=True)
+class ModelAssistResult:
+    model_id: str
+    memo: dict[str, Any]
+    note: str
+
+
+def model_id_for_engine(engine: str) -> str | None:
+    choice = MODEL_CHOICES.get(engine)
+    if not choice:
+        return None
+    model_id = choice["model_id"]
+    return str(model_id) if model_id else None
+
+
+def run_model_assist(
+    *,
+    engine: str,
+    result: AnalysisResult,
+    narrative_text: str,
+    client: ChatClient | None = None,
+) -> ModelAssistResult:
+    """Ask the selected small model for a concise memo grounded in visible text."""
+
+    model_id = model_id_for_engine(engine)
+    if not model_id:
+        raise ValueError(f"No model is configured for analysis engine {engine!r}.")
+
+    prompt = build_model_prompt(result, narrative_text)
+    if client is None:
+        from huggingface_hub import InferenceClient
+
+        inference_client = InferenceClient(
+            model=model_id,
+            provider=os.getenv("TRACE_FIELD_NOTES_INFERENCE_PROVIDER") or None,
+            token=os.getenv("HF_TOKEN") or None,
+            timeout=float(os.getenv("TRACE_FIELD_NOTES_MODEL_TIMEOUT", "45")),
+        )
+    else:
+        inference_client = client
+    response = inference_client.chat_completion(
+        messages=[
+            {
+                "role": "system",
+                "content": (
+                    "You analyze visible coding-agent narrative messages. "
+                    "Do not infer hidden reasoning. Return JSON only."
+                ),
+            },
+            {"role": "user", "content": prompt},
+        ],
+        model=model_id,
+        max_tokens=900,
+        temperature=0.2,
+        response_format={"type": "json_object"},
+    )
+    content = extract_chat_content(response)
+    memo = parse_model_json(content)
+    return ModelAssistResult(
+        model_id=model_id,
+        memo=memo,
+        note=f"Small-model assist completed with {model_id}.",
+    )
+
+
+def build_model_prompt(result: AnalysisResult, narrative_text: str) -> str:
+    deterministic_json = json.dumps(result.to_dict(), ensure_ascii=False, indent=2)
+    narrative_excerpt = narrative_text[:12000]
+    return f"""Use the deterministic codebook analysis and redacted visible narrative below.
+
+Return JSON with exactly these keys:
+- executive_memo: 4-6 sentences for a developer
+- detour_memo: 2-4 sentences about productive detours vs wandering
+- outcome_audit_memo: 2-4 sentences about completion claims and caveats
+- caveats: array of short strings
+
+Rules:
+- Analyze only visible narrative messages.
+- Do not claim to know hidden reasoning.
+- Cite episode IDs where useful.
+- Do not include raw secrets, tool outputs, or long quotes.
+
+Deterministic analysis:
+{deterministic_json}
+
+Redacted narrative excerpt:
+{narrative_excerpt}
+"""
+
+
+def extract_chat_content(response: Any) -> str:
+    try:
+        content = response.choices[0].message.content
+    except (AttributeError, IndexError, TypeError) as exc:
+        raise ValueError("Model response did not contain chat completion content.") from exc
+    if not isinstance(content, str) or not content.strip():
+        raise ValueError("Model response content was empty.")
+    return content
+
+
+def parse_model_json(content: str) -> dict[str, Any]:
+    try:
+        parsed = json.loads(content)
+    except json.JSONDecodeError as exc:
+        raise ValueError("Model response was not valid JSON.") from exc
+
+    required = {
+        "executive_memo": str,
+        "detour_memo": str,
+        "outcome_audit_memo": str,
+        "caveats": list,
+    }
+    for key, expected_type in required.items():
+        if key not in parsed or not isinstance(parsed[key], expected_type):
+            raise ValueError(f"Model response missing {key!r} as {expected_type.__name__}.")
+    parsed["caveats"] = [str(item) for item in parsed["caveats"][:6]]
+    return parsed

report_renderer.py

@@ -26,6 +26,7 @@ def render_report(result: AnalysisResult) -> str:
     sections = [
         render_header(result),
         render_executive_summary(result),
+        render_model_memo(result),
         render_timeline(result.episodes),
         render_difficulty_map(result.episodes),
         render_detour_analysis(result.episodes),
@@ -71,6 +72,25 @@ def render_executive_summary(result: AnalysisResult) -> str:
     )
 
 
+def render_model_memo(result: AnalysisResult) -> str:
+    if not result.model_memo and not result.model_notes:
+        return ""
+
+    lines = ["## Small-Model Memo"]
+    if result.model_memo:
+        lines.append(result.model_memo.get("executive_memo", ""))
+        lines.append(f"**Detours:** {result.model_memo.get('detour_memo', '')}")
+        lines.append(f"**Outcome audit:** {result.model_memo.get('outcome_audit_memo', '')}")
+        caveats = result.model_memo.get("caveats") or []
+        if caveats:
+            lines.append("**Model caveats:**")
+            lines.extend(f"- {caveat}" for caveat in caveats)
+    if result.model_notes:
+        lines.append("**Model notes:**")
+        lines.extend(f"- {note}" for note in result.model_notes)
+    return "\n\n".join(line for line in lines if line)
+
+
 def render_timeline(episodes: list[DifficultyEpisode]) -> str:
     if not episodes:
         return "## Journey Timeline\n\nNo difficulty timeline was detected."

schemas.py

@@ -147,6 +147,8 @@ class AnalysisResult:
     narrative_message_count: int
     redaction_count: int = 0
     engine: str = "deterministic-codebook"
+    model_notes: list[str] = field(default_factory=list)
+    model_memo: dict[str, Any] = field(default_factory=dict)
 
     def to_dict(self) -> dict[str, Any]:
         return {
@@ -159,4 +161,6 @@ class AnalysisResult:
             "narrative_message_count": self.narrative_message_count,
             "redaction_count": self.redaction_count,
             "engine": self.engine,
+            "model_notes": self.model_notes,
+            "model_memo": self.model_memo,
         }

tests/test_model_runtime.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import json
+import types
+import unittest
+from pathlib import Path
+
+from analyzer import analyze_trace_file
+from model_runtime import PRIMARY_MODEL_ID, parse_model_json, run_model_assist
+
+
+class FakeChatClient:
+    def chat_completion(self, *args, **kwargs):
+        self.kwargs = kwargs
+        content = json.dumps(
+            {
+                "executive_memo": "The trace shows a visible upload-boundary correction.",
+                "detour_memo": "E01 narrows scope instead of changing the parser.",
+                "outcome_audit_memo": "The agent keeps a deployment caveat visible.",
+                "caveats": ["Model memo is based only on redacted narrative."],
+            }
+        )
+        return types.SimpleNamespace(
+            choices=[
+                types.SimpleNamespace(
+                    message=types.SimpleNamespace(content=content),
+                )
+            ]
+        )
+
+
+class ModelRuntimeTests(unittest.TestCase):
+    def test_parse_model_json_validates_required_shape(self) -> None:
+        memo = parse_model_json(
+            json.dumps(
+                {
+                    "executive_memo": "summary",
+                    "detour_memo": "detour",
+                    "outcome_audit_memo": "audit",
+                    "caveats": ["one"],
+                }
+            )
+        )
+
+        self.assertEqual(memo["executive_memo"], "summary")
+        self.assertEqual(memo["caveats"], ["one"])
+
+    def test_run_model_assist_uses_selected_model(self) -> None:
+        result, narrative = analyze_trace_file(Path("examples/sample_trace_redacted.jsonl"))
+        client = FakeChatClient()
+
+        assist = run_model_assist(
+            engine="nemotron",
+            result=result,
+            narrative_text=narrative,
+            client=client,
+        )
+
+        self.assertEqual(assist.model_id, PRIMARY_MODEL_ID)
+        self.assertIn("upload-boundary", assist.memo["executive_memo"])
+        self.assertEqual(client.kwargs["model"], PRIMARY_MODEL_ID)
+
+    def test_analyzer_records_unknown_engine_note(self) -> None:
+        result, _ = analyze_trace_file(
+            Path("examples/sample_trace_redacted.jsonl"),
+            analysis_engine="missing-engine",
+        )
+
+        self.assertTrue(result.model_notes)
+        self.assertIn("Unknown analysis engine", result.model_notes[0])
+
+
+if __name__ == "__main__":
+    unittest.main()