135 lines
5.6 KiB
Python
135 lines
5.6 KiB
Python
"""Real result preservation, bounded context, and zero-call observation receipts."""
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from artemis.config.constants import PLUGIN_PROMPT_TEXT_CHARS, PLUGIN_SUMMARY_TEXT_CHARS
|
|
from artemis.plugins.prompt_budget import bounded_result_text, read_result_text, save_result_text
|
|
from artemis.plugins.models import PluginCallResult
|
|
from artemis.plugins.runtime import PluginRuntime
|
|
|
|
|
|
def test_long_plugin_result_is_bounded_but_paged_evidence_is_lossless(tmp_path):
|
|
engine = SimpleNamespace(base_dir=tmp_path)
|
|
call_id = "a" * 32
|
|
original = "Address bar\n" + "关键控件与真实观察\n" * 10000 + "ERROR: window unavailable"
|
|
save_result_text(engine, call_id, original)
|
|
prompt = bounded_result_text(original, call_id=call_id)
|
|
assert len(prompt) <= PLUGIN_PROMPT_TEXT_CHARS
|
|
assert "Address bar" in prompt and "ERROR: window unavailable" in prompt
|
|
assert call_id in prompt and "节选" in prompt
|
|
recovered = "".join(
|
|
read_result_text(engine, call_id, offset).split("\n", 1)[1]
|
|
for offset in range(0, len(original), PLUGIN_PROMPT_TEXT_CHARS)
|
|
)
|
|
assert recovered == original
|
|
assert "已到结尾" in read_result_text(engine, call_id, len(original))
|
|
with pytest.raises(ValueError):
|
|
read_result_text(engine, call_id, len(original) + 1)
|
|
|
|
|
|
@pytest.mark.parametrize("bad", ["../outside", "a" * 31, "a" * 32 + ".txt", "A" * 32])
|
|
def test_plugin_evidence_reader_rejects_arbitrary_paths(tmp_path, bad):
|
|
with pytest.raises(ValueError):
|
|
read_result_text(SimpleNamespace(base_dir=tmp_path), bad)
|
|
|
|
|
|
def test_flash_plugin_output_is_bounded_without_losing_image_or_error():
|
|
outcome = PluginCallResult(
|
|
status="error",
|
|
plugin_id="pc",
|
|
text_ref="b" * 32,
|
|
content=[
|
|
{"type": "text", "text": "X" * 50000 + "Permission denied"},
|
|
{"type": "image", "mimeType": "image/png", "data": "AA=="},
|
|
],
|
|
)
|
|
result = PluginRuntime.content_blocks(outcome)
|
|
assert len(result[0]["text"]) <= PLUGIN_PROMPT_TEXT_CHARS
|
|
assert "Permission denied" in result[0]["text"]
|
|
assert result[1]["image_url"]["url"] == "data:image/png;base64,AA=="
|
|
assert len(outcome.content[0]["text"]) > PLUGIN_PROMPT_TEXT_CHARS
|
|
|
|
|
|
def test_read_only_observation_needs_no_summary_model(monkeypatch):
|
|
from artemis.agents.flash.summarizer import VisualStepSummarizer
|
|
|
|
model = MagicMock()
|
|
monkeypatch.setattr("artemis.agents.flash.summarizer.get_llm", lambda *a, **k: model)
|
|
engine = MagicMock()
|
|
service = VisualStepSummarizer(SimpleNamespace(data_engine=engine))
|
|
service.dispatch(
|
|
step_number=2,
|
|
action_name="plug_test_get_app_state",
|
|
action_args={},
|
|
pre_img_bytes=None,
|
|
post_img_bytes=b"actual screenshot",
|
|
exec_outcome="UI" * 20000,
|
|
data_engine_step_id="step-2",
|
|
action_key="call-2",
|
|
receipt_only=True,
|
|
)
|
|
assert not service.is_pending("call-2")
|
|
summary = service.get_summary("call-2")
|
|
assert summary and "不代表视觉验收通过" in summary
|
|
assert len(summary) < PLUGIN_SUMMARY_TEXT_CHARS + 150
|
|
assert service.get_job_payload("call-2")["post_img_bytes"] is None
|
|
assert service.get_job_payload("call-2")["data_engine_step_id"] == "step-2"
|
|
assert not service._pending_tasks
|
|
model.ainvoke.assert_not_called()
|
|
assert engine.update_step_summary.call_args.kwargs["source"] == "tool_receipt"
|
|
|
|
|
|
def test_assistant_message_count_does_not_change_exploration_context(monkeypatch):
|
|
from artemis.exploration.runtime import conversation_message
|
|
|
|
session = {"goal": "inspect website", "message_count": 10}
|
|
history = [{"sequence": 1, "role": "user", "text": "inspect website"}]
|
|
store = SimpleNamespace(
|
|
get=lambda _: session, recent_messages=lambda *a, **k: history, case=lambda _: None
|
|
)
|
|
monkeypatch.setattr("artemis.exploration.runtime.explorations", store)
|
|
monkeypatch.setattr("artemis.exploration.editorial.prior_context", lambda *a, **k: [])
|
|
ctx = SimpleNamespace(exploration_id="explore", exploration_run_id="run")
|
|
first = conversation_message(ctx).content
|
|
session["message_count"] = 99
|
|
assert conversation_message(ctx).content == first
|
|
history.append({"sequence": 99, "role": "user", "text": "return home"})
|
|
assert "return home" in conversation_message(ctx).content
|
|
|
|
|
|
def test_observation_steps_do_not_force_plan_bookkeeping():
|
|
from artemis.agents.operator.prompts import unwritten_action_streak
|
|
|
|
write = {"action_taken": "click"}
|
|
read = {
|
|
"action_taken": "get_app_state",
|
|
"last_execution_result": {"execution": [{"read_only": True}]},
|
|
}
|
|
assert unwritten_action_streak([write, read, read, read, read]) == 1
|
|
assert unwritten_action_streak([write] * 4) == 4
|
|
|
|
|
|
def test_flash_plugin_summary_caps_text_before_model_submission(monkeypatch):
|
|
from artemis.agents.flash.summarizer import VisualStepSummarizer
|
|
|
|
monkeypatch.setattr("artemis.agents.flash.summarizer.get_llm", lambda *a, **k: MagicMock())
|
|
service = VisualStepSummarizer(SimpleNamespace(data_engine=None))
|
|
submit = MagicMock()
|
|
monkeypatch.setattr(service, "submit", submit)
|
|
service.dispatch(
|
|
step_number=1,
|
|
action_name="plug_test_click",
|
|
action_args={},
|
|
pre_img_bytes=None,
|
|
post_img_bytes=b"image",
|
|
exec_outcome="target clicked" * 10000,
|
|
post_ui_elements_text="control tree" * 10000,
|
|
)
|
|
payload = submit.call_args.args[1]
|
|
assert len(payload["exec_outcome"]) <= PLUGIN_SUMMARY_TEXT_CHARS
|
|
assert len(payload["post_ui_elements_text"]) <= PLUGIN_SUMMARY_TEXT_CHARS
|
|
assert payload["post_img_bytes"] == b"image"
|