openswarm/backend/tests/test_browser_send_script.py

"""Send-script mechanism, verified without a live webview: real LinkedIn-shaped
interactives fixtures driven through a mock executor exercise the exact path the
live rig would (opener -> composer -> fill -> commit-check -> send -> clear-check),
plus every abort/honesty branch. The wall-clock a live run measures is not here;
the correctness the live run would prove is."""
import pytest

from backend.apps.agents.browser import browser_send_script as ss
from backend.apps.agents.browser import browser_send_parse as sp
from backend.apps.agents.browser.browser_agent import send_submit_index_in_state, payload_in_textbox

PROFILE = '[22]*<link "Tyler Chen Premium 1st">\n[50]*<link "Message">\n[51]<button "Follow">'
COMPOSER_EMPTY = '[2]<textbox "Write a message">\n[9]<button "Attach">'
COMPOSER_FILLED = '[2]<textbox "Write a message" value="[test] hello world r9-os">\n[14]<button "Send">'
COMPOSER_SENT = '[2]<textbox "Write a message">\n[9]<button "Attach">'  # cleared, Send gone
# The profile-overlay committed-fill (ground truth from probe r-dump): payload IS
# in the box but the Send button ranks OUT of the capped numbered list, so the
# script must fall back to click-by-name (how the model itself sends there).
COMPOSER_FILLED_NO_SEND = '[1]<textbox "I\'m looking for…">\n[24]<textbox "Write a message" value="[test] hello world r9-os">'

TASK = "go to tyler chen's linkedin hes in entrepreneurs first and text him '[test] hello world r9-os'"


def make_exec(list_script):
    """execute_tool mock: each BrowserListInteractives call returns the next
    scripted state; clicks/fills succeed and are recorded."""
    calls = {"list": 0, "clicks": []}
    states = list(list_script)

    async def execute(tool, params, bid, tid):
        if tool == "BrowserListInteractives":
            i = min(calls["list"], len(states) - 1)
            calls["list"] += 1
            return {"text": states[i]}
        calls["clicks"].append((tool, params))
        return {"ok": True}
    return execute, calls


# A full-page messaging composer: the only surface the script fires on (the live
# A/B proved firing on the profile /in/ overlay is net-negative). Tests that exercise
# the fill/send mechanism pass this; the surface-gate test passes the overlay URL.
THREAD_URL = "https://www.linkedin.com/messaging/thread/2-abc/"
PROFILE_URL = "https://www.linkedin.com/in/tylerchen1200/"


async def run(task, state0, list_script, url=THREAD_URL):
    ex, calls = make_exec(list_script)
    r = await ss.run_send_script(task, "b1", "", state0, ex, send_submit_index_in_state,
                                 payload_in_textbox, current_url=url)
    return r, calls


FEED_URL = "https://www.linkedin.com/feed/"
NO_COMPOSER = '[1]<link "Home">\n[2]<button "Search">\n[3]<link "Jobs">'
X_COMPOSER = '[3]<textbox "Post your reply">\n[8]<button "Reply">'  # X, not LinkedIn


@pytest.mark.asyncio
async def test_surface_gate_declines_when_no_composer_in_perception():
    """STRUCTURAL gate: a page whose perception has no compose-shaped textbox and no messaging
    opener declines, regardless of URL, so the script never fires where a fill would land nowhere
    useful. Stays composer-less through the poll.

    Since the structural finder went default-ON it gets ONE look here (that is the point of it, and
    it is how instagram reaches its composer at all). So "untouched" is now specifically: it may ASK
    the page whether it has an editable, and it may do nothing else. No index fill, no click, no
    send. The finder that answers "found nothing" must still put the run on the model path."""
    ex, calls = make_exec([NO_COMPOSER, NO_COMPOSER, NO_COMPOSER, NO_COMPOSER])
    r = await ss.run_send_script(TASK, "b1", "", NO_COMPOSER, ex, send_submit_index_in_state,
                                 payload_in_textbox, payload_source=TASK, current_url=FEED_URL)
    assert r is None
    assert [t for t, _ in calls["clicks"]] == ["BrowserFindComposer"]


@pytest.mark.asyncio
async def test_surface_gate_polls_for_a_late_rendering_composer():
    """The flake fix: the composer isn't in the FIRST perception (prestage snapshotted early,
    X home does this ~half the time) but lazy-renders a beat later. The gate polls a fresh
    perception before declining, so the write doesn't falsely fall to the slow model path."""
    # initial perception (passed in) has no composer; the fresh poll finds X's composer, then
    # the fill-committed and cleared states follow.
    X_FILLED = '[3]<textbox "Post your reply" value="[test] hello world r9-os">\n[8]<button "Reply">'
    X_SENT = '[3]<textbox "Post your reply">\n[1]<link "Home">'
    ex, calls = make_exec([X_COMPOSER, X_FILLED, X_SENT])
    r = await ss.run_send_script(TASK, "b1", "", NO_COMPOSER, ex, send_submit_index_in_state,
                                 payload_in_textbox, payload_source=TASK, current_url=FEED_URL)
    assert r is not None and r["sent"] is True  # it fired instead of a false decline
    assert calls["clicks"]


@pytest.mark.asyncio
async def test_surface_gate_fires_on_a_NON_linkedin_composer():
    """The whole generalization: the same send-script fires on ANY site whose perception
    carries a composer (here X's 'Post your reply'), no per-site URL gate, AND completes
    by clicking X's real submit button ('Reply') BY INDEX. Before the submit-vocabulary
    was generalized this only 'passed' because the mock succeeds on the by-name 'Send'
    fallback that doesn't exist on real X, so assert the real index path is taken."""
    X_FILLED = '[3]<textbox "Post your reply" value="[test] hello world r9-os">\n[8]<button "Reply">'
    X_SENT = '[3]<textbox "Post your reply">\n[1]<link "Home">'
    ex, calls = make_exec([X_FILLED, X_FILLED, X_SENT])
    r = await ss.run_send_script(TASK, "b1", "", X_COMPOSER, ex, send_submit_index_in_state,
                                 payload_in_textbox, payload_source=TASK,
                                 current_url="https://x.com/messages/123")
    assert r is not None and r["sent"] is True
    # the send went via BrowserClickIndex on X's real Reply submit (8), not the by-name 'Send' crutch
    assert ("BrowserClickIndex", {"index": 8}) in calls["clicks"]
    assert not any(t == "BrowserClickByName" for t, _ in calls["clicks"])


@pytest.mark.asyncio
async def test_surface_gate_allows_profile_overlay():
    """The profile /in/ overlay is winnable via click-by-name (ground truth: its Send
    ranks out of the list but is a real button), so it's back in scope, not declined."""
    ex, calls = make_exec([COMPOSER_FILLED_NO_SEND, COMPOSER_FILLED_NO_SEND, COMPOSER_SENT])
    r = await ss.run_send_script(TASK, "b1", "", COMPOSER_EMPTY, ex, send_submit_index_in_state,
                                 payload_in_textbox, payload_source=TASK, current_url=PROFILE_URL)
    assert r is not None and r["sent"] is True


@pytest.mark.asyncio
async def test_send_via_click_by_name_when_send_absent_from_ranked_list():
    """Ground truth (probe r-dump): the committed-fill state has the payload in the
    box but NO Send in the capped numbered list. The script falls back to
    click-by-name (the model's own send path there) and the receipt still passes."""
    ex, calls = make_exec([COMPOSER_FILLED_NO_SEND, COMPOSER_FILLED_NO_SEND, COMPOSER_SENT])
    r = await ss.run_send_script(TASK, "b1", "", COMPOSER_EMPTY, ex, send_submit_index_in_state,
                                 payload_in_textbox, payload_source=TASK, current_url=THREAD_URL)
    assert r is not None and r["sent"] is True
    # the send went through click-by-name, not an index click
    byname = [c for c in calls["clicks"] if c[0] == "BrowserClickByName"]
    assert byname and byname[0][1] == {"name": "Send", "role": "button"}


@pytest.mark.asyncio
async def test_opener_hop_full_success():
    """On a messaging THREAD-LIST page (full-page surface): script opens the
    conversation composer, fills, sees it commit, finds the late Send, clicks,
    sees it clear -> receipt passes."""
    # poll reads stay no-composer (PROFILE x3) -> opener path -> click Message -> composer appears
    r, calls = await run(TASK, PROFILE, [PROFILE, PROFILE, PROFILE, COMPOSER_EMPTY, COMPOSER_FILLED, COMPOSER_SENT])
    assert r is not None and r["sent"] is True
    assert r["payload"] == "[test] hello world r9-os"
    # opener click (50), fill into composer (2 w/ text), solo send click (14)
    idxs = [c[1].get("index") for c in calls["clicks"]]
    assert 50 in idxs and 2 in idxs and 14 in idxs
    fill = next(c for c in calls["clicks"] if c[1].get("text"))
    assert fill[1]["text"] == "[test] hello world r9-os"


@pytest.mark.asyncio
async def test_composer_already_open_skips_opener():
    """Prestage left the composer open: no opener click, straight to fill+send."""
    r, calls = await run(TASK, COMPOSER_EMPTY, [COMPOSER_FILLED, COMPOSER_FILLED, COMPOSER_SENT])
    assert r is not None and r["sent"] is True
    assert 50 not in [c[1].get("index") for c in calls["clicks"]]  # never clicked an opener


@pytest.mark.asyncio
async def test_abort_when_no_payload():
    """No quoted payload = the model's judgment call, never the script's."""
    r, _ = await run("open tyler chen's linkedin and message him something nice", PROFILE,
               [COMPOSER_EMPTY])
    assert r is None


@pytest.mark.asyncio
async def test_abort_when_fill_not_seen_committed():
    """Fill click ran but the textbox never shows the payload -> abort PRE-click,
    the irreversible send never fires (no false 'sent')."""
    r, calls = await run(TASK, COMPOSER_EMPTY, [COMPOSER_EMPTY, COMPOSER_EMPTY, COMPOSER_EMPTY])
    assert r is None
    # a Send-class click (index 14) must NEVER have been issued
    assert all(c[1].get("index") != 14 for c in calls["clicks"])


@pytest.mark.asyncio
async def test_post_click_unverified_yields_honest_note_not_resend():
    """Send clicked but the composer never verifiably clears -> the run returns
    sent=False with a do-not-resend note, never a silent retry."""
    r, calls = await run(TASK, COMPOSER_EMPTY, [COMPOSER_FILLED, COMPOSER_FILLED,
                                          COMPOSER_FILLED, COMPOSER_FILLED, COMPOSER_FILLED])
    assert r is not None and r["sent"] is False
    assert "do NOT send again" in r["note"] or "not send again" in r["note"].lower()
    # exactly one send-class click was issued (no blind re-fire)
    assert sum(1 for c in calls["clicks"] if c[1].get("index") == 14) == 1


@pytest.mark.asyncio
async def test_ambiguous_opener_aborts():
    """Two 'Message' openers = ambiguous = hands to the model, no guess."""
    two = '[50]*<link "Message">\n[70]*<link "Message">'
    r, _ = await run(TASK, two, [COMPOSER_EMPTY])
    assert r is None


COMPOSED = (
    f"{TASK}\n\n"
    "[routing brief from a fast pre-pass; follow it unless the live page disagrees]\n"
    'ENTRY: https://www.linkedin.com/search/results/people/\nSTEPS: click the "Tyler Chen" result, '
    'then the "Message" button, type the text, click "Send"'
)


@pytest.mark.asyncio
async def test_composed_task_brief_quotes_fire_via_payload_source():
    """The routing brief's own quoted strings made the payload ambiguous on every
    real dispatch (r242/r243 declined live); the raw user prompt rides separately."""
    ex, calls = make_exec([COMPOSER_FILLED, COMPOSER_FILLED, COMPOSER_SENT])
    r = await ss.run_send_script(COMPOSED, "b1", "", COMPOSER_EMPTY, ex,
                                 send_submit_index_in_state, payload_in_textbox,
                                 payload_source=TASK, current_url=THREAD_URL)
    assert r is not None and r["sent"] is True
    assert r["payload"] == "[test] hello world r9-os"


@pytest.mark.asyncio
async def test_brief_readonly_wording_does_not_disarm():
    """The aux brief wrote 'do not submit it' for a plain send once, false-flagging it read-only;
    the brief section is stripped before the read-only check, so the send still fires."""
    composed = (
        f"{TASK}\n\n"
        "[routing brief from a fast pre-pass; follow it unless the live page disagrees]\n"
        "STEPS: type the text into the composer, do not submit anything else, read-only elsewhere"
    )
    ex, calls = make_exec([COMPOSER_FILLED, COMPOSER_FILLED, COMPOSER_SENT])
    r = await ss.run_send_script(composed, "b1", "", COMPOSER_EMPTY, ex,
                                 send_submit_index_in_state, payload_in_textbox,
                                 payload_source=TASK, current_url=THREAD_URL)
    assert r is not None and r["sent"] is True


def test_dryrun_report_encodes_the_gate_funnel():
    """The coverage harness greps this one line for gate attribution: a staged composer
    reports composer=1, a bare page reports zeros, and armed/filled ride the booleans."""
    r = sp.dryrun_report(COMPOSER_EMPTY, armed=True, filled=True, url=THREAD_URL)
    assert "armed=1" in r and "composer=1" in r and "filled=1" in r and "textboxes=1" in r
    r0 = sp.dryrun_report(NO_COMPOSER, armed=False, filled=False)
    assert "armed=0" in r0 and "composer=0" in r0 and "opener=0" in r0 and "textboxes=0" in r0


def test_dryrun_report_counts_unmatched_textboxes():
    """The X 'Post text' class: a textbox present but name-unmatched must be visible in
    the report (textboxes>0, composer=0), or the funnel can't tell R from N failures."""
    state = '[4]<textbox "Some Novel Label">\n[7]<button "Go">'
    r = sp.dryrun_report(state, armed=True, filled=False)
    assert "composer=0" in r and "textboxes=1" in r


@pytest.mark.asyncio
async def test_readonly_probe_never_fires():
    """The send-probe quotes the very payload it checks for; without this wall the
    script DELIVERED a real message from a read-only probe (r243 live)."""
    probe = (
        "READ-ONLY verification, do NOT send, type, click any send/submit control. "
        'Check the thread for this exact text:\n"[test] hello world r9-os"\n'
        "End with OUTCOME: PAYLOAD-FOUND or PAYLOAD-NOT-FOUND."
    )
    ex, calls = make_exec([COMPOSER_FILLED, COMPOSER_FILLED, COMPOSER_SENT])
    r = await ss.run_send_script(probe, "b1", "", COMPOSER_EMPTY, ex,
                                 send_submit_index_in_state, payload_in_textbox,
                                 payload_source=TASK, current_url=THREAD_URL)
    assert r is None
    assert not calls["clicks"]


def test_looks_like_login_wall_hits_and_false_positives():
    # login/auth URLs (the live instagram/threads mis-fire was one of these)
    assert sp.looks_like_login_wall("https://www.instagram.com/accounts/login/?force_authentication", "")
    assert sp.looks_like_login_wall("https://accounts.google.com/v3/signin/identifier", "")
    assert sp.looks_like_login_wall("https://www.reddit.com/login/", "")
    assert sp.looks_like_login_wall("https://x.com/i/flow/login", "")
    # auth-form perception signals with an innocuous url
    assert sp.looks_like_login_wall("https://site.com/x", '[3]<textbox "Password">')
    assert sp.looks_like_login_wall("https://site.com/x", "Log in to X to continue")
    # false positives: a real composer page, a blog path, gmail inbox, a /author/ path
    assert not sp.looks_like_login_wall("https://x.com/home", X_COMPOSER)
    assert not sp.looks_like_login_wall("https://example.com/blog/login-tips", "")
    assert not sp.looks_like_login_wall("https://mail.google.com/mail/u/0/#inbox?compose=new", "")
    assert not sp.looks_like_login_wall("https://site.com/author/jane", "")


@pytest.mark.asyncio
async def test_login_wall_url_declines_before_any_fill():
    """A login URL declines even when the perception carries a composer and the task quotes a
    payload: a real send surface never shares a page with a login wall, and filling here types
    into the auth form (the live instagram/threads mis-fire under the reveal finder)."""
    ex, calls = make_exec([X_COMPOSER])
    r = await ss.run_send_script('post this exactly: "hello from the test x9"', "b1", "", X_COMPOSER, ex,
                                 send_submit_index_in_state, payload_in_textbox,
                                 current_url="https://x.com/i/flow/login")
    assert r is None
    assert not calls["clicks"]