diff --git a/contrib/batch_scan/__init__.py b/contrib/batch_scan/__init__.py
index 0cb112fd0..74fe1e59f 100644
--- a/contrib/batch_scan/__init__.py
+++ b/contrib/batch_scan/__init__.py
@@ -48,7 +48,7 @@
 from .api_pool import ApiKey, ApiKeyPool, PooledChatModel, create_api_key_pool_from_env
 from .detection import detect_language, detect_skill_language
 from .discovery import discover_skills
-from .gap_fill import GapFillAnalyzer, GapFillFinding, GapFillResult, run_gap_fill
+from .gap_fill import GapFillAnalyzer, GapFillError, GapFillFinding, GapFillResult, run_gap_fill
 from .runner import run_one
 
 __all__ = [
@@ -60,6 +60,7 @@
     "detect_skill_language",
     "discover_skills",
     "GapFillAnalyzer",
+    "GapFillError",
     "GapFillFinding",
     "GapFillResult",
     "is_language_compatible",
diff --git a/contrib/batch_scan/batch_scan.py b/contrib/batch_scan/batch_scan.py
index 4b96fde8a..9ed418988 100644
--- a/contrib/batch_scan/batch_scan.py
+++ b/contrib/batch_scan/batch_scan.py
@@ -116,6 +116,7 @@ def _scan_skill(
     require_llm: bool,
     api_pool=None,
     verbose: bool = False,
+    timeout: float = 90,
 ) -> tuple[dict[str, object], str | None, str]:
     """Scan a single skill through the full pipeline.
 
@@ -139,6 +140,7 @@ def _scan_skill(
         detected_language=lang,
         apply_gap_fill=True,
         api_pool=api_pool,
+        timeout=timeout,
     )
 
     return entry, error_msg, rel_name
@@ -205,6 +207,7 @@ def _scan_skill_bounded(
     startup_timeout: float = 90, **options
 ) -> tuple[dict[str, object], str | None, str]:
     """Enforce the wall-clock limit on actual work, including local analysis."""
+    options["timeout"] = timeout
     owner = uuid4().hex
     if api_pool is not None:
         options["api_pool"] = SimpleNamespace(
diff --git a/contrib/batch_scan/docs/README.md b/contrib/batch_scan/docs/README.md
index ad012f42e..8f0ddad5a 100644
--- a/contrib/batch_scan/docs/README.md
+++ b/contrib/batch_scan/docs/README.md
@@ -202,7 +202,8 @@ python -m contrib.batch_scan.batch_scan ./tests/fixtures/ -f json -o batch.json
 
 Key differences in batch output:
 - `scan_mode: "multilingual-enhanced"` — provenance marker
-- `enhancements.gap_fill_applied` — true if LLM gap-fill was used
+- `enhancements.gap_fill_applied` — true only if every gap-fill batch completed;
+  partial passes can still have `gap_fill_findings > 0`
 - `enhancements.english_keyword_rules_skipped` — count of static rules bypassed
 - `skill.language` — detected language tag
 
@@ -313,7 +314,19 @@ python -m contrib.batch_scan.batch_scan ./tests/fixtures/ --no-require-llm --no-
 |------|---------|
 | 0 | All safe (no HIGH/CRITICAL) |
 | 1 | ≥1 skill has HIGH or CRITICAL risk |
-| 2 | Scan errors occurred |
+| 2 | Scan errors or incomplete gap-fill occurred; `skills[].error` or `enhancements.gap_fill_error` explains the failure |
+
+
+Gap-fill uses the remaining per-skill budget after core analysis, leaving up to
+five seconds for reporting before the worker's 90-second hard limit. Its pooled
+calls receive that remaining budget; the provider pool otherwise uses its
+caller's timeout. Invalid responses can take up to four attempts within it.
+Failed setup and failed batches retain core findings and completed gap-fill
+findings. An incomplete pass records `enhancements.gap_fill_status: "incomplete"`,
+`gap_fill_error_reasons`, and `enhancements.gap_fill_error`, and exits 2 even when
+its ledger outcome is nonfatal. This deadline covers gap-fill only. Core analysis
+keeps its own budget and can reach the worker limit even when providers respect
+request timeouts. The worker limit also applies if a provider ignores its timeout.
 
 CI usage:
 
@@ -415,7 +428,7 @@ tests/
 ├── test_monkeypatch_fragility.py     ← Issue #2: 26 tests (guard verification)
 ├── tests-pro/
 │   ├── test_api_pool.py              ← Issue #3: 45 tests (acquire/backoff)
-│   ├── test_gap_fill.py              ← Issue #3: 41 tests (JSON parsing)
+│   ├── test_gap_fill.py              ← Issue #3: 29 tests (JSON parsing)
 │   ├── test_runner_patches.py        ← Issue #3: 24 tests (context manager)
 │   └── test_annotation.py            ← Issue #3: 10 tests (language compat)
 └── docs/
diff --git a/contrib/batch_scan/gap_fill.py b/contrib/batch_scan/gap_fill.py
index bef027a5c..6d19536b4 100644
--- a/contrib/batch_scan/gap_fill.py
+++ b/contrib/batch_scan/gap_fill.py
@@ -28,12 +28,21 @@
 
 from __future__ import annotations
 
+from collections.abc import Callable
 from typing import Literal
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, ValidationError
 
 from skillspector.constants import MODEL_CONFIG
-from skillspector.llm_analyzer_base import LLMAnalyzerBase
+from skillspector.inspection_ledger import LedgerReason
+from skillspector.llm_analyzer_base import (
+    Batch,
+    BatchExecutionResult,
+    BatchFailure,
+    LLMRuntimeLimitError,
+    LLMAnalyzerBase,
+    _StructuredResponseValidationError,
+)
 from skillspector.logging_config import get_logger
 from skillspector.models import Finding
 
@@ -87,7 +96,16 @@ def to_finding(self, file: str) -> Finding:
 class GapFillResult(BaseModel):
     """Structured LLM response for the gap-fill analyzer."""
 
-    findings: list[GapFillFinding] = Field(default_factory=list)
+    findings: list[GapFillFinding]
+
+
+class GapFillError(RuntimeError):
+    """An incomplete pass, retaining successful batches for the report."""
+
+    def __init__(self, outcome: BatchExecutionResult):
+        self.outcome = outcome
+        self.findings = [finding for _, findings in outcome.successful for finding in findings]
+        super().__init__(f"Gap-fill analysis did not complete for {len(outcome.failures)} batch(es).")
 
 
 # ---------------------------------------------------------------------------
@@ -179,12 +197,12 @@ class GapFillAnalyzer(LLMAnalyzerBase):
     # response_format.  JSON is parsed manually in parse_response().
     response_schema: type | None = None
 
-    def __init__(self, language: str, model: str | None = None, api_pool: "ApiKeyPool | None" = None):
+    def __init__(self, language: str, model: str | None = None, api_pool: "ApiKeyPool | None" = None, *, timeout: float | None | Callable[[], float | None] = None):
         self.language = language
         resolved_model = model or MODEL_CONFIG.get("default", "gpt-5.4")
         # Inject language into the base prompt before passing to parent
         prompt = GAP_FILL_ANALYZER_PROMPT.format(language=language)
-        super().__init__(base_prompt=prompt, model=resolved_model)
+        super().__init__(base_prompt=prompt, model=resolved_model, timeout=timeout)
         # Wire multi-key pool into gap-fill LLM calls
         if api_pool:
             from .api_pool import PooledChatModel
@@ -209,38 +227,12 @@ def parse_response(self, response, batch):
         Because ``response_schema`` is ``None``, *response* is a raw string
         (not a Pydantic model).  We strip markdown code fences, parse JSON,
         validate with :class:`GapFillResult`, and filter to ``confidence >= 0.7``.
+        Invalid JSON or schema raises ``_StructuredResponseValidationError`` for retry.
         """
-        text = str(response).strip()
-
-        # Strip markdown code fences if present
-        if text.startswith("```"):
-            first_nl = text.find("\n")
-            if first_nl != -1:
-                text = text[first_nl + 1:]
-            if text.rstrip().endswith("```"):
-                text = text.rstrip()[:-3].rstrip()
-
-        # Parse JSON → Pydantic for validation
-        import json
-        try:
-            data = json.loads(text)
-        except json.JSONDecodeError as exc:
-            logger.warning(
-                "GapFillAnalyzer: invalid JSON for %s: %s",
-                batch.file_label,
-                exc,
-            )
-            return []
-
-        try:
-            result = GapFillResult.model_validate(data)
-        except Exception as exc:
-            logger.warning(
-                "GapFillAnalyzer: schema validation failed for %s: %s",
-                batch.file_label,
-                exc,
-            )
-            return []
+        if isinstance(response, GapFillResult):
+            result = response
+        else:
+            result = self._parse_json_response(response)
 
         findings: list[Finding] = []
         for item in result.findings:
@@ -256,9 +248,29 @@ def parse_response(self, response, batch):
             findings.append(item.to_finding(batch.file_path))
         return findings
 
+    @staticmethod
+    def _parse_json_response(response):
+        text = str(response).lstrip("\ufeff").strip()
+
+        # Strip markdown code fences if present
+        if text.startswith("```"):
+            first_nl = text.find("\n")
+            if first_nl != -1:
+                text = text[first_nl + 1:]
+            if text.rstrip().endswith("```"):
+                text = text.rstrip()[:-3].rstrip()
+
+        # Parse JSON → Pydantic for validation
+        import json
+        try:
+            data = json.loads(text)
+            return GapFillResult.model_validate(data)
+        except (json.JSONDecodeError, ValidationError) as exc:
+            raise _StructuredResponseValidationError from exc
+
 
 # ---------------------------------------------------------------------------
-# Backward-compatible entry point
+# Entry point preserving partial results
 # ---------------------------------------------------------------------------
 
 
@@ -267,6 +279,8 @@ def run_gap_fill(
     language: str,
     model: str | None = None,
     api_pool: "ApiKeyPool | None" = None,
+    *,
+    timeout: float | None | Callable[[], float | None] = None,
 ) -> list[Finding]:
     """Run a single targeted LLM pass covering the 8 gap-fill rules.
 
@@ -283,23 +297,50 @@ def run_gap_fill(
         Detected language string (``"zh"``, ``"ja"``, ``"ko"``, ``"en"``).
     model :
         Optional model override.  Falls back to the configured default.
+    timeout :
+        Remaining gap-fill budget in seconds, or a callable returning it before
+        each request. Does not set the core graph's analysis budget.
 
     Returns
     -------
     list[Finding]
         A (possibly empty) list of gap-fill findings.  Only findings with
         ``confidence >= 0.7`` are included.
+
+    Raises
+    ------
+    GapFillError
+        Any setup or batch failure. ``outcome`` contains per-batch evidence and
+        ``findings`` retains results from completed batches.
     """
     if not file_cache:
         return []
 
     try:
-        analyzer = GapFillAnalyzer(language=language, model=model, api_pool=api_pool)
+        analyzer = GapFillAnalyzer(language=language, model=model, api_pool=api_pool, timeout=timeout)
         batches = analyzer.get_batches(list(file_cache.keys()), file_cache)
-        results = analyzer.run_batches(batches, language=language)
-        return analyzer.collect_findings(results)
-    except ValueError:
-        raise
     except Exception as exc:
-        logger.warning("Gap-fill analysis failed: %s", exc)
-        return []
+        reason = LedgerReason.RUNTIME_LIMIT if isinstance(exc, LLMRuntimeLimitError) else LedgerReason.LLM_BATCH_FAILED
+        outcome = BatchExecutionResult(failures=[
+            BatchFailure(Batch(file_path=path, content=content), error_class=type(exc).__name__, reason=reason)
+            for path, content in file_cache.items()
+        ])
+        raise GapFillError(outcome) from exc
+    outcome = BatchExecutionResult()
+    for index, batch in enumerate(batches):
+        try:
+            completed = analyzer.run_batches_detailed([batch], language=language)
+        except Exception as exc:
+            # Core deliberately propagates some provider errors. Keep earlier
+            # results and record every batch that could not finish in this pass.
+            reason = LedgerReason.RUNTIME_LIMIT if isinstance(exc, LLMRuntimeLimitError) else LedgerReason.LLM_BATCH_FAILED
+            outcome.failures.extend(
+                BatchFailure(pending, error_class=type(exc).__name__, reason=reason)
+                for pending in batches[index:]
+            )
+            raise GapFillError(outcome) from exc
+        outcome.successful.extend(completed.successful)
+        outcome.failures.extend(completed.failures)
+    if outcome.failures:
+        raise GapFillError(outcome)
+    return analyzer.collect_findings(outcome.successful)
diff --git a/contrib/batch_scan/runner.py b/contrib/batch_scan/runner.py
index 42f4945b0..eaa160033 100644
--- a/contrib/batch_scan/runner.py
+++ b/contrib/batch_scan/runner.py
@@ -34,23 +34,26 @@
 import json
 import os
 import shutil
+import time
 from datetime import UTC, datetime
 from pathlib import Path
 
 from pydantic import ValidationError
 
 from skillspector.graph import graph
+from skillspector.inspection_ledger import finalize_ledger
 from skillspector.llm_analyzer_base import (
     LLMAnalyzerBase,
     LLMAnalysisResult,
     _StructuredResponseValidationError,
+    ledger_events_for_batches,
 )
 from skillspector.logging_config import get_logger
 from skillspector.nodes.meta_analyzer import LLMMetaAnalyzer, MetaAnalyzerResult
 
 from .annotation import annotate_findings
 from .detection import detect_skill_language
-from .gap_fill import run_gap_fill
+from .gap_fill import GapFillError, run_gap_fill
 
 logger = get_logger(__name__)
 
@@ -725,6 +728,7 @@ def run_one(
     gap_fill_findings: int = 0,
     apply_gap_fill: bool = False,
     api_pool=None,
+    timeout: float = 90,
 ) -> tuple[dict[str, object], str | None]:
     """Scan a single skill through the full graph pipeline.
 
@@ -747,13 +751,20 @@ def run_one(
         Run non-English gap-fill against the graph's provider-eligible cache.
     api_pool :
         Optional API key pool for gap-fill calls.
+    timeout :
+        Per-skill worker budget in seconds. Gap-fill uses the time remaining
+        after core analysis, with reporting headroom. Core analysis keeps its
+        own budget and can still reach the worker's process timeout.
 
     Returns
     -------
     ``(entry, error_message_or_None)`` — on success *error_message*
-    is ``None``; on failure *entry* is a stub error entry and
-    *error_message* carries the exception text.
+    is ``None``. Gap-fill failure retains the core entry and adds incomplete
+    coverage and a machine-readable error. A failure before the core result
+    returns a stub error entry.
     """
+    # Leave time to serialize partial findings before the worker hard limit.
+    deadline = time.monotonic() + max(0, timeout - min(5.0, timeout / 10))
     result = None
     try:
         state = scan_state(skill_dir, use_llm=use_llm)
@@ -771,12 +782,38 @@ def run_one(
             gap_fill_findings=gap_fill_findings,
         )
         if apply_gap_fill and use_llm and detected_language != "en":
-            gap_findings = run_gap_fill(file_cache, detected_language, api_pool=api_pool)
+            gap_error = None
+            try:
+                gap_findings = run_gap_fill(
+                    file_cache, detected_language, api_pool=api_pool,
+                    timeout=lambda: deadline - time.monotonic(),
+                )
+            except GapFillError as exc:
+                gap_error = exc
+                gap_findings = exc.findings
             entry["issues"] = list(entry.get("issues", [])) + annotate_findings(
                 [finding.to_dict() for finding in gap_findings], detected_language
             )
-            entry["enhancements"]["gap_fill_applied"] = True
+            entry["enhancements"]["gap_fill_applied"] = bool(file_cache) and gap_error is None
             entry["enhancements"]["gap_fill_findings"] = len(gap_findings)
+            if gap_error is not None:
+                events, status = ledger_events_for_batches("gap_fill", gap_error.outcome)
+                completeness, _ = finalize_ledger({
+                    **result,
+                    "findings": [*(result.get("findings") or []), *gap_findings],
+                    "inspection_ledger": [*(result.get("inspection_ledger") or []), *events],
+                    "analyzer_status_events": [*(result.get("analyzer_status_events") or []), status],
+                })
+                entry["analysis_completeness"] = completeness
+                entry["execution_successful"] = completeness["execution_successful"]
+                if entry["risk_assessment"]["recommendation"] == "SAFE":
+                    entry["risk_assessment"]["recommendation"] = "CAUTION"
+                entry["enhancements"]["gap_fill_error"] = str(gap_error)
+                entry["enhancements"]["gap_fill_status"] = "incomplete"
+                entry["enhancements"]["gap_fill_error_reasons"] = sorted({
+                    failure.reason.value for failure in gap_error.outcome.failures
+                })
+                return entry, str(gap_error)
         return entry, None
     except Exception as exc:
         return entry_from_error(skill_dir, root, str(exc), detected_language), str(exc)
diff --git a/contrib/batch_scan/tests/docs/TEST_GUIDE.md b/contrib/batch_scan/tests/docs/TEST_GUIDE.md
index d884777dd..9cf78790a 100644
--- a/contrib/batch_scan/tests/docs/TEST_GUIDE.md
+++ b/contrib/batch_scan/tests/docs/TEST_GUIDE.md
@@ -39,7 +39,7 @@ tests/
 │
 └── tests-pro/
     ├── test_api_pool.py                 ← 45 tests — pool acquire/release/backoff
-    ├── test_gap_fill.py                 ← 41 tests — JSON parsing, prompt building
+    ├── test_gap_fill.py                 ← 29 tests — JSON parsing, prompt building
     ├── test_runner_patches.py           ← 24 tests — context manager, patches
     ├── test_annotation.py               ← 10 tests — language compatibility
     ├── random_numbered.py               ← main entry point (seed=42)
@@ -114,12 +114,12 @@ Answers reviewer: *"Several patches depend on internal details that can break on
 | `TestResourceLeakRecovery` | 2 | Exception between acquire/release doesn't leak slot; release(failure) doesn't leak |
 | `TestIsRateLimit` | 5 | 429 in string message; OpenAI `RateLimitError` type; `rate_limit` keyword; false for `ValueError`; false for ordinary `Exception` |
 
-### `test_gap_fill.py` — 41 tests, 11 classes
+### `test_gap_fill.py` — 29 tests, 11 classes
 
 | Class | Tests | Covers |
 |-------|-------|--------|
 | `TestParseResponseValidJSON` | 4 | Single finding; multiple findings; empty findings; default values |
-| `TestParseResponseInvalidInput` | 9 | Non-JSON; integer; list; missing `rule_id`; null bytes; BOM prefix; missing `findings` key; illegal severity → defaults |
+| `TestParseResponseInvalidInput` | 3 | Invalid JSON/schema and missing `findings` raise; BOM-prefixed valid findings are retained |
 | `TestParseResponseMarkdownFences` | 4 | Fenced with language tag; no tag; trailing whitespace; unclosed fence |
 | `TestParseResponseFiltering` | 5 | Confidence below threshold; unknown rule_id; mixed valid/invalid; all below threshold; all unknown |
 | `TestParseResponsePydanticModel` | 1 | Delegate to Pydantic model path |
diff --git a/contrib/batch_scan/tests/test_monkeypatch_invasiveness.py b/contrib/batch_scan/tests/test_monkeypatch_invasiveness.py
index 9d4617278..ab1fa7b4f 100644
--- a/contrib/batch_scan/tests/test_monkeypatch_invasiveness.py
+++ b/contrib/batch_scan/tests/test_monkeypatch_invasiveness.py
@@ -140,8 +140,9 @@ class TestImportNoSideEffect(unittest.TestCase):
     )
     def test_import_runner_leaves_original_init_untouched(self):
         """Subprocess isolation: import runner → __init__ unchanged."""
-        repo_root = str(Path(__file__).resolve().parents[4])
-        env = {**os.environ, "PYTHONPATH": repo_root}
+        repo_root = str(Path(__file__).resolve().parents[3])
+        pythonpath = os.pathsep.join(filter(None, (repo_root, os.environ.get("PYTHONPATH"))))
+        env = {**os.environ, "PYTHONPATH": pythonpath}
         result = subprocess.run(
             [
                 sys.executable, "-X", "utf8", "-c",
diff --git a/contrib/batch_scan/tests/tests-pro/test_gap_fill.py b/contrib/batch_scan/tests/tests-pro/test_gap_fill.py
index 3b36bbb87..505a17055 100644
--- a/contrib/batch_scan/tests/tests-pro/test_gap_fill.py
+++ b/contrib/batch_scan/tests/tests-pro/test_gap_fill.py
@@ -30,7 +30,7 @@
 if str(_project_root) not in sys.path:
     sys.path.insert(0, str(_project_root))
 
-from skillspector.llm_analyzer_base import Batch
+from skillspector.llm_analyzer_base import Batch, _StructuredResponseValidationError
 from skillspector.models import Finding
 
 from contrib.batch_scan.gap_fill import (
@@ -210,46 +210,27 @@ class TestParseResponseInvalidInput(unittest.TestCase):
     def setUpClass(cls):
         cls.analyzer = GapFillAnalyzer(language="ko")
 
-    def test_non_json_string_returns_empty_list(self):
-        results = self.analyzer.parse_response("This is not JSON at all.", _batch())
-        self.assertEqual(len(results), 0)
-
-    def test_empty_string_returns_empty_list(self):
-        self.assertEqual(len(self.analyzer.parse_response("", _batch())), 0)
-
-    def test_integer_input_returns_empty_list(self):
-        self.assertEqual(len(self.analyzer.parse_response(42, _batch())), 0)
-
-    def test_json_list_instead_of_object_returns_empty_list(self):
-        self.assertEqual(len(self.analyzer.parse_response("[1, 2, 3]", _batch())), 0)
-
-    def test_missing_findings_key_returns_empty_list(self):
-        self.assertEqual(
-            len(self.analyzer.parse_response(json.dumps({"other": "value"}), _batch())), 0
-        )
-
-    def test_findings_value_is_string_not_list_returns_empty_list(self):
-        self.assertEqual(
-            len(self.analyzer.parse_response(json.dumps({"findings": "not a list"}), _batch())), 0
-        )
-
-    def test_invalid_severity_literal_value_returns_empty_list(self):
-        data = {"findings": [_valid_finding(severity="CATASTROPHIC")]}
-        results = self.analyzer.parse_response(json.dumps(data), _batch())
-        self.assertEqual(len(results), 0)
-
-    def test_utf8_bom_prepended_json_does_not_crash(self):
-        """#C3: JSON with UTF-8 BOM prefix — should not crash."""
-        text = "" + json.dumps({"findings": [_valid_finding()]})
-        results = self.analyzer.parse_response(text, _batch())
-        # May or may not parse (BOM handling is platform-dependent), but must not crash
-        self.assertIsInstance(results, list)
-
-    def test_json_with_embedded_null_bytes_does_not_crash(self):
-        """Edge: null bytes in JSON string — should not crash."""
-        text = '{"findings": [\x00]}'
-        results = self.analyzer.parse_response(text, _batch())
-        self.assertIsInstance(results, list)
+    def test_invalid_responses_raise_for_core_retry_and_failure_accounting(self):
+        responses = [
+            "This is not JSON at all.", "", 42, "[1, 2, 3]",
+            json.dumps({"findings": "not a list"}),
+            json.dumps({"findings": [_valid_finding(severity="CATASTROPHIC")]}),
+            '{"findings": [\x00]}',
+        ]
+        for response in responses:
+            with self.subTest(response=response):
+                with self.assertRaises(_StructuredResponseValidationError):
+                    self.analyzer.parse_response(response, _batch())
+
+    def test_missing_findings_key_is_not_a_completed_analysis(self):
+        for response in ('{}', '{"error": "no analysis"}', '{"refusal": "no"}'):
+            with self.subTest(response=response):
+                with self.assertRaises(_StructuredResponseValidationError):
+                    self.analyzer.parse_response(response, _batch())
+
+    def test_bom_prefixed_valid_findings_are_preserved(self):
+        findings = self.analyzer.parse_response("\ufeff" + json.dumps({"findings": [_valid_finding()]}), _batch())
+        self.assertEqual([finding.rule_id for finding in findings], ["P5"])
 
 
 # ---------------------------------------------------------------------------
@@ -297,8 +278,7 @@ def test_pydantic_model_path_delegates_to_original_parse_response(self):
         results = self.analyzer.parse_response(result, _batch())
         # Should return findings (delegates to parent class behavior)
         self.assertIsInstance(results, list)
-        # At minimum, must not crash
-        self.assertGreaterEqual(len(results), 0)
+        self.assertEqual([finding.rule_id for finding in results], ["P5"])
 
 
 # ---------------------------------------------------------------------------
diff --git a/tests/test_batch_scan_security.py b/tests/test_batch_scan_security.py
index 2bfa498a1..baf6d319d 100644
--- a/tests/test_batch_scan_security.py
+++ b/tests/test_batch_scan_security.py
@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
 import multiprocessing
 import os
@@ -15,6 +16,7 @@
 from concurrent.futures import ThreadPoolExecutor
 from functools import partial
 from pathlib import Path
+from threading import Thread
 from types import SimpleNamespace
 
 import pytest
@@ -213,6 +215,121 @@ def invoke(prompt):
     assert entry["issues"] == []
 
 
+@pytest.mark.parametrize(
+    "failure_mode", ["json", "schema", "provider", "runtime_signal", "missing_findings"]
+)
+@pytest.mark.parametrize("partial_success", [False, True])
+def test_gap_fill_failure_is_incomplete_and_preserves_findings(
+    batch_skill, monkeypatch: pytest.MonkeyPatch, failure_mode, partial_success
+) -> None:
+    from contrib.batch_scan import reports
+    from skillspector.inspection_ledger import (
+        LedgerOutcome,
+        analyzer_status_for_events,
+        ledger_event,
+    )
+    from skillspector.models import Finding
+
+    skill, _ = batch_skill
+    core_finding = Finding(
+        rule_id="TM1", message="Existing evidence", severity="HIGH", file="SKILL.md"
+    )
+    if partial_success:
+        (skill / "partial.md").write_text(_SAFE_TEXT, encoding="utf-8")
+
+    def keep_core_evidence(context):
+        events = [
+            ledger_event(
+                analyzer_id="core-test",
+                phase="static",
+                path="SKILL.md",
+                outcome=LedgerOutcome.COMPLETED,
+                emitted_finding_ids=[core_finding.finding_id],
+            )
+        ]
+        context["findings"] = [core_finding]
+        context["inspection_ledger"] = [*context.get("inspection_ledger", []), *events]
+        context["analyzer_status_events"] = [analyzer_status_for_events("core-test", events)]
+
+    _mock_scan(monkeypatch, keep_core_evidence)
+    monkeypatch.setattr(runner, "run_gap_fill", run_gap_fill)
+    calls = []
+
+    def invoke(prompt):
+        calls.append(prompt)
+        if "File: partial.md" in prompt:
+            return AIMessage(
+                content=json.dumps(
+                    {
+                        "findings": [
+                            {
+                                "rule_id": "P5",
+                                "message": "Retained gap evidence",
+                                "severity": "HIGH",
+                            }
+                        ]
+                    }
+                )
+            )
+        if failure_mode == "provider":
+            raise RuntimeError("synthetic provider failure")
+        if failure_mode == "runtime_signal":
+            raise llm_analyzer_base.LLMRuntimeLimitError("deadline")
+        if failure_mode == "missing_findings":
+            return AIMessage(content='{"refusal": "No analysis returned"}')
+        return AIMessage(content="not JSON" if failure_mode == "json" else '{"findings": 1}')
+
+    monkeypatch.setattr(
+        llm_analyzer_base, "get_chat_model", lambda **kw: SimpleNamespace(invoke=invoke)
+    )
+    monkeypatch.setattr(llm_analyzer_base, "get_max_input_tokens", lambda model: 100_000)
+    monkeypatch.setattr(llm_analyzer_base, "STRUCTURED_RESPONSE_RETRY_DELAYS_SECONDS", (0, 0, 0))
+
+    entry, error = runner.run_one(
+        skill, skill.parent, use_llm=True, detected_language="zh", apply_gap_fill=True
+    )
+
+    assert error == "Gap-fill analysis did not complete for 1 batch(es)."
+    assert entry["enhancements"]["gap_fill_applied"] is False
+    assert entry["enhancements"]["gap_fill_findings"] == int(partial_success)
+    assert entry["analysis_completeness"]["is_complete"] is False
+    assert entry["risk_assessment"]["recommendation"] == "CAUTION"
+    assert entry["execution_successful"] is (failure_mode != "provider")
+    assert entry["enhancements"]["gap_fill_error"] == error
+    assert entry["enhancements"]["gap_fill_status"] == "incomplete"
+    assert any(issue["id"] == "TM1" for issue in entry["issues"])
+    assert any(issue["id"] == "P5" for issue in entry["issues"]) is partial_success
+    expected_reason = {
+        "json": "llm_structured_response_invalid",
+        "schema": "llm_structured_response_invalid",
+        "provider": "llm_batch_failed",
+        "runtime_signal": "runtime_limit",
+        "missing_findings": "llm_structured_response_invalid",
+    }[failure_mode]
+    assert any(
+        row["reason_code"] == expected_reason
+        for row in entry["analysis_completeness"]["ledger_exceptions"]
+    )
+    payload = json.loads(reports._format_json([entry]))
+    assert payload["batch"]["enhancements"]["gap_fill_applied"] == 0
+    assert payload["batch"]["inspection_completeness"]["incomplete_skills"] == 1
+    assert expected_reason in reports._format_terminal([entry])
+    assert expected_reason.replace("_", "\\_") in reports._format_markdown([entry])
+    expected_failed_calls = 4 if failure_mode in {"json", "schema", "missing_findings"} else 1
+    assert len(calls) == expected_failed_calls + int(partial_success)
+    monkeypatch.setattr(
+        batch_scan, "_scan_skill_bounded", lambda *args, **kw: (entry, error, skill.name)
+    )
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        ["batch_scan", str(skill.parent), "--no-llm", "--format", "json", "--workers", "1"],
+    )
+    with pytest.raises(SystemExit) as stopped:
+        batch_scan._main_impl()
+    assert stopped.value.code == 2
+
+
 @pytest.mark.parametrize("cache_state", ["empty", "missing"])
 def test_gap_fill_never_falls_back_to_local_content(
     batch_skill, monkeypatch: pytest.MonkeyPatch, cache_state
@@ -234,6 +351,7 @@ def remove_provider_content(context):
     assert error is None, error
     assert observed["calls"] == [({}, "zh")]
     assert entry["skill"]["language"] == "zh"
+    assert entry["enhancements"]["gap_fill_applied"] is False
 
 
 @pytest.mark.parametrize(
@@ -684,24 +802,113 @@ def test_compat_http_timeout_preserves_shorter_deadline(monkeypatch, requested):
     assert observed["timeout"].connect == (2 if requested == 2 else 8)
 
 
+@pytest.mark.parametrize("stage", ["construction", "batching"])
+def test_gap_fill_setup_failure_retains_core_findings(batch_skill, monkeypatch, stage):
+    import contrib.batch_scan.gap_fill as gap_fill
+    from skillspector.models import Finding
+
+    skill, _ = batch_skill
+    core = Finding(rule_id="TM1", message="Core evidence", severity="HIGH", file="SKILL.md")
+    _mock_scan(monkeypatch, lambda context: context.update(findings=[core]))
+    monkeypatch.setattr(runner, "run_gap_fill", run_gap_fill)
+
+    def fail(*args, **kwargs):
+        raise TypeError("synthetic setup failure")
+
+    if stage == "construction":
+        monkeypatch.setattr(gap_fill, "GapFillAnalyzer", fail)
+    else:
+        monkeypatch.setattr(llm_analyzer_base, "get_chat_model", lambda **kw: SimpleNamespace())
+        monkeypatch.setattr(llm_analyzer_base, "get_max_input_tokens", lambda model: 100_000)
+        monkeypatch.setattr(gap_fill.GapFillAnalyzer, "get_batches", fail)
+    entry, error = runner.run_one(
+        skill, skill.parent, use_llm=True, detected_language="zh", apply_gap_fill=True
+    )
+    assert error is not None and entry["enhancements"]["gap_fill_error"] == error
+    assert any(issue["id"] == "TM1" for issue in entry["issues"])
+    assert entry["risk_assessment"]["recommendation"] != "SAFE"
+    assert entry["analysis_completeness"]["is_complete"] is False
+    assert entry["execution_successful"] is False
+    assert entry["enhancements"]["gap_fill_error_reasons"] == ["llm_batch_failed"]
+    assert any(
+        row["path"] == "SKILL.md" for row in entry["analysis_completeness"]["ledger_exceptions"]
+    )
+
+
+def test_gap_fill_real_deadline_stops_retries_and_keeps_core(batch_skill, monkeypatch):
+    from skillspector.models import Finding
+
+    skill, _ = batch_skill
+    core = Finding(rule_id="TM1", message="Core evidence", severity="HIGH", file="SKILL.md")
+    _mock_scan(monkeypatch, lambda context: context.update(findings=[core]))
+    monkeypatch.setattr(runner, "run_gap_fill", run_gap_fill)
+    calls = []
+
+    def invoke(prompt):
+        calls.append(prompt)
+        return AIMessage(content="not JSON")
+
+    monkeypatch.setattr(
+        llm_analyzer_base, "get_chat_model", lambda **kw: SimpleNamespace(invoke=invoke)
+    )
+    monkeypatch.setattr(llm_analyzer_base, "get_max_input_tokens", lambda model: 100_000)
+    started = time.monotonic()
+    with runner.deepseek_compat():
+        entry, error = runner.run_one(
+            skill,
+            skill.parent,
+            use_llm=True,
+            detected_language="zh",
+            apply_gap_fill=True,
+            timeout=0.3,
+        )
+    assert time.monotonic() - started < 2
+    assert len(calls) <= 1
+    assert error is not None
+    assert any(issue["id"] == "TM1" for issue in entry["issues"])
+    assert entry["enhancements"]["gap_fill_error_reasons"] == ["runtime_limit"]
+    assert entry["analysis_completeness"]["is_complete"] is False
+
+
 @pytest.mark.parametrize("asynchronous", [False, True])
-async def test_pooled_request_slot_wait_obeys_deadline(asynchronous):
+async def test_pooled_request_slot_wait_obeys_deadline(monkeypatch, asynchronous):
     from contrib.batch_scan.api_pool import ApiKey, ApiKeyPool, PooledChatModel
 
     pool = ApiKeyPool([ApiKey("synthetic", None, "test", max_concurrent=1)])
     occupied = pool.acquire()
     model = PooledChatModel(pool, timeout=0.02)
+
+    def no_client(*args, **kwargs):
+        raise AssertionError("slot-wait test must not build a provider client")
+
+    monkeypatch.setattr(model, "_build_llm", no_client)
+    errors = []
+
+    def invoke():
+        try:
+            model.invoke("unused")
+        except BaseException as exc:
+            errors.append(exc)
+
+    worker = None
     started = time.monotonic()
     try:
-        with pytest.raises(llm_analyzer_base.LLMRuntimeLimitError):
-            if asynchronous:
-                await model.ainvoke("unused")
-            else:
-                model.invoke("unused")
+        if asynchronous:
+            with pytest.raises(llm_analyzer_base.LLMRuntimeLimitError):
+                await asyncio.wait_for(model.ainvoke("unused"), timeout=2)
+        else:
+            worker = Thread(target=invoke, daemon=True)
+            worker.start()
+            worker.join(timeout=2)
+            assert not worker.is_alive(), "pooled slot wait ignored its deadline"
+            assert len(errors) == 1
+            assert isinstance(errors[0], llm_analyzer_base.LLMRuntimeLimitError)
         assert time.monotonic() - started < 1
         assert occupied.active_requests == 1
     finally:
         pool.release(occupied)
+        if worker is not None:
+            worker.join(timeout=2)
     assert occupied.active_requests == 0
 
 
@@ -771,3 +978,55 @@ def test_pooled_client_bounds_connect_and_disables_sdk_retries(monkeypatch):
     PooledChatModel(ApiKeyPool([key]))._build_llm(key, timeout=0.1)
     assert observed["timeout"].connect == observed["timeout"].read == 0.1
     assert observed["max_retries"] == 0
+
+
+@pytest.mark.parametrize("completed_first", [False, True])
+def test_gap_fill_provider_value_error_retains_completed_findings(
+    batch_skill, monkeypatch, completed_first
+):
+    from skillspector.models import Finding
+
+    skill, _ = batch_skill
+    for name in ("notes.txt", "remaining.txt"):
+        (skill / name).write_text(_SAFE_TEXT, encoding="utf-8")
+    core = Finding(rule_id="TM1", message="Core evidence", severity="HIGH", file="SKILL.md")
+    _mock_scan(monkeypatch, lambda context: context.update(findings=[core]))
+    monkeypatch.setattr(runner, "run_gap_fill", run_gap_fill)
+    calls = []
+
+    def invoke(prompt):
+        calls.append(prompt)
+        if completed_first and len(calls) == 1:
+            return AIMessage(
+                content=json.dumps(
+                    {
+                        "findings": [
+                            {
+                                "rule_id": "P5",
+                                "message": "Gap-fill evidence",
+                                "severity": "HIGH",
+                                "confidence": 0.9,
+                            }
+                        ]
+                    }
+                )
+            )
+        raise ValueError("synthetic provider error response")
+
+    monkeypatch.setattr(
+        llm_analyzer_base, "get_chat_model", lambda **kw: SimpleNamespace(invoke=invoke)
+    )
+    monkeypatch.setattr(llm_analyzer_base, "get_max_input_tokens", lambda model: 100_000)
+    with runner.deepseek_compat():
+        entry, error = runner.run_one(
+            skill, skill.parent, use_llm=True, detected_language="zh", apply_gap_fill=True
+        )
+    assert error is not None
+    assert len(calls) == (2 if completed_first else 1)
+    ids = {issue["id"] for issue in entry["issues"]}
+    assert "TM1" in ids
+    assert ("P5" in ids) is completed_first
+    assert entry["risk_assessment"]["recommendation"] != "SAFE"
+    assert entry["enhancements"]["gap_fill_status"] == "incomplete"
+    assert entry["enhancements"]["gap_fill_error_reasons"] == ["llm_batch_failed"]
+    assert entry["analysis_completeness"]["is_complete"] is False