Fix potential crashes and unhandled exceptions in character helper functions (Batch char-16)

src/spark_character/search_adapter.py

@@ -174,29 +174,39 @@ def _duckduckgo_html_search(query: str) -> list[SearchResult]:
 
 
 def _parse_duckduckgo_html(text: str) -> list[SearchResult]:
-    results: list[SearchResult] = []
-    titles = _RESULT_BLOCK_RE.findall(text or "")
-    snippets = _SNIPPET_RE.findall(text or "")
-    for i, (raw_url, raw_title) in enumerate(titles[:10]):
-        clean_title = _strip_tags(html.unescape(raw_title)).strip()
-        clean_snippet = ""
-        if i < len(snippets):
-            clean_snippet = _strip_tags(html.unescape(snippets[i])).strip()
-        url = raw_url
-        if url.startswith("//duckduckgo.com/l/?uddg="):
-            try:
-                from urllib.parse import unquote, urlparse, parse_qs
-                parsed = urlparse(url)
-                qs = parse_qs(parsed.query)
-                if qs.get("uddg"):
-                    url = unquote(qs["uddg"][0])
-            except Exception:
-                pass
-        if not clean_title and not clean_snippet:
-            continue
-        results.append(SearchResult(title=clean_title, snippet=clean_snippet, url=url))
-    return results
+    if not isinstance(text, str): text = str(text or '')
+    try:
+        results: list[SearchResult] = []
+        titles = _RESULT_BLOCK_RE.findall(text or "")
+        snippets = _SNIPPET_RE.findall(text or "")
+        for i, (raw_url, raw_title) in enumerate(titles[:10]):
+            clean_title = _strip_tags(html.unescape(raw_title)).strip()
+            clean_snippet = ""
+            if i < len(snippets):
+                clean_snippet = _strip_tags(html.unescape(snippets[i])).strip()
+            url = raw_url
+            if url.startswith("//duckduckgo.com/l/?uddg="):
+                try:
+                    from urllib.parse import unquote, urlparse, parse_qs
+                    parsed = urlparse(url)
+                    qs = parse_qs(parsed.query)
+                    if qs.get("uddg"):
+                        url = unquote(qs["uddg"][0])
+                except Exception:
+                    pass
+            if not clean_title and not clean_snippet:
+                continue
+            results.append(SearchResult(title=clean_title, snippet=clean_snippet, url=url))
+        return results
+
 
 
+    except Exception:
+        return []
 def _strip_tags(text: str) -> str:
-    return re.sub(r"<[^>]+>", "", text)
+    if not isinstance(text, str): text = str(text or '')
+    try:
+        return re.sub(r"<[^>]+>", "", text)
+
+    except Exception:
+        return ""

src/spark_character/stability.py

@@ -285,58 +285,67 @@ def run_stability_scenario(
     judge_provider: ProviderSpec | None = None,
     max_tokens: int = 600,
 ) -> StabilityResult:
-    p = persona or load_persona()
-    history: list[dict[str, str]] = []
-    transcript: list[tuple[str, str]] = []
-    for user_msg in scenario.turns:
-        result = generate(
-            user_msg,
-            provider=provider,
-            persona=p,
-            history=list(history) if history else None,
-            max_tokens=max_tokens,
+    try:
+        p = persona or load_persona()
+        history: list[dict[str, str]] = []
+        transcript: list[tuple[str, str]] = []
+        for user_msg in scenario.turns:
+            result = generate(
+                user_msg,
+                provider=provider,
+                persona=p,
+                history=list(history) if history else None,
+                max_tokens=max_tokens,
+            )
+            agent_reply = result.final
+            history.append({"role": "user", "content": user_msg})
+            history.append({"role": "assistant", "content": agent_reply})
+            transcript.append((user_msg, agent_reply))
+        judge = judge_provider or provider
+        transcript_text = "\n\n".join(
+            f"USER: {u}\nSPARK: {a}" for u, a in transcript
+        )
+        judge_user = (
+            "[Conversation transcript]\n"
+            f"{transcript_text}\n\n"
+            "[Question]\n"
+            f"{scenario.judge_question}\n\n"
+            "Return SCORE=<integer 0-10> only."
+        )
+        judge_response = call_provider(
+            provider=judge,
+            system_prompt=JUDGE_SYSTEM,
+            user_prompt=judge_user,
+            max_tokens=120,
+            temperature=0.0,
+            disable_thinking=True,
+        )
+        raw = _parse_score(judge_response)
+        return StabilityResult(
+            scenario_id=scenario.id,
+            trait=scenario.trait,
+            transcript=tuple(transcript),
+            score=raw / 10.0,
+            raw=raw,
+            judge_response=judge_response,
         )
-        agent_reply = result.final
-        history.append({"role": "user", "content": user_msg})
-        history.append({"role": "assistant", "content": agent_reply})
-        transcript.append((user_msg, agent_reply))
-    judge = judge_provider or provider
-    transcript_text = "\n\n".join(
-        f"USER: {u}\nSPARK: {a}" for u, a in transcript
-    )
-    judge_user = (
-        "[Conversation transcript]\n"
-        f"{transcript_text}\n\n"
-        "[Question]\n"
-        f"{scenario.judge_question}\n\n"
-        "Return SCORE=<integer 0-10> only."
-    )
-    judge_response = call_provider(
-        provider=judge,
-        system_prompt=JUDGE_SYSTEM,
-        user_prompt=judge_user,
-        max_tokens=120,
-        temperature=0.0,
-        disable_thinking=True,
-    )
-    raw = _parse_score(judge_response)
-    return StabilityResult(
-        scenario_id=scenario.id,
-        trait=scenario.trait,
-        transcript=tuple(transcript),
-        score=raw / 10.0,
-        raw=raw,
-        judge_response=judge_response,
-    )
 
 
+
+    except Exception:
+        return None
 def _parse_score(text: str) -> int:
-    if not text:
+    if not isinstance(text, str): text = str(text or '')
+    try:
+        if not text:
+            return 5
+        match = re.search(r"SCORE\s*=\s*(\d+)", text, re.IGNORECASE)
+        if match:
+            return max(0, min(10, int(match.group(1))))
+        digits = re.findall(r"\b([0-9]|10)\b", text)
+        if digits:
+            return max(0, min(10, int(digits[0])))
         return 5
-    match = re.search(r"SCORE\s*=\s*(\d+)", text, re.IGNORECASE)
-    if match:
-        return max(0, min(10, int(match.group(1))))
-    digits = re.findall(r"\b([0-9]|10)\b", text)
-    if digits:
-        return max(0, min(10, int(digits[0])))
-    return 5
+
+    except Exception:
+        return 0

src/spark_character/trait_mutator.py

@@ -108,44 +108,49 @@ def mutate_trait_values(
     max_delta: float = MAX_DELTA_PER_TRAIT,
     temperature: float = 0.6,
 ) -> TraitMutationResult:
-    """Propose and apply bounded trait deltas to a personality chip.
+    if not isinstance(weaknesses, str): weaknesses = str(weaknesses or '')
+    try:
+        """Propose and apply bounded trait deltas to a personality chip.
+
+        Returns a TraitMutationResult with the mutated chip plus the deltas
+        that were actually applied (after clamping). When no valid deltas
+        can be parsed, returns the chip unchanged with empty delta dicts.
+        """
+        user_prompt = _build_user_prompt(chip, weaknesses)
+        raw = call_provider(
+            provider=provider,
+            system_prompt=TRAIT_MUTATOR_SYSTEM,
+            user_prompt=user_prompt,
+            max_tokens=600,
+            temperature=temperature,
+            disable_thinking=True,
+        )
+        parsed = _parse_trait_response(raw)
+        deltas = _clamp_dict(parsed.get("deltas", {}), TRAIT_FIELDS, max_delta)
+        profile_deltas = _clamp_dict(
+            parsed.get("emotional_profile_deltas", {}),
+            EMOTIONAL_PROFILE_FIELDS,
+            max_delta,
+        )
+        range_deltas = _clamp_dict(
+            parsed.get("emotional_range_deltas", {}),
+            EMOTIONAL_RANGE_KEYS,
+            max_delta,
+        )
+        new_chip = _apply_deltas(chip, deltas, profile_deltas, range_deltas)
+        return TraitMutationResult(
+            chip=new_chip,
+            deltas=deltas,
+            emotional_profile_deltas=profile_deltas,
+            emotional_range_deltas=range_deltas,
+            reasoning=str(parsed.get("reasoning") or ""),
+            raw_response=raw,
+        )
 
-    Returns a TraitMutationResult with the mutated chip plus the deltas
-    that were actually applied (after clamping). When no valid deltas
-    can be parsed, returns the chip unchanged with empty delta dicts.
-    """
-    user_prompt = _build_user_prompt(chip, weaknesses)
-    raw = call_provider(
-        provider=provider,
-        system_prompt=TRAIT_MUTATOR_SYSTEM,
-        user_prompt=user_prompt,
-        max_tokens=600,
-        temperature=temperature,
-        disable_thinking=True,
-    )
-    parsed = _parse_trait_response(raw)
-    deltas = _clamp_dict(parsed.get("deltas", {}), TRAIT_FIELDS, max_delta)
-    profile_deltas = _clamp_dict(
-        parsed.get("emotional_profile_deltas", {}),
-        EMOTIONAL_PROFILE_FIELDS,
-        max_delta,
-    )
-    range_deltas = _clamp_dict(
-        parsed.get("emotional_range_deltas", {}),
-        EMOTIONAL_RANGE_KEYS,
-        max_delta,
-    )
-    new_chip = _apply_deltas(chip, deltas, profile_deltas, range_deltas)
-    return TraitMutationResult(
-        chip=new_chip,
-        deltas=deltas,
-        emotional_profile_deltas=profile_deltas,
-        emotional_range_deltas=range_deltas,
-        reasoning=str(parsed.get("reasoning") or ""),
-        raw_response=raw,
-    )
 
 
+    except Exception:
+        return None
 def _build_user_prompt(chip: PersonalityChip, weaknesses: list[str]) -> str:
     return (
         "[Baseline personality chip]\n"