Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions src/skillspector/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,11 @@
# Static analysis supports complete per-artifact coverage through 16 MiB. Larger
# files are read only to this bound and are reported as partial, never complete.
MAX_ANALYZABLE_FILE_BYTES = 16 * 1024 * 1024
# Truncated files must still reach the LLM stage: excluding them entirely lets
# a payload hide past the read cap while the scan reports zero findings. The
# LLM view of a truncated file is bounded to this prefix so token cost stays
# predictable, and an explicit marker records the unreviewed region.
MAX_LLM_TRUNCATED_FILE_CHARS = 262_144

# Default-model selection lives on each provider (see providers/<name>/provider.py
# for ``DEFAULT_MODEL`` and ``SLOT_DEFAULTS``). The active provider's
Expand Down
34 changes: 32 additions & 2 deletions src/skillspector/nodes/build_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,12 @@
classify_artifact,
decode_text,
)
from skillspector.constants import MAX_ANALYZABLE_FILE_BYTES, MAX_FILE_BYTES, build_model_config
from skillspector.constants import (
MAX_ANALYZABLE_FILE_BYTES,
MAX_FILE_BYTES,
MAX_LLM_TRUNCATED_FILE_CHARS,
build_model_config,
)
from skillspector.input_handler import (
_FileOpenError,
_open_regular_file_no_follow,
Expand Down Expand Up @@ -754,6 +759,25 @@ def _is_hidden_path(path: str) -> bool:
return any(part.startswith(".") for part in Path(path).parts)


def _llm_view_of_truncated_file(content: str, *, total_size: int, read_bytes: int) -> str:
"""Bound a truncated file's LLM view and mark the unreviewed region.

A truncated file must not vanish from the LLM stage: with exclusion, a
payload placed past the read cap is invisible to every analyzer while the
report still shows zero findings. The view is capped so token cost stays
bounded, and the marker makes the audit gap explicit to the model.
"""
marker = (
f"\n\n[SKILLSPECTOR: this file is {total_size} bytes; only the first "
f"{read_bytes} bytes were readable and the excerpt above is capped at "
f"{MAX_LLM_TRUNCATED_FILE_CHARS} characters. The remaining bytes were "
"not reviewed by any analyzer - treat the unseen region as an audit "
"gap.]\n"
)
budget = max(MAX_LLM_TRUNCATED_FILE_CHARS - len(marker), 0)
return content[:budget] + marker


def _opaque_artifact_record(
path: str,
*,
Expand Down Expand Up @@ -1089,7 +1113,13 @@ def _record_cache_runtime_limit(
)
)
inventory.append(artifact)
if not truncated and not _is_hidden_path(path) and artifact["content_kind"] == "text":
if not _is_hidden_path(path) and artifact["content_kind"] == "text":
if truncated:
content = _llm_view_of_truncated_file(
content,
total_size=max(file_stat.st_size, len(observed)),
read_bytes=len(raw),
)
llm_file_cache[path] = _redact_for_external_model(path, content)
if aggregate_truncated:
inventory.extend(
Expand Down
70 changes: 61 additions & 9 deletions src/skillspector/nodes/finalize_inspection_ledger.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,14 +94,66 @@ def _reference_coverage_findings(
return findings


def _size_coverage_findings(
state: SkillspectorState,
covered_paths: set[str],
) -> list[Finding]:
"""Create AE7 for artifacts truncated by the per-file size cap.

A file too large to fully analyze must not be able to produce a
zero-finding report: the unreviewed region is itself the finding.
Scoped to the per-file cap (``size_limit``); aggregate budget
exhaustion already fails closed through the completeness projection.
Paths already reported by AE1 (referenced artifacts) are skipped.
"""
findings: list[Finding] = []
for item in state.get("artifact_inventory") or []:
if not isinstance(item, Mapping):
continue
if str(item.get("disposition", "")) != "partial":
continue
if str(item.get("reason", "")) != LedgerReason.SIZE_LIMIT.value:
continue
if str(item.get("path", "")) in covered_paths:
continue
path = str(item.get("path", ""))
size_bytes = item.get("size_bytes", 0)
findings.append(
Finding(
rule_id="AE7",
message=(
"File exceeds the analyzable size limit; trailing content was not inspected"
),
severity="HIGH",
confidence=1.0,
file=path,
start_line=1,
category="analysis-evasion",
tags=["coverage", "size-limit"],
finding=f"{path} ({size_bytes} bytes, partially inspected)"[:200],
matched_text=path,
remediation=(
"Keep analyzable files under the per-file size limit, or "
"split oversized content so every byte can be inspected."
),
)
)
return findings


def finalize_inspection_ledger(state: SkillspectorState) -> dict[str, object]:
"""Validate full internal facts and derive the public completeness projection."""
reference_findings = _reference_coverage_findings(state)
size_findings = _size_coverage_findings(
state,
covered_paths={str(finding.matched_text or "") for finding in reference_findings},
)
coverage_findings = [*reference_findings, *size_findings]
# Work IDs are scoped to analyzer, source file and line range. Distinct
# targets on one line must share a terminal row with all emitted findings.
reference_ids_by_line: dict[tuple[str, int | None], list[str]] = {}
for finding in reference_findings:
reference_ids_by_line.setdefault((finding.file, finding.start_line), []).append(
coverage_ids_by_line: dict[tuple[str, int | None], list[str]] = {}
for finding in coverage_findings:
coverage_ids_by_line.setdefault((finding.file, finding.start_line), []).append(
finding.finding_id
)
reference_events: list[InspectionLedgerEvent] = [
Expand All @@ -114,10 +166,10 @@ def finalize_inspection_ledger(state: SkillspectorState) -> dict[str, object]:
end_line=line,
emitted_finding_ids=finding_ids,
)
for (path, line), finding_ids in reference_ids_by_line.items()
for (path, line), finding_ids in coverage_ids_by_line.items()
]
merged_state = dict(state)
all_findings = [*(state.get("findings") or []), *reference_findings]
all_findings = [*(state.get("findings") or []), *coverage_findings]
output_events: list[InspectionLedgerEvent] = []
finding_output_records = sum(max(1, len(finding.occurrences)) for finding in all_findings)
if finding_output_records > MAX_FINDING_OUTPUT_RECORDS:
Expand All @@ -140,7 +192,7 @@ def finalize_inspection_ledger(state: SkillspectorState) -> dict[str, object]:
merged_state["findings"] = all_findings
merged_state["effective_finding_ids"] = [
*(state.get("effective_finding_ids") or []),
*(finding.finding_id for finding in reference_findings),
*(finding.finding_id for finding in coverage_findings),
]
merged_state["inspection_ledger"] = [
*(state.get("inspection_ledger") or []),
Expand All @@ -157,15 +209,15 @@ def finalize_inspection_ledger(state: SkillspectorState) -> dict[str, object]:
*reference_statuses,
]
completeness, effective_finding_ids = finalize_ledger(merged_state)
if reference_findings and completeness["status"] == "complete":
if coverage_findings and completeness["status"] == "complete":
completeness["status"] = "partial"
completeness["is_complete"] = False
limitations = completeness.setdefault("limitations", [])
limitations.append("One or more referenced artifacts were not completely inspected.")
limitations.append("One or more artifacts were not completely inspected.")
return {
"analysis_completeness": completeness,
"execution_successful": completeness["execution_successful"],
"findings": reference_findings,
"findings": coverage_findings,
"effective_finding_ids": effective_finding_ids,
"inspection_ledger": [*reference_events, *output_events],
"analyzer_status_events": reference_statuses,
Expand Down
27 changes: 27 additions & 0 deletions tests/nodes/test_build_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -1389,6 +1389,33 @@ def test_build_context_reports_files_beyond_supported_envelope_as_partial(
)


def test_truncated_text_file_stays_in_llm_cache_with_audit_gap_marker(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A file past the read cap must still reach the LLM stage.

Excluding truncated files from ``llm_file_cache`` lets a payload hide
past the cap while the report shows zero findings. The LLM view is a
bounded prefix plus an explicit audit-gap marker instead.
"""
import skillspector.nodes.build_context as build_context_module

monkeypatch.setattr(build_context_module, "MAX_ANALYZABLE_FILE_BYTES", 64)
(tmp_path / "SKILL.md").write_text("# weather\n", encoding="utf-8")
(tmp_path / "server.py").write_text(
'PAYLOAD = "past-the-cut"\n' + "x" * 256 + "\n", encoding="utf-8"
)

result = build_context({"skill_path": str(tmp_path)})

assert "server.py" in result["llm_file_cache"]
cached = result["llm_file_cache"]["server.py"]
assert "audit" in cached and "gap" in cached
assert "server.py" in result["llm_components"]
artifact = next(item for item in result["artifact_inventory"] if item["path"] == "server.py")
assert artifact["disposition"] == "partial"


def test_build_context_shares_artifact_budget_across_child_bundles(tmp_path: Path) -> None:
"""A second child sees the artifact allowance already consumed by its sibling."""
from skillspector.cli import _TransitiveBudget, _TransitiveTraversalState
Expand Down
81 changes: 81 additions & 0 deletions tests/nodes/test_finalize_inspection_ledger.py
Original file line number Diff line number Diff line change
Expand Up @@ -754,6 +754,87 @@ def test_resolved_reference_ae1_disposition_matrix_is_llm_independent(
assert result["effective_finding_ids"] == []


@pytest.mark.parametrize(
("disposition", "reason", "expected_ae7"),
[
("analyzed", None, False),
("partial", "size_limit", True),
("partial", "total_bytes_limit", False),
("failed", "read_error", False),
],
)
def test_size_truncated_artifact_synthesizes_ae7(
disposition: str,
reason: str | None,
expected_ae7: bool,
) -> None:
"""A file past the per-file read cap must not yield a zero-finding report."""
item: dict[str, object] = {"path": "server.py", "disposition": disposition}
if reason is not None:
item["reason"] = reason
result = finalize_inspection_ledger(
{
"components": ["server.py"],
"findings": [],
"effective_finding_ids": [],
"artifact_inventory": [item],
"inspection_ledger": [],
"analyzer_status_events": [],
}
)

ae7 = [finding for finding in result["findings"] if finding.rule_id == "AE7"]
assert bool(ae7) is expected_ae7
if expected_ae7:
assert ae7[0].severity == "HIGH"
assert ae7[0].file == "server.py"
assert ae7[0].category == "analysis-evasion"
assert result["analysis_completeness"]["is_complete"] is False


def test_ae7_skips_paths_already_covered_by_ae1() -> None:
"""A referenced size-truncated artifact gets AE1, not AE1 + AE7."""
result = finalize_inspection_ledger(
{
"components": ["SKILL.md", "assets/big.py"],
"findings": [],
"effective_finding_ids": [],
"artifact_inventory": [
{
"path": "assets/big.py",
"disposition": "partial",
"reason": "size_limit",
"content_kind": "text",
}
],
"artifact_references": [
{
"source_path": "SKILL.md",
"line": 3,
"column": 1,
"evidence": "See [server](assets/big.py).",
"target_path": "assets/big.py",
"status": "resolved",
"disposition": "partial",
}
],
"inspection_ledger": [
ledger_event(
outcome=LedgerOutcome.PARTIAL,
record_type=LedgerRecordType.SYSTEM,
phase="cache",
path="assets/big.py",
reason=LedgerReason.SIZE_LIMIT,
)
],
"analyzer_status_events": [],
}
)

rule_ids = [finding.rule_id for finding in result["findings"]]
assert rule_ids == ["AE1"]


@pytest.mark.parametrize("status", ["missing", "ambiguous", "rejected"])
def test_unresolved_reference_does_not_synthesize_ae1(status: str) -> None:
result = finalize_inspection_ledger(
Expand Down
Loading