wehub-resource-sync
341 lines
12 KiB
Python
341 lines
12 KiB
Python
"""Opt-in provider-view history dedup lever.
|
|
|
|
Covers OPENSQUILLA_PROVIDER_HISTORY_DEDUP (off by default). Motivation: long
|
|
single-turn episodes re-run the same
|
|
read/grep/diff commands and full-history replay resends every byte-identical
|
|
tool_result on every iteration, paying quadratic cost. When enabled, older
|
|
byte-identical tool results are replaced in the provider request projection
|
|
with a short back-reference to the surviving newest copy; persisted history is
|
|
never mutated.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterator
|
|
from typing import Any
|
|
|
|
from opensquilla.engine import Agent, AgentConfig
|
|
from opensquilla.provider import (
|
|
ContentBlockToolResult,
|
|
ContentBlockToolUse,
|
|
Message,
|
|
)
|
|
from opensquilla.provider.types import ChatConfig
|
|
|
|
|
|
class _StubProvider:
|
|
provider_name = "fake"
|
|
|
|
def chat(
|
|
self,
|
|
messages: list[Message],
|
|
tools: list[Any] | None = None,
|
|
config: ChatConfig | None = None,
|
|
) -> AsyncIterator[Any]:
|
|
return self._stream()
|
|
|
|
async def _stream(self) -> AsyncIterator[Any]:
|
|
return
|
|
yield
|
|
|
|
async def list_models(self) -> list[Any]:
|
|
return []
|
|
|
|
|
|
# A payload comfortably above _PROVIDER_HISTORY_DEDUP_MIN_CHARS (400).
|
|
BIG_RESULT = "line of grep output\n" * 60
|
|
|
|
|
|
def _pair(use_id: str, content: str, *, is_error: bool = False) -> list[Message]:
|
|
return [
|
|
Message(
|
|
role="assistant",
|
|
content=[
|
|
ContentBlockToolUse(
|
|
id=use_id, name="grep_search", input={"pattern": "foo"}
|
|
),
|
|
],
|
|
),
|
|
Message(
|
|
role="user",
|
|
content=[
|
|
ContentBlockToolResult(
|
|
tool_use_id=use_id, content=content, is_error=is_error
|
|
)
|
|
],
|
|
),
|
|
]
|
|
|
|
|
|
def _history(*contents: tuple[str, str]) -> list[Message]:
|
|
messages: list[Message] = [Message(role="user", content="find the bug")]
|
|
for use_id, content in contents:
|
|
messages.extend(_pair(use_id, content))
|
|
return messages
|
|
|
|
|
|
def _tool_results(messages: list[Message]) -> list[ContentBlockToolResult]:
|
|
results: list[ContentBlockToolResult] = []
|
|
for message in messages:
|
|
if not isinstance(message.content, list):
|
|
continue
|
|
for block in message.content:
|
|
if isinstance(block, ContentBlockToolResult):
|
|
results.append(block)
|
|
return results
|
|
|
|
|
|
def test_dedup_off_by_default() -> None:
|
|
assert AgentConfig().provider_history_dedup_enabled is False
|
|
agent = Agent(provider=_StubProvider(), config=AgentConfig())
|
|
history = _history(("g1", BIG_RESULT), ("g2", BIG_RESULT), ("g3", BIG_RESULT))
|
|
projected = agent._dedup_repeated_tool_results_for_provider(history)
|
|
contents = [r.content for r in _tool_results(projected)]
|
|
assert contents == [BIG_RESULT, BIG_RESULT, BIG_RESULT]
|
|
|
|
|
|
def test_dedup_elides_older_identical_results_keeps_newest() -> None:
|
|
agent = Agent(
|
|
provider=_StubProvider(),
|
|
config=AgentConfig(provider_history_dedup_enabled=True),
|
|
)
|
|
# g1..g3 are byte-identical; g4 is unique. The recent-tail guard covers the
|
|
# last two results (g3, g4), so g1 and g2 are elided and g3 survives full.
|
|
history = _history(
|
|
("g1", BIG_RESULT),
|
|
("g2", BIG_RESULT),
|
|
("g3", BIG_RESULT),
|
|
("g4", "unique tail result " * 40),
|
|
)
|
|
projected = agent._dedup_repeated_tool_results_for_provider(history)
|
|
results = _tool_results(projected)
|
|
for stub in (results[0], results[1]):
|
|
assert stub.content.startswith("[duplicate_tool_result_elided]\n")
|
|
assert "identical_to_tool_use_id: g3" in stub.content
|
|
# The surviving newest identical copy (g3) stays full.
|
|
assert results[2].content == BIG_RESULT
|
|
assert agent.config.metadata["provider_history_dedup_elided"] == 2
|
|
assert agent.config.metadata["provider_history_dedup_chars_saved"] > 0
|
|
|
|
|
|
def test_dedup_is_idempotent_across_iterations() -> None:
|
|
agent = Agent(
|
|
provider=_StubProvider(),
|
|
config=AgentConfig(provider_history_dedup_enabled=True),
|
|
)
|
|
history = _history(
|
|
("g1", BIG_RESULT),
|
|
("g2", BIG_RESULT),
|
|
("g3", BIG_RESULT),
|
|
("g4", "unique tail result " * 40),
|
|
)
|
|
once = agent._dedup_repeated_tool_results_for_provider(history)
|
|
twice = agent._dedup_repeated_tool_results_for_provider(once)
|
|
assert [r.content for r in _tool_results(twice)] == [
|
|
r.content for r in _tool_results(once)
|
|
]
|
|
|
|
|
|
def test_dedup_leaves_recent_tail_duplicates_alone() -> None:
|
|
agent = Agent(
|
|
provider=_StubProvider(),
|
|
config=AgentConfig(provider_history_dedup_enabled=True),
|
|
)
|
|
# The only duplicates are the two most recent results - the model is
|
|
# actively looking at them, so nothing is elided.
|
|
history = _history(
|
|
("u1", "unique first result " * 40),
|
|
("g1", BIG_RESULT),
|
|
("g2", BIG_RESULT),
|
|
)
|
|
projected = agent._dedup_repeated_tool_results_for_provider(history)
|
|
contents = [r.content for r in _tool_results(projected)]
|
|
assert contents.count(BIG_RESULT) == 2
|
|
|
|
|
|
def test_dedup_does_not_mutate_persisted_history() -> None:
|
|
agent = Agent(
|
|
provider=_StubProvider(),
|
|
config=AgentConfig(provider_history_dedup_enabled=True),
|
|
)
|
|
history = _history(
|
|
("g1", BIG_RESULT),
|
|
("g2", BIG_RESULT),
|
|
("g3", BIG_RESULT),
|
|
("g4", "tail " * 120),
|
|
)
|
|
before = [r.content for r in _tool_results(history)]
|
|
agent._dedup_repeated_tool_results_for_provider(history)
|
|
after = [r.content for r in _tool_results(history)]
|
|
assert before == after
|
|
|
|
|
|
def test_dedup_skips_error_results() -> None:
|
|
agent = Agent(
|
|
provider=_StubProvider(),
|
|
config=AgentConfig(provider_history_dedup_enabled=True),
|
|
)
|
|
history = _history(
|
|
("e1", BIG_RESULT),
|
|
("e2", BIG_RESULT),
|
|
("e3", BIG_RESULT),
|
|
("t1", "tail " * 120),
|
|
)
|
|
# Mark the first three as errors post-hoc.
|
|
for use_id in ("e1", "e2", "e3"):
|
|
for message in history:
|
|
if isinstance(message.content, list):
|
|
for block in message.content:
|
|
if (
|
|
isinstance(block, ContentBlockToolResult)
|
|
and block.tool_use_id == use_id
|
|
):
|
|
block.is_error = True
|
|
projected = agent._dedup_repeated_tool_results_for_provider(history)
|
|
contents = [r.content for r in _tool_results(projected)]
|
|
assert all(not c.startswith("[duplicate_tool_result_elided]") for c in contents)
|
|
|
|
|
|
def test_dedup_skips_small_results_below_min_chars() -> None:
|
|
agent = Agent(
|
|
provider=_StubProvider(),
|
|
config=AgentConfig(provider_history_dedup_enabled=True),
|
|
)
|
|
small = "tiny" # < 400 chars
|
|
history = _history(
|
|
("s1", small),
|
|
("s2", small),
|
|
("s3", small),
|
|
("t1", "tail " * 120),
|
|
)
|
|
projected = agent._dedup_repeated_tool_results_for_provider(history)
|
|
contents = [r.content for r in _tool_results(projected)]
|
|
assert contents.count(small) == 3
|
|
|
|
|
|
def test_dedup_respects_min_repeats_threshold() -> None:
|
|
# With min_repeats=3, two copies are not enough to elide.
|
|
agent = Agent(
|
|
provider=_StubProvider(),
|
|
config=AgentConfig(
|
|
provider_history_dedup_enabled=True,
|
|
provider_history_dedup_min_repeats=3,
|
|
),
|
|
)
|
|
history = _history(
|
|
("g1", BIG_RESULT),
|
|
("g2", BIG_RESULT),
|
|
("t1", "tail " * 120),
|
|
("t2", "other " * 120),
|
|
)
|
|
projected = agent._dedup_repeated_tool_results_for_provider(history)
|
|
contents = [r.content for r in _tool_results(projected)]
|
|
assert contents.count(BIG_RESULT) == 2
|
|
|
|
|
|
def test_dedup_groups_across_iterations_despite_frozen_survivor() -> None:
|
|
# A block already frozen full (shown in a prior request) must still count
|
|
# toward the digest group so newer duplicates get elided — it must only
|
|
# be excluded from being re-stubbed itself.
|
|
agent = Agent(
|
|
provider=_StubProvider(),
|
|
config=AgentConfig(
|
|
provider_history_dedup_enabled=True,
|
|
provider_history_dedup_min_repeats=3,
|
|
),
|
|
)
|
|
iter1_history = _history(("g1", BIG_RESULT))
|
|
iter1_projected = agent._dedup_repeated_tool_results_for_provider(iter1_history)
|
|
agent._remember_provider_visible_tool_results(iter1_projected)
|
|
assert "g1" in agent._provider_tool_result_frozen_full_ids
|
|
|
|
iter2_history = _history(
|
|
("g1", BIG_RESULT),
|
|
("g2", BIG_RESULT),
|
|
("g3", BIG_RESULT),
|
|
("u1", "unique final result " * 40),
|
|
)
|
|
projected = agent._dedup_repeated_tool_results_for_provider(iter2_history)
|
|
results = {r.tool_use_id: r.content for r in _tool_results(projected)}
|
|
# g1 stays full (frozen, protected from retroactive downgrade).
|
|
assert results["g1"] == BIG_RESULT
|
|
# g2 is now correctly elided against the newest survivor g3.
|
|
assert results["g2"].startswith("[duplicate_tool_result_elided]\n")
|
|
assert "identical_to_tool_use_id: g3" in results["g2"]
|
|
assert results["g3"] == BIG_RESULT
|
|
|
|
|
|
def test_dedup_stub_is_not_permanently_frozen() -> None:
|
|
# Elision depends on another block's state (its survivor), so it must
|
|
# never be frozen into a permanent override — only recomputed fresh.
|
|
agent = Agent(
|
|
provider=_StubProvider(),
|
|
config=AgentConfig(provider_history_dedup_enabled=True),
|
|
)
|
|
history = _history(
|
|
("g1", BIG_RESULT),
|
|
("g2", BIG_RESULT),
|
|
("g3", BIG_RESULT),
|
|
("u1", "unique final result " * 40),
|
|
)
|
|
projected = agent._dedup_repeated_tool_results_for_provider(history)
|
|
agent._remember_provider_visible_tool_results(projected)
|
|
|
|
assert "g1" not in agent._provider_tool_result_frozen_overrides
|
|
assert "g1" not in agent._provider_tool_result_frozen_full_ids
|
|
assert "g2" not in agent._provider_tool_result_frozen_overrides
|
|
assert "g2" not in agent._provider_tool_result_frozen_full_ids
|
|
|
|
|
|
def test_compact_does_not_recompact_dedup_survivor_in_same_request() -> None:
|
|
# The dedup survivor is the one full copy other stubs point back to; the
|
|
# aggregate compaction pass running right after must not turn it into a
|
|
# compacted stub too, or the back-references become dangling.
|
|
agent = Agent(
|
|
provider=_StubProvider(),
|
|
config=AgentConfig(
|
|
provider_history_dedup_enabled=True,
|
|
context_window_tokens=100,
|
|
),
|
|
)
|
|
history = _history(
|
|
("g1", BIG_RESULT),
|
|
("g2", BIG_RESULT),
|
|
("g3", BIG_RESULT),
|
|
("u1", "unique tail result one " * 40),
|
|
("u2", "unique tail result two " * 40),
|
|
)
|
|
deduped = agent._dedup_repeated_tool_results_for_provider(history)
|
|
assert agent._provider_history_dedup_survivor_ids == {"g3"}
|
|
|
|
compacted = agent._compact_aggregate_tool_results_for_provider(deduped)
|
|
results = {r.tool_use_id: r.content for r in _tool_results(compacted)}
|
|
assert results["g3"] == BIG_RESULT
|
|
|
|
|
|
def test_dedup_env_plumbing(monkeypatch) -> None:
|
|
from opensquilla.engine.turn_runner.agent_bootstrap_stage import (
|
|
_bool_from_env,
|
|
_positive_int_from_env,
|
|
)
|
|
|
|
assert _bool_from_env("OPENSQUILLA_PROVIDER_HISTORY_DEDUP", False) is False
|
|
assert _positive_int_from_env("OPENSQUILLA_PROVIDER_HISTORY_DEDUP_MIN_REPEATS", 2) == 2
|
|
monkeypatch.setenv("OPENSQUILLA_PROVIDER_HISTORY_DEDUP", "1")
|
|
monkeypatch.setenv("OPENSQUILLA_PROVIDER_HISTORY_DEDUP_MIN_REPEATS", "4")
|
|
assert _bool_from_env("OPENSQUILLA_PROVIDER_HISTORY_DEDUP", False) is True
|
|
assert _positive_int_from_env("OPENSQUILLA_PROVIDER_HISTORY_DEDUP_MIN_REPEATS", 2) == 4
|
|
|
|
|
|
def test_tool_result_artifact_detection_is_cached() -> None:
|
|
from opensquilla.engine.agent import _tool_result_content_has_artifact
|
|
|
|
content = '{"status": "published", "marker": "cache-test-unique-9f3"}'
|
|
_tool_result_content_has_artifact.cache_clear()
|
|
before = _tool_result_content_has_artifact.cache_info()
|
|
assert _tool_result_content_has_artifact(content) is True
|
|
assert _tool_result_content_has_artifact(content) is True
|
|
after = _tool_result_content_has_artifact.cache_info()
|
|
assert after.hits == before.hits + 1
|
|
assert after.misses == before.misses + 1
|