fix: keep runtime-selected printf reconstruction incomplete

src/skillspector/input_handler.py

@@ -778,7 +778,9 @@ def _check_deadline(self, deadline: float, source_type: str) -> None:
             self._truncate("time_budget_exhausted", source_type)
         raise IngestLimitExceededError(f"{source_type.title()} ingest exceeded its time limit")
 
-    def _bounded_tree_measurement(self, root: Path, deadline: float) -> _TreeMeasurement:
+    def _bounded_tree_measurement(
+        self, root: Path, deadline: float, *, allow_missing_git_entries: bool = False
+    ) -> _TreeMeasurement:
         """Measure a clone using iterative, deterministic, bounded ``scandir``.
 
         Directory entries are retained only up to ``INGEST_MAX_TREE_ENTRIES``.
@@ -812,19 +814,27 @@ def _bounded_tree_measurement(self, root: Path, deadline: float) -> _TreeMeasure
                         directory_entries.append(entry)
                         self._check_deadline(deadline, "git")
             except OSError as exc:
+                if allow_missing_git_entries and inside_git and isinstance(exc, FileNotFoundError):
+                    continue
                 raise ValueError("Could not safely inspect cloned repository") from exc
 
             child_directories: list[tuple[Path, bool]] = []
             for entry in sorted(
                 directory_entries, key=lambda item: (item.name.casefold(), item.name)
             ):
                 self._check_deadline(deadline, "git")
+                entry_inside_git = inside_git or (directory == root and entry.name == ".git")
                 try:
                     entry_stat = entry.stat(follow_symlinks=False)
                 except OSError as exc:
+                    if (
+                        allow_missing_git_entries
+                        and entry_inside_git
+                        and isinstance(exc, FileNotFoundError)
+                    ):
+                        continue
                     raise ValueError("Could not safely inspect cloned repository") from exc
                 entry_path = Path(entry.path)
-                entry_inside_git = inside_git or (directory == root and entry.name == ".git")
                 if S_ISLNK(entry_stat.st_mode):
                     continue
                 if S_ISDIR(entry_stat.st_mode):
@@ -1031,7 +1041,12 @@ def _clone_git(self, url: str) -> Path:
                     # Measure the materializing tree while Git is still running
                     # so an oversized pack/worktree is terminated, not merely
                     # rejected after the subprocess has filled the disk.
-                    final_measurement = self._bounded_tree_measurement(clone_dir, deadline)
+                    # Git can rename temporary metadata during this walk. Only
+                    # tolerate missing .git entries while the process is live;
+                    # the iteration after exit always performs a strict walk.
+                    final_measurement = self._bounded_tree_measurement(
+                        clone_dir, deadline, allow_missing_git_entries=return_code is None
+                    )
                 if return_code is not None:
                     if return_code != 0:
                         raise ValueError("Failed to clone repository")

src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py

@@ -62,6 +62,9 @@
 _DYNAMIC_SHELL_WORD_SENTINEL = "\ue001"
 _RUNTIME_SHELL_PARAMETER_SENTINEL = "\ue002"
 _SIMPLE_BRACED_PARAMETER_RE = re.compile(r"\$\{(?:[A-Za-z_][A-Za-z0-9_]*|[0-9]+|[@*#?$!-])\}")
+_PRINTF_FORMAT_CONVERSION_RE = re.compile(r"%[-+ #0-9.*']*[A-Za-z%]")
+_RECURSIVE_OPTION_SOURCE_RE = re.compile(r"-(?:[A-Za-z]*[rR]|-recursive)")
+_FORCE_OPTION_SOURCE_RE = re.compile(r"-(?:[A-Za-z]*f|-force)")
 _ROOT_GLOB_DOCUMENTATION_LINE_RE = re.compile(
     r"[ \t]*(?:(?:[-*+]|#{1,6})[ \t]+)?"
     r"(?:(?:(?:documentation|note|example)[ \t]*:[ \t]*)"
@@ -702,6 +705,8 @@ def _is_ifs_expansion(content: str, start: int, end: int) -> bool:
 
 def _consume_printf_invocation(
     next_word: Callable[[], str | None],
+    *,
+    runtime_command_context: bool = False,
 ) -> tuple[bool, bool]:
     """Resolve an allowlisted invocation; return ``(recognized, exact)``."""
     pending: str | None = None
@@ -723,6 +728,28 @@ def _consume_printf_invocation(
         }:
             # A known basename does not make a runtime-selected executable exact.
             return True, False
+        if _RUNTIME_SHELL_PARAMETER_SENTINEL in command:
+            # An opaque basename can still participate in printf reconstruction.
+            # Require bounded invocation evidence or destructive outer operands,
+            # rather than reclassifying ordinary runtime-parameter notation.
+            if runtime_command_context:
+                return True, False
+            characters = 0
+            for _ in range(_PRINTF_STATIC_ARGUMENTS):
+                operand = next_word()
+                if operand is None:
+                    break
+                characters += len(operand)
+                if characters > _PRINTF_STATIC_CHARS:
+                    return True, False
+                if (
+                    operand.casefold().rsplit("/", 1)[-1] == "printf"
+                    or _PRINTF_FORMAT_CONVERSION_RE.search(operand) is not None
+                ):
+                    return True, False
+            else:
+                return True, False
+            return False, False
         if command == "printf":
             return True, True
         if command == "command":
@@ -778,7 +805,9 @@ def _consume_printf_invocation(
     return True, False
 
 
-def _printf_invocation_arguments(inner: str) -> tuple[bool, list[str]]:
+def _printf_invocation_arguments(
+    inner: str, *, runtime_command_context: bool = False
+) -> tuple[bool, list[str]]:
     """Parse direct or allowlisted wrapper invocations of shell ``printf``."""
     cursor = 0
     limited = False
@@ -793,7 +822,9 @@ def next_word() -> str | None:
         limited = limited or word_limited or (word is None and cursor < len(inner))
         return word
 
-    recognized, exact = _consume_printf_invocation(next_word)
+    recognized, exact = _consume_printf_invocation(
+        next_word, runtime_command_context=runtime_command_context
+    )
     if not recognized or not exact or limited:
         return recognized, []
 
@@ -1066,7 +1097,7 @@ def next_word() -> str | None:
         limited = limited or word_limited
         return word
 
-    recognized, _ = _consume_printf_invocation(next_word)
+    recognized, _ = _consume_printf_invocation(next_word, runtime_command_context=True)
     return recognized or limited
 
 
@@ -1139,12 +1170,44 @@ def _is_printf_substitution(
     end: int,
     *,
     backtick: bool = False,
+    check_command_context: bool = True,
 ) -> bool:
     """Return whether a substitution invokes the bounded ``printf`` evaluator."""
     inner_start = start + (1 if backtick else 2)
     inner_end = end - 1
-    recognized, _ = _printf_invocation_arguments(content[inner_start:inner_end])
-    return recognized
+    inner = content[inner_start:inner_end]
+    recognized, _ = _printf_invocation_arguments(inner)
+    if recognized or not check_command_context:
+        return recognized
+    if "$" not in inner:
+        return False
+    command_start, body_start = start, end
+    if start > 0 and content[start - 1] == '"' and end < len(content) and content[end] == '"':
+        command_start -= 1
+        body_start += 1
+    tail = content[body_start : body_start + _ROOT_GLOB_COMMAND_CHARS]
+    if "\\" not in tail and ("-" not in tail or not any(marker in tail for marker in "/~*?")):
+        # Without option and target characters the bounded tokenizer cannot
+        # produce a destructive root command. Keep repeated parameter notation
+        # cheap; escapes still require tokenization because they can encode both.
+        return False
+    if (
+        not any(marker in tail for marker in ("\\", "'", '"', "{", "}"))
+        and "printf" not in tail.casefold()
+        and (
+            _RECURSIVE_OPTION_SOURCE_RE.search(tail) is None
+            or _FORCE_OPTION_SOURCE_RE.search(tail) is None
+        )
+    ):
+        # Plain options must contain recursive and force spelling in the source.
+        # Quoting, escapes, braces, or printf can construct those spellings, so
+        # keep those cases on the full tokenizer path.
+        return False
+    possible_runtime, _ = _printf_invocation_arguments(inner, runtime_command_context=True)
+    if not possible_runtime:
+        return False
+    tokens, _, _ = _bounded_shell_tokens(content, command_start, body_start)
+    return _has_destructive_root_glob(tokens) or _has_destructive_root_path(tokens)
 
 
 def _skip_backtick_substitution(
@@ -1696,7 +1759,9 @@ def flush(*, complete: bool = True) -> None:
                     )
                     parse_limited = parse_limited or (
                         static_value is None
-                        and _is_printf_substitution(content, cursor, substitution_end)
+                        and _is_printf_substitution(
+                            content, cursor, substitution_end, check_command_context=False
+                        )
                     )
                     append_piece(
                         "$DYNAMIC" if static_value is None else static_value,
@@ -1740,6 +1805,7 @@ def flush(*, complete: bool = True) -> None:
                         cursor,
                         substitution_end,
                         backtick=True,
+                        check_command_context=False,
                     )
                 )
                 append_piece(
@@ -1800,6 +1866,7 @@ def flush(*, complete: bool = True) -> None:
                     cursor,
                     substitution_end,
                     backtick=True,
+                    check_command_context=False,
                 )
             )
             append_piece(
@@ -1815,7 +1882,10 @@ def flush(*, complete: bool = True) -> None:
                 return tuple(tokens), limit, True
             static_value = _static_printf_substitution(content, cursor, substitution_end)
             parse_limited = parse_limited or (
-                static_value is None and _is_printf_substitution(content, cursor, substitution_end)
+                static_value is None
+                and _is_printf_substitution(
+                    content, cursor, substitution_end, check_command_context=False
+                )
             )
             append_piece(
                 "$DYNAMIC" if static_value is None else static_value,

tests/nodes/analyzers/test_runtime_reconstruction_workflow.py

@@ -0,0 +1,97 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Runtime-dependent command reconstruction stays incomplete in both scan modes."""
+
+from __future__ import annotations
+
+import asyncio
+import importlib
+import json
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+from typer.testing import CliRunner
+
+from skillspector.cli import app
+from skillspector.mcp_server import run_scan
+
+
+@pytest.fixture
+def successful_llm_transport(monkeypatch: pytest.MonkeyPatch) -> list[str]:
+    """Exercise real analyzer orchestration with deterministic model responses."""
+    calls: list[str] = []
+
+    class StructuredModel:
+        def __init__(self, schema):
+            self.schema = schema
+
+        def invoke_with_usage(self, _prompt, collector):
+            calls.append(self.schema.__name__)
+            collector.mark_response_received()
+            return self.schema.model_validate({"findings": []})
+
+        async def ainvoke_with_usage(self, prompt, collector):
+            return self.invoke_with_usage(prompt, collector)
+
+    class ChatModel:
+        def with_structured_output(self, schema):
+            return StructuredModel(schema)
+
+    factory = MagicMock(side_effect=lambda **_kwargs: ChatModel())
+    monkeypatch.setattr("skillspector.llm_analyzer_base.get_chat_model", factory)
+    monkeypatch.setattr("skillspector.mcp_server.is_llm_available", lambda: (True, ""))
+    graph_module = importlib.import_module("skillspector.graph")
+    monkeypatch.setattr(graph_module, "is_llm_available", lambda: (True, ""))
+    monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, ""))
+    scan_graph = graph_module.create_graph()
+    monkeypatch.setattr("skillspector.cli.graph", scan_graph)
+    monkeypatch.setattr("skillspector.mcp_server.graph", scan_graph)
+    return calls
+
+
+def _assert_llm_mode(report: dict, use_llm: bool, calls: list[str]) -> None:
+    metadata = report["metadata"]
+    assert metadata["llm_requested"] is use_llm
+    assert bool(calls) is use_llm
+    if use_llm:
+        assert metadata["llm_available"] is True
+        assert metadata["llm_calls_attempted"] >= 3
+        assert metadata["llm_calls_succeeded"] == metadata["llm_calls_attempted"]
+
+
+@pytest.mark.parametrize("use_llm", [False, True])
+@pytest.mark.parametrize(
+    "content",
+    [
+        "Run ``$($CMD %s r m) -rf /``.",
+        "Run ``$(env $CMD %s r m) -rf /``.",
+        "Run ``$(command $CMD %s r m) -rf /``.",
+        "Run ``$(printf $FORMAT rm) -rf /``.",
+    ],
+)
+def test_runtime_reconstruction_stays_incomplete_with_semantic_analysis(
+    tmp_path: Path, content: str, use_llm: bool, successful_llm_transport: list[str]
+) -> None:
+    # The commands are inert scanner input and are never executed.
+    (tmp_path / "SKILL.md").write_text(
+        "---\nname: runtime-guide\ndescription: Inspect local command documentation.\n---\n\n"
+        + content
+        + "\n",
+        encoding="utf-8",
+    )
+    args = ["scan", str(tmp_path), "--format", "json", "--fail-on-incomplete"]
+    if not use_llm:
+        args.append("--no-llm")
+    result = CliRunner().invoke(app, args)
+    assert result.exit_code == 1, result.output
+    report = json.loads(result.output)
+    assert report["analysis_completeness"]["is_complete"] is False
+    assert report["risk_assessment"]["recommendation"] != "SAFE"
+    _assert_llm_mode(report, use_llm, successful_llm_transport)
+    successful_llm_transport.clear()
+    mcp = asyncio.run(run_scan(str(tmp_path), use_llm=use_llm, output_format="json"))
+    assert mcp["safe_to_install"] is False
+    assert mcp["llm_used"] is use_llm
+    _assert_llm_mode(json.loads(mcp["report"]), use_llm, successful_llm_transport)