148 lines
7.2 KiB
Python
148 lines
7.2 KiB
Python
"""Structured transport fails closed and never repairs or retries an action."""
|
|
import json
|
|
from types import SimpleNamespace as Obj
|
|
|
|
import pytest
|
|
from pydantic import BaseModel, ConfigDict, ValidationError
|
|
|
|
from backend.browser import diagnostics, llm_factory, structured_output as contract
|
|
from backend.prospecting import usage
|
|
|
|
|
|
class Click(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
index: int
|
|
|
|
|
|
class Action(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
click: Click
|
|
|
|
|
|
class Output(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
action: list[Action]
|
|
|
|
|
|
def tool(arguments='{"action":[{"click":{"index":3}}]}', *, name=contract.FUNCTION_NAME):
|
|
return Obj(type="function", function=Obj(name=name, arguments=arguments))
|
|
|
|
|
|
@pytest.mark.parametrize("message,reason", [
|
|
(Obj(content='{"action":[{"click":{"index":3}}]}'), "missing_tool_call"),
|
|
(Obj(tool_calls=[tool(), tool()]), "multiple_tool_calls"),
|
|
(Obj(tool_calls=[tool(name="click")]), "unexpected_tool_call"),
|
|
(Obj(tool_calls=[Obj(type="computer", function=Obj(name=contract.FUNCTION_NAME))]), "unexpected_tool_call"),
|
|
(Obj(tool_calls=[tool("")]), "missing_arguments"),
|
|
(Obj(tool_calls=[tool('{"action":')]), "malformed_arguments"),
|
|
(Obj(tool_calls=[tool('```json\n{"action":[]}\n```')]), "malformed_arguments"),
|
|
(Obj(tool_calls=[tool('{"action":[],"action":[{"click":{"index":3}}]}')]), "duplicate_argument_key"),
|
|
(Obj(tool_calls=[tool('{"action":[{"click":{"index":1,"index":2}}]}')]), "duplicate_argument_key"),
|
|
(Obj(tool_calls=[tool('{"action":[{"click":{"index":NaN}}]}')]), "nonfinite_arguments"),
|
|
(Obj(tool_calls=[tool('[]')]), "arguments_not_object"),
|
|
(Obj(tool_calls=[tool('{"action":[]}')]), "empty_actions"),
|
|
])
|
|
def test_ambiguous_or_malformed_action_never_falls_back_to_prose(message, reason):
|
|
with pytest.raises(contract.ResponseContractError) as error:
|
|
contract.validated_tool_result(message, Output)
|
|
assert error.value.reason == reason
|
|
|
|
|
|
def test_one_requested_tool_is_authoritative_and_content_is_not_a_second_action():
|
|
result = contract.validated_tool_result(Obj(content='{"action":[{"click":{"index":999}}]}', tool_calls=[tool()]), Output)
|
|
assert result.action[0].click.index == 3
|
|
|
|
|
|
@pytest.mark.parametrize("arguments", ['{"memory":"done"}', '{"action":[{"click":{}}]}',
|
|
'{"action":[{"click":{"index":3},"done":{}}]}'])
|
|
def test_full_schema_is_still_required_without_guessing_fields(arguments):
|
|
with pytest.raises(ValidationError):
|
|
contract.validated_tool_result(Obj(tool_calls=[tool(arguments)]), Output)
|
|
|
|
|
|
def test_real_browser_action_union_and_agent_envelope_share_the_transport():
|
|
from browser_use.agent.views import AgentOutput
|
|
from backend.browser.tools import build_prospecting_tools
|
|
actions = build_prospecting_tools('contract_fixture').registry.create_action_model()
|
|
output = AgentOutput.type_with_custom_actions(actions)
|
|
function, instruction = contract.request_contract(output)
|
|
parameters = function['function']['parameters']
|
|
original = output.model_json_schema()
|
|
assert parameters['$defs'] == original['$defs']
|
|
assert parameters['properties']['action']['minItems'] == 1
|
|
assert parameters['properties']['action']['items']['maxProperties'] == 1
|
|
assert 'min_items' not in parameters['properties']['action']
|
|
assert 'minItems' not in original['properties']['action']
|
|
assert 'action' in function['function']['parameters']['required']
|
|
arguments = json.dumps({'memory': '当前页面', 'action': [{'inspect_source_candidates': {'max_comments': 1}}]})
|
|
result = contract.validated_tool_result(Obj(tool_calls=[tool(arguments)]), output)
|
|
assert result.action[0].model_dump(exclude_none=True) == {'inspect_source_candidates': {'max_comments': 1}}
|
|
assert '非空数组' in instruction
|
|
|
|
|
|
def test_contract_is_appended_to_system_without_reordering_or_mutating_multimodal_context():
|
|
original = [{'role':'system','content':[{'type':'text','text':'framework instructions'}]},
|
|
{'role':'assistant','content':'history'},
|
|
{'role':'user','content':[{'type':'text','text':'page'}, {'type':'image_url','image_url':{'url':'data:image/png;base64,AA=='}}]}]
|
|
before = json.dumps(original)
|
|
prepared = contract.attach_instruction(original, 'safe contract')
|
|
assert [row['role'] for row in prepared] == ['system','assistant','user']
|
|
assert prepared[-1] == original[-1] and prepared[0]['content'][-1]['text'] == 'safe contract'
|
|
assert json.dumps(original) == before
|
|
|
|
|
|
@pytest.fixture
|
|
def isolated(tmp_path, monkeypatch):
|
|
def directory(tid):
|
|
path = tmp_path / tid
|
|
path.mkdir(exist_ok=True)
|
|
return path
|
|
monkeypatch.setattr(usage, 'task_doc_dir', directory)
|
|
monkeypatch.setattr(diagnostics, 'task_doc_dir', directory)
|
|
return tmp_path
|
|
|
|
|
|
async def test_protocol_failure_is_not_a_retryable_502_and_keeps_actual_usage(isolated, monkeypatch):
|
|
import openai
|
|
from browser_use.llm.exceptions import ModelProviderError
|
|
from browser_use.llm.messages import UserMessage
|
|
sent = []
|
|
class Client:
|
|
def __init__(self, **options):
|
|
assert options['max_retries'] == 0
|
|
self.chat = Obj(completions=Obj(create=self.create))
|
|
async def __aenter__(self): return self
|
|
async def __aexit__(self, *_args): return False
|
|
async def create(self, **params):
|
|
sent.append(params)
|
|
return Obj(model='deepseek-flash', choices=[Obj(message=Obj(content='PRIVATE-RESPONSE', tool_calls=[]), finish_reason='stop')],
|
|
usage=Obj(prompt_tokens=41, completion_tokens=7, total_tokens=48))
|
|
monkeypatch.setattr(openai, 'AsyncOpenAI', Client)
|
|
with usage.usage_scope('protocol', 'browser'), diagnostics.run_scope('protocol', 'run') as state:
|
|
with pytest.raises(ModelProviderError) as caught:
|
|
await llm_factory.DeepSeekBrowserChat('deepseek-flash', 'PRIVATE-KEY').ainvoke([UserMessage(content='PRIVATE-PROMPT')], Output)
|
|
assert caught.value.status_code == 422
|
|
assert diagnostics.consecutive_schema_errors() == 1
|
|
assert 'missing_tool_call' in diagnostics.pending_correction_hint()
|
|
assert 'PRIVATE' not in diagnostics.pending_correction_hint()
|
|
diagnostics.validated_response()
|
|
assert not diagnostics.pending_correction_hint()
|
|
assert len(sent) == 1 and 'response_format' not in sent[0]
|
|
assert usage.snapshot('protocol')['total_tokens'] == 48
|
|
assert usage.snapshot('protocol')['parse_failed_requests'] == 1
|
|
assert 'PRIVATE' not in json.dumps(diagnostics.summary(state))
|
|
assert 'PRIVATE' not in (isolated/'protocol'/'model-diagnostics.jsonl').read_text()
|
|
|
|
|
|
def test_pending_safe_hint_does_not_cross_run_or_success(isolated):
|
|
with diagnostics.run_scope('one', 'first'):
|
|
diagnostics.response_contract_error(None, 'missing_tool_call')
|
|
assert diagnostics.pending_correction_hint()
|
|
with diagnostics.run_scope('two', 'other'):
|
|
assert not diagnostics.pending_correction_hint()
|
|
assert diagnostics.pending_correction_hint()
|
|
diagnostics.validated_response()
|
|
assert not diagnostics.pending_correction_hint()
|
|
with diagnostics.run_scope('one', 'second'):
|
|
assert not diagnostics.pending_correction_hint()
|