fix: exclude binary and large files from LLM analysis

src/skillspector/llm_analyzer_base.py

@@ -436,7 +436,10 @@ def get_batches(
 
         batches: list[Batch] = []
         for path in file_paths:
-            content = file_cache.get(path) or "No content available for this file."
+            if path not in file_cache:
+                logger.info("Skipping file absent from LLM file cache: %s", path)
+                continue
+            content = file_cache[path] or "No content available for this file."
             file_findings = findings_by_file.get(path, [])
 
             extra = self._estimate_extra_overhead(file_findings)

src/skillspector/nodes/analyzers/semantic_developer_intent.py

@@ -31,7 +31,12 @@
 )
 from skillspector.llm_utils import run_async
 from skillspector.logging_config import get_logger
-from skillspector.state import AnalyzerNodeResponse, SkillspectorState, llm_call_record
+from skillspector.state import (
+    AnalyzerNodeResponse,
+    SkillspectorState,
+    get_llm_file_cache,
+    llm_call_record,
+)
 
 ANALYZER_ID = "semantic_developer_intent"
 logger = get_logger(__name__)
@@ -173,7 +178,7 @@ def node(state: SkillspectorState) -> AnalyzerNodeResponse:
             ],
         }
 
-    file_cache: dict[str, str] = state.get("file_cache") or {}
+    file_cache = get_llm_file_cache(state)
     if not file_cache:
         return {
             "findings": [],

src/skillspector/nodes/analyzers/semantic_quality_policy.py

@@ -31,7 +31,12 @@
 )
 from skillspector.llm_utils import run_async
 from skillspector.logging_config import get_logger
-from skillspector.state import AnalyzerNodeResponse, SkillspectorState, llm_call_record
+from skillspector.state import (
+    AnalyzerNodeResponse,
+    SkillspectorState,
+    get_llm_file_cache,
+    llm_call_record,
+)
 
 ANALYZER_ID = "semantic_quality_policy"
 logger = get_logger(__name__)
@@ -146,7 +151,7 @@ def node(state: SkillspectorState) -> AnalyzerNodeResponse:
             ],
         }
 
-    file_cache: dict[str, str] = state.get("file_cache") or {}
+    file_cache = get_llm_file_cache(state)
     files = sorted(file_cache.keys())
     if not files:
         return {

src/skillspector/nodes/analyzers/semantic_security_discovery.py

@@ -34,7 +34,12 @@
     ledger_events_for_batches,
 )
 from skillspector.logging_config import get_logger
-from skillspector.state import AnalyzerNodeResponse, SkillspectorState, llm_call_record
+from skillspector.state import (
+    AnalyzerNodeResponse,
+    SkillspectorState,
+    get_llm_file_cache,
+    llm_call_record,
+)
 
 ANALYZER_ID = "semantic_security_discovery"
 logger = get_logger(__name__)
@@ -96,7 +101,7 @@ def node(state: SkillspectorState) -> AnalyzerNodeResponse:
             ],
         }
 
-    file_cache: dict[str, str] = state.get("file_cache") or {}
+    file_cache = get_llm_file_cache(state)
     components: list[str] = state.get("components") or sorted(file_cache.keys())
     if not components:
         return {

src/skillspector/nodes/build_context.py

@@ -23,6 +23,7 @@
 
 import os
 import re
+from dataclasses import dataclass
 from pathlib import Path
 from stat import S_ISREG
 
@@ -70,6 +71,111 @@
 )
 
 
+@dataclass(frozen=True)
+class _FileInspection:
+    """Content classification used to decide whether LLM analyzers can read a file."""
+
+    content: str
+    content_kind: str
+    llm_analysis_status: str
+    llm_skip_reason: str | None = None
+
+
+def _has_media_signature(data: bytes) -> bool:
+    """Return whether bytes have a recognized image, audio, or video signature."""
+    if data.startswith(
+        (
+            b"\x89PNG\r\n\x1a\n",
+            b"\xff\xd8\xff",
+            b"GIF87a",
+            b"GIF89a",
+            b"BM",
+            b"\x00\x00\x01\x00",
+            b"II*\x00",
+            b"MM\x00*",
+            b"fLaC",
+            b"OggS",
+            b"\x1aE\xdf\xa3",
+            b"\x00\x00\x01\xba",
+            b"\x00\x00\x01\xb3",
+        )
+    ):
+        return True
+    if data.startswith(b"RIFF") and data[8:12] in {b"WEBP", b"WAVE", b"AVI "}:
+        return True
+    if data.startswith(b"ID3") or (
+        len(data) >= 2 and data[0] == 0xFF and data[1] & 0xE6 in {0xE0, 0xE2, 0xE4, 0xE6}
+    ):
+        return True
+    if len(data) >= 12 and data[4:8] == b"ftyp":
+        return data[8:12].lower() in {
+            b"3gp4",
+            b"3gp5",
+            b"avif",
+            b"avis",
+            b"heic",
+            b"heix",
+            b"hevc",
+            b"hevx",
+            b"isom",
+            b"m4a ",
+            b"m4v ",
+            b"mif1",
+            b"mp41",
+            b"mp42",
+            b"msf1",
+            b"qt  ",
+        }
+    return False
+
+
+def _has_abnormal_controls(content: str) -> bool:
+    """Return whether decoded text contains controls not used for normal layout."""
+    return any(
+        (ord(char) < 32 and char not in "\t\n\r\f") or 127 <= ord(char) <= 159 for char in content
+    )
+
+
+def _is_strong_binary(data: bytes) -> bool:
+    """Detect strong binary evidence while leaving uncertain encodings analyzable."""
+    if b"\x00" in data:
+        return True
+    if not data:
+        return False
+    abnormal_controls = sum(
+        (byte < 32 and byte not in {9, 10, 12, 13}) or byte == 127 for byte in data
+    )
+    return abnormal_controls / len(data) >= 0.3
+
+
+def _inspect_bytes(data: bytes) -> _FileInspection:
+    """Classify bytes, preferring analyzable text and failing open when uncertain."""
+    try:
+        strict_content = data.decode("utf-8")
+    except UnicodeDecodeError:
+        strict_content = None
+
+    if strict_content is not None and not _has_abnormal_controls(strict_content):
+        return _FileInspection(strict_content, "text", "included")
+    if _has_media_signature(data):
+        return _FileInspection(
+            data.decode("utf-8", errors="replace"),
+            "media",
+            "excluded",
+            "media_content",
+        )
+    if _is_strong_binary(data):
+        return _FileInspection(
+            data.decode("utf-8", errors="replace"),
+            "binary",
+            "excluded",
+            "binary_content",
+        )
+
+    # Invalid or unusual text that is not confidently binary remains in scope.
+    return _FileInspection(data.decode("utf-8", errors="replace"), "text", "included")
+
+
 def _resolve_skill_dir(state: SkillspectorState) -> Path:
     """Resolve state skill_path to an existing directory Path."""
     skill_path = state.get("skill_path")
@@ -145,7 +251,10 @@ def _infer_file_type(path: str) -> str:
 
 
 def _build_component_metadata(
-    skill_dir: Path, components: list[str], file_cache: dict[str, str]
+    skill_dir: Path,
+    components: list[str],
+    inspections: dict[str, _FileInspection],
+    file_cache: dict[str, str],
 ) -> tuple[list[dict[str, object]], bool]:
     """Build component_metadata list and has_executable_scripts from paths."""
     metadata: list[dict[str, object]] = []
@@ -156,6 +265,7 @@ def _build_component_metadata(
         file_type = _infer_file_type(path)
         content = file_cache.get(path)
         lines = len(content.splitlines()) if content is not None else 0
+        inspection = inspections.get(path)
         executable = suffix in _EXECUTABLE_EXTENSIONS
         if executable:
             has_executable = True
@@ -171,16 +281,28 @@ def _build_component_metadata(
                 "lines": lines,
                 "executable": executable,
                 "size_bytes": size_bytes,
+                "content_kind": inspection.content_kind if inspection else "unknown",
+                "llm_analysis_status": (
+                    inspection.llm_analysis_status if inspection else "excluded"
+                ),
+                "llm_skip_reason": inspection.llm_skip_reason if inspection else None,
             }
         )
     return metadata, has_executable
 
 
 def _read_file_cache(
     skill_dir: Path, components: list[str]
-) -> tuple[dict[str, str], list[InspectionLedgerEvent]]:
-    """Build readable file content and terminal events for cache failures."""
+) -> tuple[
+    dict[str, str],
+    dict[str, str],
+    dict[str, _FileInspection],
+    list[InspectionLedgerEvent],
+]:
+    """Build shared/LLM caches, classifications, and terminal cache evidence."""
     file_cache: dict[str, str] = {}
+    llm_file_cache: dict[str, str] = {}
+    inspections: dict[str, _FileInspection] = {}
     ledger_events: list[InspectionLedgerEvent] = []
     for path in components:
         full = skill_dir / path
@@ -222,8 +344,7 @@ def _read_file_cache(
             )
             continue
         try:
-            content = full.read_text(encoding="utf-8", errors="replace")
-            file_cache[path] = content
+            inspection = _inspect_bytes(full.read_bytes())
         except FileNotFoundError as exc:
             ledger_events.append(
                 ledger_event(
@@ -235,6 +356,7 @@ def _read_file_cache(
                     error_class=type(exc).__name__,
                 )
             )
+            continue
         except OSError as exc:
             logger.debug("Could not read file: %s", path)
             ledger_events.append(
@@ -247,7 +369,27 @@ def _read_file_cache(
                     error_class=type(exc).__name__,
                 )
             )
-    return file_cache, ledger_events
+            continue
+
+        inspections[path] = inspection
+        file_cache[path] = inspection.content
+        if inspection.llm_analysis_status == "included":
+            llm_file_cache[path] = inspection.content
+            continue
+
+        logger.info("Excluding %s from LLM analysis: %s", path, inspection.llm_skip_reason)
+        ledger_events.append(
+            ledger_event(
+                outcome=LedgerOutcome.SKIPPED,
+                record_type=LedgerRecordType.SYSTEM,
+                phase="llm_eligibility",
+                path=path,
+                reason=LedgerReason.BINARY_CONTENT,
+                observed_bytes=file_stat.st_size,
+            )
+        )
+
+    return file_cache, llm_file_cache, inspections, ledger_events
 
 
 def _parse_manifest(skill_dir: Path) -> dict[str, object]:
@@ -319,15 +461,16 @@ def build_context(state: SkillspectorState) -> dict[str, object]:
     skill_dir = _resolve_skill_dir(state)
 
     components, discovery_events = _walk_skill_files(skill_dir)
-    file_cache, cache_events = _read_file_cache(skill_dir, components)
+    file_cache, llm_file_cache, inspections, cache_events = _read_file_cache(skill_dir, components)
     manifest = _parse_manifest(skill_dir)
     component_metadata, has_executable_scripts = _build_component_metadata(
-        skill_dir, components, file_cache
+        skill_dir, components, inspections, file_cache
     )
 
     return {
         "components": components,
         "file_cache": file_cache,
+        "llm_file_cache": llm_file_cache,
         "inspection_ledger": [*discovery_events, *cache_events],
         "ast_cache": {},
         "manifest": manifest,

src/skillspector/nodes/meta_analyzer.py

@@ -51,7 +51,12 @@
     get_explanation,
     get_remediation,
 )
-from skillspector.state import MetaAnalyzerResponse, SkillspectorState, llm_call_record
+from skillspector.state import (
+    MetaAnalyzerResponse,
+    SkillspectorState,
+    get_llm_file_cache,
+    llm_call_record,
+)
 
 logger = get_logger(__name__)
 
@@ -632,7 +637,7 @@ def meta_analyzer(state: SkillspectorState) -> MetaAnalyzerResponse:
             ],
         }
 
-    file_cache: dict[str, str] = state.get("file_cache") or {}
+    file_cache = get_llm_file_cache(state)
     manifest: dict[str, object] = state.get("manifest") or {}
     model_config: dict[str, str] = state.get("model_config") or {}
     model = (

src/skillspector/state.py

@@ -58,6 +58,9 @@ class SkillspectorState(TypedDict, total=False):
     # build_context node populates these
     components: list[str]
     file_cache: dict[str, str]
+    # Text files eligible for LLM analysis. Missing on legacy/test states, where
+    # LLM nodes fall back to file_cache for compatibility.
+    llm_file_cache: dict[str, str]
     ast_cache: dict[str, str]
     manifest: dict[str, object]
     previous_manifest: dict[str, object] | None
@@ -148,6 +151,14 @@ class AnalyzerNodeResponse(TypedDict):
     llm_call_log: NotRequired[list[LLMCallRecord]]
 
 
+def get_llm_file_cache(state: SkillspectorState) -> dict[str, str]:
+    """Return the LLM-eligible cache, preserving compatibility with legacy states."""
+    cache = state.get("llm_file_cache")
+    if cache is not None:
+        return cache
+    return state.get("file_cache") or {}
+
+
 class MetaAnalyzerResponse(TypedDict):
     """Meta-analyzer payload with canonical findings and ID selection."""