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

"""
apply_harness.py — accuracy harness. Runs the master-key engine (apply_engine) in NO-SUBMIT mode
over a list of real apply URLs and tallies the outcome per job. This is how we measure the precise
accuracy on the offsite set (does the generic engine actually fill arbitrary boards?).

SAFE: always no-submit (these are real employers — we never fire applications). Headless, fresh
browser per job. Reads URLs from /tmp/picks.json (list of {company, ats, apply_url, title}).

Run:  python apply_harness.py
"""
from __future__ import annotations

import asyncio
import inspect
import json
import os
import re
import time

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
from adapters import pick_adapter
from apply_engine import build_task, APPLY_MODEL  # build_task uses AUTO_SUBMIT=False by default

load_dotenv()
HERE = os.path.dirname(os.path.abspath(__file__))
TAG = os.environ.get("TAG", "")
RESULTS = os.path.join(HERE, f"harness_results{('_' + TAG) if TAG else ''}.jsonl")
MAX_STEPS = int(os.environ.get("MAX_STEPS", "38"))
USE_VISION = os.environ.get("USE_VISION", "1") != "0"


def parse_status(text: str) -> dict:
    """Pull the engine's final JSON (or at least its status) out of the agent's final message."""
    if not text:
        return {"status": "unknown"}
    m = re.search(r"\{.*\}", text, re.DOTALL)
    if m:
        try:
            return json.loads(m.group(0))
        except json.JSONDecodeError:
            pass
    m = re.search(r'status"?\s*[:=]\s*"?(\w+)', text)
    return {"status": m.group(1) if m else "unknown"}


async def run_one(cand: dict, job: dict, idx: int) -> dict:
    url = job["apply_url"]
    adapter = pick_adapter(url)
    session = BrowserSession(browser_profile=make_profile(headless=True, persist=False))
    agent = Agent(
        task=build_task(cand, url, adapter.hints()),
        llm=ChatOpenAI(model=APPLY_MODEL),
        browser_session=session,
        available_file_paths=[RESUME_PATH],
        calculate_cost=True,
        use_vision=USE_VISION,
    )
    out = {"company": job.get("company"), "ats": job.get("ats"), "adapter": adapter.name, "url": url}
    t0 = time.perf_counter()
    try:
        hist = await agent.run(max_steps=MAX_STEPS)
        out["elapsed_s"] = round(time.perf_counter() - t0, 1)
        out["steps"] = len(getattr(hist, "history", []) or [])
        res = parse_status(hist.final_result())
        out["status"] = res.get("status", "unknown")
        out["all_required_filled"] = res.get("all_required_filled")
        out["notes"] = (res.get("notes") or "")[:160]
        try:
            summ = agent.token_cost_service.get_usage_summary()
            if inspect.isawaitable(summ):
                summ = await summ
            get = (lambda o, k: (getattr(o, k, None) if not isinstance(o, dict) else o.get(k)))
            out["cost_usd"] = round(get(summ, "total_cost") or 0, 4)
            out["tokens"] = get(summ, "total_tokens") or get(summ, "total_prompt_tokens")
        except Exception as e:
            out["cost_err"] = str(e)[:80]
        try:
            shot = os.path.join(HERE, f"harness_{idx}_{job.get('company')}.png")
            await session.take_screenshot(path=shot, full_page=True)
            out["screenshot"] = os.path.basename(shot)
        except Exception:
            pass
    except Exception as e:
        out["elapsed_s"] = round(time.perf_counter() - t0, 1)
        out["status"] = "engine_error"
        out["notes"] = f"{type(e).__name__}: {str(e)[:140]}"
    finally:
        await session.kill()
    with open(RESULTS, "a") as f:
        f.write(json.dumps(out) + "\n")
    return out


async def main():
    jobs = json.load(open(os.environ.get("PICKS", "/tmp/picks.json")))[:int(os.environ.get("LIMIT", "5"))]
    open(RESULTS, "w").close()  # reset
    print(f"Harness: {len(jobs)} jobs, NO-SUBMIT, model={APPLY_MODEL}")
    cand = await build_candidate()
    print(f"Candidate: {cand['profile'].get('full_name')} · {cand['profile'].get('years_experience')} yrs\n")

    results = []
    for i, job in enumerate(jobs):
        print(f"[{i+1}/{len(jobs)}] {job.get('company')} ({job.get('ats')}) -> running...")
        r = await run_one(cand, job, i)
        print(f"        status={r['status']}  filled={r.get('all_required_filled')}  "
              f"cost=${r.get('cost_usd') or 0:.4f}  time={r.get('elapsed_s') or 0:.0f}s  steps={r.get('steps','?')}")
        results.append(r)

    print("\n================ SUMMARY ================")
    from collections import Counter
    c = Counter(r["status"] for r in results)
    print(f"model: {APPLY_MODEL} | vision: {'ON' if USE_VISION else 'OFF'}")
    print("status tally:", dict(c))
    ok = sum(1 for r in results if r["status"] in ("ready_for_review", "submitted") and r.get("all_required_filled"))
    total_cost = sum((r.get("cost_usd") or 0) for r in results)
    total_time = sum((r.get("elapsed_s") or 0) for r in results)
    print(f"fully-filled (gate passed): {ok}/{len(results)}")
    print(f"TOTAL COST: ${total_cost:.4f}  (avg ${total_cost/max(1,len(results)):.4f}/job)")
    print(f"TOTAL TIME: {total_time:.0f}s  (avg {total_time/max(1,len(results)):.0f}s/job)")
    for r in results:
        print(f"  - {r['company']:11} [{r['adapter']:7}] {r['status']:16} "
              f"${r.get('cost_usd') or 0:.4f}  {r.get('elapsed_s') or 0:.0f}s  "
              f"{r.get('steps','?')}st  {r.get('notes','')[:40]}")


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