unbroker

1"""Automation-tier selection and per-subject action planning.2 3Tiers:4  T0  fully automated, no verification loop5  T1  automated submit + automated verification (email mode B/C, or backend-cleared captcha)6  T2  automated submit, verification needs a human (hard captcha / phone callback / account)7  T3  human-required end-to-end (gov ID, fax, mail, voice-only phone)8"""9from __future__ import annotations10 11import dossier as dossier_mod12import vectors as vectors_mod13 14HARD_HUMAN = ("gov_id", "fax", "mail", "phone_voice")15 16 17def select_tier(broker: dict, email_mode: str = "draft_only",18                browser_clears_captcha: bool = False) -> str:19    req = ((broker.get("optout") or {}).get("requires")) or {}20    if not isinstance(req, dict):21        req = {}  # defensive: a malformed record (e.g. requires as a list) must not crash planning22 23    if any(req.get(k) for k in HARD_HUMAN):24        return "T3"25    if req.get("account"):26        return "T2"27 28    captcha = bool(req.get("captcha"))29    if (captcha and not browser_clears_captcha) or req.get("phone_callback"):30        return "T2"31 32    if req.get("email_verification"):33        return "T1" if email_mode in ("programmatic", "alias") else "T2"34 35    if captcha and browser_clears_captcha:36        return "T1"37    return "T0"38 39 40def plan(subject_dossier: dict, brokers_list: list[dict], cfg: dict,41         browser_clears_captcha: bool = False) -> list[dict]:42    email_mode = (subject_dossier.get("preferences") or {}).get("email_mode") \43        or cfg.get("email_mode", "draft_only")44    actions: list[dict] = []45    for b in brokers_list:46        opt = b.get("optout") or {}47        search = b.get("search") or {}48        # Defensive shape coercion: a subagent may have written a malformed record (requires as a49        # list, quirks as a string). Normalize here so nothing downstream crashes on a bad broker file.50        req = opt.get("requires") if isinstance(opt.get("requires"), dict) else {}51        q = opt.get("quirks")52        quirks = q if isinstance(q, list) else ([q] if isinstance(q, str) and q else [])53        tier = select_tier(b, email_mode, browser_clears_captcha)54        disclosure = dossier_mod.select_disclosure(subject_dossier, opt.get("inputs", []))55        svectors = vectors_mod.search_vectors(subject_dossier, b)56        # Pre-warn (don't discover mid-flow): a broker whose identity gate hard-requires DOB will57        # force a human touchpoint if DOB was not collected at intake (§4.1). Surface it now.58        prewarn: list[str] = []59        if req.get("dob") and not (subject_dossier.get("identity") or {}).get("date_of_birth"):60            prewarn.append("date_of_birth: this broker's identity gate requires DOB to match records; "61                           "collect it up front (intake --dob) or expect a mid-flow human pause")62        actions.append({63            "broker_id": b.get("id"),64            "broker_name": b.get("name"),65            "priority": b.get("priority"),66            "method": opt.get("method"),67            "tier": tier,68            "human_required": tier == "T3",69            "search_url": search.get("url"),70            "fetch": search.get("fetch", "web_extract"),71            "antibot": search.get("antibot"),72            "search_by": vectors_mod.supported_by(b),73            "search_vectors": svectors,74            "optout_url": opt.get("url"),75            "optout_email": opt.get("email"),76            "disclosure_fields": sorted(disclosure.keys()),77            "needs_operator_input": prewarn,78            "owns": b.get("owns") or [],79            "notes": opt.get("notes", ""),80            "optout_quirks": quirks,81            "optout_requires": req,82            # The DELETION lane (right-to-delete), distinct from listing suppression. Structured so83            # the autopilot can route to it: {via: email|in_flow|web_form, email?, url?, kinds?, notes?}84            "deletion": opt.get("deletion") or {},85            # Exact ordered opt-out steps maintained IN the broker record (field-verified knowledge86            # lives with the data, not in code).87            "optout_playbook": opt.get("playbook") or [],88        })89    return actions90 91 92def fanout(brokers_list: list[dict], batch_size: int = 5) -> dict:93    """Group brokers into batches for parallel `delegate_task` scan subagents.94 95    Scanning many brokers serially is slow and burns context; above `batch_size`96    the agent is expected to spawn one subagent per batch (see SKILL.md).97    """98    ids = [b.get("id") for b in brokers_list if b.get("id")]99    batches = [ids[i:i + batch_size] for i in range(0, len(ids), batch_size)]100    return {101        "broker_count": len(ids),102        "batch_size": batch_size,103        "should_fanout": len(ids) > batch_size,104        "batches": batches,105    }106 107 108# States that mean "the crawl reached a verdict for this broker".109_SCANNED_STATES = {"found", "not_found", "indirect_exposure", "blocked", "submitted",110                   "verification_pending", "awaiting_processing", "confirmed_removed", "reappeared",111                   "action_selected", "human_task_queued"}112# States that still need a deletion action taken.113_ACTIONABLE_STATES = {"found", "indirect_exposure", "reappeared", "action_selected"}114 115 116def batch_plan(subject_dossier: dict, brokers_list: list[dict], cfg: dict,117               ledger: dict | None = None, browser_clears_captcha: bool = False) -> dict:118    """Reduce the per-broker plan into a phase-oriented batch view.119 120    Overlays the current ledger state on each broker, groups by what the operator121    should DO next, and collapses ownership clusters so a parent removal that clears122    children is ONE action, not N. Read-only: computes, never mutates the ledger.123    """124    ledger = ledger or {}125    actions = plan(subject_dossier, brokers_list, cfg, browser_clears_captcha)126 127    # child id -> parent id (only for parents present in this plan set)128    child_to_parent: dict[str, str] = {}129    for a in actions:130        for child in a.get("owns") or []:131            child_to_parent[child] = a["broker_id"]132 133    def state_of(bid: str) -> str:134        return (ledger.get(bid) or {}).get("state", "new")135 136    groups: dict[str, list[dict]] = {137        "unscanned": [],        # no verdict yet -> Phase 1 crawl138        "found": [],            # direct removable listing -> Phase 2 opt-out (incl. reappeared/action_selected)139        "indirect_exposure": [],# PII on a third party's record -> CCPA/GDPR delete email140        "blocked": [],          # anti-bot / needs stealth browser -> requeue141        "in_progress": [],      # submitted / verification_pending / awaiting_processing142        "human": [],            # human_task_queued -> the end-of-run digest, NOT re-scanning143        "done": [],             # confirmed_removed144        "not_found": [],145    }146    covered_by_parent: dict[str, list[str]] = {}147 148    for a in actions:149        bid = a["broker_id"]150        st = state_of(bid)151        # cluster collapse: if a parent in this set is already actioned, the child is covered152        parent = child_to_parent.get(bid)153        if parent and state_of(parent) in ("found", "reappeared", "action_selected", "submitted",154                                           "verification_pending", "awaiting_processing",155                                           "confirmed_removed", "human_task_queued"):156            covered_by_parent.setdefault(parent, []).append(bid)157            continue158 159        row = {"broker_id": bid, "broker_name": a["broker_name"], "priority": a["priority"],160               "tier": a["tier"], "method": a["method"], "state": st,161               "optout_url": a["optout_url"], "optout_email": a.get("optout_email"),162               "clears_children": a.get("owns") or [],163               "optout_requires": a.get("optout_requires") or {},164               "optout_quirks": a.get("optout_quirks") or [],165               "deletion": a.get("deletion") or {},166               "optout_playbook": a.get("optout_playbook") or [],167               "notes": a.get("notes", "")}168        if st in ("submitted", "verification_pending", "awaiting_processing"):169            groups["in_progress"].append(row)170        elif st == "confirmed_removed":171            groups["done"].append(row)172        elif st in ("reappeared", "action_selected"):173            groups["found"].append(row)   # still needs the opt-out action174        elif st == "human_task_queued":175            groups["human"].append(row)   # parked for the digest; never re-queued as work176        elif st in groups:177            groups[st].append(row)178        elif st not in _SCANNED_STATES:179            groups["unscanned"].append(row)180        else:181            groups.setdefault(st, []).append(row)182 183    # PARENTS FIRST: within the actionable 'found' group, order cluster parents (a removal184    # that clears children) ahead of standalone listings, most-children first. Working a185    # parent before its children is what makes the cluster dedup real -- do them in this order.186    groups["found"].sort(key=lambda r: (-len(r.get("clears_children") or []),187                                        {"T0": 0, "T1": 1, "T2": 2, "T3": 3}.get(r.get("tier") or "", 9),188                                        r["broker_id"]))189 190    return {191        "subject": subject_dossier.get("subject_id"),192        "phase": "discover" if groups["unscanned"] else "delete",193        "counts": {k: len(v) for k, v in groups.items()},194        "groups": groups,195        "cluster_savings": {p: kids for p, kids in covered_by_parent.items()},196        "parent_playbook": _parent_playbook(groups["found"]),197        "next_actions": _batch_next(groups, covered_by_parent),198    }199 200 201def synthesize_steps(r: dict) -> list[str]:202    """Generic ordered opt-out steps derived from an optout record's structured fields.203 204    Used for any broker without a hand-verified `optout.playbook`. Bespoke, field-verified205    step lists live IN the broker JSON (`optout.playbook`) - single source of truth that206    accrues knowledge as live runs discover mechanics (see methods.md logging rule).207    """208    steps = [f"Opt out at {r.get('optout_url') or r.get('optout_email') or '(see broker record)'}"209             + (f" -- clears {', '.join(r['clears_children'])}." if r.get("clears_children") else ".")]210    req = r.get("optout_requires") or {}211    if req.get("profile_url"):212        steps.append("Needs the confirmed profile_url (paste the listing URL you recorded).")213    if req.get("email_verification"):214        steps.append("Email verification: the same browser/inbox must open the confirmation link.")215    if req.get("phone_callback"):216        steps.append("Phone-callback code required; queue a human task if no operator is available.")217    if req.get("gov_id"):218        steps.append("Government ID demanded (T3): human task; never send SSN or a full ID number.")219    d = r.get("deletion") or {}220    if d.get("email"):221        steps.append(f"DELETION lane: a right-to-delete request can be emailed to {d['email']}"222                     + (f" ({d['notes']})" if d.get("notes") else "")223                     + " -- prefer deletion over suppression.")224    if r.get("notes"):225        steps.append(str(r["notes"]))226    for q in (r.get("optout_quirks") or [])[:3]:227        steps.append(str(q))228    return steps229 230 231def _parent_playbook(found_rows: list[dict]) -> list[dict]:232    """Tailored, ordered opt-out instructions for each cluster PARENT in the found group.233 234    Steps come from the broker record's own `optout.playbook` (field-verified, maintained with235    the data) with a synthesised fallback so the guidance is never empty. Standalone listings236    are intentionally omitted -- the playbook exists to make the parents-first order concrete.237    """238    playbook: list[dict] = []239    for i, r in enumerate([x for x in found_rows if x.get("clears_children")], start=1):240        steps = list(r.get("optout_playbook") or []) or synthesize_steps(r)241        playbook.append({242            "order": i,243            "broker_id": r["broker_id"],244            "broker_name": r["broker_name"],245            "tier": r["tier"],246            "clears_children": r["clears_children"],247            "optout_url": r.get("optout_url"),248            "optout_email": r.get("optout_email"),249            "deletion": r.get("deletion") or {},250            "steps": steps,251        })252    return playbook253 254 255def _batch_next(groups: dict, covered: dict) -> list[str]:256    tips: list[str] = []257    if groups["unscanned"]:258        tips.append(f"PHASE 1 (crawl): {len(groups['unscanned'])} broker(s) unscanned -- run `fanout` and "259                    "scan read-only before any deletion.")260    if groups["found"]:261        parents = [r for r in groups["found"] if r.get("clears_children")]262        if parents:263            order = " -> ".join(r["broker_id"] for r in parents)264            tips.append(f"PHASE 2 (opt-out): {len(groups['found'])} direct listing(s). DO CLUSTER PARENTS "265                        f"FIRST, in this order: {order} (see `parent_playbook` for tailored per-parent "266                        "steps), then the standalone listings.")267        else:268            tips.append(f"PHASE 2 (opt-out): {len(groups['found'])} direct listing(s) to remove.")269    if groups["indirect_exposure"]:270        tips.append(f"{len(groups['indirect_exposure'])} indirect-exposure case(s): send a targeted "271                    "CCPA/GDPR delete-my-PII email (render-email --kind ccpa_indirect), do NOT use the opt-out form.")272    if groups["blocked"]:273        tips.append(f"{len(groups['blocked'])} blocked (anti-bot): requeue for a stealth/cloud browser "274                    "pass; don't burn subagent time fighting CAPTCHAs.")275    if covered:276        n = sum(len(v) for v in covered.values())277        tips.append(f"Cluster dedup: {n} child site(s) covered by parent removals -- skip separate opt-outs.")278    if groups["in_progress"]:279        tips.append(f"{len(groups['in_progress'])} in progress: resolve verification links, then confirm removal.")280    if groups.get("human"):281        tips.append(f"{len(groups['human'])} parked human task(s): present via `tasks` at end of run "282                    "(do not re-scan or re-queue them).")283    return tips284