feat: configurable interview panel size with stratified sampling

app/config.py

@@ -25,6 +25,26 @@ def _get_bool_env(name: str, default: bool = False) -> bool:
     return value.strip().lower() in {"1", "true", "yes", "on"}
 
 
+def _get_int_env(name: str, default: int, min_value: int = 1) -> int:
+    """Parse an int env var, falling back (with a warning) on bad values.
+
+    Raising at import time from the Config class body produces an opaque
+    crash before validate() can run, so malformed values degrade to the
+    default instead.
+    """
+    raw = os.environ.get(name)
+    if raw is None or not raw.strip():
+        return default
+    try:
+        value = int(raw.strip())
+    except ValueError:
+        import logging
+        logging.getLogger("mirofish.config").warning(
+            "Ignoring invalid %s=%r; using default %s", name, raw, default)
+        return default
+    return max(min_value, value)
+
+
 def _resolve_path(default_path: str, env_name: str) -> str:
     raw_value = os.environ.get(env_name, default_path)
     return os.path.abspath(raw_value)
@@ -40,6 +60,15 @@ class Config:
 
     LLM_PROVIDER = os.environ.get("LLM_PROVIDER", "claude-cli").strip().lower()
 
+    # Post-simulation interviews for the final report. Each interviewed agent
+    # costs one LLM call per platform, so the default stays small; raise it to
+    # build a larger synthetic panel (>=6 also enables stratified sampling of
+    # vocal / relevant / silent agents). EXTRA_QUESTIONS (';'-separated) are
+    # appended to the generated questionnaire — useful for closed questions
+    # whose answers can be coded into distributions.
+    INTERVIEW_MAX_AGENTS = _get_int_env("MIROFISH_INTERVIEW_MAX_AGENTS", 5)
+    INTERVIEW_EXTRA_QUESTIONS = os.environ.get("MIROFISH_INTERVIEW_EXTRA_QUESTIONS", "").strip()
+
     DATA_DIR = _resolve_path(os.path.join(os.path.dirname(__file__), "../data/graphs"), "DATA_DIR")
 
     MAX_CONTENT_LENGTH = 50 * 1024 * 1024

app/services/graph_tools.py

@@ -11,6 +11,7 @@
 from .graph_db import GraphDatabase
 from .graph_storage import GraphStorage
 
+from ..config import Config
 from ..utils.logger import get_logger
 from ..utils.llm_client import LLMClient
 
@@ -296,9 +297,16 @@ def interview_agents(
         simulation_requirement: str = "",
         max_agents: int = 5,
         custom_questions: List[str] = None,
+        activity_by_agent: Optional[Dict[int, int]] = None,
     ) -> InterviewResult:
-        """Interview simulated agents via OASIS IPC."""
+        """Interview simulated agents via OASIS IPC.
+
+        activity_by_agent maps agent index -> action count; when provided (and
+        the panel is large enough) it enables stratified sampling so the panel
+        mixes vocal, relevant and silent agents instead of only top posters.
+        """
         from .simulation_runner import SimulationRunner
+        from ..utils.llm_client import CLI_CALL_TIMEOUT_SECONDS
 
         result = InterviewResult(
             interview_topic=interview_requirement,
@@ -317,6 +325,7 @@ def interview_agents(
             interview_requirement=interview_requirement,
             simulation_requirement=simulation_requirement,
             max_agents=max_agents,
+            activity_by_agent=activity_by_agent,
         )
         result.selected_agents = selected_agents
         result.selection_reasoning = selection_reasoning
@@ -328,6 +337,11 @@ def interview_agents(
                 selected_agents=selected_agents,
             )
 
+        if Config.INTERVIEW_EXTRA_QUESTIONS:
+            result.interview_questions.extend(
+                q.strip() for q in Config.INTERVIEW_EXTRA_QUESTIONS.split(";") if q.strip()
+            )
+
         combined_prompt = "\n".join([f"{i+1}. {q}" for i, q in enumerate(result.interview_questions)])
         optimized_prompt = (
             "You are being interviewed. Please draw on your persona, all past memories, and actions "
@@ -348,20 +362,40 @@ def interview_agents(
                 for idx in selected_indices
             ]
 
-            api_result = SimulationRunner.interview_agents_batch(
-                simulation_id=simulation_id,
-                interviews=interviews_request,
-                platform=None,
-                timeout=180.0,
-            )
-
-            if not api_result.get("success", False):
-                result.summary = f"Interview API call failed: {api_result.get('error', 'Unknown error')}"
+            # Chunked batches, one full CLI-call ceiling per interviewed agent
+            # (a fixed 180s budget starved whole batches to 0/N). Chunking
+            # keeps each IPC transaction small so a single hung interview
+            # loses only its chunk instead of the entire panel, and bounds
+            # the worst-case wall clock per transaction.
+            CHUNK_SIZE = 5
+            results_dict: Dict[str, Any] = {}
+            any_chunk_ok = False
+            for start in range(0, len(interviews_request), CHUNK_SIZE):
+                chunk = interviews_request[start:start + CHUNK_SIZE]
+                try:
+                    api_result = SimulationRunner.interview_agents_batch(
+                        simulation_id=simulation_id,
+                        interviews=chunk,
+                        platform=None,
+                        timeout=float(CLI_CALL_TIMEOUT_SECONDS * len(chunk)),
+                    )
+                except Exception as chunk_exc:
+                    logger.warning(f"Interview chunk {start // CHUNK_SIZE + 1} failed (non-fatal): {chunk_exc}")
+                    continue
+                if api_result.get("success", False):
+                    any_chunk_ok = True
+                    api_data = api_result.get("result", {})
+                    if isinstance(api_data, dict):
+                        results_dict.update(api_data.get("results", {}) or {})
+                else:
+                    logger.warning(
+                        f"Interview chunk {start // CHUNK_SIZE + 1} API failure: "
+                        f"{api_result.get('error', 'Unknown error')}")
+
+            if not any_chunk_ok:
+                result.summary = "Interview API call failed for every chunk"
                 return result
 
-            api_data = api_result.get("result", {})
-            results_dict = api_data.get("results", {}) if isinstance(api_data, dict) else {}
-
             for i, agent_idx in enumerate(selected_indices):
                 agent = selected_agents[i]
                 agent_name = agent.get("realname", agent.get("username", f"Agent_{agent_idx}"))
@@ -473,7 +507,43 @@ def _load_agent_profiles(self, simulation_id: str) -> List[Dict[str, Any]]:
 
         return []
 
-    def _select_agents_for_interview(self, profiles, interview_requirement, simulation_requirement, max_agents):
+    def _select_agents_for_interview(self, profiles, interview_requirement, simulation_requirement, max_agents,
+                                     activity_by_agent=None):
+        # Stratified sampling for larger panels: pure top-poster selection
+        # over-represents the vocal minority, while silent agents often hold
+        # the most informative objections (they saw everything and engaged
+        # with none of it). Quotas: ~40% most vocal, ~30% LLM-picked for
+        # relevance from the remainder, ~30% silent non-participants.
+        if activity_by_agent and max_agents >= 6 and len(profiles) > max_agents:
+            n_vocal = max(1, round(max_agents * 0.4))
+            n_silent = max(1, round(max_agents * 0.3))
+            n_llm = max(0, max_agents - n_vocal - n_silent)
+
+            by_activity = sorted(range(len(profiles)), key=lambda i: -activity_by_agent.get(i, 0))
+            vocal = [i for i in by_activity if activity_by_agent.get(i, 0) > 0][:n_vocal]
+            silent_pool = [i for i in range(len(profiles))
+                           if activity_by_agent.get(i, 0) == 0 and i not in vocal]
+            stride = max(1, len(silent_pool) // n_silent) if silent_pool else 1
+            silent = silent_pool[::stride][:n_silent]
+
+            chosen = set(vocal) | set(silent)
+            remainder_idx = [i for i in range(len(profiles)) if i not in chosen]
+            llm_sel = []
+            if n_llm > 0 and remainder_idx:
+                remainder = [profiles[i] for i in remainder_idx]
+                _, rel_idx, _ = self._select_agents_for_interview(
+                    remainder, interview_requirement, simulation_requirement, n_llm
+                )
+                llm_sel = [remainder_idx[i] for i in rel_idx]
+
+            indices = vocal + llm_sel + silent
+            agents = [profiles[i] for i in indices]
+            reasoning = (
+                f"Stratified panel of {len(indices)}: {len(vocal)} most vocal, "
+                f"{len(llm_sel)} LLM-picked for relevance, {len(silent)} silent non-participants"
+            )
+            return agents, indices, reasoning
+
         agent_summaries = [
             {
                 "index": i,
@@ -502,7 +572,14 @@ def _select_agents_for_interview(self, profiles, interview_requirement, simulati
                 ],
                 temperature=0.3,
             )
-            indices = response.get("selected_indices", [])[:max_agents]
+            # Dedupe preserving order and coerce/validate: the LLM sometimes
+            # returns duplicates or stringified indices
+            raw_indices = response.get("selected_indices", [])
+            indices = list(dict.fromkeys(
+                int(i) for i in raw_indices
+                if isinstance(i, (int, float, str)) and str(i).strip().lstrip("-").isdigit()
+                and 0 <= int(i) < len(profiles)
+            ))[:max_agents]
             reasoning = response.get("reasoning", "Selected based on relevance")
             agents = [profiles[i] for i in indices if 0 <= i < len(profiles)]
             valid_indices = [i for i in indices if 0 <= i < len(profiles)]

app/services/report_agent.py

@@ -185,16 +185,22 @@ def generate_report_fast(
                     f"[Round {a.round_num}] [{a.platform}] {a.agent_name} ({a.action_type}): {content_preview}"
                 )
 
-            # Interview top agents
+            # Interview a panel of agents (size configurable; with activity
+            # data the selector stratifies vocal / relevant / silent agents)
             interview_text = ""
             try:
-                top_agents = sorted(agent_stats, key=lambda x: x.get("total_actions", 0), reverse=True)[:5]
-                if top_agents:
+                if agent_stats:
+                    activity_by_agent = {
+                        int(s["agent_id"]): int(s.get("total_actions", 0))
+                        for s in agent_stats
+                        if s.get("agent_id") is not None
+                    }
                     interview_result = self.graph_tools.interview_agents(
                         simulation_id=self.simulation_id,
                         interview_requirement=f"What is your prediction for how {self.simulation_requirement}? What surprised you during the simulation?",
                         simulation_requirement=self.simulation_requirement,
-                        max_agents=5,
+                        max_agents=Config.INTERVIEW_MAX_AGENTS,
+                        activity_by_agent=activity_by_agent,
                     )
                     if hasattr(interview_result, "to_text"):
                         interview_text = interview_result.to_text()

app/utils/llm_client.py

@@ -2,10 +2,13 @@
 LLM Client — CLI-only providers (Claude Code, Codex).
 """
 
+import functools
 import json
 import re
+import shutil
 import subprocess
 import time
+import tempfile
 from typing import Optional, Dict, Any, List
 
 from ..config import Config
@@ -16,6 +19,39 @@
 MAX_RETRIES = 3
 RETRY_BASE_DELAY = 2.0  # seconds
 
+# Wall-clock ceiling for a single CLI provider call. Long prompts (large
+# knowledge-graph context, interview batches) can legitimately take several
+# minutes through a subprocess-based provider. Callers that wait on multiple
+# sequential calls (e.g. interview batches) should budget per call using this
+# constant rather than hardcoding their own number.
+CLI_CALL_TIMEOUT_SECONDS = 600
+
+
+class LLMTimeoutError(RuntimeError):
+    """A CLI call exhausted its wall-clock ceiling.
+
+    Kept distinct from generic RuntimeError so the retry loop can skip it:
+    a prompt that exhausts the full ceiling will almost certainly exhaust it
+    again, and retrying only multiplies the wait (3x600s + backoff ~= 30 min
+    blocked on a single call).
+    """
+
+
+@functools.lru_cache(maxsize=None)
+def _resolve_cli(binary: str) -> str:
+    """Resolve a provider CLI to an absolute path, failing with a clear error.
+
+    On Windows, invoking a bare name only works for .exe; npm-style .cmd shims
+    need the resolved path. shutil.which also gives us a clean early failure
+    instead of an opaque FileNotFoundError mid-simulation.
+    """
+    resolved = shutil.which(binary)
+    if resolved is None:
+        raise RuntimeError(
+            f"'{binary}' CLI not found on PATH. Install it and log in before running simulations."
+        )
+    return resolved
+
 
 class LLMClient:
     """LLM Client — supports claude-cli and codex-cli."""
@@ -59,6 +95,9 @@ def chat(
                 if self.provider == "codex-cli":
                     return self._chat_codex_cli(messages, temperature, max_tokens, response_format)
                 return self._chat_claude_cli(messages, temperature, max_tokens, response_format)
+            except LLMTimeoutError:
+                # No retry: see LLMTimeoutError docstring
+                raise
             except RuntimeError as exc:
                 last_error = exc
                 if attempt < MAX_RETRIES - 1:
@@ -92,15 +131,20 @@ def _chat_claude_cli(
 
         try:
             result = subprocess.run(
-                ["claude", "-p", "--output-format", "json", prompt],
-                capture_output=True, text=True, timeout=300,
-                cwd="/tmp"
+                [_resolve_cli("claude"), "-p", "--output-format", "json"],
+                input=prompt,
+                capture_output=True, text=True, timeout=CLI_CALL_TIMEOUT_SECONDS,
+                encoding="utf-8", errors="replace",
+                cwd=tempfile.gettempdir()
             )
 
             if result.returncode != 0:
                 logger.error(f"Claude CLI error: {result.stderr[:200]}")
                 raise RuntimeError(f"Claude CLI failed: {result.stderr[:200]}")
 
+            if "�" in result.stdout:
+                logger.warning("Claude CLI output contained undecodable bytes; replaced with U+FFFD")
+
             try:
                 output = json.loads(result.stdout)
                 content = output.get("result", result.stdout)
@@ -110,7 +154,7 @@ def _chat_claude_cli(
             return self._clean_content(content)
 
         except subprocess.TimeoutExpired:
-            raise RuntimeError("Claude CLI timed out after 300s")
+            raise LLMTimeoutError(f"Claude CLI timed out after {CLI_CALL_TIMEOUT_SECONDS}s")
 
     def _chat_codex_cli(
         self,
@@ -137,16 +181,20 @@ def _chat_codex_cli(
 
         try:
             result = subprocess.run(
-                ["codex", "exec", "--skip-git-repo-check"],
+                [_resolve_cli("codex"), "exec", "--skip-git-repo-check"],
                 input=prompt,
-                capture_output=True, text=True, timeout=180,
-                cwd="/tmp"
+                capture_output=True, text=True, timeout=CLI_CALL_TIMEOUT_SECONDS,
+                encoding="utf-8", errors="replace",
+                cwd=tempfile.gettempdir()
             )
 
             if result.returncode != 0:
                 logger.error(f"Codex CLI error: {result.stderr[:200]}")
                 raise RuntimeError(f"Codex CLI failed: {result.stderr[:200]}")
 
+            if "�" in result.stdout:
+                logger.warning("Codex CLI output contained undecodable bytes; replaced with U+FFFD")
+
             raw = result.stdout.strip()
             parts = raw.split("\ncodex\n")
             if len(parts) > 1:
@@ -163,7 +211,7 @@ def _chat_codex_cli(
             return self._clean_content(content)
 
         except subprocess.TimeoutExpired:
-            raise RuntimeError("Codex CLI timed out after 180s")
+            raise LLMTimeoutError(f"Codex CLI timed out after {CLI_CALL_TIMEOUT_SECONDS}s")
 
     def chat_json(
         self,

pyproject.toml

@@ -43,3 +43,9 @@ dev = [
 
 [tool.hatch.build.targets.wheel]
 packages = ["app"]
+
+[tool.uv]
+# rapidfuzz >=3.14 ships no Windows cp312 wheels (and no sdist in-range), which
+# breaks `uv sync` under requires-python <3.13 on Windows. Scoped to win32 so
+# macOS/Linux keep the resolver's pick; drop once upstream publishes wheels.
+override-dependencies = ["rapidfuzz==3.13.0; sys_platform == 'win32'"]