feat(context): spill oversized tool output to disk with a retrieval tool

Truncating a large tool result to a head+tail preview loses the middle,
which may hold the one line that matters (a buried match, a stack frame, a
credential). Instead of dropping it, persist the full output and let the
agent page back to it on demand.

- output_store.py: bound_and_store() writes the complete output to a
  per-scan store and embeds an output_id in the truncation notice;
  read_stored_output() serves it back in validated, paginated chunks
  (output_id is a 32-char hex token, guarding against path traversal).
- read_tool_output tool: lets the agent retrieve any elided output by id,
  paging via offset/limit.
- factory.py: bounded tool results now spill via bound_and_store; the new
  tool is registered for every scan agent.
- runner.py: point the store at the run's .state/tool-output directory so
  spilled output lives beside the rest of the scan state.

Falls back to a plain head+tail preview if the spill write fails.
This commit is contained in:
Ahmed Allam
2026-07-26 00:16:02 +00:00
committed by Devin AI
parent 4eedfc64b4
commit 1da3140d84
6 changed files with 223 additions and 21 deletions
+61 -2
View File
@@ -1,10 +1,20 @@
"""Tests for per-tool-output bounding."""
"""Tests for per-tool-output bounding and the durable spill store."""
from __future__ import annotations
import re
from typing import TYPE_CHECKING
from strix.tools.output_store import bound_text
from strix.tools.output_store import (
bound_and_store,
bound_text,
configure_output_store,
read_stored_output,
)
if TYPE_CHECKING:
from pathlib import Path
def test_small_output_passes_through_unchanged() -> None:
@@ -63,3 +73,52 @@ def test_dropped_line_count_accounts_for_byte_trimming() -> None:
assert dropped == 200 - len(kept)
# The naive middle-only count (max_lines split evenly) would under-report.
assert dropped > 200 - 20
def test_bound_and_store_small_output_not_spilled(tmp_path: Path) -> None:
configure_output_store(tmp_path)
text = "just a few lines\nsecond line"
assert bound_and_store(text, max_lines=100, max_bytes=10_000) == text
assert list(tmp_path.iterdir()) == []
def test_bound_and_store_spills_full_output_and_is_retrievable(tmp_path: Path) -> None:
configure_output_store(tmp_path)
text = "\n".join(f"secret-line-{i}" for i in range(1000))
bounded = bound_and_store(text, max_lines=10, max_bytes=1_000_000)
match = re.search(r'output_id="([0-9a-f]{32})"', bounded)
assert match is not None, bounded
output_id = match.group(1)
# The full, untruncated output round-trips through the store.
full = read_stored_output(output_id, offset=0, limit=10_000)
assert full.splitlines() == text.splitlines()
# A buried line elided from the preview is retrievable.
assert "secret-line-500" not in bounded
assert "secret-line-500" in full
def test_read_stored_output_paginates(tmp_path: Path) -> None:
configure_output_store(tmp_path)
text = "\n".join(str(i) for i in range(100))
output_id = re.search(
r'output_id="([0-9a-f]{32})"',
bound_and_store(text, max_lines=4, max_bytes=1_000_000),
)
assert output_id is not None
page = read_stored_output(output_id.group(1), offset=0, limit=10)
assert page.startswith("0\n1")
assert "more lines" in page
assert "offset=10" in page
def test_read_stored_output_rejects_traversal(tmp_path: Path) -> None:
configure_output_store(tmp_path)
assert "Invalid output_id" in read_stored_output("../../etc/passwd")
def test_read_stored_output_missing_id(tmp_path: Path) -> None:
configure_output_store(tmp_path)
assert "No stored output" in read_stored_output("0" * 32)