fix: align Responses messages and video exports

This commit is contained in:
chenmingxuan committed 2026-09-29 00:12:35 +08:00
1 parent e1e26772d9
commit c818a09a49
10 files changed
+277 -18

No files matched your search

+5 -3
View File
@@ -632,15 +632,17 @@ def session_videos(report: dict) -> list[tuple[Path, str]]:
if not candidate: if not candidate:
return [] return []
source = Path(str(candidate)).resolve() source = Path(str(candidate)).resolve()
session_root = (TRACES_PATH / identifier(report["id"])).resolve() trace_root = TRACES_PATH.resolve()
if not source.is_relative_to(session_root): # Workers save recordings in their finalized web_<time>_<id>_PASS/FAIL
# folder. The recording row already ties that path to this session.
if not source.is_relative_to(trace_root):
return [] return []
# A resumed recording lives under resume_NNNN; the initial recording is # A resumed recording lives under resume_NNNN; the initial recording is
# in its parent, with any later segments in sibling resume folders. # in its parent, with any later segments in sibling resume folders.
directory = source.parent directory = source.parent
if directory.name.startswith("resume_"): if directory.name.startswith("resume_"):
directory = directory.parent directory = directory.parent
if not directory.is_dir() or not directory.is_relative_to(session_root): if not directory.is_dir() or not directory.is_relative_to(trace_root):
return [] return []
paths = [ paths = [
p p
+4 -1
View File
@@ -78,6 +78,7 @@ from artemis.controllers.unified_controller import UnifiedMobileController
from artemis.data_engine.trace import trace from artemis.data_engine.trace import trace
from artemis.graph.perception import _check_injected_instruction_file from artemis.graph.perception import _check_injected_instruction_file
from artemis.graph.state import State from artemis.graph.state import State
from artemis.llm.router import ModelEndpoint
from artemis.llm.structured import ParseFailure, parse_structured from artemis.llm.structured import ParseFailure, parse_structured
from artemis.mcp.action_executor import McpActionExecutor from artemis.mcp.action_executor import McpActionExecutor
from artemis.mcp.observation import observe from artemis.mcp.observation import observe
@@ -288,7 +289,9 @@ class FlashRunner:
logger.warning(f"Failed to get operator LLM from config, using default: {e}") logger.warning(f"Failed to get operator LLM from config, using default: {e}")
return RobustChatModelWrapper( return RobustChatModelWrapper(
get_openai_llm(model_name=settings.OPENAI_MODEL), self.ctx get_openai_llm(model_name=settings.OPENAI_MODEL),
self.ctx,
endpoint=ModelEndpoint(model_name=settings.OPENAI_MODEL),
) )
def _render_system_prompt(self, tools_declaration: list) -> str: def _render_system_prompt(self, tools_declaration: list) -> str:
+3 -1
View File
@@ -36,6 +36,7 @@ from jinja2 import Template
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from artemis.context import ArtemisContext from artemis.context import ArtemisContext
from artemis.llm.messages import adapt_responses_messages
from artemis.memory.step_memory import JobKey, StepMemoryService from artemis.memory.step_memory import JobKey, StepMemoryService
from artemis.services.llm import RobustChatModelWrapper, get_openai_llm, get_llm from artemis.services.llm import RobustChatModelWrapper, get_openai_llm, get_llm
from artemis.services.token_meter import record_llm_usage from artemis.services.token_meter import record_llm_usage
@@ -462,7 +463,8 @@ class VisualStepSummarizer(StepMemoryService):
] ]
response = await asyncio.wait_for( response = await asyncio.wait_for(
self._llm.ainvoke(messages), timeout=self._timeout_seconds self._llm.ainvoke(adapt_responses_messages(messages, self._model_name)),
timeout=self._timeout_seconds,
) )
self._meter_lens_call(response) self._meter_lens_call(response)
summary_raw = response.content if isinstance(response.content, str) else "" summary_raw = response.content if isinstance(response.content, str) else ""
+69
View File
@@ -0,0 +1,69 @@
"""Adapt LangChain messages for Responses-compatible model families."""
from collections.abc import Sequence
from typing import Any
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
_USER_ROLE_MODELS = {"mimo-v2.6-pro", "mimo-v2.6-flash"}
def adapt_responses_messages(
messages: Sequence[BaseMessage], model_name: str | None
) -> Sequence[BaseMessage]:
"""Send selected models user-role instructions without changing stored history.
Full Artemis requests to GPT models can fail with system-role input. The
equivalent user-role request works with GPT and the two tested Mimo models.
Other models keep their existing message roles.
"""
if not model_name:
return messages
model = model_name.strip().lower()
if not (model.startswith("gpt") or model in _USER_ROLE_MODELS):
return messages
if not any(isinstance(message, SystemMessage) for message in messages):
return messages
adapted: list[BaseMessage] = []
pending_blocks: list[dict[str, Any]] = []
for message in messages:
if isinstance(message, SystemMessage):
content = message.content
if isinstance(content, str):
pending_blocks.append({"type": "text", "text": content})
else:
pending_blocks.extend(
{"type": "text", "text": block} if isinstance(block, str) else block
for block in content
)
continue
if pending_blocks:
if isinstance(message, HumanMessage):
content = message.content
blocks = (
[{"type": "text", "text": content}]
if isinstance(content, str)
else list(content)
)
adapted.append(
message.model_copy(
update={
"content": [
*pending_blocks,
*blocks,
]
}
)
)
pending_blocks = []
continue
adapted.append(HumanMessage(content=pending_blocks))
pending_blocks = []
adapted.append(message)
if pending_blocks:
adapted.append(HumanMessage(content=pending_blocks))
return adapted
+3 -1
View File
@@ -46,6 +46,7 @@ from uuid import uuid4
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from artemis.llm.messages import adapt_responses_messages
from artemis.memory.step_memory import JobKey, StepLens, StepMemoryService from artemis.memory.step_memory import JobKey, StepLens, StepMemoryService
from artemis.memory.transcript import format_session_offset from artemis.memory.transcript import format_session_offset
from artemis.utils.logger import get_logger from artemis.utils.logger import get_logger
@@ -503,7 +504,8 @@ class StepCapsuleLens(StepLens):
return parsed return parsed
async def _invoke(self, llm: Any, messages: list[BaseMessage], model_label: str): async def _invoke(self, llm: Any, messages: list[BaseMessage], model_label: str):
response = await asyncio.wait_for(llm.ainvoke(messages), timeout=90.0) request_messages = adapt_responses_messages(messages, model_label)
response = await asyncio.wait_for(llm.ainvoke(request_messages), timeout=90.0)
# Raw-model bypass metering (gateway-wrapped models meter themselves); # Raw-model bypass metering (gateway-wrapped models meter themselves);
# capsule prompts are small and are kept out of the session meter's # capsule prompts are small and are kept out of the session meter's
# last_prompt_tokens (the compaction base itself is operator-owned on # last_prompt_tokens (the compaction base itself is operator-owned on
+22
View File
@@ -54,6 +54,7 @@ from artemis.llm.reliability import (
classify_failure, classify_failure,
retry_policy_for, retry_policy_for,
) )
from artemis.llm.messages import adapt_responses_messages
from artemis.llm.router import ModelEndpoint, ModelFactory, ModelProvider from artemis.llm.router import ModelEndpoint, ModelFactory, ModelProvider
from artemis.services.token_meter import record_llm_usage from artemis.services.token_meter import record_llm_usage
from artemis.llm.structured import ( from artemis.llm.structured import (
@@ -568,6 +569,25 @@ class RobustChatModelWrapper:
args, kwargs = _inject_parent_trace_id(trace_id, *args, **kwargs) args, kwargs = _inject_parent_trace_id(trace_id, *args, **kwargs)
return args, kwargs, trace_id return args, kwargs, trace_id
def _prepare_model_call(self, args: tuple, kwargs: dict) -> tuple[tuple, dict]:
model_name = self.endpoint.model_name if self.endpoint else None
if (
args
and isinstance(args[0], (list, tuple))
and all(isinstance(message, BaseMessage) for message in args[0])
):
return (adapt_responses_messages(args[0], model_name), *args[1:]), kwargs
if (
"input" in kwargs
and isinstance(kwargs["input"], (list, tuple))
and all(isinstance(message, BaseMessage) for message in kwargs["input"])
):
return args, {
**kwargs,
"input": adapt_responses_messages(kwargs["input"], model_name),
}
return args, kwargs
def _meter_usage(self, response) -> None: def _meter_usage(self, response) -> None:
"""Record measured token usage for this call (M0 metering, best-effort).""" """Record measured token usage for this call (M0 metering, best-effort)."""
engine = self.ctx.data_engine if self.ctx else None engine = self.ctx.data_engine if self.ctx else None
@@ -576,6 +596,7 @@ class RobustChatModelWrapper:
record_llm_usage(engine, response, source=self._endpoint_key()) record_llm_usage(engine, response, source=self._endpoint_key())
async def ainvoke(self, *args, **kwargs): async def ainvoke(self, *args, **kwargs):
args, kwargs = self._prepare_model_call(args, kwargs)
args, kwargs, _ = self._traced_call(args, kwargs) args, kwargs, _ = self._traced_call(args, kwargs)
response = await _run_with_recovery( response = await _run_with_recovery(
functools.partial(self.base_model.ainvoke, *args, **kwargs), functools.partial(self.base_model.ainvoke, *args, **kwargs),
@@ -597,6 +618,7 @@ class RobustChatModelWrapper:
signals the UI to drop it, and retries the whole call per policy; signals the UI to drop it, and retries the whole call per policy;
partial or duplicated chunks can never reach message history. partial or duplicated chunks can never reach message history.
""" """
args, kwargs = self._prepare_model_call(args, kwargs)
args, kwargs, _ = self._traced_call(args, kwargs) args, kwargs, _ = self._traced_call(args, kwargs)
emit_deltas = bool(self.ctx and self.ctx.data_engine) emit_deltas = bool(self.ctx and self.ctx.data_engine)
response = await _run_with_recovery( response = await _run_with_recovery(
@@ -544,6 +544,43 @@ def test_task_material_zip_contains_ordered_evidence_and_reports(
assert "1.1-all.zip" in response.headers["content-disposition"] assert "1.1-all.zip" in response.headers["content-disposition"]
def test_video_export_accepts_finalized_worker_trace_and_rejects_external_path(
isolated, tmp_path, monkeypatch
):
job = isolated.jobs(service.create_batch(payload())["id"])[0]
session_id = job["session_id"]
recording_dir = reports.TRACES_PATH / f"web_123_{session_id[:8]}_PASS_2026-09-28"
recording_dir.mkdir(parents=True)
first = recording_dir / "recording.mp4"
first.write_bytes(b"first segment")
resumed = recording_dir / "resume_0001"
resumed.mkdir()
second = resumed / "recording.mp4"
second.write_bytes(b"second segment")
repository = Mock()
repository.get_video_recording_for_session.return_value = {"local_video_path": str(first)}
repository.get_session_by_id.return_value = {"video_filepath": str(first)}
monkeypatch.setattr(reports, "SessionRepository", lambda: repository)
archive_path = reports.export_session_media(session_id, "video")
with zipfile.ZipFile(archive_path) as archive:
names = archive.namelist()
assert names == [
"任务录屏/0001_web_123_" + session_id[:8] + "_PASS_2026-09-28_recording.mp4",
"任务录屏/0002_resume_0001_recording.mp4",
"文件清单.json",
]
assert archive.read(names[0]) == b"first segment"
assert archive.read(names[1]) == b"second segment"
external = tmp_path / "outside" / "recording.mp4"
external.parent.mkdir()
external.write_bytes(b"unrelated")
repository.get_video_recording_for_session.return_value = {"local_video_path": str(external)}
assert reports.session_videos({"id": session_id}) == []
def test_history_filters_all_persisted_rows_and_escapes_like(client, isolated): def test_history_filters_all_persisted_rows_and_escapes_like(client, isolated):
batch = service.create_batch(payload()) batch = service.create_batch(payload())
first, second = isolated.jobs(batch["id"]) first, second = isolated.jobs(batch["id"])
+21 -10
View File
@@ -39,6 +39,14 @@ def mock_context():
return ctx return ctx
def _user_role_prompt(messages):
"""Return the instruction text and observation blocks sent to the model."""
assert len(messages) == 1
assert isinstance(messages[0], HumanMessage)
blocks = messages[0].content
return blocks[0]["text"], blocks[1:]
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_summarizer_dispatch_and_caching(mock_context): async def test_summarizer_dispatch_and_caching(mock_context):
"""Verify non-blocking dispatch and summary generation.""" """Verify non-blocking dispatch and summary generation."""
@@ -425,8 +433,8 @@ async def test_single_frame_dispatch_uses_single_image_prompt_variant(mock_conte
await summarizer.flush() await summarizer.flush()
assert summarizer.has_summary(1) assert summarizer.has_summary(1)
system_text = str(captured[0][0].content) system_text, human_blocks = _user_role_prompt(captured[0])
human_text = str(captured[0][1].content) human_text = str(human_blocks)
assert "Exactly ONE screenshot" in system_text assert "Exactly ONE screenshot" in system_text
assert "no independent post-action evidence" in system_text assert "no independent post-action evidence" in system_text
assert "DECISION FRAME" in human_text assert "DECISION FRAME" in human_text
@@ -464,8 +472,8 @@ async def test_dual_frame_dispatch_keeps_transition_prompt(mock_context):
await summarizer.flush() await summarizer.flush()
assert summarizer.has_summary(1) assert summarizer.has_summary(1)
system_text = str(captured[0][0].content) system_text, human_blocks = _user_role_prompt(captured[0])
human_text = str(captured[0][1].content) human_text = str(human_blocks)
assert "Before/After screenshots" in system_text assert "Before/After screenshots" in system_text
assert "--- [1] BEFORE ACTION SCREEN (Action Marked Visually in Red) ---" in human_text assert "--- [1] BEFORE ACTION SCREEN (Action Marked Visually in Red) ---" in human_text
assert "--- [2] AFTER ACTION SCREEN ---" in human_text assert "--- [2] AFTER ACTION SCREEN ---" in human_text
@@ -628,8 +636,8 @@ async def test_focus_block_and_provenance_marked_action_phrase_reach_the_lens(mo
await summarizer.flush() await summarizer.flush()
assert summarizer.has_summary(3) assert summarizer.has_summary(3)
system_text = str(captured[0][0].content) system_text, human_blocks = _user_role_prompt(captured[0])
lead_text = captured[0][1].content[0]["text"] lead_text = human_blocks[0]["text"]
assert "# OPERATOR FOCUS" in system_text assert "# OPERATOR FOCUS" in system_text
assert "Absence is a fact" in system_text assert "Absence is a fact" in system_text
assert lead_text.startswith( assert lead_text.startswith(
@@ -643,7 +651,7 @@ async def test_focus_block_and_provenance_marked_action_phrase_reach_the_lens(mo
assert "User instruction at this step: Do not join any network yet." in lead_text assert "User instruction at this step: Do not join any network yet." in lead_text
assert "expect it to read On." in lead_text assert "expect it to read On." in lead_text
# The focus precedes the frames: context first, evidence second. # The focus precedes the frames: context first, evidence second.
assert captured[0][1].content[1]["text"].startswith("--- [1] BEFORE ACTION SCREEN") assert human_blocks[1]["text"].startswith("--- [1] BEFORE ACTION SCREEN")
@pytest.mark.asyncio @pytest.mark.asyncio
@@ -672,7 +680,8 @@ async def test_dispatch_without_focus_carries_no_focus_block(mock_context):
) )
await summarizer.flush() await summarizer.flush()
lead_text = captured[0][1].content[0]["text"] _, human_blocks = _user_role_prompt(captured[0])
lead_text = human_blocks[0]["text"]
assert lead_text.startswith("Step 1 Physical Action: Tapped 'Search' at [500, 600]\n") assert lead_text.startswith("Step 1 Physical Action: Tapped 'Search' at [500, 600]\n")
assert "(self-described)" not in lead_text assert "(self-described)" not in lead_text
assert FOCUS_BLOCK_HEADER not in lead_text assert FOCUS_BLOCK_HEADER not in lead_text
@@ -706,7 +715,8 @@ async def test_single_frame_focus_label_forbids_describing_the_outcome(mock_cont
) )
await summarizer.flush() await summarizer.flush()
lead_text = captured[0][1].content[0]["text"] _, human_blocks = _user_role_prompt(captured[0])
lead_text = human_blocks[0]["text"]
assert "the expected outcome is NOT in this frame and must not be described" in lead_text assert "the expected outcome is NOT in this frame and must not be described" in lead_text
assert "what screen change it expected" not in lead_text assert "what screen change it expected" not in lead_text
# The focus text is released once the summary landed, like the frames. # The focus text is released once the summary landed, like the frames.
@@ -798,7 +808,8 @@ async def test_verbatim_argument_line_appears_only_when_the_phrase_folds_argumen
lead_by_step = {} lead_by_step = {}
for messages in captured: for messages in captured:
text = messages[1].content[0]["text"] _, human_blocks = _user_role_prompt(messages)
text = human_blocks[0]["text"]
lead_by_step[text.split(" ")[1]] = text lead_by_step[text.split(" ")[1]] = text
launch = lead_by_step["1"] launch = lead_by_step["1"]
assert launch.startswith("Step 1 Physical Action: Launched app 'Settings'\nController Outcome:") assert launch.startswith("Step 1 Physical Action: Launched app 'Settings'\nController Outcome:")
+26 -1
View File
@@ -2,10 +2,11 @@
from types import SimpleNamespace from types import SimpleNamespace
from langchain_core.messages import AIMessage, AIMessageChunk from langchain_core.messages import AIMessage, AIMessageChunk, HumanMessage, SystemMessage
import pytest import pytest
from artemis.llm.reliability import LLMExhaustedError, LLMPermanentError from artemis.llm.reliability import LLMExhaustedError, LLMPermanentError
from artemis.llm.router import ModelEndpoint
from artemis.services import llm as llm_service from artemis.services import llm as llm_service
from artemis.services.llm import ( from artemis.services.llm import (
RobustChatModelWrapper, RobustChatModelWrapper,
@@ -54,6 +55,30 @@ async def test_mid_stream_failure_retries_without_duplicating_chunks():
assert result.content == "Hello world" assert result.content == "Hello world"
@pytest.mark.asyncio
async def test_gpt_stream_uses_adapted_messages_without_mutating_history():
class RecordingStream:
messages = None
async def astream(self, messages, **kwargs):
self.messages = messages
yield AIMessageChunk(content="OK")
base = RecordingStream()
wrapper = RobustChatModelWrapper(
base, endpoint=ModelEndpoint(model_name="gpt-6-luna", max_retries=0)
)
original = [SystemMessage(content="Use tools"), HumanMessage(content="Inspect screen")]
result = await wrapper.complete(original)
assert result.content == "OK"
assert len(base.messages) == 1
assert isinstance(base.messages[0], HumanMessage)
assert base.messages[0].content[0]["text"] == "Use tools"
assert base.messages[0].content[1]["text"] == "Inspect screen"
assert isinstance(original[0], SystemMessage)
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_auth_error_fails_fast_without_retry_or_pause(tmp_path): async def test_auth_error_fails_fast_without_retry_or_pause(tmp_path):
class AuthFails: class AuthFails:
+87 -1
View File
@@ -5,10 +5,11 @@ from unittest.mock import MagicMock
import httpx import httpx
import pytest import pytest
from langchain_core.messages import HumanMessage, ToolMessage from langchain_core.messages import HumanMessage, SystemMessage, ToolMessage
from pydantic import BaseModel, SecretStr, ValidationError from pydantic import BaseModel, SecretStr, ValidationError
from artemis.config.settings import settings from artemis.config.settings import settings
from artemis.llm.messages import adapt_responses_messages
from artemis.llm.router import ModelEndpoint, ModelFactory, ModelProvider from artemis.llm.router import ModelEndpoint, ModelFactory, ModelProvider
from artemis.services.llm import RobustChatModelWrapper from artemis.services.llm import RobustChatModelWrapper
@@ -85,6 +86,91 @@ async def test_text_and_image_use_responses_transport(transport_model):
assert result.usage_metadata["total_tokens"] == 13 assert result.usage_metadata["total_tokens"] == 13
@pytest.mark.asyncio
@pytest.mark.parametrize(
("model_name", "reasoning_effort"),
[
("mimo-v2.6-pro", "high"),
("mimo-v2.6-flash", "high"),
("gpt-6-luna", "max"),
("gpt-6-luna", "high"),
],
)
async def test_selected_models_put_instructions_in_user_input(
transport_model, model_name, reasoning_effort
):
_, requests, _ = transport_model
endpoint = ModelEndpoint(
model_name=model_name,
reasoning_effort=reasoning_effort,
timeout_seconds=23,
max_retries=0,
)
model = ModelFactory.create_model(endpoint)
wrapper = RobustChatModelWrapper(model, endpoint=endpoint)
tool = {
"type": "function",
"function": {
"name": "inspect_screen",
"description": "Inspect the current screen",
"parameters": {"type": "object", "properties": {}},
},
}
await wrapper.bind_tools([tool]).ainvoke(
[
SystemMessage(content="Use the available tools."),
HumanMessage(
content=[
{"type": "text", "text": "Inspect this screen."},
{"type": "image_url", "image_url": {"url": "data:image/png;base64,AA=="}},
]
),
]
)
payload = json.loads(requests[-1].content)
assert payload["model"] == model_name
assert payload["reasoning"]["effort"] == reasoning_effort
assert payload["input"][0]["role"] == "user"
assert payload["input"][0]["content"][0]["text"] == "Use the available tools."
assert payload["input"][0]["content"][2]["type"] == "input_image"
assert payload["tools"][0]["name"] == "inspect_screen"
assert all(item.get("role") != "system" for item in payload["input"])
@pytest.mark.parametrize("model_name", ["gpt-6-luna", "GPT-5.6-luna"])
def test_gpt_message_adaptation_keeps_history_and_images(model_name):
image = {"type": "image_url", "image_url": {"url": "data:image/png;base64,AA=="}}
original = [
SystemMessage(content="First instruction"),
SystemMessage(content="Second instruction"),
HumanMessage(content=[{"type": "text", "text": "Observation"}, image]),
SystemMessage(content="Follow-up instruction"),
]
adapted = adapt_responses_messages(original, model_name)
assert len(adapted) == 2
assert isinstance(adapted[0], HumanMessage)
assert adapted[0].content[0]["text"] == "First instruction"
assert adapted[0].content[1]["text"] == "Second instruction"
assert adapted[0].content[3] == image
assert isinstance(adapted[1], HumanMessage)
assert adapted[1].content == [{"type": "text", "text": "Follow-up instruction"}]
assert isinstance(original[0], SystemMessage)
assert len(original[2].content) == 2
@pytest.mark.parametrize("model_name", ["mimo-v2.6-pro", "mimo-v2.6-flash"])
def test_mimo_messages_use_same_user_role_format(model_name):
messages = [SystemMessage(content="Instructions"), HumanMessage(content="Task")]
adapted = adapt_responses_messages(messages, model_name)
assert len(adapted) == 1
assert isinstance(adapted[0], HumanMessage)
assert adapted[0].content[0]["text"] == "Instructions"
assert adapted[0].content[1]["text"] == "Task"
assert isinstance(messages[0], SystemMessage)
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_max_reasoning_is_sent_unchanged_to_responses(transport_model): async def test_max_reasoning_is_sent_unchanged_to_responses(transport_model):
model, requests, _ = transport_model model, requests, _ = transport_model