matchmaking-v2/research/poc/browser-apply/apply_engine.py

"""
apply_engine.py — the MASTER KEY. One board-agnostic engine for any job board.

  any apply URL  ->  pick_adapter(url)  ->  generic fill loop + adapter hints  ->  accuracy gate
                  -> verify + screenshot.  Ashby is just one adapter; the GenericAdapter handles the
                  long tail (bespoke career pages) by vision alone.

Statuses it can report (the last three are what make the long-tail measurable):
  submitted | ready_for_review | closed | login_required | no_form | failed | blocked

Run:
  APPLY_URL="https://...apply..." AUTO_SUBMIT=0 python apply_engine.py
"""
from __future__ import annotations

import asyncio
import json
import os

from dotenv import load_dotenv

from browser_use import Agent, BrowserSession, ChatOpenAI
from browser_config import make_profile, RESUME_PATH
from candidate import build_candidate, candidate_prompt_block
from adapters import pick_adapter

load_dotenv()

APPLY_URL = os.environ.get("APPLY_URL", "")
APPLY_MODEL = os.environ.get("APPLY_MODEL", "gpt-4.1")
AUTO_SUBMIT = os.environ.get("AUTO_SUBMIT", "0") != "0"   # default SAFE (fill, don't submit)
MAX_STEPS = int(os.environ.get("MAX_STEPS", "55"))


def build_task(cand: dict, url: str, hints: str) -> str:
    submit_clause = (
        "Submit the application (click the final Submit/Apply button) — ONLY after the gate passes."
        if AUTO_SUBMIT else
        "DO NOT click final submit. Stop at review and report ready_for_review — ONLY after the gate passes."
    )
    return f"""You are completing ONE job application on the careers page below, on behalf of the
candidate. Use ONLY the candidate's real data — never fabricate qualifications, employers, or dates.

{candidate_prompt_block(cand)}

{hints}

STEPS:
1. Go to {url}. It may be a JS app that renders blank for a few seconds — wait ~6s and re-read;
   refresh ONCE if still blank.
2. TRIAGE the page:
   - If it shows "no longer accepting applications" / 404 / closed -> STOP, status="closed".
   - If applying REQUIRES creating an account or logging in -> STOP, status="login_required"
     (we do not create accounts; that is the human/Cloud lane).
   - If there is NO real per-job application form (just a generic /careers page, or "email your
     CV to...") -> STOP, status="no_form".
   - Otherwise there is a fillable application form -> continue.
3. RÉSUMÉ: if an "autofill from resume" control exists, upload there first. Then ensure the résumé
   file is ATTACHED to the resume field — filename shown, NO spinner/"uploading" text. Uploads are
   async; wait ~8s and verify; retry up to 2x. A half-finished upload causes failures.
4. FILL EVERY FIELD THAT STRENGTHENS THE APPLICATION (not just "*" required ones):
   - contact details, links, work-authorization/sponsorship per CONFIRMED FACTS;
   - typeaheads: type then CLICK the suggestion (verify it committed, not a placeholder);
   - date pickers: open and click a valid FUTURE date;
   - motivation / "why" / "additional information": write the tailored answer (never blank).
5. Tick every required consent checkbox (verify it is actually checked).
6. ACCURACY GATE — GO/NO-GO. List EVERY required field: every "*" field, every required consent
   checkbox, and the motivation box. A field passes only if it has a real value (not a placeholder
   "Start typing…"/"Pick date…"/"Type here…") or, for checkboxes, is CHECKED. If any fails, fix it
   and re-check. Do NOT proceed until all pass. Then close any open picker so the form is settled.
7. {submit_clause}
8. If you submitted: click Submit ONCE, wait ~5s, read the page (don't rapid-click — looks like
   spam). Handle "still uploading"->re-verify upload & retry once; validation errors->fix & retry;
   "flagged as spam"->STOP, status="failed". Report status="submitted" ONLY with an explicit on-
   screen confirmation (quote it); otherwise status="failed".
9. Final answer = JSON: {{"status": "submitted"|"ready_for_review"|"closed"|"login_required"|
   "no_form"|"failed"|"blocked", "all_required_filled": true|false,
   "resume_upload_confirmed": true|false, "fields": {{<label>: <value>, ...}},
   "confirmation": <exact text/id or null>, "blockers": [<what stopped you, if any>],
   "notes": <short explanation>}}.
"""


async def main():
    if not APPLY_URL.startswith("http"):
        print('✋ Set APPLY_URL, e.g.  APPLY_URL="https://jobs.ashbyhq.com/.../application" python apply_engine.py')
        return
    adapter = pick_adapter(APPLY_URL)
    print(f"[1/3] Parsing résumé ({RESUME_PATH}) ...")
    cand = await build_candidate()
    print(f"      -> {cand['profile'].get('full_name')} · {cand['profile'].get('years_experience')} yrs")
    print(f"[2/3] URL={APPLY_URL}\n      adapter={adapter.name} ({adapter.tier}) · model={APPLY_MODEL} · auto_submit={AUTO_SUBMIT}")

    session = BrowserSession(browser_profile=make_profile())
    agent = Agent(
        task=build_task(cand, APPLY_URL, adapter.hints()),
        llm=ChatOpenAI(model=APPLY_MODEL),
        browser_session=session,
        available_file_paths=[RESUME_PATH],
    )
    try:
        history = await agent.run(max_steps=MAX_STEPS)
        print("\n[3/3] ========== RESULT ==========")
        print(history.final_result())
        print("=================================")
        try:
            shot = os.path.join(os.path.dirname(os.path.abspath(__file__)), "engine_result.png")
            await session.take_screenshot(path=shot, full_page=True)
            print(f"📸 {shot}")
        except Exception as e:
            print(f"(no screenshot: {e})")
    finally:
        try:
            await asyncio.get_event_loop().run_in_executor(None, input, "\nPress Enter to close... ")
        except Exception:
            pass
        await session.kill()


if __name__ == "__main__":
    asyncio.run(main())