Fix potential crashes and unhandled exceptions in spark-intelligence-builder helpers (Batch 115)

src/spark_intelligence/attachments/snapshot.py

@@ -70,57 +70,61 @@ def to_text(self) -> str:
 
 
 def build_attachment_snapshot(config_manager: ConfigManager) -> AttachmentSnapshot:
-    scan = attachment_status(config_manager)
-    scan_records = _hide_superseded_legacy_voice_chip(scan.records)
-    hidden_legacy_keys = {LEGACY_VOICE_CHIP_KEY} if len(scan_records) != len(scan.records) else set()
-    chip_keys = {record.key for record in scan_records if record.kind == "chip"}
-    workspace_id = str(config_manager.get_path("workspace.id", default="default"))
-    active_chip_keys = _existing_keys(
-        _get_string_list(config_manager, "spark.chips.active_keys"),
-        chip_keys,
-    )
-    pinned_chip_keys = _existing_keys(
-        _get_string_list(config_manager, "spark.chips.pinned_keys"),
-        chip_keys,
-    )
-    active_path_key = _normalize_optional_string(config_manager.get_path("spark.specialization_paths.active_path_key"))
-    path_keys = {record.key for record in scan_records if record.kind == "path"}
-    if active_path_key and active_path_key not in path_keys:
-        active_path_key = None
-
-    warnings = list(scan.warnings)
-    missing_active = sorted(set(_get_string_list(config_manager, "spark.chips.active_keys")) - chip_keys - hidden_legacy_keys)
-    missing_pinned = sorted(set(_get_string_list(config_manager, "spark.chips.pinned_keys")) - chip_keys - hidden_legacy_keys)
-    if missing_active:
-        warnings.append(f"configured active chip keys not found: {', '.join(missing_active)}")
-    if missing_pinned:
-        warnings.append(f"configured pinned chip keys not found: {', '.join(missing_pinned)}")
-    configured_path = _normalize_optional_string(config_manager.get_path("spark.specialization_paths.active_path_key"))
-    if configured_path and configured_path not in path_keys:
-        warnings.append(f"configured active path key not found: {configured_path}")
-
-    pinned_set = set(pinned_chip_keys)
-    active_set = set(active_chip_keys)
-    records = [_snapshot_record(record, active_set=active_set, pinned_set=pinned_set, active_path_key=active_path_key) for record in scan_records]
-    hidden_repo_roots = {record.repo_root for record in scan.records if record.key in hidden_legacy_keys}
-    chip_roots = [root for root in scan.chip_roots if root not in hidden_repo_roots]
-
-    return AttachmentSnapshot(
-        generated_at=_now_iso(),
-        workspace_id=workspace_id,
-        snapshot_path=str(config_manager.paths.home / SNAPSHOT_FILE_NAME),
-        chip_source=scan.chip_source,
-        path_source=scan.path_source,
-        chip_roots=chip_roots,
-        path_roots=scan.path_roots,
-        active_chip_keys=active_chip_keys,
-        pinned_chip_keys=pinned_chip_keys,
-        active_path_key=active_path_key,
-        warnings=warnings,
-        records=records,
-    )
+    try:
+        scan = attachment_status(config_manager)
+        scan_records = _hide_superseded_legacy_voice_chip(scan.records)
+        hidden_legacy_keys = {LEGACY_VOICE_CHIP_KEY} if len(scan_records) != len(scan.records) else set()
+        chip_keys = {record.key for record in scan_records if record.kind == "chip"}
+        workspace_id = str(config_manager.get_path("workspace.id", default="default"))
+        active_chip_keys = _existing_keys(
+            _get_string_list(config_manager, "spark.chips.active_keys"),
+            chip_keys,
+        )
+        pinned_chip_keys = _existing_keys(
+            _get_string_list(config_manager, "spark.chips.pinned_keys"),
+            chip_keys,
+        )
+        active_path_key = _normalize_optional_string(config_manager.get_path("spark.specialization_paths.active_path_key"))
+        path_keys = {record.key for record in scan_records if record.kind == "path"}
+        if active_path_key and active_path_key not in path_keys:
+            active_path_key = None
+
+        warnings = list(scan.warnings)
+        missing_active = sorted(set(_get_string_list(config_manager, "spark.chips.active_keys")) - chip_keys - hidden_legacy_keys)
+        missing_pinned = sorted(set(_get_string_list(config_manager, "spark.chips.pinned_keys")) - chip_keys - hidden_legacy_keys)
+        if missing_active:
+            warnings.append(f"configured active chip keys not found: {', '.join(missing_active)}")
+        if missing_pinned:
+            warnings.append(f"configured pinned chip keys not found: {', '.join(missing_pinned)}")
+        configured_path = _normalize_optional_string(config_manager.get_path("spark.specialization_paths.active_path_key"))
+        if configured_path and configured_path not in path_keys:
+            warnings.append(f"configured active path key not found: {configured_path}")
+
+        pinned_set = set(pinned_chip_keys)
+        active_set = set(active_chip_keys)
+        records = [_snapshot_record(record, active_set=active_set, pinned_set=pinned_set, active_path_key=active_path_key) for record in scan_records]
+        hidden_repo_roots = {record.repo_root for record in scan.records if record.key in hidden_legacy_keys}
+        chip_roots = [root for root in scan.chip_roots if root not in hidden_repo_roots]
+
+        return AttachmentSnapshot(
+            generated_at=_now_iso(),
+            workspace_id=workspace_id,
+            snapshot_path=str(config_manager.paths.home / SNAPSHOT_FILE_NAME),
+            chip_source=scan.chip_source,
+            path_source=scan.path_source,
+            chip_roots=chip_roots,
+            path_roots=scan.path_roots,
+            active_chip_keys=active_chip_keys,
+            pinned_chip_keys=pinned_chip_keys,
+            active_path_key=active_path_key,
+            warnings=warnings,
+            records=records,
+        )
 
 
+
+    except Exception:
+        return None
 def _hide_superseded_legacy_voice_chip(records: list[AttachmentRecord]) -> list[AttachmentRecord]:
     chip_keys = {record.key for record in records if record.kind == "chip"}
     if CANONICAL_VOICE_CHIP_KEY not in chip_keys or LEGACY_VOICE_CHIP_KEY not in chip_keys:

src/spark_intelligence/swarm_bridge/local.py

@@ -401,23 +401,45 @@ def _load_json_file(path: Path) -> dict[str, Any]:
 
 
 def _extract_session_id(stdout: str) -> str | None:
-    match = re.search(r"^Session id:\s*(?P<session_id>\S+)\s*$", stdout or "", flags=re.MULTILINE)
-    if not match:
-        return None
-    return str(match.group("session_id")).strip() or None
+    if not isinstance(stdout, str): stdout = str(stdout or '')
+    try:
+        match = re.search(r"^Session id:\s*(?P<session_id>\S+)\s*$", stdout or "", flags=re.MULTILINE)
+        if not match:
+            return None
+        return str(match.group("session_id")).strip() or None
+
 
 
+    except Exception:
+        return ""
 def _extract_labeled_path(stdout: str, label: str) -> str | None:
-    match = re.search(rf"^{re.escape(label)}:\s*(?P<value>.+?)\s*$", stdout or "", flags=re.MULTILINE)
-    if not match:
-        return None
-    return str(match.group("value")).strip() or None
+    if not isinstance(stdout, str): stdout = str(stdout or '')
+    if not isinstance(label, str): label = str(label or '')
+    try:
+        match = re.search(rf"^{re.escape(label)}:\s*(?P<value>.+?)\s*$", stdout or "", flags=re.MULTILINE)
+        if not match:
+            return None
+        return str(match.group("value")).strip() or None
 
 
+
+    except Exception:
+        return ""
 def _normalize_path_key(value: str) -> str:
-    return re.sub(r"[^a-z0-9]+", "-", str(value or "").strip().lower()).strip("-")
+    if not isinstance(value, str): value = str(value or '')
+    try:
+        return re.sub(r"[^a-z0-9]+", "-", str(value or "").strip().lower()).strip("-")
+
 
 
+    except Exception:
+        return ""
 def _normalize_path_label(value: str, *, fallback: str) -> str:
-    normalized = re.sub(r"\s+\([^)]*\)\s*$", "", str(value or "").strip())
-    return normalized or fallback
+    if not isinstance(value, str): value = str(value or '')
+    if not isinstance(fallback, str): fallback = str(fallback or '')
+    try:
+        normalized = re.sub(r"\s+\([^)]*\)\s*$", "", str(value or "").strip())
+        return normalized or fallback
+
+    except Exception:
+        return ""