"""connectors_meta.py — the Meta Ads (Marketing API) connector: identity, credentials, and the schema PROBE that W31-T40 measures with. ⭐ WHY A PROBE AND NOT A FIELD LIST. Owner ruling R8: the token exists to *"measure the schema now, with nobody logging in via Facebook yet"*, and R2 requires the spawned grids to carry **every field the API returns** — with T40's measured list as T47's only oracle (absent 0 / extra 0). Meta's own Insights documentation declines to enumerate its fields, so a docs-derived list is not an answer to either ruling. This file asks the API what it has. ⛔ THE MEASUREMENT IS `?metadata=1`, WHICH IS GRAPH'S OWN INTROSPECTION, and that is the whole design. Requesting a node without `fields=` returns Meta's DEFAULT subset — a fraction of the schema, and a list that changes when they change their defaults. `metadata=1` returns the node's FIELD TABLE (name + type + description) whether or not this account populates it, so the answer is a property of the API rather than of our ad account's data. Insights has no such table, so its catalog is measured the only other honest way: request a candidate set and record which names the API ACCEPTS and which it REJECTS BY NAME. Both halves are written to the asset, so a reader can see the question as well as the answer. ⛔ SECRETS. `META_ADS_ACCESS_TOKEN` lives in gitignored `platform/.env`. It is read here, sent as a POST-shaped query parameter to Graph over TLS, and **never printed, never logged, never written to the asset** — the asset records the URL PATH and the field names, not the query string. Every call is a read-only GET. python aios-web/api/connectors_meta.py --probe # writes the wave asset python aios-web/api/connectors_meta.py --check # credentials only, no schema walk """ from __future__ import annotations import json import os import sys import time import urllib.error import urllib.parse import urllib.request from pathlib import Path #: The connector's identity in the platform's registries. ⛔ `ut_meta_` is declared ONCE, in #: `routes_connectors.CONNECTED_PREFIXES`, and read from there — a second spelling here is the #: [[one-question-two-normalizers]] shape and `verify_meta` asserts it does not exist. KEY = "meta_ads" LABEL = "Meta Ads" #: The keychain entry type an admin stores a token under. ⚠ `core.keychain.ENTRY_TYPES` must learn #: this or `add_entry` 400s — that is W31-T48's half, not this file's. ENTRY_TYPE = "meta_ads" #: ⚠ VERSIONED, AND THE VERSION IS REPORTED RATHER THAN ASSUMED. Graph deprecates versions on a #: schedule; a pinned literal that has aged out fails with an error that reads like a bad token. #: Overridable so a probe can be re-run against a newer version without an edit. GRAPH_VERSION = os.environ.get("META_GRAPH_VERSION") or "v21.0" GRAPH = f"https://graph.facebook.com/{GRAPH_VERSION}" #: The five entity levels R2 names, in the order a person reads them. `edge` is the connection on #: its parent; `node` is what `metadata=1` is asked about. LEVELS = [ ("adaccount", "Ad Account", None), ("campaign", "Campaign", "campaigns"), ("adset", "Ad Set", "adsets"), ("ad", "Ad", "ads"), ("adcreative", "Creative", "adcreatives"), ] #: ⚠ INPUTS, NOT ANSWERS — the names OFFERED to the API per entity level. The measurement is #: which of them Graph ACCEPTS (see `node_fields`); a name missing from a list here is "not #: asked", never "not available", and the asset says so on its own face. #: ⛔ THESE EXIST BECAUSE `?metadata=1` RETURNS NOTHING ON v21.0 — measured, not assumed. Graph #: withdrew the introspection table that used to make a candidate list unnecessary. ENTITY_CANDIDATES = { "adaccount": [ "account_id", "account_status", "age", "amount_spent", "balance", "business_city", "business_country_code", "business_name", "business_state", "business_street", "business_street2", "business_zip", "capabilities", "created_time", "currency", "disable_reason", "end_advertiser", "end_advertiser_name", "funding_source", "has_migrated_permissions", "id", "io_number", "is_attribution_spec_system_default", "is_direct_deals_enabled", "is_notifications_enabled", "is_personal", "is_prepay_account", "is_tax_id_required", "line_numbers", "media_agency", "min_campaign_group_spend_cap", "min_daily_budget", "name", "offsite_pixels_tos_accepted", "owner", "partner", "spend_cap", "tax_id", "tax_id_status", "tax_id_type", "timezone_id", "timezone_name", "timezone_offset_hours_utc", "tos_accepted", "user_tasks", "user_tos_accepted"], "campaign": [ "account_id", "bid_strategy", "boosted_object_id", "budget_rebalance_flag", "budget_remaining", "buying_type", "campaign_group_active_time", "can_create_brand_lift_study", "can_use_spend_cap", "configured_status", "created_time", "daily_budget", "effective_status", "id", "is_skadnetwork_attribution", "issues_info", "last_budget_toggling_time", "lifetime_budget", "name", "objective", "pacing_type", "primary_attribution", "promoted_object", "smart_promotion_type", "source_campaign", "source_campaign_id", "special_ad_categories", "special_ad_category", "special_ad_category_country", "spend_cap", "start_time", "status", "stop_time", "topline_id", "updated_time"], "adset": [ "account_id", "adlabels", "adset_schedule", "asset_feed_id", "attribution_spec", "bid_adjustments", "bid_amount", "bid_constraints", "bid_info", "bid_strategy", "billing_event", "budget_remaining", "campaign", "campaign_active_time", "campaign_attribution", "campaign_id", "configured_status", "created_time", "creative_sequence", "daily_budget", "daily_min_spend_target", "daily_spend_cap", "destination_type", "effective_status", "end_time", "frequency_control_specs", "id", "instagram_actor_id", "is_dynamic_creative", "issues_info", "learning_stage_info", "lifetime_budget", "lifetime_imps", "lifetime_min_spend_target", "lifetime_spend_cap", "multi_optimization_goal_weight", "name", "optimization_goal", "optimization_sub_event", "pacing_type", "promoted_object", "recurring_budget_semantics", "review_feedback", "rf_prediction_id", "source_adset", "source_adset_id", "start_time", "status", "targeting", "targeting_optimization_types", "time_based_ad_rotation_id_blocks", "time_based_ad_rotation_intervals", "updated_time", "use_new_app_click"], "ad": [ "account_id", "ad_active_time", "ad_review_feedback", "ad_schedule_end_time", "ad_schedule_start_time", "adlabels", "adset", "adset_id", "bid_amount", "bid_info", "bid_type", "campaign", "campaign_id", "configured_status", "conversion_domain", "created_time", "creative", "demolink_hash", "display_sequence", "effective_status", "engagement_audience", "failed_delivery_checks", "id", "issues_info", "last_updated_by_app_id", "name", "preview_shareable_link", "priority", "recommendations", "source_ad", "source_ad_id", "status", "targeting", "tracking_and_conversion_with_defaults", "tracking_specs", "updated_time"], "adcreative": [ "account_id", "actor_id", "adlabels", "applink_treatment", "asset_feed_spec", "authorization_category", "body", "branded_content_sponsor_page_id", "bundle_folder_id", "call_to_action_type", "categorization_criteria", "category_media_source", "collaborative_ads_lsb_image_bank_id", "degrees_of_freedom_spec", "destination_set_id", "dynamic_ad_voice", "effective_authorization_category", "effective_instagram_media_id", "effective_object_story_id", "enable_direct_install", "enable_launch_instant_app", "id", "image_crops", "image_hash", "image_url", "instagram_actor_id", "instagram_permalink_url", "instagram_story_id", "instagram_user_id", "interactive_components_spec", "link_deep_link_url", "link_destination_display_url", "link_og_id", "link_url", "messenger_sponsored_message", "name", "object_id", "object_store_url", "object_story_id", "object_story_spec", "object_type", "object_url", "page_link", "page_message", "place_page_set_id", "platform_customizations", "playable_asset_id", "portrait_customizations", "product_set_id", "recommender_settings", "source_instagram_media_id", "status", "template_url", "template_url_spec", "thumbnail_id", "thumbnail_url", "title", "url_tags", "use_page_actor_override", "video_id"], } #: ⚠ AN INPUT, NOT AN ANSWER — and labelled as one in the asset. Insights has no `metadata=1` #: table, so the catalog is measured by ACCEPTANCE: every name here is offered to the API and the #: probe records which it takes and which it names in an error. A name missing from this list is #: therefore "not asked", never "not available", and the asset says so. INSIGHTS_CANDIDATES = [ "account_id", "account_name", "account_currency", "campaign_id", "campaign_name", "adset_id", "adset_name", "ad_id", "ad_name", "date_start", "date_stop", "impressions", "reach", "frequency", "clicks", "unique_clicks", "ctr", "unique_ctr", "cpc", "cpm", "cpp", "spend", "social_spend", "actions", "action_values", "conversions", "conversion_values", "cost_per_action_type", "cost_per_unique_click", "cost_per_inline_link_click", "cost_per_thruplay", "purchase_roas", "website_purchase_roas", "inline_link_clicks", "inline_link_click_ctr", "inline_post_engagement", "outbound_clicks", "outbound_clicks_ctr", "unique_outbound_clicks", "video_play_actions", "video_p25_watched_actions", "video_p50_watched_actions", "video_p75_watched_actions", "video_p95_watched_actions", "video_p100_watched_actions", "video_avg_time_watched_actions", "video_thruplay_watched_actions", "quality_ranking", "engagement_rate_ranking", "conversion_rate_ranking", "objective", "optimization_goal", "buying_type", "attribution_setting", "canvas_avg_view_time", "estimated_ad_recallers", "full_view_impressions", ] #: Every probe call, for the mailbox log the PRD's shared-resource table demands ("log every probe #: in mailbox/D.md; the rate limit is shared across the token"). Paths only — never a query string, #: because the token rides in one. CALLS: list = [] class MetaError(RuntimeError): """A Graph refusal, carrying Meta's own message. ⛔ Its `str()` is safe to print: the token is stripped from every URL before it reaches here.""" def _token(): """The Meta token from the environment, or from gitignored `platform/.env`. Returns "" when absent — the caller REPORTS that rather than crashing, because "no token" is a real state of this deployment and it is the state as of 2026-08-12.""" tok = os.environ.get("META_ADS_ACCESS_TOKEN") or "" if tok: return tok.strip() env = Path(__file__).resolve().parents[2] / "platform" / ".env" if env.exists(): for line in env.read_text(encoding="utf-8", errors="replace").splitlines(): if line.strip().startswith("META_ADS_ACCESS_TOKEN"): _, _, v = line.partition("=") return v.strip().strip('"').strip("'") return "" def _safe(url): """`url` with every query value redacted — the only form allowed near a log or an asset.""" parts = urllib.parse.urlsplit(url) q = urllib.parse.parse_qsl(parts.query, keep_blank_values=True) shown = "&".join(f"{k}=" if k == "access_token" else f"{k}={v}" for k, v in q) return urllib.parse.urlunsplit((parts.scheme, parts.netloc, parts.path, shown, "")) def get(path, token, **params): """One read-only Graph GET. Returns the decoded body, or raises `MetaError` with Meta's reason. ⚠ A 400 FROM GRAPH IS AN ANSWER, NOT A CRASH — an unknown field name comes back as a 400 whose message NAMES the field, and that is precisely how the Insights catalog is measured below. """ params["access_token"] = token url = f"{GRAPH}/{path.lstrip('/')}?" + urllib.parse.urlencode(params) CALLS.append({"path": path.lstrip("/"), "params": sorted(k for k in params if k != "access_token")}) req = urllib.request.Request(url, headers={"User-Agent": "aios-meta-probe/1"}) try: with urllib.request.urlopen(req, timeout=60) as r: return json.loads(r.read().decode("utf-8", "replace")) except urllib.error.HTTPError as e: body = e.read().decode("utf-8", "replace") try: err = (json.loads(body).get("error") or {}) msg = err.get("message") or body[:300] code = err.get("code") except Exception: # noqa: BLE001 msg, code = body[:300], None raise MetaError(f"HTTP {e.code} (code {code}) on {_safe(url)}: {msg}") from None except Exception as e: # noqa: BLE001 raise MetaError(f"{type(e).__name__} on {_safe(url)}: {e}") from None def node_fields(node_id, token, candidates, batch=60): """`(accepted, rejected, gated)` for ONE node, measured by ACCEPTANCE + BISECT. ⛔⛔ `?metadata=1` IS DEAD AND THAT WAS MEASURED, NOT ASSUMED. This function used to ask Graph for its published introspection table, which is the documented way to enumerate a node's fields. On **v21.0 it returns nothing**: `GET /act_…?metadata=1` answers with keys `['account_id','id']` and `GET /?metadata=1` with `['id']` — `metadata.fields` is EMPTY in both. The probe reported "0 fields" for all five levels and looked like a broken probe; it was a withdrawn API. So the entity levels are measured exactly as Insights already was: offer names, keep what the API takes. ⛔⛔ **ACCEPTED IS NOT RETURNED, AND THIS IS THE ONE THAT BREAKS R2 SILENTLY.** A 20-field batch on a live campaign answers **HTTP 200 with 14 keys** — Graph omits nulls from the response. A catalog built from the keys that came BACK loses roughly a third of the schema, which is R2's *"do not drop any column"* violated with no gate anywhere going red. The catalog is therefore what was **ASKED and ACCEPTED**, and this function never inspects the payload's keys. ⛔ **A 400 NAMES THE OFFENDER; A 403 NAMES NOTHING.** `(#100) Tried accessing nonexisting field (x)` tells us exactly which name to drop, so a batch converges in one retry. But the Ad Account level answers `(#200) Requires business_management permission` — no field named — and dropping the batch there reports "Ad Account: 0 fields", which is the honest-looking wrong answer. An unnamed refusal is BISECTED instead: 11 calls classified all 46, finding exactly one gated column (`owner`). """ accepted, rejected, gated = [], {}, {} def ask(names): """`(ok, err)` — one batch. `ok` means every name in it is a real, readable field.""" try: get(str(node_id), token, fields=",".join(names)) return True, "" except MetaError as e: return False, str(e) def resolve(names, depth=0): if not names: return ok, err = ask(names) if ok: accepted.extend(names) return named = [f for f in names if f"({f})" in err or f"field ({f})" in err] if named and len(named) < len(names): for f in named: rejected[f] = err[:200] resolve([f for f in names if f not in named], depth + 1) return if len(names) == 1: # a single name that still refuses: classify it by WHAT the refusal was. (rejected if "(#100)" in err else gated)[names[0]] = err[:200] return half = len(names) // 2 # unnamed refusal ⇒ bisect, never drop resolve(names[:half], depth + 1) resolve(names[half:], depth + 1) time.sleep(0.15) # the rate limit is shared across this token for start in range(0, len(candidates), batch): resolve(list(candidates[start:start + batch])) return sorted(set(accepted)), rejected, gated def insights_catalog(act_id, token, candidates=None): """`(accepted, rejected)` for the daily Insights field catalog, measured by ACCEPTANCE. Asks in batches; on a 400 that names a field, drops that name and retries. What survives is what the API takes at `time_increment=1` — which is the only claim R2's Insights grids need. ⚠ `rejected` is reported too: "we asked and Meta refused" and "we never asked" are different facts, and only the asset can keep them apart. """ remaining = list(candidates if candidates is not None else INSIGHTS_CANDIDATES) rejected, accepted = {}, [] for start in range(0, len(remaining), 25): batch = remaining[start:start + 25] while batch: try: get(f"act_{act_id}/insights", token, fields=",".join(batch), time_increment="1", date_preset="last_7d", limit="1") accepted.extend(batch) break except MetaError as e: bad = [f for f in batch if f in str(e)] if not bad: # a refusal that names no field: report, do not loop rejected["(batch)"] = str(e) break for f in bad: rejected[f] = str(e)[:200] batch = [f for f in batch if f not in bad] time.sleep(0.2) # the rate limit is shared across this token return sorted(set(accepted)), rejected def probe(): """Walk the whole schema and return the report dict. Never raises on a Meta refusal — a refusal IS the finding, and a probe that dies on one produces no asset at all.""" token = _token() report = {"graphVersion": GRAPH_VERSION, "token": bool(token), "accounts": [], "levels": {}, "insights": {}, "errors": []} if not token: report["errors"].append( "META_ADS_ACCESS_TOKEN is not set in the environment or in platform/.env. " "Nothing was measured. This is the state of this box as of 2026-08-12: the PRD " "records the token as 'recorded in platform/.env' and it is not there.") return report try: me = get("me/adaccounts", token, fields="id,account_id,name,account_status,currency,timezone_name,business_name") report["accounts"] = me.get("data") or [] except MetaError as e: report["errors"].append(f"reaching /me/adaccounts: {e}") return report if not report["accounts"]: report["errors"].append("the token authenticates but reaches ZERO ad accounts — a " "permissions/scope answer, not an empty business") return report act = report["accounts"][0] act_id = str(act.get("account_id") or str(act.get("id") or "").replace("act_", "")) node_of = {"adaccount": f"act_{act_id}"} for key, _label, edge in LEVELS: if edge is None: continue try: got = get(f"act_{act_id}/{edge}", token, fields="id", limit="1") rows = got.get("data") or [] if rows: node_of[key] = rows[0]["id"] else: report["errors"].append(f"{key}: this account has no {edge}, so no real row could " f"be read (the field TABLE is still measured below)") except MetaError as e: report["errors"].append(f"listing {edge}: {e}") for key, label, _edge in LEVELS: nid = node_of.get(key) cands = ENTITY_CANDIDATES.get(key) or [] if not nid: report["levels"][key] = {"label": label, "accepted": [], "rejected": {}, "gated": {}, "asked": len(cands), "node": "", "note": "no node of this level was reachable, so nothing " "could be offered to the API"} continue try: accepted, rejected, gated = node_fields(nid, token, cands) report["levels"][key] = {"label": label, "node": str(nid), "asked": len(cands), "accepted": accepted, "rejected": rejected, "gated": gated} except MetaError as e: report["levels"][key] = {"label": label, "node": str(nid), "asked": len(cands), "accepted": [], "rejected": {}, "gated": {}, "note": str(e)} try: accepted, rejected = insights_catalog(act_id, token) report["insights"] = {"accepted": accepted, "rejected": rejected, "candidatesAsked": len(INSIGHTS_CANDIDATES)} except MetaError as e: report["errors"].append(f"insights catalog: {e}") return report def as_markdown(report): """The wave asset. ⛔ Field NAMES and TYPES only — no row values, no account identifiers beyond the ad account name the owner already knows, and no query strings.""" L = ["# W31-T40 — the Meta Ads schema, MEASURED", "", f"Graph version: `{report['graphVersion']}` · token present: " f"**{'yes' if report['token'] else 'NO'}** · probe calls: **{len(CALLS)}**", ""] if report["errors"]: L += ["## What could not be measured", ""] L += [f"- {e}" for e in report["errors"]] + [""] if not report["token"]: L += ["⛔ **NOTHING BELOW WAS MEASURED.** This asset exists so the absence is a recorded " "fact rather than a missing file: the probe is built, runs, and reports honestly " "that it has no credential. Re-run `python aios-web/api/connectors_meta.py --probe` " "the moment a token lands and this file fills itself in.", ""] return "\n".join(L) L += ["## Reachable ad accounts", "", "| account_id | name | status | currency | timezone |", "|---|---|---|---|---|"] for a in report["accounts"]: L.append(f"| `{a.get('account_id','')}` | {a.get('name','')} | " f"{a.get('account_status','')} | {a.get('currency','')} | " f"{a.get('timezone_name','')} |") L.append("") L += ["## Method — and why it is not `?metadata=1`", "", "⛔ **Graph v21.0 publishes no introspection table.** `GET /act_…?metadata=1` returns " "keys `['account_id','id']` and `GET /?metadata=1` returns `['id']` — " "`metadata.fields` is EMPTY. The first run of this probe reported *0 fields* for all " "five levels and looked like a broken probe; it was a withdrawn API.", "", "⭐ So every level below is measured by **ACCEPTANCE**: the names in " "`connectors_meta.ENTITY_CANDIDATES` are offered, and what Graph takes is the catalog. " "A `(#100)` refusal NAMES the offending field, so a batch converges in one retry; a " "`(#200)` permission refusal names nothing, so the batch is **bisected** rather than " "dropped — otherwise the Ad Account level reports a confident, wrong `0 fields`.", "", "⛔⛔ **ACCEPTED IS NOT RETURNED.** A 20-field batch on a live campaign answers HTTP 200 " "with 14 keys, because Graph omits nulls. Every list below is what was **asked and " "accepted** — the response's own keys are never inspected. Building columns from " "returned keys drops roughly a third of the schema, which is R2's *do not drop any " "column* broken with nothing going red.", ""] for key, label, _edge in LEVELS: lv = report["levels"].get(key) or {} fields = lv.get("accepted") or [] L += [f"## {label} — **{len(fields)} accepted** of {lv.get('asked', 0)} asked", ""] if lv.get("note"): L += [f"⚠ {lv['note']}", ""] if fields: L += ["```", ", ".join(fields), "```", ""] if lv.get("rejected"): L += ["**Not a field on this account** (refused BY NAME, recorded rather than " "guessed at):", ""] L += [f"- `{k}`" for k in sorted(lv["rejected"])] + [""] if lv.get("gated"): L += ["⚠ **Permission-gated — a REPORTED limit, not a silent omission** (R6's second " "sentence):", ""] L += [f"- `{k}` — {v}" for k, v in sorted(lv["gated"].items())] + [""] ins = report.get("insights") or {} if ins: L += [f"## Insights at `time_increment=1` — {len(ins.get('accepted') or [])} accepted of " f"{ins.get('candidatesAsked', 0)} asked", "", "⚠ Insights publishes no `metadata=1` field table, so this catalog is measured by " "ACCEPTANCE: every name below was offered to the API and taken. A name absent from " "this file was **not asked**, which is not the same as unavailable.", "", "```", ", ".join(ins.get("accepted") or []), "```", ""] if ins.get("rejected"): L += ["### Refused by name", ""] L += [f"- `{k}` — {v}" for k, v in sorted((ins.get("rejected") or {}).items())] + [""] return "\n".join(L) def probe_credential(fields, timeout=8): """`{ok, message}` for a stored `meta_ads` credential — the keychain's liveness probe. ⛔ IT LIVES HERE, IN THE API LAYER, AND IS REGISTERED INTO `core.keychain` RATHER THAN IMPORTED BY IT. `core` never imports up (ARCHITECTURE.md), and the first draft of this arm did `sys.path.insert` + `import connectors_meta` from inside `core/keychain.py` — an inversion with no gate on it. `keychain.register_prober` is the slot, borrowed from `datastore.set_paused_probe`, which exists for exactly this reason one layer down. ⚠ ZERO reachable ad accounts is reported as NOT ok, deliberately: a token that authenticates but sees no account is a SCOPE answer — the single likeliest failure for a Marketing API credential — and `ok: true` would hide the one thing the admin needs to fix. """ tok = str((fields or {}).get("access_token") or (fields or {}).get("token") or (fields or {}).get("api_key") or "") if not tok: return {"ok": False, "message": "store the token under `access_token` (or `token` / `api_key`)"} try: got = get("me/adaccounts", tok, fields="id", limit="1") except MetaError as e: return {"ok": False, "message": str(e)[:200]} n = len(got.get("data") or []) return {"ok": n > 0, "message": (f"Meta answered — {n} ad account(s) reachable" if n else "the token authenticates but reaches NO ad account: that is a " "scope/permission answer, not an empty business")} def register(): """Fill the keychain's probe slot for this connector. Called at import by the API layer's connector registry (`routes_connectors`), so any process that can serve the directory has also registered the prober. Never raises: a probe slot that could not be filled costs a liveness CHECK, never the ability to store a credential.""" try: import core.keychain as kc return kc.register_prober(ENTRY_TYPE, probe_credential) except Exception: # noqa: BLE001 return False def main(argv): check_only = "--check" in argv token = _token() if check_only: print(f"META token present: {'yes' if token else 'NO'} " f"(never printed; length is not disclosed)") if not token: return 2 try: me = get("me/adaccounts", token, fields="id,name", limit="5") print(f"authenticated; reachable ad accounts: {len(me.get('data') or [])}") return 0 except MetaError as e: print(f"REFUSED: {e}") return 1 report = probe() out = (Path(__file__).resolve().parents[2] / ".claude" / "wiki" / "research" / "waves" / "wave31" / "proto" / "meta-ads-schema.md") out.parent.mkdir(parents=True, exist_ok=True) out.write_text(as_markdown(report), encoding="utf-8") print(f"wrote {out}") for c in CALLS: print(f" probe: GET /{c['path']} [{', '.join(c['params'])}]") return 0 if report["token"] and not report["errors"] else 2 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))