WEBDIENSTE/openswarm
1
0
Fork 0
mirror of https://github.com/openswarm-ai/openswarm.git synced 2026-09-11 20:27:44 +02:00
Code Issues Packages Projects Releases Wiki Activity

[haik]: refactor: apply p_/P_ naming convention to browser_agent.py and 31 test files. In browser_agent.py, rename 6 module-internal functions from leading-underscore to p_ prefix (_strip_lone_surrogates -> p_strip_lone_surrogates, _format_tool_result -> p_format_tool_result, _delta_state -> p_delta_state, _send_index_in_state -> p_send_index_in_state, _is_composer_fill -> p_is_composer_fill) and update all call sites. Across 31 test files, rename test helper functions from _func to p_func or descriptive names (_tu -> p_tu, _rp -> p_rp, _install -> p_install, _log -> action_log, _isolated_skills -> isolated_skills), drop leading underscores from local variables (_nm -> nm, _send_browser_command -> send_browser_command), and update module attribute references from BH._browser_history/BH._domain_notes to BH.BROWSER_HISTORY/BH.P_DOMAIN_NOTES. ~1023 insertions, ~1024 deletions, no behavioral changes

Browse Source
This commit is contained in:
haikdc
2026-06-14 01:11:48 -07:00
parent 5c99dbbd80
commit a28ac6b642
33 changed files with 1023 additions and 1024 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/agents/browser/browser_agent.py
+11 -11
View File
@@ -129,7 +129,7 @@ def _extract_domain(url: str) -> str | None:
return None
def _strip_lone_surrogates(s: str) -> str:
def p_strip_lone_surrogates(s: str) -> str:
# The JS/webview hands us page text as UTF-16, so an emoji can arrive as half
# of its surrogate pair; Python carries the orphan but .encode('utf-8') later
# (the SDK serializing the request to the LLM) detonates with "surrogates not
@@ -137,10 +137,10 @@ def _strip_lone_surrogates(s: str) -> str:
return re.sub(r"[\ud800-\udfff]", "�", s) if s else s
def _format_tool_result(result: dict, tool_name: str) -> list[dict]:
def p_format_tool_result(result: dict, tool_name: str) -> list[dict]:
"""Convert a browser command result dict into Anthropic API content blocks."""
if "error" in result:
return [{"type": "text", "text": _strip_lone_surrogates(f"Error: {result['error']}")}]
return [{"type": "text", "text": p_strip_lone_surrogates(f"Error: {result['error']}")}]
if tool_name == "BrowserScreenshot" and result.get("image"):
blocks = [
@@ -157,7 +157,7 @@ def _format_tool_result(result: dict, tool_name: str) -> list[dict]:
return blocks
text = result.get("text", json.dumps(result))
return [{"type": "text", "text": _strip_lone_surrogates(str(text))}]
return [{"type": "text", "text": p_strip_lone_surrogates(str(text))}]
# Mutating tools whose results get fresh page state attached (the browser-use
@@ -190,7 +190,7 @@ def _truncate_state(text: str, max_lines: int = _AUTO_STATE_MAX_LINES) -> str:
)
def _delta_state(text: str, seen_lines: set[str]) -> str:
def p_delta_state(text: str, seen_lines: set[str]) -> str:
"""Shrink an attached element list to the rows that changed since the last
attach; stable indices make a line's identity meaningful, so re-sending 30
unchanged rows every action is pure token burn. Mutates `seen_lines` to the
@@ -219,7 +219,7 @@ def _delta_state(text: str, seen_lines: set[str]) -> str:
_SEND_ROW_RE = re.compile(r'\[(\d+)\]\*?<\s*button\s+"([^"]*)"', re.I)
def _send_index_in_state(state_text: str):
def p_send_index_in_state(state_text: str):
"""(index, name) of a real Send button in an interactives list, or None.
Strict exact match so it never grabs an upsell or a profile 'Send a message' link."""
for line in (state_text or "").splitlines():
@@ -229,7 +229,7 @@ def _send_index_in_state(state_text: str):
return None
def _is_composer_fill(tool_name: str, tool_input: dict) -> bool:
def p_is_composer_fill(tool_name: str, tool_input: dict) -> bool:
"""True if this action typed a message into a composer (the moment the Send
button is about to matter). Covers the solo fill, BrowserType, and a batched
fill, the three ways the model composes."""
@@ -262,7 +262,7 @@ async def _post_action_state(
)
if settle.get("hung"):
return ""
_composer_fill = _is_composer_fill(tool_name, tool_input)
_composer_fill = p_is_composer_fill(tool_name, tool_input)
params = {"goal": goal} if goal else {}
lst = None
_send_si = None
@@ -281,7 +281,7 @@ async def _post_action_state(
break
if isinstance(_l, dict) and "error" not in _l and _l.get("text"):
lst = _l
_send_si = _send_index_in_state(_l["text"])
_send_si = p_send_index_in_state(_l["text"])
if _send_si:
break
if time.monotonic() >= _deadline:
@@ -296,7 +296,7 @@ async def _post_action_state(
return ""
if not isinstance(lst, dict) or "error" in lst or not lst.get("text"):
return ""
state = lst["text"] if seen_lines is None else _delta_state(lst["text"], seen_lines)
state = lst["text"] if seen_lines is None else p_delta_state(lst["text"], seen_lines)
out = f"\n\n{PAGE_STATE_MARKER}\n{_truncate_state(state)}"
# Hand the Send button's index over so the model clicks it directly instead of
# scanning the list or hunting via CSS/JS/screenshots (the polled list above is what
@@ -1781,7 +1781,7 @@ async def run_browser_agent(
if len(recent_tool_calls) > LOOP_WINDOW_SIZE * 2:
recent_tool_calls = recent_tool_calls[-LOOP_WINDOW_SIZE * 2:]
content_blocks = _format_tool_result(result, tu.name)
content_blocks = p_format_tool_result(result, tu.name)
try:
url = result.get("url") or (tu.input or {}).get("url")
if url:
backend/tests/test_browser_agent_loop.py
+266 -266
View File
File diff suppressed because it is too large Load Diff
backend/tests/test_browser_batch_replay.py
+6 -6
View File
@@ -170,27 +170,27 @@ SEEN = {
}
def _click_idx(i):
def p_click_idx(i):
return {"type": "click_index", "params": {"index": i}}
def test_guard_blocks_send_click_index_resolved_from_state():
why = br.live_batch_guard([_click_idx(4)], SEEN)
why = br.live_batch_guard([p_click_idx(4)], SEEN)
assert "irreversible" in why and "Send" in why
def test_guard_blocks_connect_but_allows_message_composer_opener():
assert br.live_batch_guard([_click_idx(12)], SEEN) != ""
assert br.live_batch_guard([_click_idx(7)], SEEN) == ""
assert br.live_batch_guard([p_click_idx(12)], SEEN) != ""
assert br.live_batch_guard([p_click_idx(7)], SEEN) == ""
def test_guard_index_prefix_does_not_collide():
# [4] is "Send" but [41] is "Next page"; clicking 41 must pass
assert br.live_batch_guard([_click_idx(41)], SEEN) == ""
assert br.live_batch_guard([p_click_idx(41)], SEEN) == ""
def test_guard_allows_unresolvable_index_and_garbage():
assert br.live_batch_guard([_click_idx(99)], SEEN) == ""
assert br.live_batch_guard([p_click_idx(99)], SEEN) == ""
assert br.live_batch_guard([{"type": "click_index"}, "junk", None], SEEN) == ""
assert br.live_batch_guard(None, set()) == ""
backend/tests/test_browser_command_timeout.py
+22 -22
View File
@@ -13,33 +13,33 @@ import pytest
from backend.apps.agents.core import ws_manager as wsm
class _FakeSock:
class p_FakeSock:
async def send_text(self, _):
return None
def _mgr():
def p_mgr():
m = wsm.ConnectionManager()
m.global_connections = [_FakeSock()] # get past the 'no dashboard' guard
m.global_connections = [p_FakeSock()] # get past the 'no dashboard' guard
return m
def test_timeout_map_reads_are_short_navigation_longer():
# reads/clicks act on a loaded page -> short; navigation loads network -> longer
assert wsm._BROWSER_CMD_TIMEOUT_DEFAULT <= 15
assert wsm._BROWSER_CMD_TIMEOUTS["navigate"] <= 25
assert wsm._BROWSER_CMD_TIMEOUTS["navigate"] > wsm._BROWSER_CMD_TIMEOUT_DEFAULT
assert wsm.P_BROWSER_CMD_TIMEOUT_DEFAULT <= 15 # p-private-ignore: P_BROWSER_CMD_TIMEOUT_DEFAULT
assert wsm.P_BROWSER_CMD_TIMEOUTS["navigate"] <= 25 # p-private-ignore: P_BROWSER_CMD_TIMEOUTS
assert wsm.P_BROWSER_CMD_TIMEOUTS["navigate"] > wsm.P_BROWSER_CMD_TIMEOUT_DEFAULT # p-private-ignore: P_BROWSER_CMD_TIMEOUTS, P_BROWSER_CMD_TIMEOUT_DEFAULT
# the old flat 30s is gone for the common path
assert wsm._BROWSER_CMD_TIMEOUT_DEFAULT < 30
assert wsm.P_BROWSER_CMD_TIMEOUT_DEFAULT < 30 # p-private-ignore: P_BROWSER_CMD_TIMEOUT_DEFAULT
@pytest.mark.asyncio
async def test_hung_command_returns_fast_at_the_bound(monkeypatch):
# shrink the bounds so the test is quick, then never resolve the future:
# the command must return a timeout error at ~the (default) bound, not hang.
monkeypatch.setattr(wsm, "_BROWSER_CMD_TIMEOUT_DEFAULT", 0.3)
monkeypatch.setattr(wsm, "_BROWSER_CMD_TIMEOUTS", {"navigate": 0.6})
m = _mgr()
monkeypatch.setattr(wsm, "P_BROWSER_CMD_TIMEOUT_DEFAULT", 0.3)
monkeypatch.setattr(wsm, "P_BROWSER_CMD_TIMEOUTS", {"navigate": 0.6})
m = p_mgr()
t0 = time.monotonic()
res = await m.send_browser_command("rid1", "get_text", "b1", {}) # never resolved
elapsed = time.monotonic() - t0
@@ -49,9 +49,9 @@ async def test_hung_command_returns_fast_at_the_bound(monkeypatch):
@pytest.mark.asyncio
async def test_navigate_gets_the_longer_leash(monkeypatch):
monkeypatch.setattr(wsm, "_BROWSER_CMD_TIMEOUT_DEFAULT", 0.3)
monkeypatch.setattr(wsm, "_BROWSER_CMD_TIMEOUTS", {"navigate": 0.7})
m = _mgr()
monkeypatch.setattr(wsm, "P_BROWSER_CMD_TIMEOUT_DEFAULT", 0.3)
monkeypatch.setattr(wsm, "P_BROWSER_CMD_TIMEOUTS", {"navigate": 0.7})
m = p_mgr()
t0 = time.monotonic()
await m.send_browser_command("rid2", "navigate", "b1", {"url": "x"})
elapsed = time.monotonic() - t0
@@ -62,19 +62,19 @@ async def test_navigate_gets_the_longer_leash(monkeypatch):
async def test_lost_first_delivery_heals_via_rebroadcast(monkeypatch):
# a silently-dead socket eats the first broadcast; the re-send after the
# rebroadcast interval must reach the (reconnected) client and succeed
monkeypatch.setattr(wsm, "_BROWSER_CMD_TIMEOUT_DEFAULT", 5.0)
monkeypatch.setattr(wsm, "_BROWSER_CMD_REBROADCAST_S", 0.1)
m = _mgr()
monkeypatch.setattr(wsm, "P_BROWSER_CMD_TIMEOUT_DEFAULT", 5.0)
monkeypatch.setattr(wsm, "P_BROWSER_CMD_REBROADCAST_S", 0.1)
m = p_mgr()
sends = []
class _CountingSock:
class p_CountingSock:
async def send_text(self, payload):
sends.append(payload)
if len(sends) >= 2: # first delivery "lost", second lands
rid = next(iter(m.browser_futures))
m.resolve_browser_command(rid, {"text": "ok"})
m.global_connections = [_CountingSock()]
m.global_connections = [p_CountingSock()]
res = await m.send_browser_command("rid4", "get_text", "b1", {})
assert res == {"text": "ok"}
assert len(sends) >= 2, "command must be re-broadcast until a client answers"
@@ -83,16 +83,16 @@ async def test_lost_first_delivery_heals_via_rebroadcast(monkeypatch):
@pytest.mark.asyncio
async def test_a_resolved_command_returns_immediately(monkeypatch):
# a healthy command returns the moment the renderer resolves it, not at the bound
monkeypatch.setattr(wsm, "_BROWSER_CMD_TIMEOUT_DEFAULT", 5.0)
m = _mgr()
monkeypatch.setattr(wsm, "P_BROWSER_CMD_TIMEOUT_DEFAULT", 5.0)
m = p_mgr()
async def _resolve_soon():
async def p_resolve_soon():
await asyncio.sleep(0.05)
# find the pending future and resolve it like the renderer would
rid = next(iter(m.browser_futures))
m.resolve_browser_command(rid, {"text": "ok", "url": "u"})
asyncio.create_task(_resolve_soon())
asyncio.create_task(p_resolve_soon())
t0 = time.monotonic()
res = await m.send_browser_command("rid3", "get_text", "b1", {})
elapsed = time.monotonic() - t0
backend/tests/test_browser_fast_path.py
+17 -17
View File
@@ -1,6 +1,6 @@
from backend.apps.agents.browser.browser_fast_path import (
_normalize_for_classifier,
_parse_verdict_and_brief,
from backend.apps.agents.browser.browser_fast_path import ( # p-private-ignore: p_normalize_for_classifier, p_parse_verdict_and_brief
p_normalize_for_classifier,
p_parse_verdict_and_brief,
compose_task,
dispatch_failed,
fast_path_eligible,
@@ -30,15 +30,15 @@ def test_non_browsy_or_gated_messages_fall_through():
def test_verdict_parsing_is_strict():
v, brief = _parse_verdict_and_brief("READ\nENTRY: https://news.ycombinator.com\n1. read top story")
v, brief = p_parse_verdict_and_brief("READ\nENTRY: https://news.ycombinator.com\n1. read top story")
assert v == "read" and brief.startswith("ENTRY:") and "top story" in brief
assert _parse_verdict_and_brief("ACT\nENTRY: https://x.com")[0] == "act"
assert _parse_verdict_and_brief("yes") == ("act", "")
assert _parse_verdict_and_brief("NO") == ("no", "")
assert _parse_verdict_and_brief("Maybe\nENTRY: x") == ("no", "")
assert _parse_verdict_and_brief("") == ("no", "")
assert p_parse_verdict_and_brief("ACT\nENTRY: https://x.com")[0] == "act"
assert p_parse_verdict_and_brief("yes") == ("act", "")
assert p_parse_verdict_and_brief("NO") == ("no", "")
assert p_parse_verdict_and_brief("Maybe\nENTRY: x") == ("no", "")
assert p_parse_verdict_and_brief("") == ("no", "")
long_brief = "ACT\n" + "x" * 2000
assert len(_parse_verdict_and_brief(long_brief)[1]) == 700
assert len(p_parse_verdict_and_brief(long_brief)[1]) == 700
def test_fast_read_entry_extraction_and_thin_detection():
@@ -84,14 +84,14 @@ def test_recovery_task_verifies_before_repeating():
def test_text_normalizes_to_message_without_phone_number():
assert (
_normalize_for_classifier("go to maya's linkedin and text her thanks")
p_normalize_for_classifier("go to maya's linkedin and text her thanks")
== "go to maya's linkedin and message her thanks"
)
assert _normalize_for_classifier("keep texting until he replies").startswith("keep message")
assert p_normalize_for_classifier("keep texting until he replies").startswith("keep message")
sms = "text 4085551234 saying im running late"
assert _normalize_for_classifier(sms) == sms
assert p_normalize_for_classifier(sms) == sms
count = "count messages containing the exact text r10-os"
assert "message r10-os" in _normalize_for_classifier(count)
assert "message r10-os" in p_normalize_for_classifier(count)
def test_dispatch_refused_when_no_dashboard_connected(monkeypatch):
@@ -163,7 +163,7 @@ def test_entry_url_extracted_from_brief():
def test_results_url_shapes():
from backend.apps.agents.browser.browser_agent import _RESULTS_URL_RE
from backend.apps.agents.browser import browser_agent as bagent
hits = [
"https://www.linkedin.com/search/results/people/?keywords=tyler+chen",
"https://www.google.com/search?q=anything",
@@ -176,6 +176,6 @@ def test_results_url_shapes():
"https://www.linkedin.com/messaging/thread/abc123/",
]
for u in hits:
assert _RESULTS_URL_RE.search(u), u
assert bagent._RESULTS_URL_RE.search(u), u
for u in misses:
assert not _RESULTS_URL_RE.search(u), u
assert not bagent._RESULTS_URL_RE.search(u), u
backend/tests/test_browser_hotpath_waste.py
+11 -11
View File
@@ -11,13 +11,13 @@ import os
import backend.apps.agents.browser.browser_metrics as M
from backend.apps.agents.browser.browser_loop import (
_detect_loop,
_LOOP_DETECTION_EXCLUDED_TOOLS,
detect_loop,
LOOP_DETECTION_EXCLUDED_TOOLS,
)
def test_metrics_dir_is_cached_makedirs_runs_once(monkeypatch):
M._metrics_dir_cache = None
M.P_METRICS_DIR_CACHE = None
calls = {"n": 0}
real = os.makedirs
@@ -26,9 +26,9 @@ def test_metrics_dir_is_cached_makedirs_runs_once(monkeypatch):
return real(*a, **k)
monkeypatch.setattr(os, "makedirs", counting)
d1 = M._metrics_dir()
d2 = M._metrics_dir()
d3 = M._metrics_dir()
d1 = M.metrics_dir()
d2 = M.metrics_dir()
d3 = M.metrics_dir()
assert d1 == d2 == d3
assert calls["n"] == 1, f"makedirs must run once, ran {calls['n']}x"
@@ -37,9 +37,9 @@ def test_excluded_tools_never_register_a_loop():
# The invariant the hash-skip relies on: for every excluded tool, even ten
# identical calls in a row are NOT a loop, so computing/storing the hash for
# them was dead work. Setting is_loop=False directly is therefore equivalent.
for tool in _LOOP_DETECTION_EXCLUDED_TOOLS:
for tool in LOOP_DETECTION_EXCLUDED_TOOLS:
key = (tool, "in", "out")
assert _detect_loop([key] * 10, key) is False, f"{tool} wrongly looped"
assert detect_loop([key] * 10, key) is False, f"{tool} wrongly looped"
def test_non_excluded_tool_still_loops_after_threshold():
@@ -47,6 +47,6 @@ def test_non_excluded_tool_still_loops_after_threshold():
# that need it (clicks/types/etc.).
key = ("BrowserClick", '{"selector":"#x"}', "clicked")
# below threshold -> not a loop; at/over threshold within the window -> loop
assert _detect_loop([], key) is False # 1st occurrence: not yet a wall
assert _detect_loop([key], key) is True # 2nd identical (threshold=2): a wall
assert _detect_loop([key] * 5, key) is True
assert detect_loop([], key) is False # 1st occurrence: not yet a wall
assert detect_loop([key], key) is True # 2nd identical (threshold=2): a wall
assert detect_loop([key] * 5, key) is True
backend/tests/test_browser_memory_endpoints.py
+7 -7
View File
@@ -15,14 +15,14 @@ from backend.apps.agents.browser import browser_playbook as pb
from backend.apps.agents.browser import browser_skills as sk
def _seed_skill(host, task):
def seed_skill(host, task):
sk.record_skill(host, task, [
{"tool": "BrowserClickIndex", "input": {}, "ok": True,
"clicked_role": "button", "clicked_name": "Go"},
])
async def _seed_strategy(host, *bullets):
async def seed_strategy(host, *bullets):
resp = SimpleNamespace(
content=[SimpleNamespace(text=json.dumps({"playbook": list(bullets)}))]
)
@@ -32,9 +32,9 @@ async def _seed_strategy(host, *bullets):
def test_list_browser_memory_groups_skills_and_strategy_by_site():
sk.clear(); pb.clear(wipe_disk=True)
_seed_skill("shop.com", "search now")
asyncio.run(_seed_strategy("shop.com", "use the search box at the top"))
asyncio.run(_seed_strategy("docs.com", "share lives behind the blue button"))
seed_skill("shop.com", "search now")
asyncio.run(seed_strategy("shop.com", "use the search box at the top"))
asyncio.run(seed_strategy("docs.com", "share lives behind the blue button"))
out = asyncio.run(agents_mod.list_browser_memory())
sites = {s["host"]: s for s in out["sites"]}
@@ -46,8 +46,8 @@ def test_list_browser_memory_groups_skills_and_strategy_by_site():
def test_forget_clears_both_tiers_for_a_site():
sk.clear(); pb.clear(wipe_disk=True)
_seed_skill("gone.com", "do it now")
asyncio.run(_seed_strategy("gone.com", "a strategy bullet"))
seed_skill("gone.com", "do it now")
asyncio.run(seed_strategy("gone.com", "a strategy bullet"))
# sanity: present
assert pb.get_playbook("gone.com") and sk.list_skills("gone.com")
backend/tests/test_browser_meta_playbook.py
+1 -1
View File
@@ -8,7 +8,7 @@ from backend.apps.agents.browser import browser_meta_playbook as meta
@pytest.fixture(autouse=True)
def _isolated(monkeypatch):
def isolated(monkeypatch):
monkeypatch.setenv("OPENSWARM_BROWSER_META_DIR", tempfile.mkdtemp(prefix="meta_test_"))
meta.clear(wipe_disk=True)
yield
backend/tests/test_browser_metrics.py
+11 -12
View File
@@ -14,11 +14,11 @@ def metrics(monkeypatch):
from backend.apps.agents.browser import browser_metrics as bm
# The dir is memoized once for the prod hot path; drop the cache so each test
# re-resolves to its own temp dir instead of inheriting a prior test's.
bm._metrics_dir_cache = None
bm.P_METRICS_DIR_CACHE = None
return bm, d
def _read(d, name):
def read_rows(d, name):
p = os.path.join(d, name)
if not os.path.exists(p):
return []
@@ -41,7 +41,7 @@ def test_record_tool_writes_event(metrics):
bm, d = metrics
bm.record_tool("s1", "b1", 2, "BrowserListInteractives", 18,
ok=True, error="", is_loop=False, stagnation_streak=0, result_len=120)
events = _read(d, "events.jsonl")
events = read_rows(d, "events.jsonl")
assert len(events) == 1
e = events[0]
assert e["tool"] == "BrowserListInteractives" and e["tier"] == "t3_action_surface"
@@ -53,7 +53,7 @@ def test_record_tool_captures_error(metrics):
bm.record_tool("s1", "b1", 3, "BrowserClickIndex", 9,
ok=False, error="Index 4 is no longer valid", is_loop=True,
stagnation_streak=2, result_len=40)
e = _read(d, "events.jsonl")[0]
e = read_rows(d, "events.jsonl")[0]
assert e["ok"] is False and "no longer valid" in e["error"]
assert e["is_loop"] is True and e["stagnation_streak"] == 2
@@ -75,7 +75,7 @@ def test_record_task_summary_and_rollups(metrics):
assert summary["by_tier"]["t5_vision"]["calls"] == 1
assert summary["total_ms"] >= 1000 # ~1.2s elapsed
assert any("not found" in err[0].lower() for err in summary["recurring_errors"])
tasks = _read(d, "tasks.jsonl")
tasks = read_rows(d, "tasks.jsonl")
assert len(tasks) == 1 and tasks[0]["status"] == "completed"
@@ -83,7 +83,7 @@ def test_metrics_never_raises_on_bad_dir(monkeypatch):
# An unwritable dir must not throw into the agent loop.
monkeypatch.setenv("OPENSWARM_BROWSER_METRICS_DIR", "/proc/cannot/write/here")
from backend.apps.agents.browser import browser_metrics as bm
bm._metrics_dir_cache = None # re-resolve so we actually hit the bad dir
bm.P_METRICS_DIR_CACHE = None # re-resolve so we actually hit the bad dir
bm.record_tool("s", "b", 1, "BrowserScreenshot", 5, ok=True, error="",
is_loop=False, stagnation_streak=0, result_len=1) # must not raise
bm.record_task("s", "b", "t", "error", __import__("time").time(), 1, [], {})
@@ -91,15 +91,14 @@ def test_metrics_never_raises_on_bad_dir(monkeypatch):
def test_task_secrets_are_scrubbed_from_tasks_jsonl(tmp_path, monkeypatch):
from backend.apps.agents.browser import browser_metrics as bm
import os as _os
import time as _time
import time
monkeypatch.setenv("OPENSWARM_BROWSER_METRICS_DIR", str(tmp_path))
bm._metrics_dir_cache = None
bm.P_METRICS_DIR_CACHE = None
bm.record_task("s1", "b1", "log into acme with password hunter2 then post sk-abc12345678901234567",
"completed", _time.time() - 1, 2, [], {})
line = open(_os.path.join(str(tmp_path), "tasks.jsonl")).read()
"completed", time.time() - 1, 2, [], {})
line = open(os.path.join(str(tmp_path), "tasks.jsonl")).read()
assert "hunter2" not in line and "sk-abc" not in line
assert "password [redacted]" in line
# owner-only file perms
mode = _os.stat(_os.path.join(str(tmp_path), "tasks.jsonl")).st_mode & 0o777
mode = os.stat(os.path.join(str(tmp_path), "tasks.jsonl")).st_mode & 0o777
assert mode == 0o600
backend/tests/test_browser_orchestrator_routing.py
+2 -2
View File
@@ -12,7 +12,7 @@ from types import SimpleNamespace
from backend.apps.agents.manager.prompt import prompt_context as pc
def _fake_dashboard(monkeypatch):
def fake_dashboard(monkeypatch):
# _build_browser_context loads the dashboard; give it a minimal one so it
# gets past the load and emits the static delegation guidance.
import backend.apps.dashboards.dashboards as dash
@@ -22,7 +22,7 @@ def _fake_dashboard(monkeypatch):
def test_orchestrator_routes_same_flow_batches_to_one_agent(monkeypatch):
_fake_dashboard(monkeypatch)
fake_dashboard(monkeypatch)
ctx = pc._build_browser_context("dash-1", selected_browser_ids=[])
assert ctx is not None
# the key guidance: one agent + the whole list, not one agent per item
backend/tests/test_browser_playbook.py
+23 -23
View File
@@ -13,14 +13,14 @@ from backend.apps.agents.browser import browser_playbook as pb
# --- a fake aux client that returns a scripted JSON playbook -----------------
class _Blk:
class Blk:
def __init__(self, text):
self.text = text
class _Resp:
class Resp:
def __init__(self, text):
self.content = [_Blk(text)]
self.content = [Blk(text)]
class FakeAux:
@@ -37,14 +37,14 @@ class FakeAux:
prompt = kw["messages"][0]["content"]
self.prompts.append(prompt)
r = self.reply(prompt) if callable(self.reply) else self.reply
return _Resp(r)
return Resp(r)
def _pb(*bullets):
def pb_json(*bullets):
return json.dumps({"playbook": list(bullets)})
async def _distill(host, task, mem, summary, aux):
async def distill(host, task, mem, summary, aux):
return await pb.distill_and_store(host, task, mem, summary, aux, "aux-model")
@@ -52,8 +52,8 @@ async def _distill(host, task, mem, summary, aux):
@pytest.mark.asyncio
async def test_first_success_creates_a_playbook():
pb.clear(wipe_disk=True)
aux = FakeAux(_pb("generic 'design engineer' = hardware; add React or a company name"))
changed = await _distill("linkedin.com", "find design engineers", "notes", "done", aux)
aux = FakeAux(pb_json("generic 'design engineer' = hardware; add React or a company name"))
changed = await distill("linkedin.com", "find design engineers", "notes", "done", aux)
assert changed
# the distill must use the configured aux model on a bounded token budget
assert aux.calls[0]["model"] == "aux-model"
@@ -67,15 +67,15 @@ async def test_second_run_accumulates_not_overwrites():
# THE BUG THIS FIXES: the old domain-note store overwrote. The reconcile must
# ACCUMULATE: run 2's reply (which the aux builds from existing+new) grows it.
pb.clear(wipe_disk=True)
await _distill("linkedin.com", "t1", "m1", "s1", FakeAux(_pb("Vercel/Linear+React surfaces real design engineers")))
await distill("linkedin.com", "t1", "m1", "s1", FakeAux(pb_json("Vercel/Linear+React surfaces real design engineers")))
# the aux on run 2 is handed the existing bullet (we assert that), and returns existing + a new one
seen = {}
def reply(prompt):
seen["prompt"] = prompt
return _pb("Vercel/Linear+React surfaces real design engineers",
return pb_json("Vercel/Linear+React surfaces real design engineers",
"the add-a-role wall is fine, read the top card")
await _distill("linkedin.com", "t2", "m2", "s2", FakeAux(reply))
await distill("linkedin.com", "t2", "m2", "s2", FakeAux(reply))
bullets = pb.get_playbook("linkedin.com")
assert len(bullets) == 2
# the existing bullet was actually given to the aux so it could reconcile
@@ -86,8 +86,8 @@ async def test_second_run_accumulates_not_overwrites():
async def test_reconcile_can_drop_a_contradicted_bullet():
# mem0 DELETE: the aux returns a list WITHOUT the stale bullet -> it's gone.
pb.clear(wipe_disk=True)
await _distill("x.com", "t", "m", "s", FakeAux(_pb("old way: click the big blue button")))
await _distill("x.com", "t", "m", "s", FakeAux(_pb("new way: use the keyboard shortcut /")))
await distill("x.com", "t", "m", "s", FakeAux(pb_json("old way: click the big blue button")))
await distill("x.com", "t", "m", "s", FakeAux(pb_json("new way: use the keyboard shortcut /")))
bullets = pb.get_playbook("x.com")
assert bullets == ["new way: use the keyboard shortcut /"]
@@ -95,8 +95,8 @@ async def test_reconcile_can_drop_a_contradicted_bullet():
@pytest.mark.asyncio
async def test_secrets_are_scrubbed_before_persisting():
pb.clear(wipe_disk=True)
aux = FakeAux(_pb("log in works", "the account email is eric@example.com", "token sk-ant-api03-abc lives in header"))
await _distill("site.com", "log in", "m", "s", aux)
aux = FakeAux(pb_json("log in works", "the account email is eric@example.com", "token sk-ant-api03-abc lives in header"))
await distill("site.com", "log in", "m", "s", aux)
bullets = pb.get_playbook("site.com")
blob = " ".join(bullets)
assert "eric@example.com" not in blob and "sk-ant-api03" not in blob
@@ -107,14 +107,14 @@ async def test_secrets_are_scrubbed_before_persisting():
async def test_playbook_is_capped():
pb.clear(wipe_disk=True)
many = [f"strategy bullet number {i}" for i in range(20)]
await _distill("big.com", "t", "m", "s", FakeAux(_pb(*many)))
await distill("big.com", "t", "m", "s", FakeAux(pb_json(*many)))
assert len(pb.get_playbook("big.com")) <= pb._MAX_BULLETS
@pytest.mark.asyncio
async def test_dedup_identical_bullets():
pb.clear(wipe_disk=True)
await _distill("d.com", "t", "m", "s", FakeAux(_pb("same thing", "same thing", "Same Thing")))
await distill("d.com", "t", "m", "s", FakeAux(pb_json("same thing", "same thing", "Same Thing")))
assert len(pb.get_playbook("d.com")) == 1
@@ -122,7 +122,7 @@ async def test_dedup_identical_bullets():
@pytest.mark.asyncio
async def test_playbook_survives_a_restart():
pb.clear(wipe_disk=True)
await _distill("persist.com", "t", "m", "s", FakeAux(_pb("durable lesson one", "durable lesson two")))
await distill("persist.com", "t", "m", "s", FakeAux(pb_json("durable lesson one", "durable lesson two")))
pb.clear(wipe_disk=False) # restart: memory gone, disk intact
assert not pb._cache
bullets = pb.get_playbook("persist.com")
@@ -132,10 +132,10 @@ async def test_playbook_survives_a_restart():
@pytest.mark.asyncio
async def test_garbage_aux_reply_leaves_playbook_untouched():
pb.clear(wipe_disk=True)
await _distill("safe.com", "t", "m", "s", FakeAux(_pb("good bullet")))
await distill("safe.com", "t", "m", "s", FakeAux(pb_json("good bullet")))
before = pb.get_playbook("safe.com")
# aux returns non-JSON prose -> must NOT wipe the existing playbook
changed = await _distill("safe.com", "t", "m", "s", FakeAux("sorry, I cannot help with that"))
changed = await distill("safe.com", "t", "m", "s", FakeAux("sorry, I cannot help with that"))
assert changed is False
assert pb.get_playbook("safe.com") == before
@@ -158,7 +158,7 @@ def test_should_learn_only_on_substantive_verified_success():
async def test_format_for_prompt_seeds_bullets():
pb.clear(wipe_disk=True)
assert pb.format_for_prompt("seed.com") == "" # nothing yet -> no block
await _distill("seed.com", "t", "m", "s", FakeAux(_pb("do X before Y", "avoid Z")))
await distill("seed.com", "t", "m", "s", FakeAux(pb_json("do X before Y", "avoid Z")))
block = pb.format_for_prompt("seed.com")
assert "What you learned about seed.com" in block and "do X before Y" in block
assert "re-verify" in block # honesty hedge present
@@ -167,8 +167,8 @@ async def test_format_for_prompt_seeds_bullets():
@pytest.mark.asyncio
async def test_forget_and_list_hosts_for_ux():
pb.clear(wipe_disk=True)
await _distill("a.com", "t", "m", "s", FakeAux(_pb("a lesson")))
await _distill("b.com", "t", "m", "s", FakeAux(_pb("b lesson")))
await distill("a.com", "t", "m", "s", FakeAux(pb_json("a lesson")))
await distill("b.com", "t", "m", "s", FakeAux(pb_json("b lesson")))
hosts = {h["host"] for h in pb.list_hosts()}
assert hosts == {"a.com", "b.com"}
assert pb.forget("a.com") is True
backend/tests/test_browser_save.py
+9 -9
View File
@@ -4,7 +4,7 @@ and extension allowlist are tested hard, including hostile filenames."""
import json
import os
from backend.apps.agents.browser.browser_save import save_page_data, _ALLOWED_EXT, _MAX_BYTES, _SUBDIR
from backend.apps.agents.browser.browser_save import save_page_data, P_ALLOWED_EXT, P_MAX_BYTES, P_SUBDIR # p-private-ignore: P_ALLOWED_EXT, P_MAX_BYTES, P_SUBDIR
def test_happy_path_writes_into_browser_data_subdir(tmp_path):
@@ -12,7 +12,7 @@ def test_happy_path_writes_into_browser_data_subdir(tmp_path):
msg = save_page_data(str(tmp_path), "sid", "rows.json", payload)
assert msg.startswith("Saved")
assert "3 items" in msg
out = tmp_path / _SUBDIR / "rows.json"
out = tmp_path / P_SUBDIR / "rows.json"
assert out.is_file()
assert json.loads(out.read_text()) == json.loads(payload)
@@ -26,7 +26,7 @@ def test_traversal_filename_is_confined_not_escaped(tmp_path):
# a '../../evil.json' must NOT land outside the sandbox; basename flattens it
msg = save_page_data(str(tmp_path), "sid", "../../evil.json", "[]")
assert msg.startswith("Saved")
assert (tmp_path / _SUBDIR / "evil.json").is_file()
assert (tmp_path / P_SUBDIR / "evil.json").is_file()
# nothing was written two levels up
assert not (tmp_path.parent.parent / "evil.json").exists()
@@ -34,7 +34,7 @@ def test_traversal_filename_is_confined_not_escaped(tmp_path):
def test_absolute_path_filename_is_confined(tmp_path):
msg = save_page_data(str(tmp_path), "sid", "/etc/evil.json", "[]")
assert msg.startswith("Saved")
assert (tmp_path / _SUBDIR / "evil.json").is_file()
assert (tmp_path / P_SUBDIR / "evil.json").is_file()
assert not os.path.exists("/etc/evil.json")
@@ -43,7 +43,7 @@ def test_disallowed_extension_is_rejected(tmp_path):
msg = save_page_data(str(tmp_path), "sid", bad, "data")
assert msg.startswith("Save failed"), bad
# the allowed ones all pass
for good in sorted(_ALLOWED_EXT):
for good in sorted(P_ALLOWED_EXT):
msg = save_page_data(str(tmp_path), "sid", f"file{good}", "x")
assert msg.startswith("Saved"), good
@@ -54,21 +54,21 @@ def test_empty_filename_is_rejected(tmp_path):
def test_oversize_payload_is_rejected(tmp_path):
big = "x" * (_MAX_BYTES + 1)
big = "x" * (P_MAX_BYTES + 1)
msg = save_page_data(str(tmp_path), "sid", "big.txt", big)
assert msg.startswith("Save failed")
assert not (tmp_path / _SUBDIR / "big.txt").exists()
assert not (tmp_path / P_SUBDIR / "big.txt").exists()
def test_falls_back_to_home_workspace_when_no_cwd(tmp_path, monkeypatch):
monkeypatch.setattr(os.path, "expanduser", lambda p: str(tmp_path))
msg = save_page_data(None, "sess-xyz", "f.json", "[]")
assert msg.startswith("Saved")
assert (tmp_path / ".openswarm" / "workspaces" / "sess-xyz" / _SUBDIR / "f.json").is_file()
assert (tmp_path / ".openswarm" / "workspaces" / "sess-xyz" / P_SUBDIR / "f.json").is_file()
def test_non_json_content_still_saves_without_count(tmp_path):
msg = save_page_data(str(tmp_path), "sid", "notes.txt", "just some text")
assert msg.startswith("Saved")
assert "items" not in msg and "keys" not in msg
assert (tmp_path / _SUBDIR / "notes.txt").read_text() == "just some text"
assert (tmp_path / P_SUBDIR / "notes.txt").read_text() == "just some text"
backend/tests/test_browser_screenshot_prune.py
+41 -41
View File
@@ -6,25 +6,25 @@ re-prefilled image tokens without losing the agent's memory (URL + ReportProgres
text stay). These pin the keep-set, the in-place mutation, and tool_result safety.
"""
from backend.apps.agents.browser.browser_history import (
from backend.apps.agents.browser.browser_history import ( # p-private-ignore: P_OMITTED_SCREENSHOT_STUB
prune_old_screenshots,
_OMITTED_SCREENSHOT_STUB,
P_OMITTED_SCREENSHOT_STUB,
)
def _img(tag):
def img(tag):
return {"type": "image", "source": {"type": "base64", "media_type": "image/jpeg", "data": tag}}
def _shot_turn(tag, url):
def shot_turn(tag, url):
# mirrors _format_tool_result for BrowserScreenshot: [image, text(url)]
return {"role": "user", "content": [{
"type": "tool_result", "tool_use_id": f"t_{tag}",
"content": [_img(tag), {"type": "text", "text": f"Screenshot captured. URL: {url}"}],
"content": [img(tag), {"type": "text", "text": f"Screenshot captured. URL: {url}"}],
}]}
def _count_images(messages):
def count_images(messages):
n = 0
for m in messages:
for b in m.get("content", []):
@@ -37,10 +37,10 @@ def _count_images(messages):
def test_keeps_first_and_last_two_collapses_middle():
msgs = [_shot_turn(str(i), f"https://site/{i}") for i in range(5)] # images 0..4
msgs = [shot_turn(str(i), f"https://site/{i}") for i in range(5)] # images 0..4
collapsed = prune_old_screenshots(msgs)
assert collapsed == 2 # 5 images - (first + last 2) = 2 stubbed
assert _count_images(msgs) == 3
assert count_images(msgs) == 3
# image tags that survive are 0 (first), 3 and 4 (last two)
surviving = [b["source"]["data"] for m in msgs for tr in m["content"]
for b in tr["content"] if b.get("type") == "image"]
@@ -48,46 +48,46 @@ def test_keeps_first_and_last_two_collapses_middle():
def test_three_or_fewer_is_a_noop():
msgs = [_shot_turn(str(i), f"u{i}") for i in range(3)]
msgs = [shot_turn(str(i), f"u{i}") for i in range(3)]
assert prune_old_screenshots(msgs) == 0
assert _count_images(msgs) == 3
assert count_images(msgs) == 3
def test_stub_preserves_the_url_text_block():
msgs = [_shot_turn(str(i), f"https://site/{i}") for i in range(4)]
msgs = [shot_turn(str(i), f"https://site/{i}") for i in range(4)]
prune_old_screenshots(msgs)
# the collapsed shot (#1) keeps its "URL:" text, only the image became a stub
collapsed_tr = msgs[1]["content"][0]["content"]
assert any(b.get("text") == _OMITTED_SCREENSHOT_STUB for b in collapsed_tr)
assert any(b.get("text") == P_OMITTED_SCREENSHOT_STUB for b in collapsed_tr)
assert any("URL: https://site/1" in b.get("text", "") for b in collapsed_tr)
def test_handles_direct_image_blocks_too():
msgs = [
{"role": "user", "content": [_img("a"), {"type": "text", "text": "hi"}]},
{"role": "user", "content": [_img("b")]},
{"role": "user", "content": [_img("c")]},
{"role": "user", "content": [_img("d")]},
{"role": "user", "content": [img("a"), {"type": "text", "text": "hi"}]},
{"role": "user", "content": [img("b")]},
{"role": "user", "content": [img("c")]},
{"role": "user", "content": [img("d")]},
]
collapsed = prune_old_screenshots(msgs)
assert collapsed == 1 # keep a (first), c+d (last two); stub b
assert msgs[1]["content"][0] == {"type": "text", "text": _OMITTED_SCREENSHOT_STUB}
assert msgs[1]["content"][0] == {"type": "text", "text": P_OMITTED_SCREENSHOT_STUB}
def test_keep_recent_is_tunable():
msgs = [_shot_turn(str(i), f"u{i}") for i in range(6)]
msgs = [shot_turn(str(i), f"u{i}") for i in range(6)]
prune_old_screenshots(msgs, keep_first=False, keep_recent=1)
# only the most recent survives
assert _count_images(msgs) == 1
assert count_images(msgs) == 1
def _tool_use_msg(tu_id, name):
def tool_use_msg(tu_id, name):
return {"role": "assistant", "content": [
{"type": "tool_use", "id": tu_id, "name": name, "input": {}},
]}
def _tool_result_msg(tu_id, text):
def tool_result_msg(tu_id, text):
return {"role": "user", "content": [
{"type": "tool_result", "tool_use_id": tu_id,
"content": [{"type": "text", "text": text}]},
@@ -100,8 +100,8 @@ def test_prune_stale_page_state_keeps_last_two_attachments():
)
msgs = []
for i in range(4):
msgs.append(_tool_use_msg(f"t{i}", "BrowserClickIndex"))
msgs.append(_tool_result_msg(
msgs.append(tool_use_msg(f"t{i}", "BrowserClickIndex"))
msgs.append(tool_result_msg(
f"t{i}", f"Clicked [{i}]\n\n{PAGE_STATE_MARKER}\n[1]<button \"A{i}\">",
))
pruned = prune_stale_page_state(msgs)
@@ -119,10 +119,10 @@ def test_prune_stale_page_state_collapses_old_heavy_reads_only():
big = "28 interactive elements\n" + "\n".join(f"[{i}]<button \"x\">" for i in range(60))
msgs = []
for i in range(3):
msgs.append(_tool_use_msg(f"r{i}", "BrowserListInteractives"))
msgs.append(_tool_result_msg(f"r{i}", big))
msgs.append(_tool_use_msg("nav", "BrowserNavigate"))
msgs.append(_tool_result_msg("nav", "Navigated to https://example.com"))
msgs.append(tool_use_msg(f"r{i}", "BrowserListInteractives"))
msgs.append(tool_result_msg(f"r{i}", big))
msgs.append(tool_use_msg("nav", "BrowserNavigate"))
msgs.append(tool_result_msg("nav", "Navigated to https://example.com"))
pruned = prune_stale_page_state(msgs)
assert pruned == 1
first = msgs[1]["content"][0]["content"][0]["text"]
@@ -136,8 +136,8 @@ def test_cache_marker_places_one_at_depth_and_strips_old():
from backend.apps.agents.browser.browser_history import place_cache_marker
msgs = []
for i in range(12):
msgs.append(_tool_use_msg(f"t{i}", "BrowserClickIndex"))
msgs.append(_tool_result_msg(f"t{i}", f"Clicked [{i}]"))
msgs.append(tool_use_msg(f"t{i}", "BrowserClickIndex"))
msgs.append(tool_result_msg(f"t{i}", f"Clicked [{i}]"))
place_cache_marker(msgs)
place_cache_marker(msgs) # second pass must not accumulate markers
marked = [
@@ -151,7 +151,7 @@ def test_cache_marker_places_one_at_depth_and_strips_old():
def test_cache_marker_skips_short_conversations():
from backend.apps.agents.browser.browser_history import place_cache_marker
msgs = [_tool_use_msg("t0", "BrowserClickIndex"), _tool_result_msg("t0", "Clicked")]
msgs = [tool_use_msg("t0", "BrowserClickIndex"), tool_result_msg("t0", "Clicked")]
place_cache_marker(msgs)
assert all(
"cache_control" not in b
@@ -164,32 +164,32 @@ def test_cache_marker_prefix_stays_stable_across_a_simulated_turn():
from backend.apps.agents.browser.browser_history import (
PAGE_STATE_MARKER, place_cache_marker, prune_stale_page_state,
)
import copy as _copy
import json as _json
import copy
import json
def _stripped(ms):
ms = _copy.deepcopy(ms)
def stripped(ms):
ms = copy.deepcopy(ms)
for m in ms:
for b in (m["content"] if isinstance(m["content"], list) else []):
if isinstance(b, dict):
b.pop("cache_control", None)
return _json.dumps(ms, sort_keys=True)
return json.dumps(ms, sort_keys=True)
msgs = []
for i in range(10):
msgs.append(_tool_use_msg(f"t{i}", "BrowserClickIndex"))
msgs.append(_tool_result_msg(
msgs.append(tool_use_msg(f"t{i}", "BrowserClickIndex"))
msgs.append(tool_result_msg(
f"t{i}", f"Clicked [{i}]\n\n{PAGE_STATE_MARKER}\n[1]<button \"A{i}\">",
))
prune_stale_page_state(msgs)
place_cache_marker(msgs)
cut = len(msgs) - 8
before = _stripped(msgs[:cut])
before = stripped(msgs[:cut])
# next turn: a new attachment arrives, pruning collapses the one falling out
msgs.append(_tool_use_msg("t10", "BrowserClickIndex"))
msgs.append(_tool_result_msg(
msgs.append(tool_use_msg("t10", "BrowserClickIndex"))
msgs.append(tool_result_msg(
"t10", f"Clicked [10]\n\n{PAGE_STATE_MARKER}\n[1]<button \"A10\">",
))
prune_stale_page_state(msgs)
place_cache_marker(msgs)
assert _stripped(msgs[:cut]) == before
assert stripped(msgs[:cut]) == before
backend/tests/test_browser_self_audit.py
+14 -14
View File
@@ -8,7 +8,7 @@ import tempfile
from backend.apps.agents.browser import browser_self_audit as audit
def _write(d, name, rows):
def write_rows(d, name, rows):
with open(os.path.join(d, name), "w", encoding="utf-8") as f:
for r in rows:
f.write(json.dumps(r) + "\n")
@@ -22,8 +22,8 @@ def test_thrash_is_flagged_only_without_a_promote():
# a healthy skill: edited a couple times THEN promoted = not thrash
ev += [{"kind": "edit", "host": "y.com", "task_sig": "s2"},
{"kind": "promote", "host": "y.com", "task_sig": "s2"}]
_write(d, "skill_events.jsonl", ev)
_write(d, "tasks.jsonl", [])
write_rows(d, "skill_events.jsonl", ev)
write_rows(d, "tasks.jsonl", [])
r = audit.audit(d)
thrash = [f for f in r["findings"] if f["kind"] == "thrash"]
assert len(thrash) == 1 and thrash[0]["host"] == "x.com"
@@ -34,8 +34,8 @@ def test_stall_flags_runs_far_above_the_host_norm():
# six fast runs (norm ~4) and two big spikes on the same card
tasks = [{"browser_id": "b1", "turns": n} for n in (4, 4, 5, 3, 4, 4)]
tasks += [{"browser_id": "b1", "turns": 18}, {"browser_id": "b1", "turns": 20}]
_write(d, "tasks.jsonl", tasks)
_write(d, "skill_events.jsonl", [])
write_rows(d, "tasks.jsonl", tasks)
write_rows(d, "skill_events.jsonl", [])
r = audit.audit(d)
assert any(f["kind"] == "stall" for f in r["findings"])
@@ -44,16 +44,16 @@ def test_error_rate_flags_a_systemically_failing_host():
d = tempfile.mkdtemp()
tasks = [{"browser_id": "b9", "tool_calls": 30,
"recurring_errors": {"index no longer valid": 12}}]
_write(d, "tasks.jsonl", tasks)
_write(d, "skill_events.jsonl", [])
write_rows(d, "tasks.jsonl", tasks)
write_rows(d, "skill_events.jsonl", [])
r = audit.audit(d)
assert any(f["kind"] == "error_rate" for f in r["findings"])
def test_clean_history_proposes_nothing():
d = tempfile.mkdtemp()
_write(d, "tasks.jsonl", [{"browser_id": "b1", "turns": 4, "tool_calls": 5} for _ in range(6)])
_write(d, "skill_events.jsonl", [{"kind": "learn", "host": "x.com", "task_sig": "s"},
write_rows(d, "tasks.jsonl", [{"browser_id": "b1", "turns": 4, "tool_calls": 5} for _ in range(6)])
write_rows(d, "skill_events.jsonl", [{"kind": "learn", "host": "x.com", "task_sig": "s"},
{"kind": "promote", "host": "x.com", "task_sig": "s"}])
r = audit.audit(d)
assert r["findings"] == []
@@ -65,9 +65,9 @@ def test_audit_fires_every_n_finished_tasks(monkeypatch, tmp_path, mocker):
# threads synchronous so the test is deterministic, and use a small N.
from backend.apps.agents.browser import browser_metrics as m
monkeypatch.setenv("OPENSWARM_BROWSER_METRICS_DIR", str(tmp_path))
m._metrics_dir_cache = None
m._task_count = 0
monkeypatch.setattr(m, "_AUDIT_EVERY_N", 5)
m.P_METRICS_DIR_CACHE = None
m.P_TASK_COUNT = 0
monkeypatch.setattr(m, "P_AUDIT_EVERY_N", 5)
# autospec keeps the stub honest to threading.Thread's real signature (target/
# name/daemon); start() just runs the captured target synchronously.
@@ -87,8 +87,8 @@ def test_audit_fires_every_n_finished_tasks(monkeypatch, tmp_path, mocker):
def test_run_and_write_emits_a_report_file_and_never_raises():
d = tempfile.mkdtemp()
_write(d, "tasks.jsonl", [])
_write(d, "skill_events.jsonl", [])
write_rows(d, "tasks.jsonl", [])
write_rows(d, "skill_events.jsonl", [])
path = audit.run_and_write(d)
assert path and os.path.exists(path)
# also safe on a totally missing dir
backend/tests/test_browser_skills.py
+92 -92
View File
@@ -9,7 +9,7 @@ from backend.apps.agents.browser import browser_skills as sk
@pytest.fixture(autouse=True)
def _isolated_skills(monkeypatch):
def isolated_skills(monkeypatch):
# Persist to a throwaway dir so tests never touch the real DATA_ROOT.
d = tempfile.mkdtemp(prefix="skills_test_")
monkeypatch.setenv("OPENSWARM_BROWSER_SKILLS_DIR", d)
@@ -31,7 +31,7 @@ def test_host_of():
assert sk.host_of("https://docs.google.com/x") == "docs.google.com"
def _log():
def action_log():
return [
{"tool": "BrowserScreenshot", "input": {}, "ok": False},
{"tool": "BrowserNavigate", "input": {"url": "http://h/form"}, "ok": True},
@@ -43,7 +43,7 @@ def _log():
def test_distill_builds_robust_steps():
steps = sk.distill_steps(_log())
steps = sk.distill_steps(action_log())
tools = [s["tool"] for s in steps]
# reads/screenshots dropped; click_index becomes a robust click-by-name
assert tools == ["BrowserNavigate", "BrowserType", "BrowserClickByName"]
@@ -101,14 +101,14 @@ def test_distill_bails_on_batched_click_index():
def test_record_and_find_roundtrip():
assert sk.record_skill("localhost:8901", "type hello and click Send", _log()) is True
assert sk.record_skill("localhost:8901", "type hello and click Send", action_log()) is True
found = sk.find_skill("localhost:8901", "Please type hello and click Send")
assert found is not None
assert [s["tool"] for s in found["steps"]] == ["BrowserNavigate", "BrowserType", "BrowserClickByName"]
def test_find_is_host_scoped():
sk.record_skill("a.com", "do thing now", _log())
sk.record_skill("a.com", "do thing now", action_log())
assert sk.find_skill("b.com", "do thing now") is None
@@ -119,10 +119,10 @@ def test_record_refuses_unrecordable_run():
# --- persistence + redaction ----------------------------------------------
def test_skill_persists_across_restart(_isolated_skills):
def test_skill_persists_across_restart(isolated_skills):
# record, then simulate a process restart by wiping ONLY the in-memory cache;
# find must re-load it from disk.
assert sk.record_skill("localhost:8901", "type hello and click Send", _log()) is True
assert sk.record_skill("localhost:8901", "type hello and click Send", action_log()) is True
sk.clear(wipe_disk=False) # in-memory gone, disk intact (== restart)
assert not sk._skills # cache truly empty
found = sk.find_skill("localhost:8901", "type hello and click Send")
@@ -130,7 +130,7 @@ def test_skill_persists_across_restart(_isolated_skills):
assert [s["tool"] for s in found["steps"]] == ["BrowserNavigate", "BrowserType", "BrowserClickByName"]
def test_sensitive_text_is_NOT_persisted(_isolated_skills):
def test_sensitive_text_is_NOT_persisted(isolated_skills):
# a skill that types an email/password must stay in-memory only (no disk file)
log = [
{"tool": "BrowserType", "input": {"selector": "#email", "text": "eric@example.com"}, "ok": True},
@@ -145,7 +145,7 @@ def test_sensitive_text_is_NOT_persisted(_isolated_skills):
assert sk.find_skill("site.com", "enter email and submit") is None
def test_password_field_selector_blocks_persistence(_isolated_skills):
def test_password_field_selector_blocks_persistence(isolated_skills):
log = [
{"tool": "BrowserType", "input": {"selector": "input#password", "text": "hunter2"}, "ok": True},
{"tool": "BrowserClickIndex", "input": {}, "ok": True, "clicked_role": "button", "clicked_name": "Log in"},
@@ -165,7 +165,7 @@ def test_sensitivity_detector():
assert not sk._looks_sensitive("openswarm", selector="#search")
def test_navigate_url_userinfo_and_fragment_stripped_on_disk(_isolated_skills):
def test_navigate_url_userinfo_and_fragment_stripped_on_disk(isolated_skills):
log = [
{"tool": "BrowserNavigate", "input": {"url": "https://user:pw@site.com/app?q=1#frag"}, "ok": True},
{"tool": "BrowserType", "input": {"selector": "#q", "text": "shoes"}, "ok": True},
@@ -186,15 +186,15 @@ def test_navigate_url_userinfo_and_fragment_stripped_on_disk(_isolated_skills):
assert "#section" not in nav["params"]["url"]
def test_format_version_mismatch_is_ignored(_isolated_skills, monkeypatch):
sk.record_skill("v.com", "do a thing now", _log())
def test_format_version_mismatch_is_ignored(isolated_skills, monkeypatch):
sk.record_skill("v.com", "do a thing now", action_log())
sk.clear(wipe_disk=False)
monkeypatch.setattr(sk, "_SKILL_FORMAT_VERSION", 999) # pretend the format moved on
assert sk.find_skill("v.com", "do a thing now") is None
# --- parameterization: "same task, different input" -----------------------
def test_quoted_value_becomes_a_slot_and_reuses_across_inputs(_isolated_skills):
def test_quoted_value_becomes_a_slot_and_reuses_across_inputs(isolated_skills):
# learn from a task with a quoted value
log = [
{"tool": "BrowserNavigate", "input": {"url": "https://shop.com/search"}, "ok": True},
@@ -210,7 +210,7 @@ def test_quoted_value_becomes_a_slot_and_reuses_across_inputs(_isolated_skills):
assert type_step["params"]["text"] == "winter boots" # filled from the NEW task
def test_parameterized_value_is_not_persisted(_isolated_skills):
def test_parameterized_value_is_not_persisted(isolated_skills):
log = [
{"tool": "BrowserType", "input": {"selector": "#q", "text": "running shoes"}, "ok": True},
{"tool": "BrowserClickIndex", "input": {}, "ok": True, "clicked_role": "button", "clicked_name": "Search"},
@@ -222,7 +222,7 @@ def test_parameterized_value_is_not_persisted(_isolated_skills):
assert '"value_slot": 0' in blob or '"value_slot":0' in blob
def test_rehydrate_aborts_when_slot_cannot_be_filled(_isolated_skills):
def test_rehydrate_aborts_when_slot_cannot_be_filled(isolated_skills):
log = [
{"tool": "BrowserType", "input": {"selector": "#q", "text": "shoes"}, "ok": True},
{"tool": "BrowserClickIndex", "input": {}, "ok": True, "clicked_role": "button", "clicked_name": "Go"},
@@ -234,9 +234,9 @@ def test_rehydrate_aborts_when_slot_cannot_be_filled(_isolated_skills):
assert sk.rehydrate(found, "search for shoes") is None
def test_unquoted_text_stays_literal_backward_compatible(_isolated_skills):
def test_unquoted_text_stays_literal_backward_compatible(isolated_skills):
# no quotes -> behaves exactly as before (literal text, exact-ish key)
assert sk.record_skill("localhost:8901", "type hello and click Send", _log()) is True
assert sk.record_skill("localhost:8901", "type hello and click Send", action_log()) is True
found = sk.find_skill("localhost:8901", "Please type hello and click Send")
assert found is not None
concrete = sk.rehydrate(found, "Please type hello and click Send")
@@ -245,24 +245,24 @@ def test_unquoted_text_stays_literal_backward_compatible(_isolated_skills):
# --- skill self-awareness (list / deprecate) ------------------------------
def test_list_skills_for_host(_isolated_skills):
sk.record_skill("shop.com", "search for shoes now", _log())
sk.record_skill("shop.com", "add item to the cart now", _log())
sk.record_skill("other.com", "do a thing now", _log())
def test_list_skills_for_host(isolated_skills):
sk.record_skill("shop.com", "search for shoes now", action_log())
sk.record_skill("shop.com", "add item to the cart now", action_log())
sk.record_skill("other.com", "do a thing now", action_log())
listed = sk.list_skills("shop.com")
tasks = {x["task"] for x in listed}
assert len(listed) == 2 and all("steps" in x and "replays" in x for x in listed)
assert not any(t for t in tasks if t in sk.list_skills("other.com")) # host-scoped
def test_list_skills_reads_disk_after_restart(_isolated_skills):
sk.record_skill("shop.com", "search for shoes now", _log())
def test_list_skills_reads_disk_after_restart(isolated_skills):
sk.record_skill("shop.com", "search for shoes now", action_log())
sk.clear(wipe_disk=False) # restart: memory gone, disk intact
assert len(sk.list_skills("shop.com")) == 1
def test_deprecate_removes_skill_from_memory_and_disk(_isolated_skills):
sk.record_skill("shop.com", "search for shoes now", _log())
def test_deprecate_removes_skill_from_memory_and_disk(isolated_skills):
sk.record_skill("shop.com", "search for shoes now", action_log())
sig = sk._sig("search for shoes now")
assert os.path.exists(sk._skill_path("shop.com", sig))
# deprecate using the task_sig as list_skills would surface it
@@ -271,7 +271,7 @@ def test_deprecate_removes_skill_from_memory_and_disk(_isolated_skills):
assert sk.find_skill("shop.com", "search for shoes now") is None
def test_deprecate_unknown_is_false(_isolated_skills):
def test_deprecate_unknown_is_false(isolated_skills):
assert sk.deprecate_skill("shop.com", "never recorded this") is False
@@ -280,21 +280,21 @@ def test_deprecate_unknown_is_false(_isolated_skills):
# fails is quarantined (never replayed again) so a lossy skill can't ghost-succeed
# or run slower-than-baseline; re-deriving different steps is a re-versioned EDIT.
def test_new_skill_starts_on_probation(_isolated_skills):
sk.record_skill("shop.com", "do a thing now", _log())
def test_new_skill_starts_on_probation(isolated_skills):
sk.record_skill("shop.com", "do a thing now", action_log())
s = sk.find_skill("shop.com", "do a thing now")
assert s["state"] == sk._PROBATION and s["rev"] == 1 and s["replays"] == 0
def test_replay_success_promotes_probation_to_trusted(_isolated_skills):
sk.record_skill("shop.com", "do a thing now", _log())
def test_replay_success_promotes_probation_to_trusted(isolated_skills):
sk.record_skill("shop.com", "do a thing now", action_log())
sk.mark_replay_succeeded("shop.com", "do a thing now")
s = sk.find_skill("shop.com", "do a thing now")
assert s["state"] == sk._TRUSTED and s["replays"] == 1 and s["fails"] == 0
def test_probation_failure_quarantines_and_blocks_future_replay(_isolated_skills):
sk.record_skill("shop.com", "do a thing now", _log()) # probation
def test_probation_failure_quarantines_and_blocks_future_replay(isolated_skills):
sk.record_skill("shop.com", "do a thing now", action_log()) # probation
verdict = sk.mark_replay_failed("shop.com", "do a thing now")
assert verdict == "quarantined"
# the ghost guard: a quarantined skill is NEVER handed back for replay...
@@ -304,30 +304,30 @@ def test_probation_failure_quarantines_and_blocks_future_replay(_isolated_skills
assert len(listed) == 1 and listed[0]["state"] == sk._QUARANTINE
def test_quarantined_skill_re_recorded_identical_stays_quarantined(_isolated_skills):
sk.record_skill("shop.com", "do a thing now", _log())
def test_quarantined_skill_re_recorded_identical_stays_quarantined(isolated_skills):
sk.record_skill("shop.com", "do a thing now", action_log())
sk.mark_replay_failed("shop.com", "do a thing now") # quarantined
# the full LLM agent re-runs and distills the SAME (still-lossy) steps:
sk.record_skill("shop.com", "do a thing now", _log())
sk.record_skill("shop.com", "do a thing now", action_log())
# it must stay quarantined -> pure-LLM baseline, never a wasted replay again
assert sk.find_skill("shop.com", "do a thing now") is None
assert sk.list_skills("shop.com")[0]["state"] == sk._QUARANTINE
def test_quarantined_skill_unquarantines_on_a_real_edit(_isolated_skills):
sk.record_skill("shop.com", "do a thing now", _log())
def test_quarantined_skill_unquarantines_on_a_real_edit(isolated_skills):
sk.record_skill("shop.com", "do a thing now", action_log())
sk.mark_replay_failed("shop.com", "do a thing now") # quarantined
# now the page changed and the LLM derives a DIFFERENT click -> a real edit,
# which earns the skill another chance (back on probation, re-versioned)
edited = _log()[:-1] + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
edited = action_log()[:-1] + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
"clicked_role": "button", "clicked_name": "Submit"}]
sk.record_skill("shop.com", "do a thing now", edited)
s = sk.find_skill("shop.com", "do a thing now")
assert s is not None and s["state"] == sk._PROBATION and s["rev"] == 2
def test_trusted_skill_tolerates_one_transient_miss_then_demotes(_isolated_skills):
sk.record_skill("shop.com", "do a thing now", _log())
def test_trusted_skill_tolerates_one_transient_miss_then_demotes(isolated_skills):
sk.record_skill("shop.com", "do a thing now", action_log())
sk.mark_replay_succeeded("shop.com", "do a thing now") # trusted
assert sk.mark_replay_failed("shop.com", "do a thing now") == "kept"
s = sk.find_skill("shop.com", "do a thing now")
@@ -336,19 +336,19 @@ def test_trusted_skill_tolerates_one_transient_miss_then_demotes(_isolated_skill
assert sk.find_skill("shop.com", "do a thing now")["state"] == sk._PROBATION
def test_re_record_identical_keeps_trust_and_rev(_isolated_skills):
sk.record_skill("shop.com", "do a thing now", _log())
def test_re_record_identical_keeps_trust_and_rev(isolated_skills):
sk.record_skill("shop.com", "do a thing now", action_log())
sk.mark_replay_succeeded("shop.com", "do a thing now")
sk.find_skill("shop.com", "do a thing now")["replays"] = 5 # pretend reused a lot
sk.record_skill("shop.com", "do a thing now", _log()) # identical re-derive
sk.record_skill("shop.com", "do a thing now", action_log()) # identical re-derive
s = sk.find_skill("shop.com", "do a thing now")
assert s["state"] == sk._TRUSTED and s["rev"] == 1 and s["replays"] == 5
def test_re_record_different_is_an_edit_that_reversions_to_probation(_isolated_skills):
sk.record_skill("shop.com", "do a thing now", _log())
def test_re_record_different_is_an_edit_that_reversions_to_probation(isolated_skills):
sk.record_skill("shop.com", "do a thing now", action_log())
sk.mark_replay_succeeded("shop.com", "do a thing now") # trusted, rev 1
edited = _log()[:-1] + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
edited = action_log()[:-1] + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
"clicked_role": "button", "clicked_name": "Submit"}]
sk.record_skill("shop.com", "do a thing now", edited) # different -> EDIT
s = sk.find_skill("shop.com", "do a thing now")
@@ -357,10 +357,10 @@ def test_re_record_different_is_an_edit_that_reversions_to_probation(_isolated_s
assert cbn["params"]["name"] == "Submit" # the new step stuck
def test_rev_and_state_persist_across_restart(_isolated_skills):
sk.record_skill("shop.com", "do a thing now", _log())
def test_rev_and_state_persist_across_restart(isolated_skills):
sk.record_skill("shop.com", "do a thing now", action_log())
sk.mark_replay_succeeded("shop.com", "do a thing now")
edited = _log()[:-1] + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
edited = action_log()[:-1] + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
"clicked_role": "button", "clicked_name": "Submit"}]
sk.record_skill("shop.com", "do a thing now", edited) # rev 2, probation
sk.clear(wipe_disk=False) # restart
@@ -380,13 +380,13 @@ def test_steps_equal_distinguishes_slot_from_literal_and_changed_click():
assert not sk._steps_equal([send], [submit]) # renamed button IS an edit
def test_mark_replay_helpers_on_unknown_are_safe(_isolated_skills):
def test_mark_replay_helpers_on_unknown_are_safe(isolated_skills):
sk.mark_replay_succeeded("shop.com", "never recorded") # no raise
assert sk.mark_replay_failed("shop.com", "never recorded") == "none"
def test_demoted_skill_can_be_re_proven(_isolated_skills):
sk.record_skill("shop.com", "do a thing now", _log())
def test_demoted_skill_can_be_re_proven(isolated_skills):
sk.record_skill("shop.com", "do a thing now", action_log())
sk.mark_replay_succeeded("shop.com", "do a thing now") # trusted
sk.mark_replay_failed("shop.com", "do a thing now")
sk.mark_replay_failed("shop.com", "do a thing now") # demoted to probation
@@ -397,34 +397,34 @@ def test_demoted_skill_can_be_re_proven(_isolated_skills):
# --- composition: build on what's already proven, propagate staleness -------
def _log_plus():
# distills to _log()'s 3 steps PLUS a 4th click -> a strict superset sequence
return _log() + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
def action_log_plus():
# distills to action_log()'s 3 steps PLUS a 4th click -> a strict superset sequence
return action_log() + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
"clicked_role": "button", "clicked_name": "Checkout"}]
def _trust(host, task, log):
def trust(host, task, log):
sk.record_skill(host, task, log)
sk.mark_replay_succeeded(host, task)
def test_composition_links_to_trusted_sub_skill(_isolated_skills):
_trust("shop.com", "search shoes now", _log()) # trusted foundation
sk.record_skill("shop.com", "search shoes and checkout now", _log_plus())
def test_composition_links_to_trusted_sub_skill(isolated_skills):
trust("shop.com", "search shoes now", action_log()) # trusted foundation
sk.record_skill("shop.com", "search shoes and checkout now", action_log_plus())
c = sk.find_skill("shop.com", "search shoes and checkout now")
assert c["composed_of"] == [sk._sig("search shoes now")]
def test_composition_ignores_untrusted_foundation(_isolated_skills):
sk.record_skill("shop.com", "search shoes now", _log()) # probation, NOT trusted
sk.record_skill("shop.com", "search shoes and checkout now", _log_plus())
def test_composition_ignores_untrusted_foundation(isolated_skills):
sk.record_skill("shop.com", "search shoes now", action_log()) # probation, NOT trusted
sk.record_skill("shop.com", "search shoes and checkout now", action_log_plus())
c = sk.find_skill("shop.com", "search shoes and checkout now")
assert c["composed_of"] == [] # only a PROVEN sub-skill is built upon
def test_deprecating_a_foundation_demotes_everything_built_on_it(_isolated_skills):
_trust("shop.com", "search shoes now", _log())
_trust("shop.com", "search shoes and checkout now", _log_plus()) # composed + trusted
def test_deprecating_a_foundation_demotes_everything_built_on_it(isolated_skills):
trust("shop.com", "search shoes now", action_log())
trust("shop.com", "search shoes and checkout now", action_log_plus()) # composed + trusted
assert sk.find_skill("shop.com", "search shoes and checkout now")["state"] == sk._TRUSTED
sk.deprecate_skill("shop.com", "search shoes now") # foundation pulled
# the ghost guard for composition: the dependent must NOT stay trusted on a
@@ -432,26 +432,26 @@ def test_deprecating_a_foundation_demotes_everything_built_on_it(_isolated_skill
assert sk.find_skill("shop.com", "search shoes and checkout now")["state"] == sk._PROBATION
def test_demoting_a_foundation_demotes_its_dependents(_isolated_skills):
_trust("shop.com", "search shoes now", _log())
_trust("shop.com", "search shoes and checkout now", _log_plus())
def test_demoting_a_foundation_demotes_its_dependents(isolated_skills):
trust("shop.com", "search shoes now", action_log())
trust("shop.com", "search shoes and checkout now", action_log_plus())
sk.mark_replay_failed("shop.com", "search shoes now")
sk.mark_replay_failed("shop.com", "search shoes now") # foundation demoted
assert sk.find_skill("shop.com", "search shoes and checkout now")["state"] == sk._PROBATION
def test_editing_a_foundation_demotes_its_dependents(_isolated_skills):
_trust("shop.com", "search shoes now", _log())
_trust("shop.com", "search shoes and checkout now", _log_plus())
edited = _log()[:-1] + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
def test_editing_a_foundation_demotes_its_dependents(isolated_skills):
trust("shop.com", "search shoes now", action_log())
trust("shop.com", "search shoes and checkout now", action_log_plus())
edited = action_log()[:-1] + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
"clicked_role": "button", "clicked_name": "Find"}]
sk.record_skill("shop.com", "search shoes now", edited) # foundation changed
assert sk.find_skill("shop.com", "search shoes and checkout now")["state"] == sk._PROBATION
def test_list_skills_surfaces_state_rev_and_builds_on(_isolated_skills):
_trust("shop.com", "search shoes now", _log())
sk.record_skill("shop.com", "search shoes and checkout now", _log_plus())
def test_list_skills_surfaces_state_rev_and_builds_on(isolated_skills):
trust("shop.com", "search shoes now", action_log())
sk.record_skill("shop.com", "search shoes and checkout now", action_log_plus())
listed = {x["task"]: x for x in sk.list_skills("shop.com")}
foundation = listed[sk._sig("search shoes now")]
composed = listed[sk._sig("search shoes and checkout now")]
@@ -477,12 +477,12 @@ def test_replay_safety_refuses_send_steps_and_passes_reads():
assert ok is False and "irreversible" in why
def test_extract_first_json_strips_fences_and_prose():
from backend.apps.agents.browser.browser_extract import _first_json
assert _first_json('```json\n{"a": 1}\n```') == '{"a": 1}'
assert _first_json('Here you go: [{"n": "x"}] hope that helps') == '[{"n": "x"}]'
assert _first_json("no json here") == ""
assert _first_json('{"broken": ') == ""
def test_extractp_first_json_strips_fences_and_prose():
from backend.apps.agents.browser.browser_extract import p_first_json # p-private-ignore: p_first_json
assert p_first_json('```json\n{"a": 1}\n```') == '{"a": 1}'
assert p_first_json('Here you go: [{"n": "x"}] hope that helps') == '[{"n": "x"}]'
assert p_first_json("no json here") == ""
assert p_first_json('{"broken": ') == ""
def test_widened_redaction_catches_audit_bypasses():
@@ -517,13 +517,13 @@ def test_first_unsafe_step_splits_send_skills():
def test_template_task_ignores_possessive_apostrophes():
from backend.apps.agents.browser.browser_skills import template_task, _sig
from backend.apps.agents.browser.browser_skills import template_task
r14 = "go to tyler chen's linkedin hes in entrepreneurs first and text him '[test] hello world r14-os'"
r15 = "go to tyler chen's linkedin hes in entrepreneurs first and text him '[test] hello world r15-os'"
t14, v14 = template_task(r14)
assert v14 == ["[test] hello world r14-os"]
assert "chen's linkedin" in t14
assert _sig(r14) == _sig(r15)
assert sk._sig(r14) == sk._sig(r15)
assert template_task("no quotes here at all") == ("no quotes here at all", [])
@@ -612,7 +612,7 @@ def test_distill_batch_aborted_tail_and_missing_identities():
# --- route hints (advisory reuse when replay can't run) ---------------------
def _record_dm_skill(host="www.linkedin.com"):
def record_dm_skill(host="www.linkedin.com"):
log = [
{"tool": "BrowserNavigate", "input": {"url": f"https://{host}/search/results/people/?keywords=tyler+chen"}, "ok": True},
{"tool": "BrowserClickIndex", "input": {"index": 7}, "ok": True,
@@ -628,8 +628,8 @@ def _record_dm_skill(host="www.linkedin.com"):
return host, task
def test_find_similar_skill_exact_and_variant(_isolated_skills):
host, task = _record_dm_skill()
def test_find_similar_skill_exact_and_variant(isolated_skills):
host, task = record_dm_skill()
s, score = sk.find_similar_skill(host, task)
assert s is not None and score == 1.0
# different quoted payload = same sig (slot), still exact
@@ -646,15 +646,15 @@ def test_find_similar_skill_exact_and_variant(_isolated_skills):
assert s5 is None
def test_find_similar_skill_skips_quarantined(_isolated_skills):
host, task = _record_dm_skill()
def test_find_similar_skill_skips_quarantined(isolated_skills):
host, task = record_dm_skill()
sk.mark_replay_failed(host, task) # probation -> quarantine
s, _ = sk.find_similar_skill(host, task)
assert s is None
def test_render_route_hint_fills_slots_and_flags_send(_isolated_skills):
host, task = _record_dm_skill()
def test_render_route_hint_fills_slots_and_flags_send(isolated_skills):
host, task = record_dm_skill()
s, score = sk.find_similar_skill(host, "go to tyler chen's linkedin and text him 'fresh payload r9'")
hint, keys = sk.render_route_hint(s, "go to tyler chen's linkedin and text him 'fresh payload r9'", score)
assert "route hint" in hint and len(keys) == 5
@@ -668,8 +668,8 @@ def test_render_route_hint_fills_slots_and_flags_send(_isolated_skills):
assert "BrowserBatch" in hint
def test_route_hint_adoption_matching(_isolated_skills):
host, task = _record_dm_skill()
def test_route_hint_adoption_matching(isolated_skills):
host, task = record_dm_skill()
s, score = sk.find_similar_skill(host, task)
_, keys = sk.render_route_hint(s, task, score)
run_log = [
backend/tests/test_browser_stagnation.py
+27 -27
View File
@@ -1,9 +1,9 @@
"""Deterministic stagnation detection for the browser sub-agent."""
from backend.apps.agents.browser.browser_loop import (
_STAGNATION_ESCALATION_AT,
_STAGNATION_MAX,
_looks_like_failure,
from backend.apps.agents.browser.browser_loop import ( # p-private-ignore: P_STAGNATION_ESCALATION_AT, P_STAGNATION_MAX, p_looks_like_failure
P_STAGNATION_ESCALATION_AT,
P_STAGNATION_MAX,
p_looks_like_failure,
advance_stagnation,
card_is_unavailable,
completion_is_honest,
@@ -14,19 +14,19 @@ from backend.apps.agents.browser.browser_loop import (
)
def _fail(url="https://a.com"):
def fail(url="https://a.com"):
return {"text": "Element not found: '.x'", "url": url}
def test_looks_like_failure_positive():
assert _looks_like_failure("Element not found: '.foo'")
assert _looks_like_failure("Index 4 is no longer valid")
assert _looks_like_failure("Error: something broke")
def testp_looks_like_failure_positive():
assert p_looks_like_failure("Element not found: '.foo'")
assert p_looks_like_failure("Index 4 is no longer valid")
assert p_looks_like_failure("Error: something broke")
def test_looks_like_failure_negative():
assert not _looks_like_failure("Clicked element: button#submit")
assert not _looks_like_failure("Typed into: input#email")
def testp_looks_like_failure_negative():
assert not p_looks_like_failure("Clicked element: button#submit")
assert not p_looks_like_failure("Typed into: input#email")
def test_error_result_is_unproductive():
@@ -63,25 +63,25 @@ def test_neutral_read_tools_never_count():
def test_nudge_mentions_human_intervention_only_at_max():
assert "RequestHumanIntervention" not in stagnation_nudge(3)
assert "RequestHumanIntervention" in stagnation_nudge(_STAGNATION_MAX)
assert "RequestHumanIntervention" in stagnation_nudge(P_STAGNATION_MAX)
assert "ladder" in stagnation_nudge(3)
def test_advance_increments_on_failures_and_nudges_at_threshold():
streak, url, text, nudge = 0, "", "", None
nudges = []
for _ in range(_STAGNATION_ESCALATION_AT):
streak, url, text, nudge = advance_stagnation(streak, url, text, "BrowserClick", _fail())
for _ in range(P_STAGNATION_ESCALATION_AT):
streak, url, text, nudge = advance_stagnation(streak, url, text, "BrowserClick", fail())
nudges.append(nudge)
assert streak == _STAGNATION_ESCALATION_AT
assert streak == P_STAGNATION_ESCALATION_AT
assert nudges[-1] is not None # nudge fires exactly when the threshold is hit
assert nudges[0] is None and nudges[1] is None
def test_advance_resets_on_progress():
# two failures, then a navigation (URL change) clears the streak
streak, url, text, _ = advance_stagnation(0, "", "", "BrowserClick", _fail("https://a.com"))
streak, url, text, _ = advance_stagnation(streak, url, text, "BrowserClick", _fail("https://a.com"))
streak, url, text, _ = advance_stagnation(0, "", "", "BrowserClick", fail("https://a.com"))
streak, url, text, _ = advance_stagnation(streak, url, text, "BrowserClick", fail("https://a.com"))
assert streak == 2
streak, url, text, _ = advance_stagnation(
streak, url, text, "BrowserNavigate", {"text": "Navigated", "url": "https://b.com"},
@@ -97,9 +97,9 @@ def test_advance_neutral_tools_pass_through_unchanged():
def test_advance_fires_again_at_max():
streak, url, text = _STAGNATION_MAX - 1, "https://a.com", "prev different"
streak, url, text, nudge = advance_stagnation(streak, url, text, "BrowserClick", _fail())
assert streak == _STAGNATION_MAX
streak, url, text = P_STAGNATION_MAX - 1, "https://a.com", "prev different"
streak, url, text, nudge = advance_stagnation(streak, url, text, "BrowserClick", fail())
assert streak == P_STAGNATION_MAX
assert nudge is not None and "RequestHumanIntervention" in nudge
assert stagnation_exhausted(streak)
@@ -108,23 +108,23 @@ def test_advance_fires_again_at_max():
# Catches the worst measured ghost: multi-minute runs, every tool errored, still
# reported 'completed'. Must NOT cry wolf on real successes (it overrides status).
def _ok(tool, summary="done"):
def ok(tool, summary="done"):
return {"tool": tool, "ok": True, "result_summary": summary}
def _err(tool):
def err(tool):
return {"tool": tool, "ok": False, "result_summary": "Element not found: '.x'"}
def test_completion_honest_when_an_action_succeeded():
log = [_ok("BrowserListInteractives", "1 button"), _ok("BrowserClickIndex", "Clicked")]
log = [ok("BrowserListInteractives", "1 button"), ok("BrowserClickIndex", "Clicked")]
honest, reason = completion_is_honest(log)
assert honest and reason == ""
def test_completion_ghost_when_every_action_errored():
# the exact LinkedIn ghost: 8 tools, all errored, model said 'completed'
log = [_err("BrowserClick") for _ in range(8)]
log = [err("BrowserClick") for _ in range(8)]
honest, reason = completion_is_honest(log)
assert not honest and "every state-changing action failed" in reason
@@ -143,14 +143,14 @@ def test_completion_ghost_when_only_looked_around_with_no_content():
def test_completion_honest_for_a_read_only_task_that_returned_content():
# a legit "tell me what's on the page" task: no action, but a read got content
log = [_ok("BrowserGetText", "The page says hello world")]
log = [ok("BrowserGetText", "The page says hello world")]
honest, reason = completion_is_honest(log)
assert honest and reason == ""
def test_completion_honest_when_some_errors_but_an_action_landed():
# partial failure is fine as long as a real action ultimately succeeded
log = [_err("BrowserClick"), _err("BrowserClick"), _ok("BrowserClickIndex", "Clicked Submit")]
log = [err("BrowserClick"), err("BrowserClick"), ok("BrowserClickIndex", "Clicked Submit")]
honest, reason = completion_is_honest(log)
assert honest
backend/tests/test_browser_validator.py
+15 -15
View File
@@ -2,38 +2,38 @@
import asyncio
from backend.apps.agents.browser.browser_validator import adjudicate_stuck, _extract_text
from backend.apps.agents.browser.browser_validator import adjudicate_stuck, p_extract_text # p-private-ignore: p_extract_text
class _Block:
class Block:
def __init__(self, type_, text=""):
self.type = type_
self.text = text
class _Resp:
class Resp:
def __init__(self, blocks):
self.content = blocks
class _FakeClient:
class FakeClient:
"""Minimal Anthropic-shaped client: client.messages.create(...)."""
def __init__(self, resp=None, raise_exc=None):
self._resp = resp
self._raise = raise_exc
self.resp = resp
self.raise_exc = raise_exc
self.calls = []
self.messages = self
async def create(self, **kwargs):
self.calls.append(kwargs)
if self._raise:
raise self._raise
return self._resp
if self.raise_exc:
raise self.raise_exc
return self.resp
def test_returns_extracted_guidance_and_assembles_prompt():
fc = _FakeClient(resp=_Resp([_Block("text", "Press Tab then Enter to focus the field.")]))
fc = FakeClient(resp=Resp([Block("text", "Press Tab then Enter to focus the field.")]))
out = asyncio.run(adjudicate_stuck(fc, "cheap-model", "share the doc", "- click -> not found", "the page"))
assert out == "Press Tab then Enter to focus the field."
call = fc.calls[0]
@@ -45,18 +45,18 @@ def test_returns_extracted_guidance_and_assembles_prompt():
def test_swallows_provider_error_and_returns_empty():
fc = _FakeClient(raise_exc=RuntimeError("429 rate limited"))
fc = FakeClient(raise_exc=RuntimeError("429 rate limited"))
out = asyncio.run(adjudicate_stuck(fc, "m", "g", "r", "p"))
assert out == ""
def test_extract_text_joins_text_blocks_and_ignores_others():
resp = _Resp([_Block("text", "First."), _Block("tool_use"), _Block("text", "Second.")])
assert _extract_text(resp) == "First. Second."
def testp_extract_text_joins_text_blocks_and_ignores_others():
resp = Resp([Block("text", "First."), Block("tool_use"), Block("text", "Second.")])
assert p_extract_text(resp) == "First. Second."
def test_handles_empty_inputs_without_crashing():
fc = _FakeClient(resp=_Resp([_Block("text", "ok")]))
fc = FakeClient(resp=Resp([Block("text", "ok")]))
out = asyncio.run(adjudicate_stuck(fc, "m", "", "", ""))
assert out == "ok"
# placeholders keep the prompt well-formed
backend/tests/test_browser_wait.py
+14 -14
View File
@@ -49,7 +49,7 @@ def test_decide_stop_handles_missing_signals():
# --- the async loop with a scripted probe -----------------------------------
def _probe(ready, quiet, elems=1000, found=False):
def probe(ready, quiet, elems=1000, found=False):
return {"text": json.dumps({"ready": ready, "quiet": quiet, "elems": elems, "found": found}),
"url": "https://x.com"}
@@ -68,13 +68,13 @@ class HangingExec:
async def __call__(self, tool, params, bid, tid):
self.calls.append((tool, params, bid, tid))
await asyncio.sleep(self.block_s)
return _probe(False, 0)
return probe(False, 0)
@pytest.mark.asyncio
async def test_returns_early_once_settled():
# first probe: still loading; second: settled -> should stop well under the cap
ex = AsyncMock(side_effect=[_probe(False, 0), _probe(True, 999)])
ex = AsyncMock(side_effect=[probe(False, 0), probe(True, 999)])
out = await bw.smart_wait(ex, "b", "", 5000, poll_ms=20, floor_ms=20, quiet_window_ms=50)
assert out["settled"] is True and out["found"] is False
assert out["waited_ms"] < 5000
@@ -86,7 +86,7 @@ async def test_returns_early_once_settled():
@pytest.mark.asyncio
async def test_rides_to_cap_when_page_never_settles():
# an SPA that keeps fetching (quiet always small) -> never settles -> caps out
ex = AsyncMock(return_value=_probe(True, 10))
ex = AsyncMock(return_value=probe(True, 10))
out = await bw.smart_wait(ex, "b", "", 200, poll_ms=20, floor_ms=20, quiet_window_ms=400)
assert out["settled"] is False
assert out["waited_ms"] >= 180 # ~the cap
@@ -97,7 +97,7 @@ async def test_rides_to_cap_when_page_never_settles():
async def test_settles_on_dom_stable_when_network_never_idles():
# the LinkedIn case: network always busy (quiet tiny) but the DOM count is
# constant -> DOM-settle fires instead of riding to the cap
ex = AsyncMock(return_value=_probe(True, 5, elems=500))
ex = AsyncMock(return_value=probe(True, 5, elems=500))
out = await bw.smart_wait(ex, "b", "", 3000, poll_ms=20, floor_ms=20, quiet_window_ms=200)
assert out["settled"] is True and out["waited_ms"] < 3000
assert "page settled" in out["text"]
@@ -107,8 +107,8 @@ async def test_settles_on_dom_stable_when_network_never_idles():
async def test_returns_the_instant_target_is_found():
# network busy AND DOM churning, but the agent's target appears on probe 2 ->
# stop immediately, bypassing even the floor
ex = AsyncMock(side_effect=[_probe(False, 5, elems=100, found=False),
_probe(False, 5, elems=200, found=True)])
ex = AsyncMock(side_effect=[probe(False, 5, elems=100, found=False),
probe(False, 5, elems=200, found=True)])
out = await bw.smart_wait(ex, "b", "", 5000, until="Send",
poll_ms=20, floor_ms=800, quiet_window_ms=999)
assert out["settled"] is True and out["found"] is True and "found target" in out["text"]
@@ -118,7 +118,7 @@ async def test_returns_the_instant_target_is_found():
@pytest.mark.asyncio
async def test_never_returns_before_the_floor():
# settled from the very first probe, but the floor must still be respected
ex = AsyncMock(return_value=_probe(True, 9999))
ex = AsyncMock(return_value=probe(True, 9999))
out = await bw.smart_wait(ex, "b", "", 5000, poll_ms=10, floor_ms=200, quiet_window_ms=50)
assert out["waited_ms"] >= 200, "must not read a page before the settle floor"
assert out["settled"] is True
@@ -136,14 +136,14 @@ async def test_probe_error_during_navigation_keeps_waiting_then_settles():
# while the page is navigating, evaluate errors; we must keep polling, not bail
ex = AsyncMock(side_effect=[{"error": "Cannot evaluate, page navigating"},
{"error": "still navigating"},
_probe(True, 999)])
probe(True, 999)])
out = await bw.smart_wait(ex, "b", "", 5000, poll_ms=15, floor_ms=15, quiet_window_ms=50)
assert out["settled"] is True and ex.await_count >= 3
@pytest.mark.asyncio
async def test_garbage_probe_text_does_not_crash():
ex = AsyncMock(side_effect=[{"text": "not json", "url": "u"}, _probe(True, 999)])
ex = AsyncMock(side_effect=[{"text": "not json", "url": "u"}, probe(True, 999)])
out = await bw.smart_wait(ex, "b", "", 3000, poll_ms=15, floor_ms=15, quiet_window_ms=50)
assert out["settled"] is True
@@ -167,15 +167,15 @@ async def test_hung_tab_returns_fast_not_after_the_full_command_timeout():
@pytest.mark.asyncio
async def test_a_single_slow_probe_then_settle_is_not_flagged_hung():
# one slow probe (under the threshold count) shouldn't trip 'hung'; it recovers
class _OneSlow:
class OneSlow:
def __init__(self):
self.calls = []
async def __call__(self, tool, params, bid, tid):
self.calls.append((tool, params, bid, tid))
if len(self.calls) == 1:
await asyncio.sleep(0.5) # one slow poll
return _probe(False, 0)
return _probe(True, 999)
out = await bw.smart_wait(_OneSlow(), "b", "", 5000, poll_ms=10, floor_ms=10,
return probe(False, 0)
return probe(True, 999)
out = await bw.smart_wait(OneSlow(), "b", "", 5000, poll_ms=10, floor_ms=10,
quiet_window_ms=50, probe_timeout_s=0.2)
assert out["hung"] is False and out["settled"] is True
backend/tests/test_disconnect_resilience.py
+51 -51
View File
@@ -43,16 +43,16 @@ from fastapi.testclient import TestClient
# the persistence dir for terminal events lives under our control.
# ---------------------------------------------------------------------------
_TMPROOT = tempfile.mkdtemp(prefix="openswarm-disconnect-test-")
os.environ.setdefault("OPENSWARM_DATA_DIR", _TMPROOT)
TMPROOT = tempfile.mkdtemp(prefix="openswarm-disconnect-test-")
os.environ.setdefault("OPENSWARM_DATA_DIR", TMPROOT)
# Push the SEQ_LOG persist dir to a deterministic location too.
_SEQ_DIR = os.path.join(_TMPROOT, "seq_terminals")
os.makedirs(_SEQ_DIR, exist_ok=True)
SEQ_DIR = os.path.join(TMPROOT, "seq_terminals")
os.makedirs(SEQ_DIR, exist_ok=True)
@pytest.fixture(autouse=True)
def _patch_persist_dir():
def patch_persist_dir():
"""Point the shared SEQ_LOG singleton at our tmp dir so we can assert on disk state.
ws_manager imports SEQ_LOG by value, so both modules hold the same object;
@@ -61,8 +61,8 @@ def _patch_persist_dir():
"""
from backend.apps.agents.core.seq_log import SEQ_LOG
os.makedirs(_SEQ_DIR, exist_ok=True)
with patch.object(SEQ_LOG, "_persist_dir", _SEQ_DIR):
os.makedirs(SEQ_DIR, exist_ok=True)
with patch.object(SEQ_LOG, "_persist_dir", SEQ_DIR):
# Fresh per-session state so each test is isolated.
SEQ_LOG._per_session.clear()
yield SEQ_LOG
@@ -77,7 +77,7 @@ def _patch_persist_dir():
# ---------------------------------------------------------------------------
def _build_app(seq_log):
def build_app(seq_log):
"""Replicates main.py's WS handler + adds a /test/emit endpoint
so the test thread can drive event emission through the same
event loop as the WS handler, avoiding the cross-loop hazards
@@ -121,13 +121,13 @@ def _build_app(seq_log):
n = int(body.get("n", 0))
terminate = body.get("terminate") # str or None
concurrent = int(body.get("concurrent", 1))
await _emit_run(session_id, n, terminate=terminate, concurrent_tasks=concurrent)
await emit_run(session_id, n, terminate=terminate, concurrent_tasks=concurrent)
return {"ok": True, "current_seq": seq_log.current_seq(session_id)}
return app
def _emit(client, session_id: str, n: int, terminate: str | None = None, concurrent: int = 1):
def emit(client, session_id: str, n: int, terminate: str | None = None, concurrent: int = 1):
"""Drive event emission via the test-only HTTP endpoint."""
r = client.post(f"/test/emit/{session_id}", json={
"n": n, "terminate": terminate, "concurrent": concurrent,
@@ -141,7 +141,7 @@ def _emit(client, session_id: str, n: int, terminate: str | None = None, concurr
# ---------------------------------------------------------------------------
async def _emit_run(session_id: str, n_events: int, terminate: str | None = "completed", concurrent_tasks: int = 1):
async def emit_run(session_id: str, n_events: int, terminate: str | None = "completed", concurrent_tasks: int = 1):
"""Emit a synthetic agent run.
`concurrent_tasks` lets the test stress the per-session lock by
@@ -187,37 +187,37 @@ async def _emit_run(session_id: str, n_events: int, terminate: str | None = "com
# ---------------------------------------------------------------------------
def test_seq_monotonic_under_concurrency(_patch_persist_dir):
def test_seq_monotonic_under_concurrency(patch_persist_dir):
"""200 concurrent broadcasts must yield strictly monotonic seq."""
app = _build_app(_patch_persist_dir)
app = build_app(patch_persist_dir)
sid = "session-conc-1"
with TestClient(app) as client:
_emit(client, sid, n=200, terminate=None, concurrent=4)
_, newest, events = _patch_persist_dir.replay(sid, 0)
emit(client, sid, n=200, terminate=None, concurrent=4)
_, newest, events = patch_persist_dir.replay(sid, 0)
assert newest == 200
seqs = [json.loads(s)["seq"] for s in events]
assert seqs == sorted(seqs)
assert len(set(seqs)) == len(seqs)
def test_terminal_event_persisted(_patch_persist_dir):
app = _build_app(_patch_persist_dir)
def test_terminal_event_persisted(patch_persist_dir):
app = build_app(patch_persist_dir)
sid = "session-term-1"
with TestClient(app) as client:
_emit(client, sid, n=0, terminate="completed")
raw = _patch_persist_dir.load_terminal(sid)
emit(client, sid, n=0, terminate="completed")
raw = patch_persist_dir.load_terminal(sid)
assert raw is not None
obj = json.loads(raw)
assert obj["event"] == "agent:status"
assert obj["data"]["status"] == "completed"
def test_replay_after_eviction_reports_gap(_patch_persist_dir):
app = _build_app(_patch_persist_dir)
def test_replay_after_eviction_reports_gap(patch_persist_dir):
app = build_app(patch_persist_dir)
sid = "session-evict-1"
with TestClient(app) as client:
_emit(client, sid, n=700, terminate=None)
oldest, newest, events = _patch_persist_dir.replay(sid, last_seq=10)
emit(client, sid, n=700, terminate=None)
oldest, newest, events = patch_persist_dir.replay(sid, last_seq=10)
assert newest == 700
assert oldest is not None and oldest > 10
# Replay only includes seqs > 10 that survived eviction.
@@ -229,9 +229,9 @@ def test_replay_after_eviction_reports_gap(_patch_persist_dir):
# ---------------------------------------------------------------------------
def test_resume_after_disconnect_recovers_all_events(_patch_persist_dir):
def test_resume_after_disconnect_recovers_all_events(patch_persist_dir):
"""Simulate a single disconnect mid-run, then a clean resume."""
app = _build_app(_patch_persist_dir)
app = build_app(patch_persist_dir)
sid = "session-res-1"
received: list[dict] = []
@@ -243,7 +243,7 @@ def test_resume_after_disconnect_recovers_all_events(_patch_persist_dir):
hello = json.loads(ws.receive_text())
assert hello["event"] == "server:hello"
# Inject a few events via the /test/emit endpoint.
_emit(client, sid, n=10, terminate=None)
emit(client, sid, n=10, terminate=None)
for _ in range(10):
received.append(json.loads(ws.receive_text()))
assert len(received) == 10
@@ -251,7 +251,7 @@ def test_resume_after_disconnect_recovers_all_events(_patch_persist_dir):
# Phase 2: between connections, the server keeps emitting. The
# agent task is alive; only the WS is gone.
_emit(client, sid, n=10, terminate="completed")
emit(client, sid, n=10, terminate="completed")
# Phase 3: reconnect with last_seq=10, expect replay of seq 11..21
# (10 deltas + 1 status), then the server:hello ack.
@@ -270,17 +270,17 @@ def test_resume_after_disconnect_recovers_all_events(_patch_persist_dir):
assert statuses[0]["data"]["status"] == "completed"
def test_terminal_event_visible_after_full_eviction(_patch_persist_dir):
def test_terminal_event_visible_after_full_eviction(patch_persist_dir):
"""If the in-memory log is wiped (process restart simulation),
a reconnecting client should still see the terminal event from
disk, never a phantom 'running' spinner."""
app = _build_app(_patch_persist_dir)
app = build_app(patch_persist_dir)
sid = "session-evict-term-1"
seq_log = _patch_persist_dir
seq_log = patch_persist_dir
with TestClient(app) as client:
_emit(client, sid, n=5, terminate="completed")
emit(client, sid, n=5, terminate="completed")
# Simulate a process restart: clear the in-memory ring buffer
# but keep the persisted terminal file.
@@ -300,16 +300,16 @@ def test_terminal_event_visible_after_full_eviction(_patch_persist_dir):
assert terminals[0]["data"]["status"] == "completed"
def test_gap_detected_when_buffer_evicted(_patch_persist_dir):
def test_gap_detected_when_buffer_evicted(patch_persist_dir):
"""A client whose lastSeq is older than the oldest buffered seq
should receive `agent:gap_detected` so it can REST-refresh,
rather than silently miss events."""
app = _build_app(_patch_persist_dir)
app = build_app(patch_persist_dir)
sid = "session-gap-1"
with TestClient(app) as client:
# Fill the buffer past its limit so seq 1..200 are evicted.
_emit(client, sid, n=700, terminate=None)
emit(client, sid, n=700, terminate=None)
with client.websocket_connect(f"/ws/agents/{sid}") as ws:
ws.send_text(json.dumps({"event": "client:hello", "data": {"last_seq": 5, "connection_uuid": "c1"}}))
saw_gap = False
@@ -325,8 +325,8 @@ def test_gap_detected_when_buffer_evicted(_patch_persist_dir):
assert saw_gap
def test_ping_pong_round_trip(_patch_persist_dir):
app = _build_app(_patch_persist_dir)
def test_ping_pong_round_trip(patch_persist_dir):
app = build_app(patch_persist_dir)
sid = "session-ping-1"
with TestClient(app) as client:
@@ -348,14 +348,14 @@ N_STRESS_ITERATIONS = int(os.environ.get("DISCONNECT_STRESS_N", "500"))
@pytest.mark.parametrize("iteration", range(N_STRESS_ITERATIONS))
def test_stress_random_disconnect(iteration, _patch_persist_dir):
def test_stress_random_disconnect(iteration, patch_persist_dir):
"""Each iteration: a random number of events, a random number of
disconnects at random points, optionally ending in a terminal
status. After all reconnects, the client must have observed
every event exactly once, in seq order, and the terminal event
if one was emitted."""
rng = random.Random(iteration) # deterministic per iteration
app = _build_app(_patch_persist_dir)
app = build_app(patch_persist_dir)
sid = f"session-stress-{iteration}"
total_events = rng.randint(5, 80)
@@ -389,7 +389,7 @@ def test_stress_random_disconnect(iteration, _patch_persist_dir):
seen[msg["seq"]] = msg
last_seq = max(last_seq, msg["seq"])
to_emit = bp - emitted_so_far
toemit = bp - emitted_so_far
emitted_so_far = bp
terminate = "completed" if (bp == total_events and will_terminate) else None
@@ -399,9 +399,9 @@ def test_stress_random_disconnect(iteration, _patch_persist_dir):
# isolated loop and re-bind the per-session asyncio.Lock
# to a different loop, which is hostile to anyio's
# blocking-portal pattern.
_emit(client, sid, n=to_emit, terminate=terminate)
emit(client, sid, n=toemit, terminate=terminate)
expected = to_emit + (1 if terminate else 0)
expected = toemit + (1 if terminate else 0)
for _ in range(expected):
msg = json.loads(ws.receive_text())
seen[msg["seq"]] = msg
@@ -427,17 +427,17 @@ def test_stress_random_disconnect(iteration, _patch_persist_dir):
@pytest.mark.parametrize("trial", range(30))
def test_concurrent_broadcast_preserves_order(trial, _patch_persist_dir):
def test_concurrent_broadcast_preserves_order(trial, patch_persist_dir):
"""8 coroutines fanning out 400 events under the per-session lock.
Drives the emit through the TestClient's portal so we use the
real event loop the rest of the WS layer runs on."""
app = _build_app(_patch_persist_dir)
app = build_app(patch_persist_dir)
sid = f"session-conc-{trial}"
with TestClient(app) as client:
_emit(client, sid, n=400, terminate="completed", concurrent=8)
emit(client, sid, n=400, terminate="completed", concurrent=8)
oldest, newest, events = _patch_persist_dir.replay(sid, last_seq=0)
oldest, newest, events = patch_persist_dir.replay(sid, last_seq=0)
assert newest == 401 # 400 deltas + 1 status
seqs = [json.loads(s)["seq"] for s in events]
assert seqs == sorted(seqs)
@@ -461,9 +461,9 @@ def test_concurrent_broadcast_preserves_order(trial, _patch_persist_dir):
@pytest.mark.parametrize("trial", range(50))
def test_terminate_during_disconnect_is_observable(trial, _patch_persist_dir):
def test_terminate_during_disconnect_is_observable(trial, patch_persist_dir):
rng = random.Random(1000 + trial)
app = _build_app(_patch_persist_dir)
app = build_app(patch_persist_dir)
sid = f"session-mid-term-{trial}"
n_pre = rng.randint(0, 40)
n_post = rng.randint(0, 40)
@@ -475,13 +475,13 @@ def test_terminate_during_disconnect_is_observable(trial, _patch_persist_dir):
ws.send_text(json.dumps({"event": "client:hello", "data": {"last_seq": 0, "connection_uuid": "c1"}}))
assert json.loads(ws.receive_text())["event"] == "server:hello"
if n_pre:
_emit(client, sid, n=n_pre, terminate=None)
emit(client, sid, n=n_pre, terminate=None)
for _ in range(n_pre):
msg = json.loads(ws.receive_text())
seen[msg["seq"]] = msg
last_seq = max(last_seq, msg["seq"])
# Disconnected. Emit the rest + terminate while WS is gone.
_emit(client, sid, n=n_post, terminate="completed")
emit(client, sid, n=n_post, terminate="completed")
# Reconnect. We expect to receive everything from last_seq+1
# through to the terminal, possibly via disk if the buffer
# rolled (it won't here; numbers are small).
@@ -511,7 +511,7 @@ def test_terminate_during_disconnect_is_observable(trial, _patch_persist_dir):
# ---------------------------------------------------------------------------
def test_disconnect_does_not_touch_agent_task(_patch_persist_dir):
def test_disconnect_does_not_touch_agent_task(patch_persist_dir):
"""If a future refactor adds task cancellation to disconnect_session,
this test will catch it. We import agent_manager lazily so the
`tasks` dict starts empty; we register a sentinel task and confirm
@@ -528,7 +528,7 @@ def test_disconnect_does_not_touch_agent_task(_patch_persist_dir):
assert "tasks" not in src
def test_main_ws_endpoints_still_gated_by_auth(_patch_persist_dir):
def test_main_ws_endpoints_still_gated_by_auth(patch_persist_dir):
src = open(os.path.join(os.path.dirname(__file__), "..", "main.py")).read()
assert "_ws_auth_ok(websocket)" in src, (
"main.py WS endpoints must still call _ws_auth_ok before accepting "
backend/tests/test_disk_caches.py
+4 -4
View File
@@ -29,7 +29,7 @@ def tools_tmp(tmp_path, monkeypatch):
return d
def _bump_mtime(path):
def bump_mtime(path):
# FAT32-style coarse clocks could hide a same-size rewrite; force a distinct mtime.
st = os.stat(path)
os.utime(path, ns=(st.st_atime_ns, st.st_mtime_ns + 1_000_000))
@@ -53,7 +53,7 @@ def test_settings_external_edit_detected(settings_tmp):
raw = json.loads(settings_tmp.read_text())
raw["theme"] = "light"
settings_tmp.write_text(json.dumps(raw))
_bump_mtime(settings_tmp)
bump_mtime(settings_tmp)
assert store.load_settings().theme == "light"
@@ -75,7 +75,7 @@ def test_tools_write_then_list_is_fresh(tools_tmp):
assert tools_lib._load_all() == []
t = ToolDefinition(name="Alpha", description="a")
tools_lib._save(t)
_bump_mtime(tools_tmp / f"{t.id}.json")
bump_mtime(tools_tmp / f"{t.id}.json")
names = [x.name for x in tools_lib._load_all()]
assert names == ["Alpha"]
@@ -98,7 +98,7 @@ def test_tools_in_place_rewrite_detected(tools_tmp):
assert [x.name for x in tools_lib._load_all()] == ["Old"]
t.name = "New"
tools_lib._save(t)
_bump_mtime(tools_tmp / f"{t.id}.json")
bump_mtime(tools_tmp / f"{t.id}.json")
assert [x.name for x in tools_lib._load_all()] == ["New"]
backend/tests/test_disk_resilience.py
+3 -3
View File
@@ -9,7 +9,7 @@ import pytest
from backend.config.json_store import read_json_or_none, atomic_write_json
class _NotSerializable:
class NotSerializable:
pass
@@ -36,7 +36,7 @@ def test_atomic_write_fully_replaces(tmp_path):
def test_atomic_write_cleans_temp_and_raises_on_bad_payload(tmp_path):
p = str(tmp_path / "x.json")
with pytest.raises(TypeError):
atomic_write_json(p, {"bad": _NotSerializable()})
atomic_write_json(p, {"bad": NotSerializable()})
assert [f for f in os.listdir(tmp_path) if f.startswith(".tmp-")] == []
assert not os.path.exists(p)
@@ -45,7 +45,7 @@ def test_atomic_write_preserves_existing_when_new_write_fails(tmp_path):
p = str(tmp_path / "x.json")
atomic_write_json(p, {"good": 1})
with pytest.raises(TypeError):
atomic_write_json(p, {"bad": _NotSerializable()})
atomic_write_json(p, {"bad": NotSerializable()})
assert read_json_or_none(p) == {"good": 1}
backend/tests/test_free_trial.py
+12 -12
View File
@@ -5,11 +5,11 @@ import backend # noqa: F401 (path sanity asserted below)
from backend.apps.settings.models import AppSettings
from backend.apps.settings.credentials import proxy_auth
from backend.apps.agents.core.error_classify import (
_is_free_trial_exhausted,
_is_transient_capacity_error,
is_free_trial_exhausted,
is_transient_capacity_error,
)
from backend.apps.agents.providers.registry import resolve_model_id_for_sdk
from backend.apps.subscription.free_trial import _has_own_model
from backend.apps.subscription.free_trial import has_own_model
def test_proxy_auth_for_each_mode():
@@ -41,17 +41,17 @@ def test_free_trial_resolves_to_a_bare_anthropic_id():
def test_exhaustion_is_classified_and_not_retried():
assert _is_free_trial_exhausted(Exception("error type free_trial_exhausted"))
assert _is_free_trial_exhausted(Exception("You've used your free OpenSwarm runs"))
assert not _is_free_trial_exhausted(Exception("overloaded, try again"))
assert is_free_trial_exhausted(Exception("error type free_trial_exhausted"))
assert is_free_trial_exhausted(Exception("You've used your free OpenSwarm runs"))
assert not is_free_trial_exhausted(Exception("overloaded, try again"))
# Must NOT look transient, or the agent loop would retry a spent trial forever.
assert not _is_transient_capacity_error(Exception("free_trial_exhausted"))
assert not is_transient_capacity_error(Exception("free_trial_exhausted"))
def test_has_own_model_never_shadows_a_real_provider():
assert not _has_own_model(AppSettings(connection_mode="free-trial", free_trial_token="x"))
assert not _has_own_model(AppSettings())
assert _has_own_model(AppSettings(anthropic_api_key="sk-ant-x"))
assert _has_own_model(
def testhas_own_model_never_shadows_a_real_provider():
assert not has_own_model(AppSettings(connection_mode="free-trial", free_trial_token="x"))
assert not has_own_model(AppSettings())
assert has_own_model(AppSettings(anthropic_api_key="sk-ant-x"))
assert has_own_model(
AppSettings(connection_mode="openswarm-pro", openswarm_bearer_token="b")
)
backend/tests/test_ghost_detector.py
+17 -17
View File
@@ -5,67 +5,67 @@ This is the guard against features that fail silently without erroring."""
import importlib.util
import os
_SCRIPT = os.path.join(os.path.dirname(__file__), "..", "..", "scripts", "analyze-browser-metrics.py")
SCRIPT = os.path.join(os.path.dirname(__file__), "..", "..", "scripts", "analyze-browser-metrics.py")
def _load():
spec = importlib.util.spec_from_file_location("abm", _SCRIPT)
def load():
spec = importlib.util.spec_from_file_location("abm", SCRIPT)
m = importlib.util.module_from_spec(spec)
spec.loader.exec_module(m)
return m
def _ev(tool, ok=True, loop=False, result_len=50):
def ev(tool, ok=True, loop=False, result_len=50):
return {"tool": tool, "ok": ok, "is_loop": loop, "result_len": result_len}
def test_zero_tools_is_ghost():
gv = _load().ghost_verdict
gv = load().ghost_verdict
is_ghost, reasons = gv({"status": "completed"}, [])
assert is_ghost and any("ZERO" in r for r in reasons)
def test_read_with_content_is_not_ghost():
# a pure read task that actually returned data is legitimate work
gv = _load().ghost_verdict
is_ghost, _ = gv({"status": "completed"}, [_ev("BrowserGetText", result_len=120)])
gv = load().ghost_verdict
is_ghost, _ = gv({"status": "completed"}, [ev("BrowserGetText", result_len=120)])
assert not is_ghost
def test_empty_reads_only_is_ghost():
gv = _load().ghost_verdict
gv = load().ghost_verdict
is_ghost, reasons = gv({"status": "completed"}, [
_ev("BrowserGetText", result_len=0), _ev("BrowserScreenshot", result_len=0)])
ev("BrowserGetText", result_len=0), ev("BrowserScreenshot", result_len=0)])
assert is_ghost
def test_all_productive_actions_errored_is_ghost():
gv = _load().ghost_verdict
gv = load().ghost_verdict
is_ghost, _ = gv({"status": "completed"}, [
_ev("BrowserClickIndex", ok=False), _ev("BrowserClickIndex", ok=False)])
ev("BrowserClickIndex", ok=False), ev("BrowserClickIndex", ok=False)])
assert is_ghost
def test_honest_action_is_not_ghost():
gv = _load().ghost_verdict
is_ghost, _ = gv({"status": "completed"}, [_ev("BrowserNavigate"), _ev("BrowserClickIndex")])
gv = load().ghost_verdict
is_ghost, _ = gv({"status": "completed"}, [ev("BrowserNavigate"), ev("BrowserClickIndex")])
assert not is_ghost
def test_loop_during_completed_is_ghost():
gv = _load().ghost_verdict
is_ghost, reasons = gv({"status": "completed"}, [_ev("BrowserNavigate"), _ev("BrowserClickIndex", loop=True)])
gv = load().ghost_verdict
is_ghost, reasons = gv({"status": "completed"}, [ev("BrowserNavigate"), ev("BrowserClickIndex", loop=True)])
assert is_ghost and any("loop" in r.lower() for r in reasons)
def test_errored_task_is_never_ghost():
# an honest failure (status=error) is not a ghost; ghosts are fake successes
gv = _load().ghost_verdict
gv = load().ghost_verdict
assert not gv({"status": "error"}, [])[0]
def test_tier_mapping_present():
# the analyzer's _PRODUCTIVE set must include the real mutation tools
m = _load()
m = load()
for t in ("BrowserClick", "BrowserClickIndex", "BrowserType", "BrowserNavigate", "BrowserReplayRoute"):
assert t in m._PRODUCTIVE
backend/tests/test_outputs_runtime_cleanup.py
+42 -42
View File
@@ -4,8 +4,8 @@ What this proves:
1. AppRuntimeManager.stop_all() reaps active runtimes.
2. AppRuntimeManager.stop_all() reaps idle (LRU) runtimes too.
3. AppRuntimeManager.stop_all() resumes SIGSTOP'd idle runtimes before reaping (otherwise the SIGTERM is queued and the process never dies).
4. _is_port_free() correctly detects collisions.
5. _write_env_value() updates a single key without clobbering siblings.
4. is_port_free() correctly detects collisions.
5. write_env_value() updates a single key without clobbering siblings.
6. _start_new_mode() rewrites .env's FRONTEND_PORT when the persisted port is in use, and the spawned child sees the rewritten value.
7. Same collision-rewrite happens for BACKEND_PORT when it's not "NONE".
@@ -24,10 +24,10 @@ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..",
from backend.apps.outputs.runtime import (
AppRuntimeManager,
_find_free_port,
_is_port_free,
_read_env_value,
_write_env_value,
find_free_port,
is_port_free,
read_env_value,
write_env_value,
)
@@ -60,7 +60,7 @@ wait $PYTHON_PID
"""
def _make_fake_workspace(tmp: str, frontend_port: int, backend_port: str = "NONE") -> str:
def make_fake_workspace(tmp: str, frontend_port: int, backend_port: str = "NONE") -> str:
ws = os.path.join(tmp, "ws")
os.makedirs(ws)
with open(os.path.join(ws, "run.sh"), "w") as f:
@@ -71,7 +71,7 @@ def _make_fake_workspace(tmp: str, frontend_port: int, backend_port: str = "NONE
return ws
def _pid_alive(pid: int) -> bool:
def pid_alive(pid: int) -> bool:
try:
os.kill(pid, 0)
return True
@@ -80,52 +80,52 @@ def _pid_alive(pid: int) -> bool:
# --- Test 1: helpers ---
def test_is_port_free():
p = _find_free_port()
assert _is_port_free(p), "freshly-allocated port should be free"
def testis_port_free():
p = find_free_port()
assert is_port_free(p), "freshly-allocated port should be free"
s = socket.socket()
s.bind(("127.0.0.1", p))
s.listen(1)
try:
assert not _is_port_free(p), "_is_port_free must return False while bound"
assert not is_port_free(p), "is_port_free must return False while bound"
finally:
s.close()
print("PASS test_is_port_free")
print("PASS testis_port_free")
def test_write_env_value():
def testwrite_env_value():
with tempfile.TemporaryDirectory() as tmp:
env = os.path.join(tmp, ".env")
with open(env, "w") as f:
f.write("A=1\nB=2\nC=3\n# comment\n")
_write_env_value(env, "B", "999")
assert _read_env_value(env, "A") == "1"
assert _read_env_value(env, "B") == "999"
assert _read_env_value(env, "C") == "3"
write_env_value(env, "B", "999")
assert read_env_value(env, "A") == "1"
assert read_env_value(env, "B") == "999"
assert read_env_value(env, "C") == "3"
# New key appended.
_write_env_value(env, "D", "new")
assert _read_env_value(env, "D") == "new"
write_env_value(env, "D", "new")
assert read_env_value(env, "D") == "new"
# Comment line + sibling values preserved.
with open(env) as f:
body = f.read()
assert "# comment" in body, "comment line dropped"
assert "A=1" in body and "C=3" in body
print("PASS test_write_env_value")
print("PASS testwrite_env_value")
# --- Test 2: stop_all reaps an active runtime (real spawn). ---
@pytest.mark.asyncio
async def test_stop_all_kills_active():
with tempfile.TemporaryDirectory() as tmp:
port = _find_free_port()
ws = _make_fake_workspace(tmp, port)
port = find_free_port()
ws = make_fake_workspace(tmp, port)
m = AppRuntimeManager()
rt = await m.attach("ws1", ws)
assert rt.running, "runtime should be running after attach"
pid = rt.process.pid
# Wait for the child python to actually bind the port.
for _ in range(40):
if not _is_port_free(port):
if not is_port_free(port):
break
await asyncio.sleep(0.05)
else:
@@ -134,14 +134,14 @@ async def test_stop_all_kills_active():
assert killed >= 1, f"stop_all reported {killed} reaped"
# Bash + python child must be gone within the grace window.
for _ in range(60):
if not _pid_alive(pid):
if not pid_alive(pid):
break
await asyncio.sleep(0.1)
else:
raise AssertionError(f"pid {pid} still alive after stop_all")
# Port must be released too.
for _ in range(40):
if _is_port_free(port):
if is_port_free(port):
break
await asyncio.sleep(0.05)
else:
@@ -154,8 +154,8 @@ async def test_stop_all_kills_active():
@pytest.mark.asyncio
async def test_stop_all_kills_idle():
with tempfile.TemporaryDirectory() as tmp:
port = _find_free_port()
ws = _make_fake_workspace(tmp, port)
port = find_free_port()
ws = make_fake_workspace(tmp, port)
m = AppRuntimeManager()
rt = await m.attach("ws-idle", ws)
pid = rt.process.pid
@@ -168,13 +168,13 @@ async def test_stop_all_kills_idle():
killed = await m.stop_all()
assert killed == 1
for _ in range(60):
if not _pid_alive(pid):
if not pid_alive(pid):
break
await asyncio.sleep(0.1)
else:
raise AssertionError("idle process never died, stop_all probably didn't SIGCONT first")
for _ in range(40):
if _is_port_free(port):
if is_port_free(port):
break
await asyncio.sleep(0.05)
else:
@@ -186,8 +186,8 @@ async def test_stop_all_kills_idle():
@pytest.mark.asyncio
async def test_port_collision_reallocates_env():
with tempfile.TemporaryDirectory() as tmp:
squatted_port = _find_free_port()
ws = _make_fake_workspace(tmp, squatted_port)
squatted_port = find_free_port()
ws = make_fake_workspace(tmp, squatted_port)
# Squat the persisted port so the runtime can't use it.
squatter = socket.socket()
squatter.bind(("127.0.0.1", squatted_port))
@@ -204,12 +204,12 @@ async def test_port_collision_reallocates_env():
f"frontend_port should have changed from {squatted_port}, got {rt.frontend_port}"
# .env should reflect the new port (so run.sh and subsequent
# restarts pick it up too).
written = _read_env_value(os.path.join(ws, ".env"), "FRONTEND_PORT")
written = read_env_value(os.path.join(ws, ".env"), "FRONTEND_PORT")
assert written == str(rt.frontend_port), \
f".env not rewritten; expected {rt.frontend_port}, found {written}"
# Sibling .env keys untouched.
assert _read_env_value(os.path.join(ws, ".env"), "SOMETHING_ELSE") == "untouched"
assert _read_env_value(os.path.join(ws, ".env"), "TRAILING") == "keep"
assert read_env_value(os.path.join(ws, ".env"), "SOMETHING_ELSE") == "untouched"
assert read_env_value(os.path.join(ws, ".env"), "TRAILING") == "keep"
await m.stop_all()
finally:
squatter.close()
@@ -235,15 +235,15 @@ async def test_descendant_tree_killed_despite_exit_only_trap():
silently and reparents the vite/uvicorn grandchild to PID 1. stop()
must walk the descendant tree to nuke the grandchild explicitly."""
with tempfile.TemporaryDirectory() as tmp:
port = _find_free_port()
ws = _make_fake_workspace(tmp, port)
port = find_free_port()
ws = make_fake_workspace(tmp, port)
m = AppRuntimeManager()
rt = await m.attach("ws-tree", ws)
bash_pid = rt.process.pid
# Wait until the python grandchild is actually listening on the port,
# so we know it exists as a separate process.
for _ in range(60):
if not _is_port_free(port):
if not is_port_free(port):
break
await asyncio.sleep(0.05)
else:
@@ -274,7 +274,7 @@ async def test_descendant_tree_killed_despite_exit_only_trap():
await m.stop_all()
# Every descendant must be gone, not just bash.
for _ in range(80):
still_alive = [p for p in all_descendants if _pid_alive(p)]
still_alive = [p for p in all_descendants if pid_alive(p)]
if not still_alive:
break
await asyncio.sleep(0.1)
@@ -284,7 +284,7 @@ async def test_descendant_tree_killed_despite_exit_only_trap():
"(EXIT-only trap let them escape)"
)
for _ in range(40):
if _is_port_free(port):
if is_port_free(port):
break
await asyncio.sleep(0.05)
else:
@@ -293,8 +293,8 @@ async def test_descendant_tree_killed_despite_exit_only_trap():
async def main():
test_is_port_free()
test_write_env_value()
testis_port_free()
testwrite_env_value()
await test_stop_all_idempotent()
await test_stop_all_kills_active()
await test_stop_all_kills_idle()
backend/tests/test_phase1_stress.py
+5 -5
View File
@@ -22,8 +22,8 @@ import pytest
# backend modules.
# ---------------------------------------------------------------------------
_TMPROOT = tempfile.mkdtemp(prefix="openswarm-phase1-stress-")
os.environ.setdefault("OPENSWARM_DATA_DIR", _TMPROOT)
TMPROOT = tempfile.mkdtemp(prefix="openswarm-phase1-stress-")
os.environ.setdefault("OPENSWARM_DATA_DIR", TMPROOT)
# ---------------------------------------------------------------------------
@@ -115,7 +115,7 @@ def test_modes_lifespan_deletes_stale_chat():
"system_prompt": "old", "tools": ["AskUserQuestion"],
}, f)
with patch.object(modes_mod, "DATA_DIR", td):
asyncio.run(_run_lifespan(modes_mod))
asyncio.run(run_lifespan(modes_mod))
assert not os.path.exists(chat_path), "stale built-in chat.json should be removed"
# User-customized: leave alone
@@ -127,11 +127,11 @@ def test_modes_lifespan_deletes_stale_chat():
"system_prompt": "user wrote this",
}, f)
with patch.object(modes_mod, "DATA_DIR", td):
asyncio.run(_run_lifespan(modes_mod))
asyncio.run(run_lifespan(modes_mod))
assert os.path.exists(chat_path), "user-customized chat.json must NOT be deleted"
async def _run_lifespan(modes_mod):
async def run_lifespan(modes_mod):
async with modes_mod.modes_lifespan():
pass
backend/tests/test_service.py
+8 -8
View File
@@ -23,8 +23,8 @@ from unittest.mock import patch
import pytest
_tmpdir = tempfile.mkdtemp()
os.environ.setdefault("OPENSWARM_DATA_DIR", _tmpdir)
tmpdir = tempfile.mkdtemp()
os.environ.setdefault("OPENSWARM_DATA_DIR", tmpdir)
@pytest.fixture(autouse=True)
@@ -44,11 +44,11 @@ def patch_settings(tmp_path):
@pytest.fixture(autouse=True)
def fresh_client(tmp_path):
import backend.apps.service.client as client
client._install_id = None
client._user_id = None
client._test_sink = None
client.P_INSTALL_ID = None
client.P_USER_ID = None
client.P_TEST_SINK = None
spool = tmp_path / "spool.db"
with patch.object(client, "_spool_path", lambda: str(spool)):
with patch.object(client, "spool_path", lambda: str(spool)):
yield
@@ -291,7 +291,7 @@ def test_install_id_persisted(sink, tmp_path):
import backend.apps.settings.store as settings_mod
settings_mod.SETTINGS_FILE = str(sf)
import backend.apps.service.client as client
client._install_id = None
client.P_INSTALL_ID = None
from backend.apps.service.client import sync
sync({})
_, body = sink[0]
@@ -349,7 +349,7 @@ async def test_endpoint_spool_count(tmp_path):
from backend.apps.service import client as svc, buffer
from backend.apps.service.service import spool_count
spool = str(tmp_path / "spool.db")
with patch.object(svc, "_spool_path", lambda: spool):
with patch.object(svc, "spool_path", lambda: spool):
buffer.enqueue(spool, "s:/x", {}, now=time.time())
result = await spool_count()
assert result == {"pending": 1}
backend/tests/test_service_legacy.py
+16 -16
View File
@@ -17,18 +17,18 @@ import tempfile
import pytest
# Sandbox the data dir before any module import touches settings on disk.
_tmpdir = tempfile.mkdtemp()
os.environ.setdefault("OPENSWARM_DATA_DIR", _tmpdir)
tmpdir = tempfile.mkdtemp()
os.environ.setdefault("OPENSWARM_DATA_DIR", tmpdir)
# Captured syncs from this test run.
_captured_syncs: list[dict] = []
captured_syncs: list[dict] = []
@pytest.fixture(autouse=True)
def reset_captured_syncs():
_captured_syncs.clear()
def resetcaptured_syncs():
captured_syncs.clear()
yield
_captured_syncs.clear()
captured_syncs.clear()
@pytest.fixture(autouse=True)
@@ -39,7 +39,7 @@ def install_sync_sink():
bag so existing tests can keep their assertions terse."""
import backend.apps.service.client as svc_client
def _sink(label: str, body: dict):
def sink_fn(label: str, body: dict):
cs = body.get("client_state") or {}
payload = body.get("d") or body.get("payload") or {}
@@ -74,20 +74,20 @@ def install_sync_sink():
props.setdefault("os", cs.get("os", ""))
props.setdefault("platform", cs.get("os", ""))
_captured_syncs.append({
captured_syncs.append({
"kind": kind,
"label": label,
"distinct_id": cs.get("install_id", ""),
"properties": props,
})
old_sink = svc_client._test_sink
old_iid = svc_client._install_id
svc_client.set_test_sink(_sink)
svc_client._install_id = "test-install-id"
old_sink = svc_client.P_TEST_SINK # p-private-ignore: P_TEST_SINK
old_iid = svc_client.P_INSTALL_ID # p-private-ignore: P_INSTALL_ID
svc_client.set_test_sink(sink_fn)
svc_client.P_INSTALL_ID = "test-install-id"
yield
svc_client.set_test_sink(old_sink)
svc_client._install_id = old_iid
svc_client.P_INSTALL_ID = old_iid
@pytest.fixture(autouse=True)
@@ -122,14 +122,14 @@ def mock_sessions_dir(tmp_path):
def syncs(kind: str | None = None) -> list[dict]:
"""Return captured syncs, optionally filtered by inferred kind."""
if kind:
return [s for s in _captured_syncs if s["kind"] == kind]
return list(_captured_syncs)
return [s for s in captured_syncs if s["kind"] == kind]
return list(captured_syncs)
def last_sync(kind: str) -> dict:
"""Return the last captured sync of a given inferred kind."""
matching = syncs(kind)
assert matching, f"No {kind} syncs captured. Got: {[s['kind'] for s in _captured_syncs]}"
assert matching, f"No {kind} syncs captured. Got: {[s['kind'] for s in captured_syncs]}"
return matching[-1]
backend/tests/test_settings_migration.py
+10 -10
View File
@@ -22,7 +22,7 @@ def settings_file(tmp_path, monkeypatch):
return f
def _write(path, obj):
def write(path, obj):
with open(path, "w", encoding="utf-8") as fh:
json.dump(obj, fh)
@@ -61,7 +61,7 @@ def test_no_file_returns_defaults():
def test_minimal_old_file_fills_missing_with_defaults(settings_file):
# An old build wrote only a couple of fields; everything else must default.
_write(settings_file, {"theme": "light"})
write(settings_file, {"theme": "light"})
s = store.load_settings()
assert s.theme == "light"
assert s.default_model == "sonnet" # filled from default
@@ -69,7 +69,7 @@ def test_minimal_old_file_fills_missing_with_defaults(settings_file):
def test_legacy_fields_migrated_end_to_end(settings_file):
_write(settings_file, {"connection_mode": "managed", "openswarm_auth_token": "tok"})
write(settings_file, {"connection_mode": "managed", "openswarm_auth_token": "tok"})
s = store.load_settings()
assert s.connection_mode == "openswarm-pro"
assert s.openswarm_bearer_token == "tok"
@@ -77,14 +77,14 @@ def test_legacy_fields_migrated_end_to_end(settings_file):
def test_install_id_and_first_opened_continuity(settings_file):
# The identity carried across upgrades must survive a load untouched.
_write(settings_file, {"installation_id": "abc-123", "first_opened_at": "2025-01-01T00:00:00Z"})
write(settings_file, {"installation_id": "abc-123", "first_opened_at": "2025-01-01T00:00:00Z"})
s = store.load_settings()
assert s.installation_id == "abc-123"
assert s.first_opened_at == "2025-01-01T00:00:00Z"
def test_null_system_prompt_backfilled(settings_file):
_write(settings_file, {"default_system_prompt": None})
write(settings_file, {"default_system_prompt": None})
assert store.load_settings().default_system_prompt == DEFAULT_SYSTEM_PROMPT
@@ -92,7 +92,7 @@ def test_null_system_prompt_backfilled(settings_file):
def test_unknown_removed_fields_are_ignored(settings_file):
# A field that existed in a future/older schema but not this one must not crash.
_write(settings_file, {"theme": "light", "a_field_we_removed": 999, "another_ghost": {"x": 1}})
write(settings_file, {"theme": "light", "a_field_we_removed": 999, "another_ghost": {"x": 1}})
s = store.load_settings()
assert s.theme == "light"
@@ -100,7 +100,7 @@ def test_unknown_removed_fields_are_ignored(settings_file):
def test_type_drifted_field_reverts_to_default_keeps_rest(settings_file):
# dismissed_mcp_suggestions is dict[str,str] now; an old build stored a list.
# The bad field must revert to its default, every valid field must survive.
_write(settings_file, {"theme": "light", "dismissed_mcp_suggestions": ["legacy", "list"]})
write(settings_file, {"theme": "light", "dismissed_mcp_suggestions": ["legacy", "list"]})
s = store.load_settings()
assert s.theme == "light"
assert s.dismissed_mcp_suggestions == {}
@@ -108,14 +108,14 @@ def test_type_drifted_field_reverts_to_default_keeps_rest(settings_file):
def test_retired_literal_value_reverts_to_default(settings_file):
# default_thinking_level is a Literal; a retired value must not brick load.
_write(settings_file, {"theme": "light", "default_thinking_level": "ultra"})
write(settings_file, {"theme": "light", "default_thinking_level": "ultra"})
s = store.load_settings()
assert s.theme == "light"
assert s.default_thinking_level == "auto"
def test_multiple_bad_fields_all_revert_valid_survive(settings_file):
_write(settings_file, {
write(settings_file, {
"theme": "light",
"default_thinking_level": "ultra", # retired literal
"dismissed_mcp_suggestions": [1, 2, 3], # wrong type
@@ -139,7 +139,7 @@ def test_corrupt_json_returns_defaults_and_preserves_file(settings_file):
def test_non_dict_top_level_returns_defaults(settings_file):
_write(settings_file, ["not", "an", "object"])
write(settings_file, ["not", "an", "object"])
s = store.load_settings()
assert s.theme == "dark"
assert os.path.exists(settings_file + ".corrupt")
backend/tests/test_settings_server_owned.py
+9 -9
View File
@@ -20,22 +20,22 @@ from backend.main import app
@pytest.fixture
def client():
import backend.auth as auth_mod
if not auth_mod._TOKEN:
if not auth_mod.TOKEN:
import secrets
auth_mod._TOKEN = secrets.token_urlsafe(32)
return TestClient(app, headers={"Authorization": f"Bearer {auth_mod._TOKEN}"})
auth_mod.TOKEN = secrets.token_urlsafe(32)
return TestClient(app, headers={"Authorization": f"Bearer {auth_mod.TOKEN}"})
@pytest.fixture
def reset_settings():
from backend.apps.settings.settings import load_settings, _save_settings
from backend.apps.settings.store import load_settings, save_settings
original = load_settings().model_copy(deep=True)
yield
_save_settings(original)
save_settings(original)
def _activate_pro(client, token="repro-bearer-0123456789abcdef"):
def activate_pro(client, token="repro-bearer-0123456789abcdef"):
"""Drive the real /api/subscription/activate with a mocked cloud /api/me."""
fake_me = AsyncMock()
fake_me.status_code = 200
@@ -62,7 +62,7 @@ def test_stale_settings_put_cannot_wipe_activation(client):
snapshot = client.get("/api/settings").json()
assert snapshot is not None
token = _activate_pro(client)
token = activate_pro(client)
from backend.apps.settings.settings import load_settings
s = load_settings()
@@ -113,7 +113,7 @@ def test_put_cannot_inject_server_owned_fields(client):
def test_dedicated_endpoints_still_mutate(client):
"""Freezing PUT must not freeze the real owners: disconnect still reverts
routing, and a fresh activate still re-connects afterwards."""
_activate_pro(client)
activate_pro(client)
r = client.post("/api/subscription/disconnect")
assert r.status_code == 200
@@ -122,7 +122,7 @@ def test_dedicated_endpoints_still_mutate(client):
assert s.connection_mode == "own_key"
assert s.openswarm_bearer_token is not None # disconnect keeps sign-in
_activate_pro(client, token="second-bearer-aaaabbbbccccdddd")
activate_pro(client, token="second-bearer-aaaabbbbccccdddd")
s = load_settings()
assert s.connection_mode == "openswarm-pro"
assert s.openswarm_bearer_token == "second-bearer-aaaabbbbccccdddd"
backend/tests/test_skill_metrics_and_analyzer.py
+42 -42
View File
@@ -14,17 +14,17 @@ import time
from backend.apps.agents.browser import browser_skills as sk
from backend.apps.agents.browser import browser_metrics as bm
_ANALYZER = os.path.join(os.path.dirname(__file__), "..", "..", "scripts", "analyze-browser-metrics.py")
ANALYZER = os.path.join(os.path.dirname(__file__), "..", "..", "scripts", "analyze-browser-metrics.py")
def _load_analyzer():
spec = importlib.util.spec_from_file_location("bma", _ANALYZER)
def load_analyzer():
spec = importlib.util.spec_from_file_location("bma", ANALYZER)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
def _log():
def action_log():
return [
{"tool": "BrowserNavigate", "input": {"url": "http://h/form"}, "ok": True},
{"tool": "BrowserType", "input": {"selector": "#q", "text": "shoes"}, "ok": True},
@@ -32,79 +32,79 @@ def _log():
]
def _task_row(sig, path, dur_s, turns=None, playbook_seeded=False):
def task_row(sig, path, dur_s, turns=None, playbook_seeded=False):
# started_at in the past makes record_task compute a realistic total_ms.
bm.record_task("s-" + sig + path + str(turns) + str(playbook_seeded), "b", sig, "completed",
time.time() - dur_s, turns if turns is not None else (0 if path == "replay" else 3),
_log(), {"input": 10, "output": 5}, path=path, task_sig=sig,
action_log(), {"input": 10, "output": 5}, path=path, task_sig=sig,
playbook_seeded=playbook_seeded)
def test_skill_events_are_emitted_for_each_transition(_metrics_dir):
def test_skill_events_are_emitted_for_each_transition(metrics_dir):
sk.clear(wipe_disk=True)
sk.record_skill("shop.com", "search now", _log()) # learn
sk.record_skill("shop.com", "search now", action_log()) # learn
sk.mark_replay_succeeded("shop.com", "search now") # promote
sk.mark_replay_failed("shop.com", "search now") # kept (trusted, 1)
sk.mark_replay_failed("shop.com", "search now") # demote
evs = _read(os.path.join(_metrics_dir, "skill_events.jsonl"))
evs = read_rows(os.path.join(metrics_dir, "skill_events.jsonl"))
kinds = [e["kind"] for e in evs]
assert "learn" in kinds and "promote" in kinds and "demote" in kinds
# every event carries enough to group + reason about it
assert all(e.get("host") and e.get("task_sig") and e.get("kind") for e in evs)
def test_analyzer_measures_replay_speedup_when_the_layer_helps(_metrics_dir, capsys):
def test_analyzer_measures_replay_speedup_when_the_layer_helps(metrics_dir, capsys):
sk.clear(wipe_disk=True)
# A repeated task: 1 slow LLM run, then 2 fast replays -> measurable speedup.
sk.record_skill("shop.com", "search now", _log())
_task_row(sk._sig("search now"), "llm", 4.0)
sk.record_skill("shop.com", "search now", action_log())
task_row(sk._sig("search now"), "llm", 4.0)
sk.mark_replay_succeeded("shop.com", "search now")
_task_row(sk._sig("search now"), "replay", 0.04)
_task_row(sk._sig("search now"), "replay", 0.05)
task_row(sk._sig("search now"), "replay", 0.04)
task_row(sk._sig("search now"), "replay", 0.05)
mod = _load_analyzer()
tasks = mod._load(os.path.join(_metrics_dir, "tasks.jsonl"))
sevs = mod._load(os.path.join(_metrics_dir, "skill_events.jsonl"))
mod = load_analyzer()
tasks = mod._load(os.path.join(metrics_dir, "tasks.jsonl"))
sevs = mod._load(os.path.join(metrics_dir, "skill_events.jsonl"))
mod.skill_layer_report(tasks, sevs)
out = capsys.readouterr().out
assert "REPLAY SPEEDUP" in out
assert "x faster" in out and "replay" in out
def test_analyzer_flags_silent_non_help_thrash(_metrics_dir, capsys):
def test_analyzer_flags_silent_non_help_thrash(metrics_dir, capsys):
sk.clear(wipe_disk=True)
# A task that keeps getting re-learned/edited and quarantined, never promoted,
# and whose runs always go via the LLM (never the fast path) = the ghost.
sk.record_skill("bad.com", "do thing now", _log()) # learn
sk.record_skill("bad.com", "do thing now", action_log()) # learn
sk.mark_replay_failed("bad.com", "do thing now") # quarantine
edited = _log()[:-1] + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
edited = action_log()[:-1] + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
"clicked_role": "button", "clicked_name": "Other"}]
sk.record_skill("bad.com", "do thing now", edited) # edit (un-quarantine)
sk.mark_replay_failed("bad.com", "do thing now") # quarantine again
_task_row(sk._sig("do thing now"), "llm", 3.0)
_task_row(sk._sig("do thing now"), "llm_fallback", 3.2)
task_row(sk._sig("do thing now"), "llm", 3.0)
task_row(sk._sig("do thing now"), "llm_fallback", 3.2)
mod = _load_analyzer()
tasks = mod._load(os.path.join(_metrics_dir, "tasks.jsonl"))
sevs = mod._load(os.path.join(_metrics_dir, "skill_events.jsonl"))
mod = load_analyzer()
tasks = mod._load(os.path.join(metrics_dir, "tasks.jsonl"))
sevs = mod._load(os.path.join(metrics_dir, "skill_events.jsonl"))
mod.skill_layer_report(tasks, sevs)
out = capsys.readouterr().out
assert "SILENT NON-HELP" in out # repeated but never replayed
assert "THRASH" in out # re-learned/edited, never promoted
def test_analyzer_reports_composition(_metrics_dir, capsys):
def test_analyzer_reports_composition(metrics_dir, capsys):
sk.clear(wipe_disk=True)
sk.record_skill("shop.com", "search now", _log())
sk.record_skill("shop.com", "search now", action_log())
sk.mark_replay_succeeded("shop.com", "search now") # trusted foundation
plus = _log() + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
plus = action_log() + [{"tool": "BrowserClickIndex", "input": {}, "ok": True,
"clicked_role": "button", "clicked_name": "Checkout"}]
sk.record_skill("shop.com", "search and checkout now", plus) # composes on foundation
sk.mark_replay_succeeded("shop.com", "search and checkout now") # dependent earns trust too
sk.deprecate_skill("shop.com", "search now") # must invalidate the TRUSTED dependent
mod = _load_analyzer()
sevs = mod._load(os.path.join(_metrics_dir, "skill_events.jsonl"))
mod = load_analyzer()
sevs = mod._load(os.path.join(metrics_dir, "skill_events.jsonl"))
# the invalidate EVENT must actually fire (end-to-end), not just the state flip
assert any(e["kind"] == "invalidate" for e in sevs)
mod.skill_layer_report([], sevs)
@@ -114,33 +114,33 @@ def test_analyzer_reports_composition(_metrics_dir, capsys):
assert "1 dependent(s) re-proofed" in out
def test_analyzer_reports_playbook_cutting_exploration_turns(_metrics_dir, capsys):
def test_analyzer_reports_playbook_cutting_exploration_turns(metrics_dir, capsys):
# tier-2 win: a cold run on a host takes many turns; once strategy is seeded,
# the same kind of task takes fewer. The analyzer must report HELPS.
sig = sk._sig("find people")
_task_row(sig, "llm", 60.0, turns=14, playbook_seeded=False) # cold
_task_row(sig, "llm", 40.0, turns=8, playbook_seeded=True) # seeded -> fewer turns
mod = _load_analyzer()
tasks = mod._load(os.path.join(_metrics_dir, "tasks.jsonl"))
task_row(sig, "llm", 60.0, turns=14, playbook_seeded=False) # cold
task_row(sig, "llm", 40.0, turns=8, playbook_seeded=True) # seeded -> fewer turns
mod = load_analyzer()
tasks = mod._load(os.path.join(metrics_dir, "tasks.jsonl"))
mod.playbook_report(tasks)
out = capsys.readouterr().out
assert "STRATEGIC PLAYBOOK" in out and "HELPS" in out and "NOT HELPING" not in out
def test_analyzer_flags_playbook_that_does_not_help(_metrics_dir, capsys):
def test_analyzer_flags_playbook_that_does_not_help(metrics_dir, capsys):
# anti-ghost: memory is active (seeded) but seeded runs are NOT cheaper -> flag.
sig = sk._sig("stubborn task")
_task_row(sig, "llm", 60.0, turns=10, playbook_seeded=False)
_task_row(sig, "llm", 60.0, turns=12, playbook_seeded=True) # seeded but MORE turns
mod = _load_analyzer()
tasks = mod._load(os.path.join(_metrics_dir, "tasks.jsonl"))
task_row(sig, "llm", 60.0, turns=10, playbook_seeded=False)
task_row(sig, "llm", 60.0, turns=12, playbook_seeded=True) # seeded but MORE turns
mod = load_analyzer()
tasks = mod._load(os.path.join(metrics_dir, "tasks.jsonl"))
mod.playbook_report(tasks)
out = capsys.readouterr().out
assert "NOT HELPING" in out
# --- helpers ---------------------------------------------------------------
def _read(path):
def read_rows(path):
import json
out = []
if os.path.exists(path):
@@ -156,6 +156,6 @@ import pytest
@pytest.fixture
def _metrics_dir():
def metrics_dir():
# the autouse conftest fixture already points metrics at a temp dir; surface it
return os.environ["OPENSWARM_BROWSER_METRICS_DIR"]
backend/tests/test_v2_invariants.py
+180 -180
View File
File diff suppressed because it is too large Load Diff
backend/tests/test_web_search_cascade.py
+24 -24
View File
@@ -22,36 +22,36 @@ from backend.apps.agents.tools.web import WebSearchTool, DDGRateLimited
@pytest.fixture(autouse=True)
def _no_network(monkeypatch):
def no_network(monkeypatch):
# Default everything to "unavailable / no network"; each test opts paths in.
monkeypatch.setattr(W, "_resolve_gemini_api_key", lambda: None)
monkeypatch.setattr(W, "_resolve_openai_api_key", lambda: None)
async def _no_subs():
async def no_subs():
return set()
monkeypatch.setattr(W, "_refresh_9r_connected", _no_subs)
monkeypatch.setattr(W, "_refresh_9r_connected", no_subs)
# subscription helpers hit localhost:20128 otherwise
monkeypatch.setattr(W, "_gemini_grounded_via_9router", AsyncMock(return_value={}))
monkeypatch.setattr(W, "_openai_websearch_via_9router", AsyncMock(return_value={}))
def _ddg_returns(monkeypatch, text):
def ddg_returns(monkeypatch, text):
monkeypatch.setattr(WebSearchTool, "_search_ddg", staticmethod(AsyncMock(return_value=text)))
def _ddg_throttled(monkeypatch):
def ddg_throttled(monkeypatch):
monkeypatch.setattr(WebSearchTool, "_search_ddg",
staticmethod(AsyncMock(side_effect=DDGRateLimited("throttled"))))
@pytest.mark.asyncio
async def test_ddg_is_tried_first_and_wins(monkeypatch):
_ddg_returns(monkeypatch, "[1] Foo\n https://foo.example")
ddg_returns(monkeypatch, "[1] Foo\n https://foo.example")
# grounded would raise if reached; prove it isn't
async def _boom(*a, **k):
async def boom(*a, **k):
raise AssertionError(f"grounded should not be called when DDG has results (args={a!r}, kwargs={k!r})")
monkeypatch.setattr(W, "_gemini_grounded_call", _boom)
monkeypatch.setattr(W, "_gemini_grounded_call", boom)
t = time.monotonic()
res = await search(SearchBody(query="foo"))
@@ -62,8 +62,8 @@ async def test_ddg_is_tried_first_and_wins(monkeypatch):
@pytest.mark.asyncio
async def test_ddg_throttled_falls_over_to_openai(monkeypatch):
_ddg_throttled(monkeypatch)
async def testddg_throttled_falls_over_to_openai(monkeypatch):
ddg_throttled(monkeypatch)
monkeypatch.setattr(W, "_resolve_openai_api_key", lambda: "okey")
monkeypatch.setattr(W, "_openai_websearch",
@@ -78,15 +78,15 @@ async def test_ddg_throttled_falls_over_to_openai(monkeypatch):
@pytest.mark.asyncio
async def test_a_hung_grounded_attempt_is_bounded(monkeypatch):
_ddg_throttled(monkeypatch)
ddg_throttled(monkeypatch)
monkeypatch.setattr(W, "_GROUNDED_ATTEMPT_TIMEOUT", 0.3)
monkeypatch.setattr(W, "_resolve_gemini_api_key", lambda: "gkey")
attempts = []
async def _hangs(*a, **k):
async def hangs(*a, **k):
attempts.append((a, k))
await asyncio.sleep(30)
monkeypatch.setattr(W, "_gemini_grounded_call", _hangs)
monkeypatch.setattr(W, "_gemini_grounded_call", hangs)
t = time.monotonic()
res = await search(SearchBody(query="x"))
@@ -99,7 +99,7 @@ async def test_a_hung_grounded_attempt_is_bounded(monkeypatch):
@pytest.mark.asyncio
async def test_primary_openai_reorders_grounded_tier(monkeypatch):
_ddg_throttled(monkeypatch)
ddg_throttled(monkeypatch)
monkeypatch.setattr(W, "_resolve_gemini_api_key", lambda: "gkey")
monkeypatch.setattr(W, "_resolve_openai_api_key", lambda: "okey")
@@ -116,7 +116,7 @@ async def test_primary_openai_reorders_grounded_tier(monkeypatch):
@pytest.mark.asyncio
async def test_everything_fails_is_honest_not_empty(monkeypatch):
_ddg_throttled(monkeypatch) # no keys, no subs (from fixture)
ddg_throttled(monkeypatch) # no keys, no subs (from fixture)
res = await search(SearchBody(query="obscure thing"))
assert res["backend"] == "none"
assert "obscure thing" in res["results"]
@@ -131,15 +131,15 @@ async def test_everything_fails_is_honest_not_empty(monkeypatch):
from backend.apps.web.web import fetch, FetchBody
from backend.apps.agents.tools.web import WebFetchTool
import backend.apps.agents.tools.ssrf_guard as _ssrf
import backend.apps.agents.tools.ssrf_guard as ssrf
@pytest.fixture(autouse=True)
def _allow_urls(monkeypatch):
monkeypatch.setattr(_ssrf, "assert_safe_url", AsyncMock(return_value=None))
def allow_urls(monkeypatch):
monkeypatch.setattr(ssrf, "assert_safe_url", AsyncMock(return_value=None))
def _local_returns(monkeypatch, text):
def local_returns(monkeypatch, text):
monkeypatch.setattr(WebFetchTool, "execute",
AsyncMock(return_value=[{"type": "text", "text": text}]))
@@ -147,10 +147,10 @@ def _local_returns(monkeypatch, text):
@pytest.mark.asyncio
async def test_fetch_local_first_wins_and_is_fast(monkeypatch):
big = "Contents of https://x.example:\n\n" + ("real article body " * 50)
_local_returns(monkeypatch, big)
async def _boom(*a, **k):
local_returns(monkeypatch, big)
async def boom(*a, **k):
raise AssertionError(f"grounded fetch should not run when local has content (args={a!r}, kwargs={k!r})")
monkeypatch.setattr(W, "_gemini_grounded_call", _boom)
monkeypatch.setattr(W, "_gemini_grounded_call", boom)
t = time.monotonic()
res = await fetch(FetchBody(url="https://x.example"))
@@ -161,7 +161,7 @@ async def test_fetch_local_first_wins_and_is_fast(monkeypatch):
@pytest.mark.asyncio
async def test_fetch_thin_local_falls_to_grounded(monkeypatch):
_local_returns(monkeypatch, "Contents of https://spa.example:\n\n") # JS wall, empty body
local_returns(monkeypatch, "Contents of https://spa.example:\n\n") # JS wall, empty body
monkeypatch.setattr(W, "_resolve_gemini_api_key", lambda: "gkey")
monkeypatch.setattr(W, "_gemini_grounded_call",
AsyncMock(return_value={"text": "rendered page text from grounding", "chunks": []}))
@@ -173,7 +173,7 @@ async def test_fetch_thin_local_falls_to_grounded(monkeypatch):
@pytest.mark.asyncio
async def test_fetch_local_error_returned_as_last_resort(monkeypatch):
_local_returns(monkeypatch, "HTTP error 403 fetching https://blocked.example")
local_returns(monkeypatch, "HTTP error 403 fetching https://blocked.example")
# no grounded keys/subs (autouse fixtures) -> all grounded skip/fail
res = await fetch(FetchBody(url="https://blocked.example"))
assert res["backend"] == "local"
backend/tests/test_web_search_ddg.py
+11 -11
View File
@@ -17,7 +17,7 @@ import pytest
from backend.apps.agents.tools.web import WebSearchTool, DDGRateLimited
class _FakeResp:
class FakeResp:
def __init__(self, status_code: int, text: str):
self.status_code = status_code
self.text = text
@@ -27,10 +27,10 @@ class _FakeResp:
raise httpx.HTTPStatusError("err", request=None, response=None)
class _FakeClient:
class FakeClient:
"""Stands in for httpx.AsyncClient; returns a canned response."""
def __init__(self, resp: _FakeResp):
self._resp = resp
def __init__(self, resp: FakeResp):
self.resp = resp
# AsyncMock accepts any (url, data=, headers=, ...) without re-declaring
# httpx's signature just to ignore it.
self.post = AsyncMock(return_value=resp)
@@ -42,12 +42,12 @@ class _FakeClient:
return False
def _patch_client(monkeypatch, resp: _FakeResp):
monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: _FakeClient(resp))
def patch_client(monkeypatch, resp: FakeResp):
monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: FakeClient(resp))
# One real organic result + one sponsored (ad) row in DDG's html markup.
_HTML_WITH_AD = """
HTML_WITH_AD = """
<div class="result results_links_deep">
<a class="result__a" href="//duckduckgo.com/l/?uddg=https%3A%2F%2Fexample.com%2Freal&amp;rut=x">Real Result Title</a>
<a class="result__snippet">A genuine snippet about the topic.</a>
@@ -61,14 +61,14 @@ _HTML_WITH_AD = """
@pytest.mark.asyncio
async def test_202_raises_rate_limited_not_empty(monkeypatch):
_patch_client(monkeypatch, _FakeResp(202, "<html>throttle challenge, no results</html>"))
patch_client(monkeypatch, FakeResp(202, "<html>throttle challenge, no results</html>"))
with pytest.raises(DDGRateLimited):
await WebSearchTool._search_ddg("anything", 5)
@pytest.mark.asyncio
async def test_execute_reports_rate_limit_clearly(monkeypatch):
_patch_client(monkeypatch, _FakeResp(202, "throttle"))
patch_client(monkeypatch, FakeResp(202, "throttle"))
parts = await WebSearchTool().execute({"query": "x", "num_results": 5}, None)
msg = parts[0]["text"].lower()
assert "rate-limit" in msg
@@ -77,7 +77,7 @@ async def test_execute_reports_rate_limit_clearly(monkeypatch):
@pytest.mark.asyncio
async def test_ads_are_stripped_real_results_kept(monkeypatch):
_patch_client(monkeypatch, _FakeResp(200, _HTML_WITH_AD))
patch_client(monkeypatch, FakeResp(200, HTML_WITH_AD))
out = await WebSearchTool._search_ddg("topic", 5)
assert "example.com/real" in out
assert "Real Result Title" in out
@@ -90,6 +90,6 @@ async def test_ads_are_stripped_real_results_kept(monkeypatch):
@pytest.mark.asyncio
async def test_genuinely_empty_is_not_a_rate_limit(monkeypatch):
# 200 with no result blocks is a real empty result set, not a throttle.
_patch_client(monkeypatch, _FakeResp(200, "<html><body>nothing here</body></html>"))
patch_client(monkeypatch, FakeResp(200, "<html><body>nothing here</body></html>"))
out = await WebSearchTool._search_ddg("zxcvqwer no hits", 5)
assert out == ""