"""automation_engine.py — wave-18 item 5 (contract C4-AUTO): the automation RUNTIME. THE SHAPE, and why it is this shape. An automation is a DEFINITION that lives in the tenant store and a RUN that lives in this process. The definition is durable, small and rarely written; the run is hot, chatty and worthless five minutes later. Conflating them is the failure this file is built to avoid, and it has two halves: * **Run state is NEVER persisted while it is running.** A `state: 'running'` written to the store outlives the process that wrote it, so a container restart mid-run leaves an automation that can never run again — the 409 sees a `running` nothing will ever clear. Hot state lives in `_RUNNING` (a process dict, cleared by definition on restart) and the store is written exactly ONCE per run, at the end. * **One coalesced store update per run per bucket.** `core/store.py` coalesces on a 20 s floor per key against a 256-commits/hour repo budget, so a per-ROW write is the wrong shape by two orders of magnitude — the S&P demo alone is ~500 rows. Every runner below computes its whole result first and commits it in a single `update` callback. WHAT IS VENDORED HERE AND WHY. `.claude/skills/browse/scrape.py` is not on `sys.path` and is not deployed, so its extraction core is copied into this file (the wave doc's instruction), ported httpx → requests (the only HTTP dependency this API already carries). The SSRF rail comes with it, HARDENED rather than merely preserved — see `guard`/`fetch`. That hardening is the point: the skill version takes a URL from a developer on a CLI; this takes one from a request body. """ from __future__ import annotations import copy import datetime as _dt import hashlib import hmac import ipaddress import json import os import re import socket import threading import time from urllib.parse import urljoin, urlparse import requests # ⭐ `providers` IS DELIBERATELY NOT IMPORTED HERE ANY MORE (wave 27 item 23). The capability # router had exactly one consumer in this file — the views top-up inside the Bright Data rung — # and that rung now lives in `connectors_ig`. So the vendor-routing seam is reached only by the # connector that routes, which is the shape the split was for: this file no longer knows that # there is more than one vendor, or that vendors cost money. Re-adding this import is therefore # a signal, not a convenience — it means something in the automation runtime started making a # vendor decision, and that belongs one layer down. try: # the parser the extraction half needs from bs4 import BeautifulSoup except Exception: # pragma: no cover — deploy lag; see mailbox ② BeautifulSoup = None # --------------------------------------------------------------------------------------------- # THE BUCKET (C4-AUTO) # --------------------------------------------------------------------------------------------- #: The per-tenant store key. Colocated with its reader, the `routes_nav._NAV_PREFS_KEY` pattern. STORE_KEY = "automations" #: The user-table bucket. ⚠ Read/written HERE through `TenantRuntime`, never through #: `core.user_tables` — that module writes the UNPREFIXED module-global key, which is correct for #: tenant #0 (empty prefix) and silently cross-tenant for every R2 tenant after it. Booked in the #: session-D mailbox as a finding against A's file rather than fixed from here. UT_STORE_KEY = "user_tables" UT_PREFIX = "ut_" MAX_AUTOMATIONS = 40 MAX_RUNS = 20 # trimmed history per automation MAX_NAME = 80 MAX_UT_ROWS = 5000 # mirrors core.user_tables.MAX_ROWS MAX_UT_TABLES = 40 #: ⚠ THE APPEND TABLES NEED A DIFFERENT CAP, and the reason is arithmetic rather than taste. #: `MAX_UT_ROWS` was sized for a scraped LIST — a page of ~500 companies that is re-read, so the #: row count is bounded by the page. The `ut_ig_*` tables are the opposite shape: every pull #: INSERTS a timestamped row (see `SNAPSHOT_FIELDS` and the compound keys below), so at R1's #: 1k-profiles-daily the snapshot table crosses 5000 rows on **day five** — and the old behaviour #: was a SILENT `skipped++`, i.e. the table would quietly stop growing and the run would still #: report success. A time series that stops after five days without saying so is worse than one #: that was never built. #: 200k is the owner-directed ceiling (R1's "~200k"). ⚠ IT BUYS VERY DIFFERENT HORIZONS FOR THE #: TWO APPEND TABLES, and the difference is worth knowing before anyone plans around it: #: ut_ig_snapshots 1 row per profile per pull -> ~200 days at 1k profiles/day #: ut_ig_post_snapshots maxPosts rows per pull -> ~8 days at 1k profiles x 24 posts #: So the post series is the one that fills, and it fills FAST. That is exactly why the breach had #: to become LOUD (a distinct `capped` count -> a `partial` run naming the table): at this rate the #: silent version would have flatlined a chart inside a fortnight with a green dot over it. #: #: ⚠ MEASURED COST AT THE CEILING (2026-08-04, this box): a full `ut_ig_post_snapshots` serialises #: to **35.8 MB** of JSON (~1.4 s). `UT_STORE_KEY` is ONE bucket for ALL of a tenant's user tables, #: so that cost is paid by every unrelated automation write in the tenant too. Booked for the B-3 #: Postgres tripwires (R2 clause b) rather than papered over — the fix is a row store, not a #: smaller number. #: (W19-C; the GENERIC loud-breach for every other table stays booked as DEBT D-11.) MAX_UT_IG_ROWS = 200_000 IG_TABLE_PREFIX = "ut_ig_" #: ⚠ THE RAISED CEILING BELONGS TO THE **APPEND** TABLES, NOT TO A NAME PREFIX (wave 20). It was #: keyed off `ut_ig_`, which was exactly right while every `ut_ig_*` table was an append table — #: and stopped being right the moment discovery added `ut_ig_candidates`, which is UPSERTED BY #: HANDLE and is bounded by how many accounts exist rather than by how often we look. It would #: have inherited a 200,000-row ceiling, and with it the measured 35.8 MB single-bucket #: serialisation cost, purely by an accident of naming. Naming the append tables is the version #: that stays true when the next `ut_ig_*` table is not one. IG_SNAPSHOTS_TABLE = "ut_ig_snapshots" IG_POSTS_TABLE = "ut_ig_posts" IG_POST_SNAPSHOTS_TABLE = "ut_ig_post_snapshots" IG_COMMENTS_TABLE = "ut_ig_comments" # --------------------------------------------------------------------------------------------- # ⭐⭐ WAVE 29 (item 7, DEBT D-9, owner rulings R1 + R2) — TIKTOK, AS A PARALLEL FAMILY # --------------------------------------------------------------------------------------------- # R1: FULL PARITY — profile AND posts AND comments, not a profile tier. # R2: a PARALLEL `ut_tt_*` family. `ut_ig_*` is untouched: zero migration, zero risk to the live # Instagram rows, and the two schemas may DIVERGE where the vendors do. # # ⛔ WHY A SECOND FAMILY RATHER THAN A `platform` COLUMN ON THE FIRST, stated here because it is # the question every reader will ask. `PRESET_PROFILE_FIELDS` carries `platform` and its identity # is `(platform, handle)` — so a TikTok PROFILE row could already have lived in `ut_ig_profile`. # The other four tables carry NO discriminator at all: `ut_ig_posts`/`ut_ig_comments` key on # `shortcode` and the snapshot tables on `influencer_key`, so an Instagram post and a TikTok video # that happened to share a code would MERGE SILENTLY. Adding `platform` to four live append tables # holding hundreds of thousands of rows is a migration on production data to buy a shared grid # nobody asked for. R2 chose the version with no migration. # # ⚠ THE ACCEPTED COST, so it is not rediscovered as a defect: the engine, rollups and grids learn # two families, and a cross-platform view needs a union. In exchange the two schemas can be HONEST # about their vendors — which is why `ut_tt_posts` has no `plays` column and `ut_tt_profile` says # `tt_id` rather than inheriting a column labelled "Instagram id". TT_TABLE_PREFIX = "ut_tt_" TT_PROFILE_TABLE = "ut_tt_profile" TT_SNAPSHOTS_TABLE = "ut_tt_snapshots" TT_POSTS_TABLE = "ut_tt_posts" TT_POST_SNAPSHOTS_TABLE = "ut_tt_post_snapshots" TT_COMMENTS_TABLE = "ut_tt_comments" AUTOMATION_RECORD_MODE = "automation" #: ⚠ THE RAISED CEILING FOLLOWS THE APPEND SHAPE, NOT THE PLATFORM. `ut_tt_snapshots` and #: `ut_tt_post_snapshots` are append tables for exactly the reason their IG twins are — one row per #: profile per pull, `maxPosts` rows per pull — so they inherit the ceiling by JOINING THIS SET, #: which is the mechanism the note above says survives the next table that is not an append table. APPEND_TABLES = frozenset({IG_SNAPSHOTS_TABLE, IG_POST_SNAPSHOTS_TABLE, TT_SNAPSHOTS_TABLE, TT_POST_SNAPSHOTS_TABLE}) #: ⭐ WAVE 24 (owner ruling R6) — `plain` IS WHAT AN AUTOMATION IS NOW, and it is the DEFAULT. #: The create wizard is deleted, so nobody picks a kind any more: a new automation is a trigger #: plus actions and NO machine step. The other three are MACHINE kinds — a scrape, an Instagram #: column, an Instagram search — and they survive on the automations that already use them #: (`discover_instagram` stays reachable through the `ig_profile_match` trigger; see C-TRIG). #: #: ⚠ ADDING A KIND HERE IS THE SMALL HALF, and the reason is worth reading before adding a fifth: #: TWO dispatches in this file used to end in a bare `else` that belonged to a SPECIFIC kind #: rather than to a default — `graph()`'s was `scrape_db`'s and `compose_sentence`'s was #: `discover_instagram`'s. A kind added here alone would have inherited another kind's whole #: description: three machine nodes it does not have, and a one-line summary about searching #: Instagram for up to 0 profiles. Both are explicit arms now, and neither has an `else`. #: ⭐ WAVE 29 (D-9 / R1) — `discover_tiktok` joins, reachable ONLY through the #: `tiktok_profile_match` trigger (the same law that makes `discover_instagram` reachable). It #: lands here IN THE SAME CHANGE as its `RUNNERS` entry and its flip law: a kind in this tuple #: with no runner is a control that must refuse. KINDS = ("plain", "scrape_db", "field_instagram", "discover_instagram", "discover_tiktok") #: ⭐⭐ WAVE 30 · T04 — THE DISCOVERY KINDS UNDER ONE NAME, and this constant is a bug fix rather #: than tidying. `discover_tiktok` shipped in wave 29 by being added to `KINDS`, `RUNNERS`, #: `clean_config` and `TRIGGER_*` — four sites that were found — while FIVE more tested the string #: `"discover_instagram"` directly and were not. The visible symptom was the owner's: picking the #: TikTok trigger 400'd with *"say how many profiles to fetch"* on the FIRST save, because the seed #: below was one of the five. The other four are silent — a canvas with no `find` panel, a flow #: table resolving to the wrong default, and a summary sentence announcing the automation would #: "do nothing yet". #: #: ⛔ SO THE RULE IS: A DISCOVERY BRANCH TESTS MEMBERSHIP OF THIS TUPLE, NEVER A KIND STRING. #: Three parallel string comparisons is precisely how the third platform gets missed twice more, #: and the misses are individually invisible — each one degrades a different surface, none of them #: raises, and every gate stays green (this whole family shipped green in wave 29). #: ⚠ The kind ↔ platform facts that genuinely DIFFER — the dataset id, the default target table, #: the field map — stay resolved per kind where they are used. This tuple answers *"is this a #: corpus search?"* and nothing else; widening it into a platform registry would just move the #: problem somewhere with a longer name. DISCOVERY_KINDS = ("discover_instagram", "discover_tiktok") #: ⭐ WAVE 30 · T06 — THE TRIGGER HALF, and it is a SEPARATE map because the two are NOT #: interchangeable, which cost a red to learn. Law 1 makes a discovery TRIGGER choose its kind, so #: `trigger ⇒ kind` always holds — but the converse does NOT: a definition can carry #: `kind: "discover_instagram"` with no trigger at all (a direct API create does exactly that, and #: `verify_automation`'s `_discover_defn` fixture is one). Testing the KIND where the rule is about #: the trigger therefore fires on definitions the picker could never produce — measured: it spawned #: a preset database under a dry-run fixture that asserts none exists. #: ⛔ SO: "did somebody PICK a corpus search in the picker?" reads THIS. "Is this stored definition a #: corpus search?" reads `DISCOVERY_KINDS`. Two questions, two maps, and the gate asserts this one #: agrees with law 1 rather than trusting that it does. DISCOVERY_TRIGGER_KIND = {"ig_profile_match": "discover_instagram", "tiktok_profile_match": "discover_tiktok"} #: ⭐ WAVE 30 · T08 — the ENRICH action kinds, one per network. Same argument as `DISCOVERY_KINDS` #: one paragraph up: these two share a validator, a selection, a cooldown and a run summary, and the #: only things that differ are which connector answers and which tables the rows land in. #: ⛔ ONE VALIDATOR, NOT TWO. `clean_actions`' enrich branch is ~40 lines of clamps whose comments #: record why each one is shaped the way it is (`submitted=False` because the client re-posts the #: whole action list; `limit` clamped twice because the run sees STORED configs). A second copy for #: TikTok would start identical and drift, and the way it fails is that one network silently accepts #: a limit the other refuses. ENRICH_KINDS = ("enrich_instagram", "enrich_tiktok") #: What a definition with no kind becomes (C-TRIG law 2) — the shape `POST /automations` stores #: when the body names none, which after R6 is every create the client makes. DEFAULT_KIND = "plain" #: R6: no NEW automation may be either of these. They still RUN, still PATCH and still validate — #: the ruling retires the door, not the two automations behind it (see `create`). #: ⚠ `discover_instagram` is NOT here: it stays creatable, through the `ig_profile_match` trigger. RETIRED_KINDS = ("scrape_db", "field_instagram") #: ⛔ DEBT D-65 — WHAT TO DO INSTEAD, said at every door that refuses one of these. The kinds were #: not deleted, they were REPLACED by actions any flow can take, and a refusal that does not name #: the replacement sends somebody looking for a bug in a decision made on purpose. One sentence #: per kind, in one place, because `create` and `clean_definition` both say it. RETIRED_KIND_REPLACEMENT = { "field_instagram": "Add the 'Enrich Instagram profile' action to any automation instead", "scrape_db": "Add a scrape step to any automation instead", } #: ⛔ D-65 — THE RETIRED KINDS THAT ACTUALLY HAVE SOMEWHERE ELSE TO GO, and the distinction is the #: whole reason this is a second, narrower tuple rather than a reuse of `RETIRED_KINDS`. #: `field_instagram` was REPLACED: W25/R4 shipped `enrich_instagram` as a `ready:true` action any #: flow can take, so refusing the kind costs a person nothing but a different click. #: ⚠ `scrape_db` IS DELIBERATELY ABSENT. Its replacement — the five `web_*` actions — is declared #: `ready:false` and is DEBT D-51, so refusing a PATCH to it would delete the only way to build a #: scrape automation and call it tidying up. W24/R6 retired its CREATE door and kept the patch #: door open on purpose, and there is a gate check that says so in those words. A kind may only be #: walled off at a door once something else answers the same need. #: (`REPLACED_KINDS` was here and is deleted with the refusal it gated — see `clean_definition`.) KIND_LABELS = {"plain": "Automation", "scrape_db": "Web page to database", "field_instagram": "Instagram profile column", "discover_instagram": "Find Instagram profiles", "discover_tiktok": "Find TikTok profiles"} EXTRACTS = ("table", "jsonld") STATES = ("idle", "running", "ok", "error", "partial") #: Which capture rung a `field_instagram` automation is allowed to reach for (R1's hybrid). #: `anonymous` = the $0 ladder only. `brightdata` = the paid rung FIRST, then the ladder as a #: fallback (unless the fallback is switched off — see `graph`'s `fallback` toggle). TIERS = ("anonymous", "brightdata") #: ⚠ STORED CONFIGS SAY `hiker`, AND THEY MEAN "THE PAID RUNG" (wave-20 D-21). The vendor swap #: must not silently answer that request with the free ladder: `clean_config` refuses an unknown #: tier by falling back to `anonymous`, so without this alias every existing Instagram automation #: would quietly stop reaching for exact counts and nothing would say so. Mapping FORWARD keeps #: the user's expressed intent (they turned the paid step ON) at a cost of ~$0.0015 a profile. TIER_ALIASES = {"hiker": "brightdata"} def clean_tier(raw): """A stored/posted tier → a tier this engine runs, or '' when it is neither.""" t = str(raw or "").strip().lower() t = TIER_ALIASES.get(t, t) return t if t in TIERS else "" def row_cap(table_key): """The row ceiling for ONE table. Per-table rather than global — see `MAX_UT_IG_ROWS`. An APPEND table (one row per subject per pull, forever) gets the raised ceiling; everything else — including the `ut_ig_` table that is an UPSERT — keeps the list-table one. """ return MAX_UT_IG_ROWS if str(table_key or "") in APPEND_TABLES else MAX_UT_ROWS #: Schedule presets the editor offers. Kept here (not in the client) so the vocabulary the UI #: shows and the vocabulary the parser accepts cannot drift. CRON_PRESETS = [ {"cron": "*/15 * * * *", "label": "Every 15 minutes"}, {"cron": "0 * * * *", "label": "Hourly"}, {"cron": "0 6 * * *", "label": "Daily at 06:00"}, {"cron": "0 6 * * 1", "label": "Weekly (Monday 06:00)"}, {"cron": "0 6 1 * *", "label": "Monthly (1st, 06:00)"}, ] UA = "Mozilla/5.0 (compatible; AIOS-automation/1.0; +https://aios.local/automation)" def _now(): return _dt.datetime.now() def _stamp(dt=None): return (dt or _now()).strftime("%Y-%m-%d %H:%M") def _iso(dt=None): """An ISO stamp **WITH its UTC offset** — `2026-08-05T14:03:11+07:00` (DEBT D-18). ⚠ THE OFFSET IS NOT COSMETIC. These stamps are the time axis of the `ut_ig_*` append tables and they are rendered by a browser, which can only subtract from an instant it can LOCATE. A naive `2026-08-05T14:03:11` is read as the *reader's* local time, so a container running UTC minted cells a Jakarta browser would place seven hours in the future — and "2 minutes ago" is not expressible at all. With the offset the same string is an instant, and relative rendering becomes possible without migrating a single stored row. ⚠ `_parse_iso` reads it BACK as naive local ON PURPOSE. Every cron / `is_due` comparison in this module is against a naive `_now()`, and mixing aware and naive datetimes raises `TypeError` — so the offset rides on the WIRE and never enters the arithmetic. """ dt = dt or _now() if dt.tzinfo is None: dt = dt.astimezone() # a naive stamp from this process IS local time return dt.isoformat(timespec="seconds") #: ⭐ WAVE 26 · R3 — the DAY out of any stamp we have ever written, for the `date`-typed columns. #: #: ⛔ IT MUST READ EVERY SHAPE THE STORE HOLDS, which is the same trap `_parse_iso` documents one #: function down: `first_found` cells exist as post-D-18 offset stamps #: (`2026-08-05T14:03:11+07:00`), as pre-D-18 naive ones (`2026-08-05T14:03:11`), as `_stamp()`'s #: space-separated minute form (`2026-08-05 14:03`) and, after this wave, as bare days. A #: converter that understood only the newest shape would blank the oldest rows — and a migration #: that empties cells is indistinguishable from one that moved them. #: ⚠ Returns "" for anything it cannot read rather than guessing a day. The migration treats "" #: as LEAVE ALONE, never as a value to write, so an unparseable cell keeps its original text and #: shows up as itself instead of disappearing. def _day(s): """`2026-08-05T14:03:11+07:00` → `2026-08-05`. "" when there is no day in there.""" raw = str(s or "").strip() if not raw: return "" head = raw.replace("T", " ").split(" ")[0] try: _dt.date.fromisoformat(head) except ValueError: dt = _parse_iso(raw) return dt.strftime("%Y-%m-%d") if dt else "" return head #: ⭐ WAVE 26 · AMENDMENT C1-a — the vendor's 0–1 engagement fraction → this product's 0–100 `pct`. #: #: MEASURED on corpus rows: `0.0074`, `0.0656`, `0.0014`, `0.0148`, `0.0274`, `0.0091`. Our `pct` #: renderer prints the stored number and appends `%`, so storing the raw fraction would show every #: creator in the book as `0.0%` — a measurement replaced by a wrong measurement, which is worse #: than the blank cell it came from ([[analyst-chart-library]]: this repo already carries a 0–1 #: dialect and a 0–100 dialect, and they meet here). #: ⚠ BLANK STAYS BLANK. `""` means the vendor did not send one — both scrape probe rows were null #: — and `0.0` would claim we measured zero engagement. def _pct100(v): """A 0–1 engagement fraction → a 0–100 percentage. "" when there is nothing to convert.""" if v is None or (isinstance(v, str) and not v.strip()): return "" try: return _s(round(float(v) * 100.0, 4)) except (TypeError, ValueError): return "" def _parse_iso(s): """A stamp → a NAIVE LOCAL datetime, with or without an offset. Both shapes exist in the store simultaneously and always will: every run committed before D-18 wrote a naive stamp, and nothing rewrites history. A parser that understood only the new shape would silently return None for them — and `is_due` reads None as "never ran", which would re-fire every schedule once. Reading both is what makes the change additive. """ raw = str(s or "").strip().replace("Z", "+00:00") dt = None try: dt = _dt.datetime.fromisoformat(raw) except Exception: # noqa: BLE001 try: dt = _dt.datetime.strptime(raw[:19], "%Y-%m-%dT%H:%M:%S") except Exception: # noqa: BLE001 return None return dt.astimezone().replace(tzinfo=None) if dt.tzinfo is not None else dt # --------------------------------------------------------------------------------------------- # THE SSRF RAIL — vendored from scrape.py `guard`, then hardened for a SERVER-SIDE fetcher # --------------------------------------------------------------------------------------------- # The skill version guards the URL a developer typed. This one guards a URL that arrived in a # request body, which changes the threat model in two ways the original does not cover: # # 1. REDIRECTS. `httpx.Client(follow_redirects=True)` / `requests.get(allow_redirects=True)` # never re-enter the guard, so `http://evil.example/x` → 302 → `http://169.254.169.254/…` # sails straight past a guard that only ever saw the first URL. The negative control ("a # localhost URL is refused") would still pass while the rail was wide open. `fetch` below # therefore takes the hops MANUALLY and re-guards every one. # 2. DNS. A perfectly public hostname may resolve to a private address. `guard` resolves the # host and checks EVERY answer, not just the literal. # # ⚠ STATED, NOT HIDDEN: this is check-then-connect, so a DNS-rebinding attacker who flips the # record between the guard and the socket is not stopped by it. Closing that needs a pinned # connection to the validated IP (a custom adapter). Out of scope for v1 and recorded here rather # than implied away — the rail refuses the realistic cases and says what it does not cover. class Refused(ValueError): """The rail refused a URL. A distinct type so a refusal is never logged as a fetch error.""" def _ip_public(ip): return not (ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved or ip.is_unspecified or ip.is_multicast) def guard(url, allowed=()): """PUBLIC-WEB-ONLY rail: http(s) only, no loopback/private/link-local host, DNS answers checked too, plus the optional `--allowed` domain rail scrape.py carries.""" p = urlparse(url or "") if p.scheme not in ("http", "https"): raise Refused(f"only http(s) allowed, not {p.scheme!r}") host = (p.hostname or "").strip() if not host: raise Refused("no host in the URL") low = host.lower() if (low in ("localhost", "localhost.localdomain") or low.endswith(".local") or low.endswith(".internal") or low.endswith(".localhost")): raise Refused(f"refused non-public host {host!r} (public web only)") try: # a bare-IP host is decided on the literal ip = ipaddress.ip_address(low) except ValueError: ip = None if ip is not None and not _ip_public(ip): raise Refused(f"refused non-public IP {host} (public web only)") if allowed and not any(low == d.lower() or low.endswith("." + d.lower()) for d in allowed): raise Refused(f"host {host!r} not in the allowed domains {list(allowed)}") if ip is None: try: infos = socket.getaddrinfo(host, None) except Exception as e: raise Refused(f"could not resolve {host!r}: {type(e).__name__}") for info in infos: addr = info[4][0] try: resolved = ipaddress.ip_address(addr.split("%")[0]) except ValueError: continue if not _ip_public(resolved): raise Refused( f"refused {host!r}: it resolves to the non-public address {resolved}") return True def fetch(url, timeout=20.0, max_kb=2048, allowed=(), max_hops=5, headers=None): """GET `url`, taking redirects BY HAND so every hop passes `guard`. Returns `(status, final_url, body_bytes)`. A hop chain longer than `max_hops` is a refusal, not a silent truncation — an endless redirect is indistinguishable from an attempt to walk the fetcher somewhere it was told not to go. """ current = url hdrs = {"User-Agent": UA, "Accept-Language": "en-US,en;q=0.9"} hdrs.update(headers or {}) for _hop in range(max_hops + 1): guard(current, allowed) r = requests.get(current, timeout=timeout, headers=hdrs, allow_redirects=False, stream=True) if r.status_code in (301, 302, 303, 307, 308): loc = r.headers.get("Location") r.close() if not loc: raise Refused(f"redirect {r.status_code} with no Location header") current = urljoin(current, loc) continue body = r.raw.read(max_kb * 1024, decode_content=True) or b"" status, final = r.status_code, str(r.url) r.close() return status, final, body raise Refused(f"more than {max_hops} redirects — refusing to follow further") def fetch_json(url, body, timeout=60.0, max_kb=8192, headers=None): """POST a JSON body through the SAME guard. Returns `(status, body_bytes)`. ⛔ NO REDIRECT FOLLOWING, AND THAT IS THE WHOLE DIFFERENCE FROM `fetch`. `fetch` walks hops by hand because a redirected GET is ordinary. A redirected POST is not: re-sending a credential-bearing body to a Location the *server* chose is precisely the hop the SSRF rail exists to refuse, and `requests`' own `allow_redirects=True` would do it without ever re-entering `guard`. So a 3xx here is a REFUSAL with the reason on it, never a second request. (The vendor calls below are the only POSTs this module makes, and they carry the API key.) """ guard(url) hdrs = {"User-Agent": UA, "Content-Type": "application/json", "Accept": "application/json"} hdrs.update(headers or {}) r = requests.post(url, timeout=timeout, headers=hdrs, allow_redirects=False, data=json.dumps(body if body is not None else {}), stream=True) try: if r.status_code in (301, 302, 303, 307, 308): raise Refused(f"the POST answered {r.status_code} — a redirected POST is refused, " f"never re-sent to a location the server picked") return r.status_code, (r.raw.read(max_kb * 1024, decode_content=True) or b"") finally: r.close() # --------------------------------------------------------------------------------------------- # EXTRACTION — vendored from scrape.py, unchanged in behaviour # --------------------------------------------------------------------------------------------- def _soup(body): if BeautifulSoup is None: raise RuntimeError( "beautifulsoup4 is not installed in this environment — the automation engine's " "extraction half needs it (see the wave-18 mailbox: add beautifulsoup4 + lxml to " "aios-web/requirements.txt).") try: return BeautifulSoup(body, "lxml") except Exception: return BeautifulSoup(body, "html.parser") def tables(soup): """Every HTML table as rows. Dict rows when the first row looks like a header.""" out = [] for t in soup.find_all("table"): rows = [] for tr in t.find_all("tr"): cells = [c.get_text(" ", strip=True) for c in tr.find_all(["th", "td"])] if cells: rows.append(cells) if not rows: continue head, body = rows[0], rows[1:] if body and len(head) == len(body[0]) and all(head): out.append([dict(zip(head, r)) for r in body if len(r) == len(head)]) else: out.append(rows) return out def jsonld(soup): out = [] for s in soup.find_all("script", attrs={"type": "application/ld+json"}): try: out.append(json.loads(s.string or s.get_text())) except Exception: pass return out def meta(soup): m = {} if soup.title and soup.title.string: m["title"] = soup.title.string.strip() for tag in soup.find_all("meta"): k = tag.get("name") or tag.get("property") v = tag.get("content") if k and v and (k in ("description", "keywords", "author") or k.startswith(("og:", "twitter:"))): m[k] = v.strip() return m def preview(url, extract="table", table_index=0, allowed=()): """The field-map preview the editor calls BEFORE anything is created: what columns does this page actually offer, and what do the first rows look like? Never writes.""" status, final, body = fetch(url, allowed=allowed) if not (200 <= status < 300): return {"ok": False, "status": status, "url": final, "note": f"the page answered {status} — it may be bot-gated or gone", "columns": [], "sample": [], "rowCount": 0} soup = _soup(body) rows = [] if extract == "jsonld": blocks = jsonld(soup) flat = [] for b in blocks: if isinstance(b, list): flat.extend([x for x in b if isinstance(x, dict)]) elif isinstance(b, dict): items = b.get("itemListElement") flat.extend([x for x in items if isinstance(x, dict)] if isinstance(items, list) else [b]) rows = [{k: _scalar(v) for k, v in d.items()} for d in flat] else: found = tables(soup) idx = max(0, min(int(table_index or 0), len(found) - 1)) if found else 0 picked = found[idx] if found else [] rows = [r for r in picked if isinstance(r, dict)] cols, seen = [], set() for r in rows[:50]: for k in r: if k not in seen: seen.add(k) cols.append(k) return {"ok": True, "status": status, "url": final, "columns": cols, "sample": rows[:8], "rowCount": len(rows), "tableCount": len(tables(soup)) if extract != "jsonld" else 0, "title": (meta(soup) or {}).get("title", "")} def _scalar(v): if isinstance(v, (str, int, float)) and not isinstance(v, bool): return str(v) if isinstance(v, bool): return "1" if v else "" if isinstance(v, dict): return str(v.get("name") or v.get("@id") or "") if isinstance(v, list): return ", ".join(_scalar(x) for x in v[:8]) return "" # --------------------------------------------------------------------------------------------- # CRON — a 5-field parser and a "what was the last fire time" walk # --------------------------------------------------------------------------------------------- # Deliberately NOT `croniter` (a dependency for ~60 lines) and deliberately NOT a forward # scheduler. The question a tick asks is backwards-looking — *"was there a scheduled minute # between the last run and now?"* — and answering it that way is what makes a missed tick # self-healing: a container that was asleep for two hours runs once on wake, not eleven times and # not never. _FIELD_RANGES = [(0, 59), (0, 23), (1, 31), (1, 12), (0, 6)] def _parse_field(spec, lo, hi): out = set() for part in str(spec).split(","): part = part.strip() if not part: raise ValueError("empty cron field part") step = 1 if "/" in part: part, _, s = part.partition("/") step = int(s) if step < 1: raise ValueError("cron step must be >= 1") if part in ("*", "?"): a, b = lo, hi elif "-" in part.lstrip("-"): a_s, _, b_s = part.partition("-") a, b = int(a_s), int(b_s) else: a = b = int(part) if a < lo or b > hi or a > b: raise ValueError(f"cron value {part!r} out of range {lo}..{hi}") out.update(range(a, b + 1, step)) return out def parse_cron(expr): """`'m h dom mon dow'` → `(minutes, hours, doms, months, dows, dom_restricted, dow_restricted)`. Raises ValueError on anything malformed — a schedule that cannot be parsed must be refused at WRITE time, because a cron string nobody can evaluate is an automation that silently never runs and reports no error anywhere. """ parts = str(expr or "").split() if len(parts) != 5: raise ValueError("a cron schedule has exactly 5 fields: minute hour day month weekday") sets = [_parse_field(p, lo, hi) for p, (lo, hi) in zip(parts, _FIELD_RANGES)] dom_r = parts[2].strip() not in ("*", "?") dow_r = parts[4].strip() not in ("*", "?") return (sets[0], sets[1], sets[2], sets[3], sets[4] | ({0} if 7 in sets[4] else set()), dom_r, dow_r) def _day_matches(day, doms, months, dows, dom_r, dow_r): if day.month not in months: return False # POSIX rule: with BOTH day-of-month and weekday restricted the match is the UNION, not the # intersection. `0 6 1 * 1` means "the 1st, and every Monday" — getting this backwards makes # a schedule that looks right fire almost never. dow = (day.weekday() + 1) % 7 # python Mon=0 -> cron Sun=0 if dom_r and dow_r: return day.day in doms or dow in dows if dom_r: return day.day in doms if dow_r: return dow in dows return True def prev_fire(expr, now=None, lookback_days=400): """The most recent scheduled minute at or before `now`, or None inside the lookback. Walks by DAY (≤400 iterations) rather than by minute (≥500k) — the day fields decide first, and only a matching day needs its hours/minutes searched. """ minutes, hours, doms, months, dows, dom_r, dow_r = parse_cron(expr) now = (now or _now()).replace(second=0, microsecond=0) for back in range(lookback_days + 1): day = (now - _dt.timedelta(days=back)).date() if not _day_matches(day, doms, months, dows, dom_r, dow_r): continue same_day = back == 0 for h in sorted(hours, reverse=True): if same_day and h > now.hour: continue for m in sorted(minutes, reverse=True): if same_day and h == now.hour and m > now.minute: continue return _dt.datetime(day.year, day.month, day.day, h, m) return None def is_due(defn, now=None): """Should the scheduler run this automation right now? Due when a scheduled minute exists strictly AFTER the reference point (the last run, else the moment the schedule was enabled, else creation) and at or before now. Anchoring on `enabledAt` rather than firing on the first tick is what stops "enable a daily 06:00 job at 14:00" from running immediately and looking like a bug. """ if not isinstance(defn, dict): return False sched = defn.get("schedule") or {} if not sched.get("enabled"): return False try: fire = prev_fire(sched.get("cron"), now=now) except ValueError: return False if fire is None: return False since = (_parse_iso((defn.get("status") or {}).get("lastRunAt")) or _parse_iso(sched.get("enabledAt")) or _parse_iso(defn.get("created"))) return since is None or fire > since def next_fire(expr, now=None, lookahead_days=400): """The next scheduled minute strictly after `now` — display only (the rail is `is_due`).""" try: minutes, hours, doms, months, dows, dom_r, dow_r = parse_cron(expr) except ValueError: return None now = (now or _now()).replace(second=0, microsecond=0) for ahead in range(lookahead_days + 1): day = (now + _dt.timedelta(days=ahead)).date() if not _day_matches(day, doms, months, dows, dom_r, dow_r): continue same_day = ahead == 0 for h in sorted(hours): if same_day and h < now.hour: continue for m in sorted(minutes): if same_day and h == now.hour and m <= now.minute: continue return _dt.datetime(day.year, day.month, day.day, h, m) return None # --------------------------------------------------------------------------------------------- # THE UPSERT — pure, so the arithmetic is testable without a store or a network # --------------------------------------------------------------------------------------------- def upsert_rows(existing, incoming, key_field, cap=None): """Merge scraped rows into a user table's rows BY KEY. Returns `(rows, counts)`. THE RULE THAT MATTERS: **an orphan is COUNTED, NEVER DELETED.** A row that has stopped appearing on the source page has not necessarily stopped existing — the page changed its filter, the fetch was partial, the site paginated. Deleting on absence turns any upstream hiccup into silent data loss, and the row may be carrying user-typed overlay values in columns the scrape never touches. So the run reports `orphans: N` and leaves them alone. Only MAPPED keys are written: a re-run never clobbers a column a user added by hand. ⚠ CALL THIS ONCE PER TABLE PER RUN, NOT ONCE PER ROW. It rebuilds the whole row dict on entry, so it is O(existing) per call — fine once, quadratic in a loop. The IG runner used to call it per POST, which was survivable only because the cap was 5000; against `MAX_UT_IG_ROWS` that same loop is hundreds of millions of dict copies and the automation simply never finishes. Raising a cap and batching the writer are ONE change, not two. (W19-C.) ⚠ `capped` IS ITS OWN COUNT, deliberately not folded into `skipped`. They are different facts: `skipped` means "this row had no key, so it could not be upserted" — a property of the DATA, and usually benign. `capped` means "this table is full and the run is now losing rows" — a property of the SYSTEM, and never benign. One number for both meant a table hitting its ceiling was indistinguishable from a page with a few blank cells, which is how a time series stops silently. The runners turn any `capped` into a `partial` run that NAMES the table. """ cap = MAX_UT_ROWS if cap is None else int(cap) rows = {str(k): dict(v or {}) for k, v in (existing or {}).items()} counts = {"inserted": 0, "updated": 0, "unchanged": 0, "skipped": 0, "duplicates": 0, "orphans": 0, "capped": 0} by_key, dupe_ids = {}, set() for rid, row in rows.items(): kv = str(row.get(key_field, "") or "").strip() if not kv: continue if kv in by_key: dupe_ids.add(rid) # a pre-existing duplicate: first id wins, second left continue by_key[kv] = rid next_id = max((int(r) for r in rows if str(r).isdigit()), default=0) + 1 seen_keys, incoming_dupes = set(), 0 for src in incoming or []: kv = str((src or {}).get(key_field, "") or "").strip() if not kv: counts["skipped"] += 1 # no key -> cannot be upserted; never guessed continue if kv in seen_keys: incoming_dupes += 1 # the SOURCE listed it twice; first wins continue seen_keys.add(kv) rid = by_key.get(kv) if rid is None: if len(rows) >= cap: counts["capped"] += 1 # LOUD: the runner turns this into a partial run continue rid = str(next_id) next_id += 1 rows[rid] = dict(src) by_key[kv] = rid counts["inserted"] += 1 continue before = rows[rid] changed = {k: v for k, v in src.items() if str(before.get(k, "")) != str(v)} if changed: before.update(src) counts["updated"] += 1 else: counts["unchanged"] += 1 counts["duplicates"] = incoming_dupes + len(dupe_ids) counts["orphans"] = sum( 1 for kv, rid in by_key.items() if kv not in seen_keys and rid not in dupe_ids) return rows, counts def dedupe_canonical_rows(existing, key_field, newest_by=""): """Collapse duplicate logical rows while preserving the lowest stable row id. Canonical entity tables use this before every upsert. Snapshot tables deliberately do not: repeated shortcodes there are new timestamped observations, not duplicate posts. Values are taken newest-first and then filled from older rows, so a sparse fresh projection does not erase a field an earlier row knew. """ rows = {str(k): dict(v or {}) for k, v in (existing or {}).items()} groups = {} for rid, row in rows.items(): identity = str(row.get(key_field) or "").strip() if identity: groups.setdefault(identity, []).append((rid, row)) removed = 0 for members in groups.values(): if len(members) < 2: continue keep = min((rid for rid, _row in members), key=lambda r: (not r.isdigit(), int(r) if r.isdigit() else r)) ordered = sorted(members, key=lambda item: str(item[1].get(newest_by) or ""), reverse=True) \ if newest_by else members merged = {} for _rid, row in ordered: for key, value in row.items(): if key not in merged or str(merged.get(key) or "").strip() == "": merged[key] = value rows[keep] = merged for rid, _row in members: if rid != keep: rows.pop(rid, None) removed += 1 return rows, removed # --------------------------------------------------------------------------------------------- # USER TABLES — read/write through the RUNTIME (never core.user_tables' module-global key) # --------------------------------------------------------------------------------------------- def _ut_slug(label): s = re.sub(r"[^a-z0-9]+", "_", str(label or "").strip().lower()).strip("_") return (s or "table")[:40] def ut_all(rt): try: return dict(rt.get(UT_STORE_KEY) or {}) except Exception: return {} def disable_for_table(rt, table_key, note="target database deleted"): """Wave 21 (item 6a, C3): a deleted table's automations are DISABLED loudly, never deleted. The definition survives with `schedule.enabled = False` + a `statusNote`, so the rail still shows what existed and why it stopped — silently deleting a user's automation because its target died would read as data loss. Returns the ids it touched.""" key = str(table_key or "") touched = [] def _up(cur): for aid, d in (cur or {}).items(): if isinstance(d, dict) and (d.get("config") or {}).get("targetTable") == key: sch = d.get("schedule") if not isinstance(sch, dict): sch = d["schedule"] = {} sch["enabled"] = False d["statusNote"] = note touched.append(str(aid)) return cur if key: _store_update(rt, _up, flush="sync") return touched def ut_get(rt, key): return ut_all(rt).get(str(key)) def retire_automation_stage_fields(rt, tables=None): """Delete obsolete Board-only fields and their hidden cells, never user fields. The authoritative selector is the engine's own ``automation.stageField`` / ``cyclesField`` metadata — labels such as “Stage” are ordinary user vocabulary and are not touched. Old generated timestamp/cycle cells are cleared too, including rows whose field definition was removed by an interrupted earlier migration. Re-running this migration is a no-op. ⭐ WAVE 29 (W29-T01) — ``tables`` LETS A CALLER LEND ITS SNAPSHOT. The SCAN below is O(all row-cells in the tenant) over a bucket whose documented ceiling is 35.8 MB / ~1.4 s to deep-copy (see this module's header), and `GET /automations` was paying for THREE independent copies of it per request. The scan is read-only, so borrowing the caller's copy is free; the WRITE below still goes through `rt.update`, which re-reads under the store lock, so a lent snapshot can never be the thing that gets written back. """ tables = ut_all(rt) if tables is None else tables stage_keys = set() for table in tables.values(): if not isinstance(table, dict): continue for field in table.get("fields") or []: auto = field.get("automation") if isinstance(field, dict) else None if isinstance(auto, dict) and (auto.get("stageField") or auto.get("cyclesField")): stage_keys.add(str(field.get("key") or "")) for row in (table.get("rows") or {}).values(): for key in (row or {}): text = str(key) if text.startswith("stage_auto_"): stage_keys.add(text.removesuffix("_at").removesuffix("_cycles")) stage_keys.discard("") if not stage_keys: return {"tables": 0, "fields": 0, "cells": 0} changed = {"tables": set(), "fields": 0, "cells": 0} def _up(cur): cur = cur if isinstance(cur, dict) else {} for table_key, table in cur.items(): if not isinstance(table, dict): continue kept, removed = [], set() for field in table.get("fields") or []: auto = field.get("automation") if isinstance(field, dict) else None key = str(field.get("key") or "") if isinstance(field, dict) else "" if key in stage_keys and isinstance(auto, dict) and \ (auto.get("stageField") or auto.get("cyclesField")): removed.add(key) changed["fields"] += 1 continue kept.append(field) if removed: table["fields"] = kept changed["tables"].add(str(table_key)) for row in (table.get("rows") or {}).values(): if not isinstance(row, dict): continue for stage_key in stage_keys: for key in (stage_key, stage_key + "_at", stage_key + "_cycles"): if key in row: row.pop(key, None) changed["cells"] += 1 changed["tables"].add(str(table_key)) return cur rt.update(UT_STORE_KEY, _up, flush="sync") return {"tables": len(changed["tables"]), "fields": changed["fields"], "cells": changed["cells"]} def bind_unbound_fields(rt): """C8's migration (wave 22, stated choice): existing automation bags WITHOUT a `flowId` are bound to the definition that already writes them — matched on the (targetTable, fieldKey) pair the definition carries, which is the binding W21-C2 established. A bag no definition references is DISABLED with the reason on it rather than deleted or guessed: the column was already dead (nothing runs a column no flow names), and now it says so. Returns `(bound, disabled)`; costs zero store commits when there is nothing to migrate.""" by_binding = {} for aid, d in all_definitions(rt).items(): cfg = d.get("config") or {} if cfg.get("fieldKey") and cfg.get("targetTable"): by_binding[(str(cfg["targetTable"]), str(cfg["fieldKey"]))] = str(aid) plan = {} for tk, t in ut_all(rt).items(): for f in (t.get("fields") or []): a = f.get("automation") # An already-DISABLED bag is a decision this migration made on a previous pass — # re-planning it every call would turn the once-per-process sweep into a write # per read. if isinstance(a, dict) and not a.get("flowId") and not a.get("stageField") \ and not a.get("disabled"): plan[(tk, str(f.get("key")))] = by_binding.get((tk, str(f.get("key")))) if not plan: return 0, 0 def _up(cur): cur = cur if isinstance(cur, dict) else {} for (tk, fk), aid in plan.items(): for f in ((cur.get(tk) or {}).get("fields") or []): if f.get("key") == fk and isinstance(f.get("automation"), dict): if aid: f["automation"]["flowId"] = aid else: f["automation"]["disabled"] = True f["automation"]["statusNote"] = ( "not bound to any flow — create an automation for this column " "or delete it") return cur rt.update(UT_STORE_KEY, _up, flush="sync") bound = sum(1 for v in plan.values() if v) return bound, len(plan) - bound def ut_key_for(label, key=None): """The table key a label resolves to. Factored out of `ut_ensure` so the DRY-RUN path (which must not create anything) resolves the same key by construction rather than by copying the derivation and drifting from it.""" k = str(key or (UT_PREFIX + _ut_slug(label))) return k if k.startswith(UT_PREFIX) else UT_PREFIX + k #: Machine names, mirrored from `core.user_tables.MACHINE_OWNERS`. A LOCAL literal for the same #: reason `UT_FIELD_TYPES` is one over there — this module must stay importable without dragging #: `core` into the API's boot path — and the two are held in step by a gate check. MACHINE_OWNERS = ("automation", "scheduler") def ut_ensure(rt, label, fields, username="automation", key=None, flow_tag="", record_mode="", lock_fields=False, tables=None): """Create the table if it is missing; return its key. Idempotent — a re-run of an automation that owns a table must not spawn `ut_x_2`, so the key is DERIVED from the label (or given) and an existing table with that key is adopted, not duplicated. ⭐⭐ WAVE 31 · T30 — `tables` LETS A CALLER LEND THE SNAPSHOT IT IS ALREADY HOLDING, and it is the SKIP TEST below that it pays for. `rt.get` deep-copies the whole tenant document on every call (documented ceiling 35.8 MB / ~1.4 s), so a caller that ensures FOUR tables in one pass paid four copies to answer four questions about one document. Owner item 7, verbatim: *"It still takes forever to change the Config for automation as well. It just say Saving..."* — measured **7,912 ms** for one save on `4258a93`. ⚠ THE LEND IS READ-ONLY AND SAFE BY INSPECTION, not by hope — the same argument `retire_automation_stage_fields` (this module, same parameter name, same contract) already makes: only the `have` lookup below reads it, and the WRITE still goes through `rt.update`, whose `_up` re-reads the live document under the store lock. **A lent snapshot can therefore never be the thing that gets written back**, and the worst a stale one can do is spend a commit that would have been skipped — never write a wrong value. Callers that ensure two tables under the SAME key in one pass pass `tables=None` for the second (see `_spawn_presets`). Fields are MERGED, never replaced: a user who added a column to an automation's table keeps it, and a new source column joins on the next run. `flow_tag` (wave 22, C7/item 5): every field THIS call adds is stamped `automation: {flowId: }` — the pre-set columns an IG automation spawns carry their provenance. Fields already on the table keep whatever tag they have (first flow wins; shared tables like ut_ig_snapshots are fed by many flows and the tag is provenance, not ownership). ⛔ AND IT STAMPS A HUMAN OWNER, WHICH IT DID NOT (wave 20, item 3). `createdBy` was whoever or WHATEVER ran the automation, so the same table belonged to a person if its first run was manual and to `"scheduler"` if the schedule got there first — and `user_tables.may_open` admits only the creator or an admin, so **whether you could open your own Instagram snapshots depended on a race you never saw**. The owner is now the automation's creator, and a table already stamped with a machine name is ADOPTED the next time a run knows a human one. Adoption is a repair, not a widening: the automation's creator is the person who asked for the table in the first place. """ key = ut_key_for(label, key) # ⭐ WAVE 26 — THE MIGRATION RIDES THE WRITE PATH, and that placement is the point. # # `ut_ensure` MERGES fields by key and never re-types an existing column, so on its own it # would leave every table already in production on the old `text` schema forever while new # ones got the honest types — the split schema R3's note warned about, arriving through the # very function the note was written on. Migrating here means a table is brought forward # immediately BEFORE anything appends to it, so no caller has to remember anything and no # tenant is left behind by a script nobody ran. # ⚠ Cheap by construction: `migrate_ig_tables` returns without a write when the table is # already current, which after the first run is every call. if key and {f.get("key") for f in (fields or [])} & set(PRESET_PROFILE_KEYS): try: migrate_ig_tables(rt, log=lambda *_a: None, only=key) except Exception: # noqa: BLE001 # A migration that cannot run must not stop the automation from writing its rows. # The old schema still reads; a refused write loses the pull we just paid for. pass wanted = [] for raw_field in (fields or []): field = dict(raw_field) if flow_tag or lock_fields: automation = dict(field.get("automation") or {}) if flow_tag: automation.setdefault("flowId", str(flow_tag)) if lock_fields: automation["preset"] = True field["automation"] = automation wanted.append(field) created = _iso() human = username if username and username not in MACHINE_OWNERS else "" # ⚠ SKIP THE WRITE WHEN NOTHING WOULD CHANGE. Without this, every re-run spends a store # commit re-writing an identical definition — against a 20 s flush floor and a 256/hr repo # budget, an idempotent helper that always writes is the same defect as a per-row insert. have = ut_get(rt, key) if tables is None else tables.get(str(key)) have_fields = {str(f.get("key") or ""): f for f in ((have or {}).get("fields") or [])} missing = [f for f in wanted if f.get("key") not in have_fields] missing_locks = [f for f in wanted if lock_fields and f.get("key") in have_fields and (not isinstance(have_fields[f.get("key")].get("automation"), dict) or have_fields[f.get("key")]["automation"].get("preset") is not True)] if (have is not None and not missing and not missing_locks and not (record_mode and have.get("recordMode") != record_mode) and not (human and (have.get("createdBy") or "") in MACHINE_OWNERS)): return key def _up(cur): cur = cur if isinstance(cur, dict) else {} t = cur.get(key) if t is None: if len(cur) >= MAX_UT_TABLES: return cur cur[key] = {"key": key, "label": str(label)[:60], "source": "Automation", "createdBy": username, "created": created, "fields": wanted, "rows": {}} if record_mode: cur[key]["recordMode"] = record_mode return cur have = {f.get("key") for f in (t.get("fields") or [])} for f in wanted: if f.get("key") not in have: t.setdefault("fields", []).append(f) have.add(f.get("key")) elif lock_fields: stored = next((g for g in (t.get("fields") or []) if g.get("key") == f.get("key")), None) if stored is not None: automation = dict(stored.get("automation") or {}) if flow_tag: automation.setdefault("flowId", str(flow_tag)) automation["preset"] = True stored["automation"] = automation # ADOPTION: a machine name is not an owner. It never overwrites a human one. if human and (t.get("createdBy") or "") in MACHINE_OWNERS: t["createdBy"] = human # An automation's table SAYS an automation owns it — the nav badge reads from this. t["source"] = t.get("source") or "Automation" if record_mode: t["recordMode"] = record_mode return cur rt.update(UT_STORE_KEY, _up, flush="sync") return key def ut_write_rows(rt, key, rows): """ONE store update for the WHOLE row set — the flush-ceiling rule (see the module header).""" def _up(cur): cur = cur if isinstance(cur, dict) else {} t = cur.get(key) if t is not None: t["rows"] = rows return cur rt.update(UT_STORE_KEY, _up, flush="sync") def ut_set_cell(rt, key, row_id, field_key, value, extra_rows=None): """Write one automation-owned cell (+ optional whole extra tables) in ONE update.""" def _up(cur): cur = cur if isinstance(cur, dict) else {} t = cur.get(key) if t is not None: t.setdefault("rows", {}).setdefault(str(row_id), {})[field_key] = value for k, rws in (extra_rows or {}).items(): tt = cur.get(k) if tt is not None: tt["rows"] = rws return cur rt.update(UT_STORE_KEY, _up, flush="sync") IG_FIELD_DESCRIPTIONS = { "alt_text": "Accessibility text attached to the Instagram post.", # ⚠ WIDENED 2026-08-10, because the column gained a second writer and the old sentence would # have made it lie. It was written for the anonymous rung, whose numbers are genuinely ROUNDED # ("10.4K followers"). A DISCOVERY observation is exact-looking and STALE instead — read off # the vendor's pre-collected corpus at a collection time we are not told. Both mean the same # thing to a reader ("do not treat this as an exact count taken at Pulled at"), and a # description that named only the first would have quietly excluded the second. "approx": "Checked when the counts are rounded, or were read from a pre-collected corpus " "rather than measured at the time shown.", "avg_comments_12": "Average comments across the 12 most recent captured posts.", "avg_engagement": "Average engagement rate reported for the profile.", "avg_likes_12": "Average likes across the 12 most recent captured posts.", "avg_plays_12": "Average plays across the 12 most recent captured posts.", "avg_views_12": "Average views across the 12 most recent captured posts.", "views": "View count for the post, as Instagram displays it.", "video_duration": "Length of the video in seconds.", "comments_disabled": "Checked when the creator has turned comments off for this post.", "plays": "Times the video started playing, including replays.", "bio": "Biography shown on the Instagram profile.", "bio_hashtags": "Hashtags listed in the profile biography.", "business_category": "Instagram's business category for the account.", "caption": "Caption published with the Instagram post.", "category": "Instagram's category for the profile.", "comment_key": "Unique ID for the captured comment record.", "commented_at": "Date the comment was posted.", "comments": "Comment count reported for the post.", "comments_captured": "Number of distinct comment records linked to this profile.", "comments_link": "Comment records linked to this record.", "country_code": "Country code reported for the profile.", "created_by": "User whose automation first added this lead.", "enriched_at": "Date the profile was last enriched.", # ⭐ ITEM 16. The description says GUESS out loud, because the column is one and the number # beside it is the only thing that says how much of one. "location_guess": "Most common city tagged across this profile's captured posts - a guess, " "not a stated location.", "location_confidence": "Share of this profile's geotagged posts that agree on the guessed " "city.", "external_url": "Website linked from the Instagram biography.", "external_url_title": "Title Instagram shows for the biography link.", "fbid": "Facebook ID associated with the Instagram account.", "first_found": "Date an automation first found this profile.", "followers": "Latest reported follower count.", "following": "Latest reported number of accounts followed.", "found_count": "Number of times discovery automations found this profile.", "full_name": "Display name shown on the Instagram profile.", "handle": "Instagram username without the @ symbol.", "has_channel": "Checked when the profile has an Instagram channel.", "hashtags": "Hashtags extracted from the post caption.", "highlights_count": "Total story highlight collections reported for the profile.", "ig_id": "Instagram's internal ID for the account.", "influencer_key": "Normalized handle linking this row to its profile.", "is_business": "Checked when Instagram marks the account as a business.", "is_joined_recently": "Checked when Instagram marks the account as recently joined.", "is_private": "Checked when the Instagram profile is private.", "is_professional": "Checked when Instagram marks the account as professional.", "last_found": "Date an automation most recently found this profile.", "likes": "Like count reported for this record.", "measured_at": "Date engagement metrics on this post were last read.", "measurements_captured": "Number of measurement rows stored for this post.", "paid_partnership": "Checked when Instagram marks the post as a paid partnership.", "partner": "Brand named in the post's paid partnership metadata.", "partner_id": "Partner ID reported for the Instagram profile.", "platform": "Social network for this profile.", "plays": "Video plays, including repeat plays, reported for the post.", "post_link": "Post record linked to this measurement or comment.", "post_measurements_captured": "Number of distinct post measurements linked to this profile.", "post_snapshot_key": "Unique ID for this post measurement.", "post_snapshots_link": "Engagement measurement rows linked to this record.", "posted_at": "Date the Instagram post was published.", "posts_captured": "Number of distinct post records captured for this profile.", "posts_count": "Total posts reported for the Instagram profile.", "posts_link": "Post records linked to this profile.", "profile_name": "Profile name returned by the data provider.", "profile_reads": "Number of profile snapshots stored for this profile.", "profile_snapshots_link": "Profile snapshot rows linked to this profile.", "profile_url": "Direct URL to the Instagram profile.", "pronouns": "Pronouns shown on the Instagram profile.", "pulled_at": "Date this snapshot was collected.", "related_accounts": "Accounts Instagram suggests alongside this profile.", "replies": "Reply count reported for the comment.", "shortcode": "Instagram shortcode that uniquely identifies the post.", "source_payload": "Full source response for this record.", "snapshot_key": "Unique ID for this profile snapshot.", "source": "Method or provider used to read this profile.", "tagged_location": "Location tagged on the post; it is not the creator's residence.", "text": "What the comment says. The commenter's name is not stored as a column.", "type": "Instagram media type: image, video, or carousel.", "url": "Direct URL to the Instagram post.", "verified": "Checked when Instagram marks the profile as verified.", } #: ⭐ WAVE 29 (R2) — TIKTOK'S OWN DESCRIPTIONS, and the reason this exists rather than reusing the #: map above is one line of `field_def`: it defaults `description` to #: `IG_FIELD_DESCRIPTIONS.get(key)`. Nine `ut_tt_*` columns share a KEY with an Instagram column #: (`shortcode`, `type`, `url`, `verified`, `caption`, `likes`, `comments`, `replies`, #: `tagged_location`), so a TikTok video's Type column would have shipped explaining "Instagram #: media type" — a wrong sentence in the header tooltip of a column nobody would think to check. #: ⚠ A KEY ABSENT HERE GETS NO DESCRIPTION AT ALL, deliberately: silence is honest, and inheriting #: the Instagram sentence is the failure this map exists to prevent. TT_FIELD_DESCRIPTIONS = { "platform": "Network this handle is on.", "handle": "TikTok @name; the unique account handle.", "full_name": "Display name shown on the TikTok profile.", "tt_id": "TikTok's own numeric id for the account.", "profile_url": "Direct URL to the TikTok profile.", "bio": "Profile biography text.", "external_url": "Link in the TikTok bio.", "verified": "Checked when TikTok marks the account as verified.", "is_private": "Checked when the account is private.", "followers": "Follower count at the time of the pull.", "following": "Accounts this profile follows.", "posts_count": "Videos published by this account.", "likes_received": "Total likes this account's videos have received.", "avg_engagement": "Average engagement rate, stored as a percentage.", "comment_engagement": "Comment engagement rate, stored as a percentage.", "like_engagement": "Like engagement rate, stored as a percentage.", "is_business": "Approximate: set when TikTok flags the account as a commerce user.", "country_code": "Two-letter country code reported for the account.", "predicted_lang": "Language TikTok predicts for this account.", "account_created_at": "When the TikTok account itself was created.", "region": "Region reported for the account.", "shortcode": "TikTok's numeric id for the video; unique per post.", "type": "TikTok post type: video or image.", "url": "Direct URL to the TikTok post.", "caption": "Post description text.", "posted_at": "When the creator published the post.", "likes": "Likes reported for the post.", "comments": "Comment count reported for the post.", "views": "Play count TikTok reports for the video.", "shares": "Times the post was shared.", "saves": "Times the post was saved to a collection.", "video_duration": "Video length in seconds.", "hashtags": "Hashtags used in the post.", "tagged_location": "Commerce location reported for the post; it is not the creator's home.", "influencer_key": "Handle of the account this record belongs to.", "comment_key": "Unique ID for this comment.", "commented_at": "When the comment was posted.", "text": "What the comment says. The commenter's name is not stored as a column.", "replies": "Reply count reported for the comment.", "snapshot_key": "Unique ID for this profile snapshot.", "post_snapshot_key": "Unique ID for this post measurement.", "pulled_at": "When this measurement was read.", "enriched_at": "When this row was last enriched.", "source": "Method or provider used to read this record.", "source_payload": "Full source response for this record.", } def field_def(text_key, label, ftype="text", **extra): """One machine-spawned column definition. ⭐ 2026-08-07 — `**extra` carries the per-field DECLARATIONS the preset set needs (`pinned`, `profile`), and it stays ONE constructor rather than growing a second for "special" fields. ⛔ Every key passed here must survive `core.user_tables._clean_field` UNCHANGED — `verify_api`'s W25-1 section pins the drift at ZERO, so a key the validator drops or rewrites turns the whole preset list red rather than failing quietly ([[default-must-pass-its-own-guard]]). ⭐ WAVE 25 — `editRole` IS DECLARED HERE, and adding it closes a bypass rather than adding a feature. `ut_ensure` writes these dicts STRAIGHT into the `user_tables` bucket, so `core.user_tables._clean_field` — the single validator every field created through the ordinary door passes — has never judged a single field this engine spawned. The difference was exactly one key: `_clean_field` emits `editRole: 'admins'` and this did not, so `_clean_field(f) != f` for every automation column in the product. ⚠ NOTHING CHANGES BEHAVIOURALLY — `may_edit_field` asks `editRole == 'everyone'`, and an absent key was already not that, so the bypass was fail-closed and therefore silent. It is stated now, and `verify_automation` asserts the equality field-by-field, so the next key `_clean_field` grows cannot go unnoticed here ([[default-must-pass-its-own-guard]]). """ description = " ".join(str(extra.pop( "description", IG_FIELD_DESCRIPTIONS.get(text_key, "")) or "").split()) out = {"key": text_key, "label": label, "type": ftype, "source": "overlay", "editRole": "admins"} if description: out["description"] = description out.update(extra) return out def tt_field_def(text_key, label, ftype="text", **extra): """One `ut_tt_*` column. `field_def` with TikTok's description map bound (wave 29, R2). ⛔ `description` IS ALWAYS SUPPLIED, even when blank, so `field_def`'s Instagram default can never be reached from here. An explicit empty string makes `field_def` omit the key, which is the same shape an undescribed column already has — the point is that the sentence a TikTok column carries is one somebody wrote about TikTok, or none at all. """ extra.setdefault("description", TT_FIELD_DESCRIPTIONS.get(text_key, "")) return field_def(text_key, label, ftype, **extra) # --------------------------------------------------------------------------------------------- # DEFINITIONS — validation and the bucket's read/write half # --------------------------------------------------------------------------------------------- def _s(v, n=200): return str(v if v is not None else "")[:n] #: ⭐ WAVE 26 · C5 / R2 — HOW MANY POSTS ONE PULL MAY KEEP, and the ceiling is the VENDOR'S. #: #: ⛔ MEASURED 2026-08-05 and recorded at `_bd_posts_count`: **a Profiles row carries the TOP 12 #: posts — a cap, not a count.** Asking for more does not fetch more; it just makes the config #: disagree with what the run can possibly do. #: ⚠ THIS REPLACES A SILENT `max(1, min(n, 200))` AT TWO CALL SITES, and the clamp was the defect #: rather than the number: a person typing 50 got a stored 50, a UI that read back 50, and twelve #: posts — with nothing anywhere saying why. A control that accepts a value it cannot honour is #: worse than one that refuses it, because the refusal is the only place the ceiling can be #: taught. So this REFUSES, and the sentence names the cap and who set it. #: #: ⭐⭐ 2026-08-09 — RAISED TO 30 (owner: *"move the max limit to 30 posts per profile"*), and the #: paragraph above needed correcting to do it honestly: **12 was OUR ROUTE'S cap, not the #: vendor's.** Re-measured the same day — the PROFILE record still returns 12 however many you ask #: for (asked 40, got 12), but the documented discover-by-url route answered `num_of_posts: 30` #: with exactly 30 rows in 90 s (23 Reels + 7 Carousels, @theresalearns). The ceiling was a #: property of the call we happened to make. #: ⛔ SO THE NUMBER MOVED AND THE CAPTURE MUST FOLLOW. Until `bd_profile_posts` is wired ahead of #: the views top-up, a `maxPosts` above 12 is honoured by the CONFIG and bounded by the PROFILE #: read at run time — the exact accept-what-you-cannot-honour shape this constant exists to #: prevent, now surviving in one place instead of two. It is recorded rather than hidden, and it #: is why `clean_post_groups` bounds a group by `maxPosts` rather than by 12. MAX_POSTS_PER_PULL = 30 DEFAULT_POSTS_PER_PULL = 10 #: ⭐⭐ THE WINDOW THE `avg_*_12` PRESET ROLLUPS AVERAGE OVER — ITS OWN CONSTANT, deliberately #: NOT `MAX_POSTS_PER_PULL`. #: #: ⛔ THEY WERE THE SAME NUMBER AND THAT WAS A LATENT BUG, caught by the gate the moment the cap #: moved: four columns are NAMED `avg_views_12` and LABELLED "Avg views · last 12 posts", so #: raising the capture cap to 30 silently made every one of them average thirty posts under a #: label that says twelve. Nobody would have looked at those columns again to check. #: ⚠ Raising the CAPTURE ceiling and redefining an EXISTING named measure are two different #: decisions, and only the first one was made. To widen the average, change this constant AND the #: four keys and labels together — a column whose meaning changes underneath its own name is #: worse than one that is merely narrow. AVG_WINDOW_POSTS = 12 #: The post kinds a group filter may name — exactly what `_bd_type` maps a vendor row onto, so a #: filter cannot ask for a category that can never match a stored row. POST_TYPES = ("video", "image", "carousel") #: ⭐ W29-T09 — the words a PERSON reads for those three keys, server-owned for the same reason #: `KIND_LABELS` and `TRIGGER_LABELS` are. The owner asked for *"the last 12 reels"*, and `video` #: is the stored key: a client that translated it locally would be a second copy of this #: vocabulary, free to drift the day a fourth kind appears or a name changes. #: ⚠ "Reels & videos", not "Reels": `_bd_type` maps every non-image, non-carousel post here, so a #: label naming only reels would over-promise on a plain video post. POST_TYPE_LABELS = {"video": "Reels & videos", "image": "Photos", "carousel": "Carousels"} def clean_post_groups(raw, max_posts): """`(groups, error)` for `config.postGroups` — "last N reels, last M images". ⭐ 2026-08-09 (owner: *"not just last 12 but by group also"*). Shape: `[{"type": "video", "limit": 12}, {"type": "image", "limit": 6}]`. ⛔ ABSENT IS OFF, and off must stay the default forever: an enrich action stored before today carries no such key, and inventing a filter for it would silently start DROPPING posts those automations have always captured. Empty list and missing are the same answer. ⚠ A MALFORMED GROUP IS REFUSED, not dropped — the opposite of `tier`/`noFallback`, and the difference is legitimate. Those are RETIRED keys that live in stored data, so refusing them would 400 old automations forever (D-65). This key is NEW: nothing stored can carry a broken one, so the only way to see one is a person typing it, and a filter that silently ignores the type you asked for is how you end up paying for reels and storing carousels. """ if raw in (None, "", [], {}): return [], None if not isinstance(raw, list): return None, "the post groups have to be a list of {type, limit} entries" out, seen = [], set() for item in raw: if not isinstance(item, dict): return None, "each post group has to be a {type, limit} entry" t = str(item.get("type") or "").strip().lower() if t not in POST_TYPES: return None, (f"'{t or 'blank'}' is not a post type — use one of " f"{', '.join(POST_TYPES)}") if t in seen: return None, f"the post groups name '{t}' twice — one limit per type" seen.add(t) try: lim = int(item.get("limit")) except (TypeError, ValueError): return None, f"how many {t} posts to keep has to be a whole number" if lim < 1: return None, f"a {t} group has to keep at least one post" # ⛔ BOUNDED BY WHAT THE RUN ACTUALLY BUYS. A group asking for 30 when the pull captures # 12 is not an error a person can act on — it is a promise the run cannot keep — so it is # refused HERE, where the number they typed is still on the screen in front of them. if lim > max_posts: return None, (f"this enrichment captures {max_posts} posts per profile, so a {t} " f"group cannot keep {lim} — raise the post count first") out.append({"type": t, "limit": lim}) return out, None def clean_max_posts(raw, default=DEFAULT_POSTS_PER_PULL, submitted=True): """`(maxPosts, error)` — refuses out-of-range rather than clamping into range. ⛔ `submitted=False` CLAMPS INSTEAD OF REFUSING, and that asymmetry is the whole reason this takes a flag. **Every automation stored before wave 26 carries `maxPosts: 24`** — the old clamp's default, which this cap now forbids. If an inherited value were refused the same way a typed one is, every one of those automations would 400 on its next Save **forever**, for a number nobody on that screen chose or can see. That is D-65's lesson exactly: a validator that refuses stored data does not protect the user from it, it locks them out of their own automation. So: a value the panel SENT is the user asking for it, and gets the sentence. A value merely INHERITED gets silently brought inside the cap it was already effectively subject to at the vendor — the run was returning 12 either way ([[default-must-pass-its-own-guard]]). """ if raw is None or (isinstance(raw, str) and not raw.strip()): return default, None try: n = int(raw) except (TypeError, ValueError): if not submitted: return default, None return None, "the post limit must be a whole number" if not submitted: return max(1, min(n, MAX_POSTS_PER_PULL)), None if n < 1: return None, "keep at least one post per profile, or turn the post capture off" if n > MAX_POSTS_PER_PULL: return None, (f"a profile pull returns at most {MAX_POSTS_PER_PULL} posts — that is the " f"vendor's cap, not ours, so {n} would store the same " f"{MAX_POSTS_PER_PULL} and read as more") return n, None def clean_config(kind, raw, previous=None): """Validate a kind's config. Returns `(config, error)`; error is a user-facing sentence. ⚠ THE NODE-SWITCH FLAGS FALL BACK TO `previous` WHEN THE KEY IS ABSENT, and that is a rail rather than a nicety. `patch` replaces the whole config, and the canvas's config panels do not edit `postMetrics` / `commentMetrics` / `dryRun` — those are node SWITCHES. So a plain "Save" from a panel that never knew about them would silently turn off the dry run, or turn ON a per-post purchase: a save that quietly changes what the automation costs. Absent ⇒ keep; present ⇒ take it, including `false`. """ raw = raw if isinstance(raw, dict) else {} prev = previous if isinstance(previous, dict) else {} def flag(name): return bool(raw[name]) if name in raw else bool(prev.get(name)) if kind == "plain": # A plain automation has no machine step, so it only needs the database its records walk. # # ⚠ THE TARGET IS OPTIONAL ON PURPOSE. An automation triggered by `record_updated` on # `ut_foo` already knows its table from the trigger — `_flow_table` resolves target-else- # trigger — so requiring a second copy of that fact here would be a mandatory field with # exactly one legal answer, and a Save that refuses until you retype what you just picked. # There is no `dryRun`: dry-run means "run the machine step but do not write", and there # is no machine step to run. target = _s(raw.get("targetTable"), 60).strip() if target and not target.startswith(UT_PREFIX): target = UT_PREFIX + target return {"targetTable": target, "targetLabel": _s(raw.get("targetLabel"), 60).strip()}, None if kind == "scrape_db": url = _s(raw.get("url"), 2000).strip() if not url: return None, "a source URL is required" try: guard(url) except Refused as e: return None, str(e) extract = raw.get("extract") if raw.get("extract") in EXTRACTS else "table" fmap = {} for k, v in list((raw.get("fieldMap") or {}).items())[:60]: tk = re.sub(r"[^a-z0-9_]+", "_", _s(v, 60).strip().lower()).strip("_") if tk: fmap[_s(k, 120)] = tk[:60] if not fmap: return None, "map at least one source column to a field" key_field = _s(raw.get("keyField"), 60).strip() if key_field not in fmap.values(): return None, "the key field must be one of the mapped fields" target = _s(raw.get("targetTable"), 60).strip() if target and not target.startswith(UT_PREFIX): target = UT_PREFIX + target return {"url": url, "extract": extract, "tableIndex": max(0, min(int(raw.get("tableIndex") or 0), 50)), "fieldMap": fmap, "keyField": key_field, "targetTable": target, "dryRun": flag("dryRun"), "targetLabel": _s(raw.get("targetLabel"), 60).strip() or "Scraped table"}, None if kind == "field_instagram": table = _s(raw.get("targetTable"), 60).strip() if not table.startswith(UT_PREFIX): return None, "an Instagram automation runs against a blank database (ut_*)" fkey = _s(raw.get("fieldKey"), 80).strip() if not fkey: return None, "pick the automation column this run writes into" # ⛔⛔ `tier` AND `noFallback` ARE ACCEPTED AND IGNORED (wave 28 / R5, contract C2). # They are read from nothing and written to nothing: a stored definition carrying either # still SAVES — it simply loses them on the next write — and neither is ever a reason to # refuse. That asymmetry is D-65's law and it is not squeamishness: refusing an unknown # key would 400 every automation a tenant stored before this wave, forever, on a screen # that gives them no way to remove it. Dropping a retired key is a migration; refusing it # is an outage. # ⚠ `clean_tier`/`TIERS`/`bd_ready` KEEP THEIR NAMES. They are VENDOR vocabulary # (`brightdata` is a provider, and `verify_automation` fences the name), not the retired # USER concept — renaming them would be a second, unrelated change wearing this one's # justification. # C5: absent ⇒ keep whatever is stored, like every other switch in this branch — a panel # that does not edit the post count must not reset it to the default on Save. And an # INHERITED value is clamped rather than refused; see `clean_max_posts`. sent = "maxPosts" in raw max_posts, perr = clean_max_posts( raw.get("maxPosts") if sent else prev.get("maxPosts"), submitted=sent) if perr: return None, perr return {"targetTable": table, "fieldKey": fkey, "urlField": _s(raw.get("urlField"), 80).strip(), # ⛔ THE ONE FLAG THAT MULTIPLIES THE BILL BY THE POST COUNT. Off unless asked # for: the Profiles row carries post IDENTITY for free but NO engagement # (measured 2026-08-05), so likes/comments cost one extra vendor record PER POST. "postMetrics": flag("postMetrics"), # The full Comments dataset can bill many rows per post. It is an explicit, # independent opt-in and never follows the per-post switch automatically. "commentMetrics": flag("commentMetrics"), "dryRun": flag("dryRun"), "maxPosts": max_posts}, None # ⭐ WAVE 29 (D-9 / R1) — BOTH discovery kinds share this branch, because they ask the vendor # the same question of two different corpora: a record ceiling, a predicate list, a join word # and where to write. ⛔ THE ONLY DIFFERENCE IS THE DEFAULT TABLE, and it is resolved from the # kind rather than hard-coded — a TikTok search falling back to `ut_ig_profile` would write # TikTok rows into the Instagram family, which is precisely what R2's parallel family exists # to prevent, and it would do it silently. if kind in DISCOVERY_KINDS: # ⭐ WAVE 30 · T05 — the inline tuple became the named one, and the two defaults below now # come from `discovery_facts` rather than being spelled out here. They were correct; they # were also the FIFTH copy of "which table does this kind write to", and the other four # were the ones wave 29 forgot to update. _, _disc_table, _disc_label, _ = discovery_facts(kind) limit = int(raw.get("recordsLimit") or 0) if limit < 1: return None, ("say how many profiles to fetch — an UNBOUNDED discovery query is the " "one shape the vendor refuses outright (NOT_ENOUGH_FUNDS)") if limit > BD_MAX_RECORDS: return None, f"a single discovery run may ask for at most {BD_MAX_RECORDS} profiles" # ⚠ THE JOIN IS RESOLVED BEFORE THE PREDICATES ARE JUDGED, because it is part of what # makes them broad or narrow (see `narrowing_refusal`). Validating them first and reading # the operator afterwards is how the OR hole survived: the guard was handed the branches # and never told they were a union. join = "or" if str(raw.get("operator") or "").lower() == "or" else "and" # ⭐ WAVE 32 · T46 (D-167) — AND THE KIND GOES WITH THEM. `clean_config` already knows which # corpus this automation searches; passing it is what stops a `discover_tiktok` being built # on the 16 Instagram fields TikTok's dataset does not carry — a search that does not error, # returns nothing, and reads as "no such creators exist" after the money is spent. preds, perr = clean_predicates(raw.get("predicates"), join, kind) if perr: return None, perr target = _s(raw.get("targetTable"), 60).strip() or _disc_table if not target.startswith(UT_PREFIX): target = UT_PREFIX + target seed, serr = clean_seed(raw.get("seed") if "seed" in raw else prev.get("seed")) if serr: return None, serr return {"recordsLimit": limit, "predicates": preds, "operator": join, "targetTable": target, "targetLabel": _s(raw.get("targetLabel"), 60).strip() or _disc_label, # C6/R5: WHERE the conditions above came from, when they were derived. Stored so # the surface can say "these were filled in from the 'Florists' view, over 12 # records" instead of presenting them as if somebody typed them. "seed": seed, "dryRun": flag("dryRun")}, None return None, f"unknown automation kind {kind!r}" def clean_seed(raw): """C6: validate `config.seed`. Returns `({}, None)` when there is none — a discovery filter somebody typed by hand has no seed, and that is the ordinary case. ⚠ `derived` AND `basis` ARE STORED AS PROVENANCE, NOT AS A SECOND FILTER. R5 is explicit that the derived conditions are "written into `config.predicates` as ordinary conditions — it is a filling-in, not a parallel filter", so the RUN never reads this bag: it reads `predicates`, like every other search. Keeping it means the surface can say where those rows came from, and a user editing them freely is exactly what is supposed to happen. """ if raw in (None, "", {}): return {}, None if not isinstance(raw, dict): return None, "the seed must be an object" source = _s(raw.get("source"), 12).strip().lower() if source not in SEED_SOURCES: return None, f"{source or 'that seed'!r} is not one of: " + ", ".join(SEED_SOURCES) table = _s(raw.get("table"), 60).strip() if table and not table.startswith(UT_PREFIX): return None, f"a seed reads a blank database (ut_*) — {table!r} is not one" basis = raw.get("basis") if isinstance(raw.get("basis"), dict) else {} derived, _err = clean_predicates(raw.get("derived") or [], "and") return {"source": source, "table": table, "id": _s(raw.get("id"), 80).strip(), # ⛔ A MALFORMED `derived` IS DROPPED, NEVER A REFUSAL. This bag is a RECORD of what # was suggested; the conditions that matter are already in `predicates` and were # validated there. Refusing a Save because a stored provenance note aged badly would # block the user from editing the very filter it describes. "derived": derived or [], "basis": {"rows": max(0, int(basis.get("rows") or 0)), "fields": [f for f in (basis.get("fields") or []) if isinstance(f, dict)][:SEED_MAX_PREDICATES], "related": basis.get("related") if isinstance(basis.get("related"), dict) else {}, "note": _s(basis.get("note"), 200)}}, None def clean_schedule(raw, previous=None): """Validate `{cron, enabled}` and stamp `enabledAt` on the OFF→ON edge (the anchor `is_due` measures from — without it, enabling a daily job fires it immediately).""" raw = raw if isinstance(raw, dict) else {} cron = _s(raw.get("cron"), 120).strip() or "0 6 * * *" parse_cron(cron) # raises -> the route answers 400 enabled = bool(raw.get("enabled")) prev = previous or {} out = {"cron": cron, "enabled": enabled} if enabled: out["enabledAt"] = (prev.get("enabledAt") if prev.get("enabled") else None) or _iso() return out # ── WAVE 23 · C5 — the ENDING, and the cycle counter that makes a loop countable. ───────────── # The owner's words: "we can make this automation a loop, so the ending should always be defined, # either it ends somewhere deterministic like Closed/Failed, or it goes to reset automatically, # or the user have to click a button to reset, or after a certain amount of time it can # automatically reset to first cycle." # # ⛔ `terminal` IS THE DEFAULT and every existing automation gets it, because a stored definition # that predates this field must not start moving records on its own the day the code ships. A # loop is a thing somebody turns on. def clean_flow(raw, previous=None): """Validate the builder's ordered action list. `(flow, error)`.""" raw = raw if isinstance(raw, dict) else {} prev = previous if isinstance(previous, dict) else {} actions, err = clean_actions(raw.get("actions") if "actions" in raw else prev.get("actions")) if err: return None, err return {"actions": actions}, None #: AMENDMENT A1 — what the `ig_profile_match` kind-flip seeds as `recordsLimit` when the #: definition carries none. #: #: ⚠ 25 BECAUSE THE INPUT SAYS 25 (2026-08-06). It was 10 — the size wave 21 proved live at #: $0.15 — while `AutomationDetail` initialises its own box to 25, and the two never met: the #: flow node read "Up to 10 profiles · about $0.025" beside a field reading 25, on a freshly #: created automation. Two numbers for one fact, in the panel, before anybody had typed #: anything. **Read off the screenshot; every assertion in the battery was green.** #: #: This is the same species as the default-versus-guard split fixed the same day: a value #: decided in one file and a value decided in another, describing the same thing. Aligning the #: constants closes the only window in which they can disagree — the seed — because every later #: state comes from the stored config. DISCOVER_SEED_RECORDS = 25 def discovery_seed_spec(kind): """-> (default table, enrich action kind) for a discovery KIND. ⭐⭐ WAVE 30, OWNER REPORT 2026-08-12, verbatim: *"'When a Tiktok profile fits a criteria' should ALWAYS have a 'Create record' EXACTLY like the Instagram one … THE ONLY DIFFERENCE IS THE COLUMNS AND DATA SCHEMA"*. They were right, and the seeding path was the one family of sites this wave widened everywhere ELSE: `clean_definition` planted step 1 and step 2 only when the trigger was `ig_profile_match`, so a TikTok search stored `flow.actions = []` and had nowhere to put what it found — MEASURED live against `auto_1` (Instagram: create_record + enrich) versus a fresh TikTok discovery (empty). ⭐ THIS IS A LOOKUP, NOT A SECOND SEEDER, and that is the whole point of the ruling. One builder plants both platforms' steps; the only things that vary are the table the record lands in and which enrich action reads it — literally "the columns and data schema". A forked `_ensure_tt_action` would start identical and drift, which is the failure `ENRICH_KINDS` and `DISCOVERY_KINDS` were introduced to prevent one screen up. ⚠ Resolved in a FUNCTION rather than a module-level dict because `DISCOVER_TABLE` is defined far below this line; a dict here would raise at import. """ if kind == "discover_tiktok": return TT_PROFILE_TABLE, "enrich_tiktok" return DISCOVER_TABLE, "enrich_instagram" def _ensure_ig_action(flow_raw, table, enrich_kind="enrich_instagram"): """Create record is PERMANENT step 1 on a DISCOVERY trigger — it cannot be deleted or moved. ⚠ THE NAME IS INSTAGRAM'S AND THE BEHAVIOUR IS BOTH PLATFORMS' (wave 30). Renaming it would churn six gate references for no behaviour change; `enrich_kind` is what makes it general, and it defaults to Instagram's so every pre-existing caller keeps its exact meaning. ⭐ OWNER RULING 2026-08-06, AND IT REVERSES WAVE 24's LAW 3. That law seeded this action once, on the edge into the trigger, and ended "deleting it is their call, not a refusal" — with a comment warning that re-seeding on every clean would be "a control that will not take no for an answer". The owner's answer: *"should ALWAYS have a 'Create record' Step 1, that can't be deleted, because the nature of that automation is that it needs to first create a record in a database somewhere from the list of profiles to fetch."* Which is right, and the earlier reasoning had the category wrong: a flow whose TRIGGER produces rows has nowhere to put them until something writes them, so an Instagram search with no Create record is not a customised automation — it is a search whose results are discarded. That is not a preference to respect. ⚠ IT KEEPS THE USER'S EDITS. Presence and POSITION are guaranteed; the table it writes to and the values it maps are theirs. An existing `create_record` further down is MOVED to the front rather than duplicated — re-seeding a second one would quietly double every run's writes. """ flow = dict(flow_raw or {}) if isinstance(flow_raw, dict) else {} actions = [a for a in (flow.get("actions") or []) if isinstance(a, dict)] if actions and actions[0].get("kind") == "create_record": return _pin_unique(_ensure_enrich_step( flow_raw if isinstance(flow_raw, dict) else flow, enrich_kind)) at = next((i for i, a in enumerate(actions) if a.get("kind") == "create_record"), -1) if at > 0: actions.insert(0, actions.pop(at)) else: # The seed must be a config `clean_actions` ACCEPTS, or the builder shows a red banner # and no card: the wave-23 header records four kinds that shipped with illegal seeds and # did exactly that. `create_record` needs a ut_-prefixed table and at least one value. actions.insert(0, { "id": "act_1", "kind": "create_record", "enabled": True, "when": None, # `config.label` follows the `review` action's precedent in `_clean_action_config` — # an action naming itself, inside the untyped config bag, so no shared TS type changes. "config": {"table": str(table or DISCOVER_TABLE), "label": "Save the profile", # C5: the picture and the real write finally agree — see `_pin_unique`. "uniqueOn": "handle", "values": {"handle": "{{handle}}"}}, }) flow["actions"] = actions return _pin_unique(_ensure_enrich_step(flow, enrich_kind)) def _ensure_enrich_step(flow_raw, enrich_kind="enrich_instagram"): """⭐ 2026-08-07 (owner ruling) — ENRICH IS STEP 2 ON A DISCOVERY SEARCH. ⚠ WAVE 30: `enrich_kind` selects the network. Instagram's is the default so every pre-existing caller means exactly what it meant before; TikTok passes `enrich_tiktok` and gets the identical step shape, which is the owner's *"only the columns and data schema differ"*. Owner: *"make it default that this enrichment action is Step 2 always, and under Config of Step 2 … we can have a toggle on or off."* Which is the same shape as step 1's ruling and for the same reason: a search that finds profiles and never reads them has done half a job. The difference is the control — step 1 is permanent because a flow without it discards its results, while this one is permanent because the TOGGLE is how you turn it off. Deleting and disabling would be two ways to say one thing, and only one of them survives a re-save. ⚠ SEEDED ON, AND ON THE FREE RUNG. `tier: "anonymous"` costs nothing, so a search that gains this step by upgrading does not quietly start spending; switching the Source to the paid provider is an explicit choice a person makes in front of the sentence that names the cost. ⚠ THE COOLDOWN IS SEEDED ON TOO (30 days). On a table nobody has enriched it changes nothing — a blank `enriched_at` is never "recent" — and the moment there IS history it stops the flow re-buying the same profile nightly. Off-by-default would make the expensive behaviour the accident. ⛔ THE EXISTENCE TEST WALKS THE FORKS (`walk_actions`). A person who moved enrichment inside an If/then branch has one; seeding a second at the top level would enrich twice and bill twice, which is the duplication `apply_actions` already paid for once with the pinned step 1. """ flow = dict(flow_raw or {}) if isinstance(flow_raw, dict) else {} actions = [a for a in (flow.get("actions") or []) if isinstance(a, dict)] if any(a.get("kind") == enrich_kind for a in walk_actions(actions)): return flow_raw if isinstance(flow_raw, dict) else flow seeded = list(actions) # Index 1 — after the pinned Create record, because there is nothing to enrich until the # profiles have been written as records. `insert` past the end is a plain append, so a flow # with only step 1 lands this at the end, which IS step 2. seeded.insert(1, { "id": "act_enrich", "kind": enrich_kind, "enabled": True, "when": None, "config": {"postMetrics": False, "commentMetrics": False, "dryRun": False, "maxPosts": DEFAULT_POSTS_PER_PULL, "fromView": "", "sortField": DEFAULT_ENRICH_SORT, "sortDir": "desc", "limit": DEFAULT_ENRICH_LIMIT, "skipRecent": True, "skipRecentDays": DEFAULT_ENRICH_COOLDOWN_DAYS}, }) return {**flow, "actions": seeded} #: The key the discovery writer really upserts on. Both discovery runners key candidates on #: `candidate_key(platform, handle)`; `handle` is the half a person can see and the half this action's #: values carry, which is why the PINNED card shows that rather than the pair — `platform` is #: supplied by the runner, never typed by a person. #: ⛔ DEBT D-73 (closed wave 29): this note used to name wave 22's C6 compound — the one that #: paired the handle with its finder — as the live key. Wave 26 · R4/R5 retired it: the identity is #: the pair above and the TENANT is the unit, while `created_by` survives as an informational #: "Found by" stamp that no run may branch on. #: ⚠ The retired pair is DESCRIBED here and not reproduced, deliberately: a gate asserts this note #: names the key `_ck` actually takes, and a verbatim quotation of the wrong one reads to that gate #: exactly like the defect. A stale comment on correct code is how the next session reintroduces a #: bug with a rationale attached. IG_PINNED_UNIQUE = "handle" def _pin_unique(flow): """C5: keep `uniqueOn: "handle"` on the PINNED step 1 across every save. ⭐ WHY IT IS RE-STAMPED RATHER THAN MERELY SEEDED. `_clean_action_config` has no `previous` — the builder posts the whole action list on every save — so a client that omitted the key would silently reset it to `""`, i.e. back to append. The pinned card is a PICTURE of the engine's own upsert (`apply_actions` skips it at runtime, see the note there), so the picture claiming "append" while the engine upserts is exactly the surface-disagrees-with-the-engine defect this module refuses everywhere else. ⛔ AND IT CANNOT MINT A CONFIG THE GUARD REFUSES — HARD RULE 11, which is the whole reason this is a function and not one line. `_clean_action_config` refuses a `uniqueOn` the action does not write, so the stamp is applied ONLY when `handle` is among the action's own values. A user who remaps the card to write different columns keeps their edit and still saves; the alternative — stamping unconditionally — is a product-seeded default that its own validator would 400, which is precisely the post-W24 hotfix this rule exists because of. ⚠ NON-MUTATING, and that is load-bearing rather than tidiness. `clean_definition` passes `prev.get("flow")` here when a PATCH carries no flow of its own — that is the STORED definition's own dict, so writing into it would edit the live store object in memory, before (and regardless of) any commit. Every touched level is copied instead. """ if not isinstance(flow, dict): return flow acts = flow.get("actions") if not isinstance(acts, list) or not acts or not isinstance(acts[0], dict): return flow first = acts[0] if first.get("kind") != "create_record": return flow cfg = first.get("config") if not isinstance(cfg, dict) or IG_PINNED_UNIQUE not in (cfg.get("values") or {}): return flow if str(cfg.get("uniqueOn") or "").strip(): return flow return {**flow, "actions": [{**first, "config": {**cfg, "uniqueOn": IG_PINNED_UNIQUE}}] + acts[1:]} #: The two steps `_ensure_ig_action` / `_ensure_enrich_step` plant, as `(kind, id)`. Named here so #: the seeder and the un-seeder cannot disagree about what "seeded" means. #: ⚠ WAVE 30 — `enrich_tiktok` shares `act_enrich`. The map is keyed by KIND, and only one enrich #: step is ever seeded per flow (the network follows the trigger), so the id cannot collide. _IG_SEED_IDS = {"create_record": "act_1", "enrich_instagram": "act_enrich", "enrich_tiktok": "act_enrich"} def _is_untouched_ig_seed(action, table, enrich_kind="enrich_instagram"): """Is this action still EXACTLY what the Instagram trigger planted? (wave 27 item 15) ⛔ THE COMPARISON IS AGAINST A FRESHLY BUILT SEED, not against a list of remembered keys, and that is the only version that stays true: `_ensure_ig_action` and `_ensure_enrich_step` build the same dicts one screen above, so a step gaining a config key next wave gains it here too. A remembered key list would quietly start calling every seeded action "edited". ⛔ AND THE FRESH SEED GOES THROUGH `clean_actions` FIRST, which cost a red check to learn. The stored action has been cleaned — `_clean_action_config` normalises it and adds the keys the kind declares (`profileField: ""` on an enrich step, for one) — so comparing against the RAW dict `_ensure_enrich_step` writes reports every seeded enrich step as "edited by a person", and the un-seeding silently never fires for it. Comparing cleaned against cleaned is the only version where the two sides are the same kind of object. ⚠ `enabled` IS DELIBERATELY NOT PART OF THE TEST for the enrich step. Its ruling says the TOGGLE is how you turn it off — so a person who switched it off has expressed an opinion about a step they still want, and an off seed is still a seed. """ if not isinstance(action, dict): return False kind = str(action.get("kind") or "") if _IG_SEED_IDS.get(kind) != str(action.get("id") or ""): return False seeded, _seed_err = clean_actions( (_ensure_ig_action({"actions": []}, table, enrich_kind).get("actions") or [])) fresh = {a.get("kind"): a for a in (seeded or [])} seed = fresh.get(kind) if not seed: return False cfg, seed_cfg = dict(action.get("config") or {}), dict(seed.get("config") or {}) if kind in ENRICH_KINDS: cfg.pop("enabled", None) seed_cfg.pop("enabled", None) return cfg == seed_cfg and (action.get("when") or None) is None def _drop_ig_seeds(flow_raw, table, enrich_kind="enrich_instagram"): """Strip the trigger's own seeded steps, keeping any the person has since made theirs. ⭐⭐ WAVE 27 ITEM 15 (owner) — *"stale pinned create_record on trigger change"*. THE COMMENT THAT USED TO SIT AT THE KIND FLIP ARGUED THE OPPOSITE and is rewritten there; it read: *"it does not delete the seeded actions … a create_record the person has since re-pointed at their own table with their own values is THEIR action now"*. That reasoning is still correct and is exactly what this function preserves — but it was applied to EVERY seeded action, including the ones nobody had ever opened, and the result is what the owner reported: switch an Instagram search to Manual and you are left holding a "Save the profile" step writing `{{handle}}` into `ut_ig_candidates` on a flow that no longer produces handles. That is not somebody's work being protected; it is the machine's own leftover, pointed at a table the automation has nothing to do with any more. ⛔ SO THE TEST IS "DID ANYBODY TOUCH IT", NOT "WAS IT SEEDED" — the same guard shape as `_vestigial_name_field` and `_retired_tracked_field`, and for the same reason: this is a branch that destroys something, so the conservative half is the load-bearing half. """ flow = dict(flow_raw or {}) if isinstance(flow_raw, dict) else {} actions = [a for a in (flow.get("actions") or []) if isinstance(a, dict)] kept = [a for a in actions if not _is_untouched_ig_seed(a, table, enrich_kind)] if len(kept) == len(actions): return flow_raw return {**flow, "actions": kept} def ig_action_pinned(defn, index): """Is this action the one that cannot be removed? Derived, never stored — a stored `pinned` flag is a second copy of the rule, and the copy is what a hand-written PATCH omits.""" # ⭐ WAVE 30 — BOTH discovery kinds. It read `== "discover_instagram"`, so TikTok's step 1 (once # seeded) would have rendered as an ordinary draggable, deletable card: the same rule the owner # asked for "EXACTLY", applied to only one network. return (defn or {}).get("kind") in DISCOVERY_KINDS and index == 0 def clean_definition(raw, previous=None, username=""): """Whole-definition validation. Returns `(defn, error)`.""" raw = raw if isinstance(raw, dict) else {} prev = previous or {} # ⭐ WAVE 24 · C-TRIG — THE TRIGGER IS RESOLVED FIRST NOW, because law 1 makes it the thing # that CHOOSES the kind. Reading `raw["trigger"]["key"]` here instead would be a second # reader of the trigger vocabulary, free to disagree with `clean_trigger` about which keys # exist and which are refused. Gate-visible consequence, recorded in AMENDMENT A1: a payload # invalid in BOTH its trigger and its config now answers with the trigger's sentence. trigger, terr = clean_trigger(raw.get("trigger") if "trigger" in raw else prev.get("trigger"), prev.get("trigger")) if terr: return None, terr # C-TRIG law 2 (wave 24): a definition that names no kind is a `plain` one. Before R6 this # refused with "unknown automation kind ''" — correct while a wizard always sent one, and a # dead end the moment the wizard was deleted and the client's create body became {name}. kind = _s(raw.get("kind") or prev.get("kind"), 40) or DEFAULT_KIND drop_ig_seeds = False if (trigger or {}).get("key") == "ig_profile_match": # LAW 1: the trigger IS how `discover_instagram` gets chosen now. Unconditional rather # than "when the kind is unset" — a definition whose trigger says Instagram-discovery and # whose kind says something else is not a preference to respect, it is two halves of one # answer disagreeing, and the trigger is the half the person actually picked. kind = "discover_instagram" elif (trigger or {}).get("key") == "tiktok_profile_match": # ⭐ WAVE 29 (D-9 / R1) — LAW 1, TIKTOK'S HALF. Same rule, same reason: the trigger is how # `discover_tiktok` gets chosen, and it is unconditional for the same reason the Instagram # arm is — a definition whose trigger says TikTok discovery and whose kind says something # else is two halves of one answer disagreeing. kind = "discover_tiktok" elif (kind == "discover_tiktok" and ((prev.get("trigger") or {}) or {}).get("key") == "tiktok_profile_match" and (trigger or {}).get("key") != "tiktok_profile_match"): # ⛔⛔ LAW 1'S INVERSE, AND ITS ABSENCE ON THE INSTAGRAM SIDE WAS A MONEY BUG (see the arm # below): with nothing to flip the kind BACK, switching the trigger to Manual left # `RUNNERS[kind]` pointing at the discovery runner, so **Run now fired a PAID corpus # search on a flow the person had just made manual.** Shipped here in the same change as # law 1 rather than discovered the same way twice. # ⚠ THE TEST IS THAT THE TRIGGER **MOVED** — comparing the RESOLVED key against the # PREVIOUS one. Asking `"trigger" in raw` would fire on every empty patch, because # `patch()` builds its raw as `dict(prev)` plus the caller's keys. kind = DEFAULT_KIND # ⭐ WAVE 30 — un-seed on the way out, exactly as the Instagram arm below does. This line # was ABSENT and harmless for as long as TikTok had no seeds to leave behind; the moment # the seeder above learned TikTok, its absence became wave-27 item 15's defect on the other # network — a flow switched to Manual still carrying a "Save the profile" step nobody # planted deliberately. Found by widening the seeder and asking what else assumed only # Instagram could have seeds. drop_ig_seeds = True elif (kind == "discover_instagram" and ((prev.get("trigger") or {}) or {}).get("key") == "ig_profile_match" and (trigger or {}).get("key") != "ig_profile_match"): # ⭐⭐ 2026-08-07 (owner report) — LAW 1 HAS AN INVERSE, AND ITS ABSENCE WAS A MONEY BUG. # # Law 1 above flips the kind TO `discover_instagram` when the trigger says Instagram # discovery. Nothing flipped it BACK. So `kind` was inherited from `prev` forever: switch # the trigger to Manual and the automation stayed `discover_instagram`, which means # # · `RUNNERS[kind]` is still `run_discover_instagram`, so pressing **Run now** on a flow # the person had just made MANUAL fired a PAID Bright Data corpus search; # · `ig_action_pinned` keys on the kind, so the seeded "Create record" stayed # UNDELETABLE — the owner's report, verbatim: *"When a trigger change from the # instagram trigger, the 'always first' create record just stuck there"*. The server # would have accepted the delete; the CLIENT hid the control, because both halves ask # the kind and the kind was lying. # # ⛔ THE TEST IS THAT THE TRIGGER **MOVED**, and the first version of this got it wrong in # a way worth recording. It asked `"trigger" in raw` — but `patch()` builds its raw as # `merged = dict(prev)` plus the caller's keys, so **`"trigger"` is present on EVERY # patch**, and an empty `patch(rt, id, {})` flipped a discovery automation to `plain`. # That turned five `section_bd` checks red and crashed the suite on a `StopIteration` # three sections later. Comparing the RESOLVED key against the PREVIOUS one is the honest # question: did this automation stop being an Instagram search? # # ⚠ NARROW ON PURPOSE, twice over. It fires only when the automation WAS on the Instagram # trigger — a `discover_instagram` created with an explicit kind and some other trigger is # somebody's deliberate state, not a mistake to correct. And only FROM # `discover_instagram`: `scrape_db` and `field_instagram` are surviving kinds whose # triggers are their own business, and clobbering them would retire them by accident. # # ⭐⭐ WAVE 27 ITEM 15 — IT NOW DROPS THE SEEDS NOBODY TOUCHED, and the paragraph this # replaces argued the other way, so here is why it was half right. It read: *"it does not # delete the seeded actions … a create_record the person has since re-pointed at their own # table with their own values is THEIR action now, and quietly destroying it is the silent # data loss this module refuses everywhere else."* Every word of that is still true and is # exactly what `_is_untouched_ig_seed` protects. What it got wrong was applying the # protection to actions NOBODY HAD EVER OPENED: the owner's report is a flow switched to # Manual still carrying a "Save the profile" step writing `{{handle}}` into # `ut_ig_candidates` — the machine's own leftover on a flow that no longer produces # handles, unpinned but still there, still runnable, and still pointed at a table that has # nothing to do with this automation. # ⚠ APPLIED BELOW, at the flow, not here: `flow_raw` is not resolved yet at this line. kind = DEFAULT_KIND drop_ig_seeds = True if kind not in KINDS: return None, f"unknown automation kind {kind!r}" # ⛔ DEBT D-65 — THE REFUSAL THAT BELONGS HERE IS NOT SHIPPED, AND THAT IS A DECISION. # D-65 offers two exits for `field_instagram`: convert a surviving definition to a `plain` # flow carrying `enrich_instagram`, or REFUSE it here with a sentence naming the replacement. # The refusal was built and then REVERTED, because W24/R6 deliberately made `patch()` the way # a retired kind legally comes into existence — `create` refuses, `patch` accepts, and the two # live automations save through this function every time their owner edits them. The gate's # own fixture helper says so in the strongest terms available: *"If R6's refusal ever moved # from `create` into `clean_definition` (the tempting simplification), this helper would go # red across a dozen sections, which is the alarm that change deserves."* It did, and the # alarm worked. # ⚠ WHAT SHIPPED INSTEAD is the half that costs nothing and was the actual complaint: every # door that DOES refuse now names the replacement (`RETIRED_KIND_REPLACEMENT`), so nobody # meets "unknown automation kind" for a decision made on purpose. The rest of D-65 is an # owner-visible behaviour change — either new `field_instagram` mints stop working, or stored # ones are rewritten under their owner — and that is a ruling, not a refactor. name = " ".join(_s(raw.get("name") or prev.get("name") or "", MAX_NAME).split()) if not name: return None, "name the automation" cfg_raw = raw.get("config") if isinstance(raw.get("config"), dict) else prev.get("config") if kind in DISCOVERY_KINDS and prev.get("kind") != kind: # ⭐⭐ WAVE 30 · T04 — WIDENED FROM `discover_instagram` TO BOTH DISCOVERY KINDS, AND THIS # ONE LINE WAS THE WHOLE OF THE OWNER'S ITEM 3. Picking "When a TikTok profile fits a # criteria" 400'd on the very first Save with *"say how many profiles to fetch — an # UNBOUNDED discovery query is the one shape the vendor refuses outright # (NOT_ENOUGH_FUNDS)"*, and the trigger was therefore never STORED, which is what the # owner saw as "the Trigger won't load". # ⛔ INSTAGRAM WAS NEVER SURVIVING ON ITS OWN MERITS: `AutomationDetail.buildConfig` sends # no `recordsLimit` for EITHER platform on a fresh automation (its `kind` is still `plain` # at that moment, so it falls through to `return { targetTable }`). IG worked only because # this seed caught it. So the defect was never "TikTok is missing something Instagram has" # — it was one hard-coded string in the single line that rescues both. # ⚠ `prev.get("kind") != kind` is the EXACT generalisation of the old # `prev.get("kind") != "discover_instagram"`, not a loosening: it still fires only on a # kind FLIP, so a later save that carries a real limit is not re-seeded (asserted). # AMENDMENT A1: the kind FLIP seeds the one field `clean_config` insists on, or the very # first Save after picking this trigger 400s — against the stored-inert-with- # `configured:false` pattern the whole picker is built on (see `clean_trigger`'s A3 note). # ⛔ THE REFUSAL ITSELF IS UNTOUCHED: an explicit 0 still gets its sentence. That guard # exists because an unbounded query is the one shape the vendor refuses outright, and # coercing a blank into a number would be exactly the silent widening it protects against. # Seeding cannot spend — `clean_schedule` defaults `enabled` False, so nothing runs until # a person presses Run now or arms the schedule. cfg_raw = dict(cfg_raw or {}) if not _ig_int(cfg_raw.get("recordsLimit")): cfg_raw["recordsLimit"] = DISCOVER_SEED_RECORDS config, err = clean_config(kind, cfg_raw, prev.get("config")) if err: return None, err if trigger: # A2: re-ask completeness now that the config is validated — `ig_profile_match`'s own # configuration IS the config, and `clean_trigger` could not see it. trigger["configured"] = _trigger_configured(trigger, config) try: schedule = clean_schedule( raw.get("schedule") if isinstance(raw.get("schedule"), dict) else prev.get("schedule"), prev.get("schedule")) except ValueError as e: return None, str(e) # (`clean_trigger` ran at the top — law 1 needs the trigger before the kind.) flow_raw = raw.get("flow") if "flow" in raw else prev.get("flow") # ⭐ EVERY CLEAN, NOT ONLY THE EDGE (owner ruling 6 — see `_ensure_ig_action`). The edge-only # call was what made the action deletable; running it on every save is what makes step 1 # permanent, and it is deliberate rather than a widened condition nobody noticed. # ⭐⭐ WAVE 30 (owner, 2026-08-12) — BOTH DISCOVERY TRIGGERS SEED, not just Instagram's. This # single `==` was the whole of *"why is it not copied EXACTLY"*: a TikTok search stored an # EMPTY flow, so the product's own rule — a search that finds profiles must have somewhere to # put them — held for one network and not the other. if (trigger or {}).get("key") in DISCOVERY_TRIGGER_KIND: _seed_table, _seed_enrich = discovery_seed_spec(kind) flow_raw = _ensure_ig_action( flow_raw, config.get("targetTable") or _seed_table, _seed_enrich) elif drop_ig_seeds: # ITEM 15: the trigger just STOPPED being a discovery one. Un-seed what the trigger # planted and nobody has since made theirs — measured against the table the seeds were # planted for, which is the PREVIOUS config's, not the one being saved. # ⚠ And against the PREVIOUS kind's enrich action, for the same reason: the seeds to strip # are TikTok's if that is what was planted, and asking "is this Instagram's seed?" of a # TikTok flow answers no and silently leaves the leftover behind. _prev_table, _prev_enrich = discovery_seed_spec(prev.get("kind")) flow_raw = _drop_ig_seeds( flow_raw, (prev.get("config") or {}).get("targetTable") or _prev_table, _prev_enrich) flow, ferr = clean_flow(flow_raw, prev.get("flow")) if ferr: return None, ferr # ⭐ WAVE 30 — widened with the seeder above: TikTok now HAS a pinned step 1, so the rule that # its `config.table` is authoritative has to reach it, or the two fields that name one database # would disagree on exactly the network that just gained the card. if (trigger or {}).get("key") in DISCOVERY_TRIGGER_KIND: # ⭐ WAVE 25 · R2 — THE CREATE RECORD ACTION'S `config.table` IS AUTHORITATIVE, and # `config.targetTable` FOLLOWS IT. Two fields have been naming the same database since # wave 24: the pinned step 1 says where profiles are written, and the config says where # the automation's records live — and for a discovery flow those are the same database by # construction. When they disagreed, every surface picked a different winner (the canvas # Write node read the config, the actual write read the action), which is a bug you can # only see by comparing two panels. # # ⛔ THE ACTION WINS BECAUSE IT IS THE ONE A PERSON EDITS. R2b puts the preset list on the # action's own configuration, so the table named there is the one they were looking at. # # ⚠ THIS IS ALSO THE MIGRATION, and it is a migration in the shape laws 4/5 established: # a stored definition carrying two different values is reconciled on its next clean, # silently and exactly once, because nothing writes the losing value back. A refusal here # would 400 every Save of a definition that was legal yesterday. # ⭐⭐ WAVE 27 ITEM 11 — WHICHEVER SIDE MOVED WINS, and R2's "the action wins" is now the # TIE-BREAK rather than the whole rule. The owner's report: change the database on the # TRIGGER, save, and the picker springs back to the old one with no message. The cause is # the branch below read unconditionally — the pinned action still named yesterday's table, # so it overwrote a change the person had just made in front of it, silently, every time. # # ⛔ R2 IS NOT REVERSED, and the distinction is which QUESTION the code asks. R2 settled # "when the two disagree, whose value is real?" — the action's, because R2b puts the # preset list on the action's own panel, so that is the one they were looking at. That # answers a definition at REST. A save is different: it carries an INTENT, and the honest # question is which side this particular save changed. So: # · the trigger-side picker moved -> it wins, and the pinned action is re-pointed to # follow it (leaving the action behind would just re-revert on the next save); # · only the action moved, or neither did and they were already inconsistent -> R2's # migration, unchanged, reconciling a stored definition exactly once. # · BOTH moved in one payload -> the action still wins. That is R2 literally, and it is # also the only reading that cannot lose a value: the action's table is the one with # the field mapping attached to it. first = (flow.get("actions") or [{}])[0] pinned = str((first.get("config") or {}).get("table") or "") \ if first.get("kind") == "create_record" else "" prev_target = str((prev.get("config") or {}).get("targetTable") or "") prev_pinned = str(((((prev.get("flow") or {}).get("actions") or [{}])[0] ).get("config") or {}).get("table") or "") target_moved = bool(prev_target) and config.get("targetTable") != prev_target action_moved = bool(prev_pinned) and pinned != prev_pinned if pinned and target_moved and not action_moved: # The person changed the trigger's database. Carry it INTO the pinned step so the two # agree, instead of throwing their edit away to keep a stale copy consistent. cfg1 = dict(first.get("config") or {}) cfg1["table"] = str(config.get("targetTable") or "") flow["actions"] = [{**first, "config": cfg1}, *(flow.get("actions") or [])[1:]] elif pinned and pinned != config.get("targetTable"): config["targetTable"] = pinned # The stored label described a DIFFERENT database, so keeping it would caption the new # one with the old one's name. Blank falls back to the key everywhere, and `ut_ensure` # never relabels a table that already exists — so an adopted hand-made database keeps # the name its owner gave it. config["targetLabel"] = "" return { "id": _s(prev.get("id") or raw.get("id"), 40), "name": name, "kind": kind, "config": config, "schedule": schedule, "trigger": trigger, # The Builder's ordered actions. An absent flow is simply an automation with no actions. "flow": flow, "status": prev.get("status") or {"state": "idle", "lastRunAt": "", "lastSummary": ""}, "runs": list(prev.get("runs") or [])[:MAX_RUNS], # ⚠ ENGINE-OWNED CONTINUATION STATE, and it is NOT config. A discovery snapshot takes ~20 # minutes to build, so a run persists its id here and the NEXT run collects it. It is # carried through `patch` untouched because a user pressing Save must not silently orphan # a set the vendor is already building (and already counting against the funds gate). "state": dict(prev.get("state") or {}), "created": prev.get("created") or _iso(), "createdBy": prev.get("createdBy") or username, }, None def set_state(rt, auto_id, patch_state): """Merge into ONE automation's engine state. A key set to None is removed. Its own tiny store write rather than a field on `_commit_run`, because the handoff must survive a run that ends in `error` — a snapshot the vendor is building does not stop existing because the run that started it failed afterwards. """ def _up(cur): cur = cur if isinstance(cur, dict) else {} d = cur.get(str(auto_id)) if d is None: return cur st = dict(d.get("state") or {}) for k, v in (patch_state or {}).items(): st.pop(k, None) if v is None else st.__setitem__(k, v) d["state"] = st return cur # ⚠ W31-T35 — THE ONE `keeps_defs=True` IN THIS MODULE. `set_state` writes `state` and never a # trigger, and `grid_hook` calls it once per CREATED RECORD; clearing the definitions memo here # would re-read the whole bucket one row later than before and undo D-134 entirely. Safe # because `grid_hook` reads exactly one state key (`rcHighwater`) and mirrors its own write # into the memo in the same statement. Read `_store_update`'s note before adding a second. _store_update(rt, _up, flush="sync", keeps_defs=True) def _without_retired_board(definition): """Return a definition with retired Board-only state removed, without mutating the store. This protects scheduled/manual execution before the next UI list request persists the migration. It removes only the former Board configuration, actions, and audit trail; normal actions and all user data remain intact. """ if not isinstance(definition, dict): return definition, False out, changed = dict(definition), False cfg = out.get("config") if isinstance(cfg, dict) and "lanes" in cfg: cfg = dict(cfg) cfg.pop("lanes", None) out["config"] = cfg changed = True def _actions(actions): nonlocal changed clean = [] for action in actions if isinstance(actions, list) else []: if not isinstance(action, dict): clean.append(action) continue if action.get("kind") == "review": changed = True continue item = dict(action) if item.get("kind") == "group" and isinstance(item.get("config"), dict): group_cfg = dict(item["config"]) branches = group_cfg.get("branches") if isinstance(branches, list): kept = [] for branch in branches: if not isinstance(branch, dict): changed = True continue next_actions = _actions(branch.get("actions")) if not next_actions: changed = True continue kept.append({**branch, "actions": next_actions}) if not kept: changed = True continue if kept != branches: changed = True group_cfg["branches"] = kept item["config"] = group_cfg elif isinstance(group_cfg.get("actions"), list): nested = _actions(group_cfg["actions"]) if not nested: changed = True continue if nested != group_cfg["actions"]: group_cfg["actions"] = nested item["config"] = group_cfg clean.append(item) return clean flow = out.get("flow") if isinstance(flow, dict): next_flow = dict(flow) actions = _actions(flow.get("actions")) if actions != flow.get("actions"): next_flow["actions"] = actions if "ending" in next_flow: next_flow.pop("ending", None) changed = True out["flow"] = next_flow if "reviews" in out: out.pop("reviews", None) changed = True return out, changed def retire_automation_board_state(rt, tables=None): """Persist the idempotent Board retirement, then remove its generated table fields. ``tables`` is the optional lent snapshot of the `user_tables` bucket (W29-T01) — it reaches the stage scan and nothing else. """ fields = retire_automation_stage_fields(rt, tables=tables) try: stored = dict(rt.get(STORE_KEY) or {}) except Exception: return {"definitions": 0, **fields} plan = {str(aid): cleaned for aid, raw in stored.items() for cleaned, changed in [_without_retired_board(raw)] if changed} if not plan: return {"definitions": 0, **fields} def _up(cur): cur = cur if isinstance(cur, dict) else {} for aid, cleaned in plan.items(): if aid in cur: cur[aid] = cleaned return cur _store_update(rt, _up, flush="sync") return {"definitions": len(plan), **fields} #: ⭐⭐ WAVE 31 · T35 (D-134) — the definitions memo, and the two things that make it safe. #: `tenant -> (monotonic_at, defs)`. Read ONLY through `all_definitions(..., cached=True)`, which #: exactly one caller uses (`grid_hook`). _DEFS_MEMO = {} #: Deliberately SHORT. This exists to collapse one burst of row events into one read, not to be a #: cache — an import of 20,000 rows arrives in far less than this, and a stale trigger for two #: seconds is bounded and recoverable where a stale one for a minute is a mystery. _DEFS_TTL = 2.0 def _store_update(rt, fn, flush="sync", keeps_defs=False): """THE one door every write to the automations bucket goes through — and the ONLY reason it exists is that the memo's invalidation must be DERIVED rather than remembered. ⚠ `keeps_defs=True` IS AN EXPLICIT, SINGLE-SITE OPT-OUT, and it is here because the safe default would otherwise make the memo useless on the exact path it was built for. `set_state` writes `state` and NEVER a trigger, and it is called by `grid_hook` itself once per created record — so clearing on it would mean a 20,000-row import re-read the bucket 20,000 times anyway, one row later than before. The opt-out is safe because `grid_hook` reads exactly one state key (`rcHighwater`) and mirrors its own write into the memo in the same statement. ⛔ The DEFAULT is the safe one, so a thirteenth writer added next wave gets invalidation without knowing this exists; only a caller that has read this paragraph can opt out. ⛔ THE ALTERNATIVE WAS TWELVE CALL SITES. `rt.update(STORE_KEY, …)` appeared twelve times in this module; hanging an invalidation on each would be a hand-maintained list, and the way that fails is silent: a thirteenth writer added next wave leaves `grid_hook` firing triggers off a definition set that no longer exists, with nothing red. Same defect class as `patch`'s hand-maintained patchable-key list, which this module already documents as *"the silent-drop seat"* [[a-constant-two-features-share]]. ⚠ IT CLEARS THE WHOLE MEMO, not this tenant's row, on purpose: identifying the tenant here would mean trusting `getattr(rt, 'key')` at a WRITE, and a wrong answer there is a stale trigger set for another tenant. The memo holds at most a handful of entries and the correct, boring thing costs nothing. """ if not keeps_defs: _DEFS_MEMO.clear() return rt.update(STORE_KEY, fn, flush=flush) def all_definitions(rt, cached=False): """Every automation definition for this tenant. ⭐⭐ `cached=True` IS D-134's FIX, AND IT IS OPT-IN FOR A REASON. `grid_hook` runs ONCE PER ROW EVENT and called this unconditionally, so a 20,000-row import performed 20,000 whole-document deep copies of the automations bucket — `Store.get` re-serialises on every call, hit or miss, under the store lock. Every other caller is a request handler that reads it once, so widening the memo to all of them would trade a real guarantee (a request sees the current store) for nothing measurable. ⛔ THE MEMO IS THE SAME OBJECT ON A HIT, and `grid_hook` MUTATES it deliberately — see the `rcHighwater` write there. That is not a leak of an implementation detail, it is the fix to the hazard a naive memo creates: `grid_hook` both READS `state.rcHighwater` and WRITES it through `set_state`, so a memo that went stale against its own write would re-fire `record_created` for records it had already handled, breaking A2(4)'s *"once per record EVER — undo-proof"*. Writes through `_store_update` drop the memo; the one write `grid_hook` makes to its OWN copy is mirrored into it in the same statement. """ if cached: key = str(getattr(rt, "key", "") or "") hit = _DEFS_MEMO.get(key) if hit is not None and (time.monotonic() - hit[0]) < _DEFS_TTL: return hit[1] try: raw = dict(rt.get(STORE_KEY) or {}) except Exception: return {} out = {str(aid): _without_retired_board(defn)[0] for aid, defn in raw.items()} if cached: _DEFS_MEMO[str(getattr(rt, "key", "") or "")] = (time.monotonic(), out) return out def _new_id(existing): n = 1 while f"auto_{n}" in existing: n += 1 return f"auto_{n}" #: C2's three answers to "which database does this automation work on?" — the FIRST question the #: create wizard asks (owner item 3: the kind is not the first thing a person picks, the data is). TARGET_MODES = ("existing", "new", "automated") def resolve_target(rt, raw, username=""): """C2: turn the wizard's `target` into a bound table key. Returns `(config_patch, error)`. - `existing` — bind a database that is already there. - `new` — mint a blank one now, so the automation has somewhere to write before its first run instead of conjuring a table the person never agreed to. - `automated` — leave it to the runner: a scraping automation MINTS its target on first run (`ut_ensure`), stamped `source: "Automation"`, which is what makes it a machine database. Nothing is created here, deliberately — an empty table created up front for a scrape that never runs is litter nobody can explain. """ if not isinstance(raw, dict) or not raw: return {}, None # no target block = the pre-wizard shape mode = _s(raw.get("mode"), 20).strip() or "existing" if mode not in TARGET_MODES: return None, f"{mode!r} is not one of: " + ", ".join(TARGET_MODES) if mode == "automated": label = " ".join(_s(raw.get("label"), 60).split()) return ({"targetLabel": label} if label else {}), None if mode == "existing": key = _s(raw.get("table"), 60).strip() if not key: return None, "choose the database this automation works on" t = ut_get(rt, key) if t is None: return None, f"{key!r} is not a database in this workspace" return {"targetTable": key, "targetLabel": t.get("label") or key}, None label = " ".join(_s(raw.get("label"), 60).split()) if not label: return None, "name the new database" import core.user_tables as _ut_new key = _ut_new.create(label, username or "automation", st=rt) if not key: return None, ("the database could not be created — this workspace may be at its table " "limit") return {"targetTable": key, "targetLabel": label}, None def _unmint(rt, key): """Delete a database this call minted moments ago, because the call is refusing (D-50). ⚠ FAIL-QUIET BY DESIGN. The caller is already returning a refusal with a sentence the user needs to read; a rollback that raised would replace that sentence with a 500 and lose the reason the create failed in the first place. The worst case of a swallowed failure here is the orphan we had before — strictly no worse, and the refusal still reaches the person. """ if not key: return try: import core.user_tables as _ut_del _ut_del.delete(key, st=rt) except Exception: # noqa: BLE001 pass def create(rt, raw, username=""): existing = all_definitions(rt) if len(existing) >= MAX_AUTOMATIONS: return None, f"this workspace is at the {MAX_AUTOMATIONS}-automation limit" raw = dict(raw or {}) # ⭐ WAVE 24 — DEBT D-50. `resolve_target` MINTS a database for `mode:"new"`, and validation # happens after it, so a correct refusal ("a source URL is required") used to leave an # orphaned `ut_*` table behind with no automation pointing at it. MEASURED at the W23 close: # three refused creates, two orphaned databases, deleted by hand. A person mis-filling a form # twice silently accumulated empty databases in their nav. # # ⛔ THE ROLLBACK IS DERIVED, NOT DECLARED, and that is the point of doing it this way. It # does not test for `mode == "new"`; it asks "did the table this call's target resolved to # exist BEFORE this call?" So it closes the hole for any future target mode that mints, and # a fifth mode cannot reopen it by forgetting a branch. Scoped to THIS call's own target # rather than "any table that appeared", so a concurrent create in another request is never # collateral. before = set(ut_all(rt) or {}) patch_cfg, terr = resolve_target(rt, raw.pop("target", None), username) if terr: return None, terr minted = str((patch_cfg or {}).get("targetTable") or "") minted = minted if minted and minted not in before else "" if patch_cfg: raw["config"] = {**(raw.get("config") or {}), **patch_cfg} # 2026-08-10 — point a targetless discovery flow at the profile database this tenant ALREADY # has, instead of letting the pure validator mint a second empty one. See # `discover_default_table`; a target the caller named is never rewritten. raw = _apply_discover_default(rt, raw) defn, err = clean_definition(raw, None, username) if err: _unmint(rt, minted) return None, err # ⭐ WAVE 24 · OWNER RULING R6 — "scrape_db and field_instagram survive on the automations # that already use them and NO NEW ONE CAN BE CREATED." # # ⛔ HERE, IN `create`, AND NEVER IN `clean_definition`. A PATCH of one of the two live # automations runs through `clean_definition` with `prev["kind"]` already set, so refusing # the kind there would make both of them unsaveable — the ruling retires the door, not the # automations behind it. Enforced on the server rather than left to the deleted wizard: a # wall that holds only because the client stopped asking is a convention, not a wall. if defn["kind"] in RETIRED_KINDS: _unmint(rt, minted) return None, (f"{KIND_LABELS.get(defn['kind'], defn['kind'])!r} automations are no " f"longer created — the ones already using it keep working. " f"{RETIRED_KIND_REPLACEMENT.get(defn['kind'], '')}") defn["id"] = _new_id(existing) def _up(cur): cur = cur if isinstance(cur, dict) else {} cur[defn["id"]] = defn return cur # `async` for the reason spelled out at `remove` and applied at `patch` — a create is the same # blocking commit inside the same request, and the person is waiting on it in the same way. # ⚠ NO PRESET GUARD HERE, deliberately: `patch` may skip the spawn because it can compare # against a PREVIOUS definition, and a create has none — R10's columns must exist before a run, # so this call stays unconditional. _store_update(rt, _up, flush="async") _presets_after_write(rt, defn, username) # R10: the columns exist before a run if defn.get("trigger"): _seed_event_state(rt, defn) return defn, None def patch(rt, auto_id, raw, username=""): existing = all_definitions(rt) prev = existing.get(str(auto_id)) if prev is None: return None, "no such automation" merged = dict(prev) # ⚠ A HAND-MAINTAINED PATCHABLE-KEY LIST, and it is the silent-drop seat of this module: a # key missing here is accepted by the route, validated by `clean_definition`, and then # DISCARDED — the client shows a saved flow that the store never received, and nothing goes # red. `flow` joined it in the same edit that created `flow` (wave 23), which is the only # ordering that never has a window where the bug exists. for k in ("name", "config", "schedule", "kind", "trigger", "flow"): if k in (raw or {}): merged[k] = raw[k] # 2026-08-10 — the PATCH door needs it too, and this is the one that actually bit: picking the # Instagram trigger on an existing automation flips the kind, and the very next Save runs a # config that has never carried a target through the pure validator, which mints the empty # `ut_ig_candidates`. `create` alone would have left the commonest path untouched. merged = _apply_discover_default(rt, merged) defn, err = clean_definition(merged, prev, username) if err: return None, err defn["id"] = str(auto_id) def _up(cur): cur = cur if isinstance(cur, dict) else {} cur[defn["id"]] = defn return cur # ⛔ WHY THIS IS `async` — owner report 2026-08-12 (wave 30), MEASURED on the deployed build. # Owner, verbatim: *"debug Automation module, its extremely slow, even renaming an Automation # takes forever"*. A rename paid a BLOCKING HF commit inside the request, and `Store.update`'s # sync branch holds `self._lock` across a strict DOWNLOAD **and** the upload — the same lock # `Store.get` takes — so one rename stalls every OTHER reader of that tenant's store for its # whole duration. That is the "module is slow" half a per-route fix never reaches. # The reasoning, the read-your-writes contract and the `user_tables.add_row` precedent are # written out in full at `remove` above (same wave, same owner report); this is that ruling # applied to the save path rather than a second argument for it. _store_update(rt, _up, flush="async") # ⛔ THE PRESET SPAWN RUNS ONLY WHEN ITS OWN INPUTS MOVED — the other half of the same report. # MEASURED as `nurilab-admin` on `4258a93`: **7,912 ms** for one rename against a **288 ms** # warm `GET /automations` on the same connection, i.e. 27x the read this wave just made fast # (T12/T13). A rename changes `name`; `_presets_after_write` reads `config`, `flow` and # `trigger` and NOTHING else, then pays a full `user_tables` read (`ut_get`/`ut_ensure`; # `rt.get` deep-copies by design, documented ceiling 35.8 MB / ~1.4 s) to re-derive columns # that cannot have changed. ⭐ The generalisable half: T12/T13 took `ut_all` off the automation # READ path this wave and every sibling WRITE path still carried it — a performance fix scoped # to "the slow page" leaves the same call in every route beside it # [[read-a-gates-predicate-for-what-it-excludes]]. # ⛔ THE DANGEROUS DIRECTION IS SKIPPING, NEVER RUNNING, so the predicate is the INPUT SET of # the function being skipped — not a hand-listed set of "cheap" edits, which is how a guard # silently un-ships W25/R10 the next time the spawn learns to read a fourth key. When none of # them moved the call is a no-op by construction (it re-derives from those same keys), and the # run-time `ut_ensure` R10 deliberately kept is still the backstop if the TABLE drifted # underneath — which is not a thing a rename should be repairing. Shaped after the trigger # comparison directly below, which already compares `prev` against the cleaned definition. # ⭐⭐ WAVE 31 · T30 — AND THE PREDICATE IS NOW THE SPAWN'S OWN ARGUMENT, which is the fix. # The four-whole-objects test above was right in principle and inert in practice: `persist` # sends a REBUILT `config`, so `prev["config"] != defn["config"]` on essentially every save and # the guard fired every time — sparing only *rename*, the one edit the owner stopped # complaining about after wave 30. `preset_inputs` is the canonical projection of exactly what # `_spawn_presets` reads, and `_spawn_presets` is handed nothing else, so this comparison # cannot go stale behind a future input the way a hand-listed key set would. The full argument, # including why this is STRONGER than what it replaces rather than a relaxation, is in # `preset_inputs`' own docstring — read that before widening or narrowing this line. if preset_inputs(prev) != preset_inputs(defn): _presets_after_write(rt, defn, username) # R10: also on the save that RE-TARGETS # A2(1): a NEW or CHANGED condition trigger starts with everything currently matching # DISARMED — enabling never fires for records already in the state. if (defn.get("trigger") or {}) != ((prev.get("trigger")) or {}): _seed_event_state(rt, defn) return defn, None def _presets_after_write(rt, defn, username): """⭐ WAVE 25 · R10 — THE PRESET COLUMNS APPEAR ON SAVE, not on the first run. Owner ruling R10: choosing the trigger / naming the Create record target spawns the preset columns immediately, "so R2b's list is real and the database is visible before anything is spent". That last clause is the point — a person can look at the database, see the sixteen columns, and decide whether to arm the schedule, instead of paying a vendor to find out what they are agreeing to. ⛔ THE RUN-TIME `ut_ensure` STAYS AS THE BACKSTOP (R10 says so explicitly), and it costs nothing to keep: `ut_ensure` skips the write entirely when no field would change, so the first run of a saved automation finds the table already correct and spends no commit. ⚠ IT SPAWNS `CANDIDATE_FIELDS`, NOT `PRESET_PROFILE_FIELDS`. The discovery target needs the search bookkeeping (`found_count`, `first_found`, `created_by`) as well as the preset set, and spawning a subset here would mean the columns changed between Save and the first run — two different answers to "what does this database look like", which is the exact disagreement R10 exists to remove. ⚠ A WORKSPACE AT ITS TABLE CEILING SPAWNS NOTHING AND THE SAVE STILL SUCCEEDS. `ut_ensure` returns the key without creating when `MAX_UT_TABLES` is reached; refusing to save an automation because the workspace is full would be a worse answer than saving one whose table arrives later. The run-time path reports that condition LOUDLY (`ut_missing`, D-11), so the honest failure still has exactly one home. """ _spawn_presets(rt, preset_inputs(defn), username) #: The enrich actions whose OWN config decides which preset columns and child databases a save #: spawns, and the exact keys read off each. DECLARED here rather than spelled inside #: `preset_inputs`, so `_spawn_presets` and the save-path guard cannot drift apart by one key. PRESET_ACTION_INPUTS = { "enrich_instagram": ("profileField",), "enrich_tiktok": ("profileField", "postMetrics", "commentMetrics"), } def preset_inputs(defn): """⭐⭐ WAVE 31 · T30 — EXACTLY the values `_spawn_presets` consumes off a definition, canonical. ⛔ THIS IS THE WHOLE POINT OF THE TICKET, so it is worth being precise about what changed and why it is SAFER than what it replaces, not weaker. Owner item 7, verbatim: *"It still takes forever to change the Config for automation as well. It just say Saving... and takes a long time for me to change options and configs etc."* Wave 30 guarded the spawn on `any(prev[k] != defn[k] for k in ("config","flow","trigger", "kind"))` and reasoned — correctly — that *"the predicate is the INPUT SET of the function being skipped, not a hand-listed set of cheap edits"*. The flaw was not the reasoning; it was that the input set was stated as four WHOLE OBJECTS while the spawn reads a handful of leaves out of them. `AutomationDetail.tsx::persist` sends `config: buildConfig()` — a value RECONSTRUCTED from React state, not the stored object — so the comparison is against something rebuilt on every save and the guard never fires on the one edit the owner is complaining about. W30's fix spared *rename* (the client omits `flow`/`trigger` entirely, and `kind`/`name` are unchanged), which is exactly the edit the owner stopped complaining about after wave 30. ⭐ WHY NARROWING IS SAFE HERE AND WAS NOT SAFE THEN: this projection is not a hand-listed allow-list that a future wave can silently outgrow. `_presets_after_write` hands this dict to `_spawn_presets` and `_spawn_presets` **never receives `defn` at all** — so it is structurally incapable of reading a fifth key that this function does not carry. Teaching the spawn to read something new REQUIRES adding it here, and the guard then follows for free. That is a stronger guarantee than wave 30's, which held only as long as somebody remembered the comment. ⚠ CANONICAL, because the input is a rebuild: each leaf is normalised exactly the way the spawn itself normalises it (`str(...).strip()` for a field name, `bool(...)` for a switch). Without that, `postMetrics: false` and a missing `postMetrics` would compare unequal while the spawn treats them identically — a guard that fires on a difference its consumer cannot see is the same defect in a smaller font. ⚠ IT DELIBERATELY COVERS THE DISCOVERY BRANCH **AND** THE ENRICH BRANCH IN ONE PROJECTION, even though today's control flow returns before the second can run for a discovery automation. That early return is D-178 (W31-T34) and it is going away; a projection built around today's branch would silently narrow the guard the moment it does. """ defn = defn or {} cfg = defn.get("config") or {} actions = ((defn.get("flow") or {}).get("actions") or []) return { "target": str(_flow_table(defn) or cfg.get("targetTable") or ""), "label": str(cfg.get("targetLabel") or ""), "flowId": str(defn.get("id") or ""), "discoveryKind": DISCOVERY_TRIGGER_KIND.get( (defn.get("trigger") or {}).get("key") or ""), "actions": { kind: [{k: (_norm_switch(a, k) if k != "profileField" else str((a.get("config") or {}).get(k) or "").strip()) for k in keys} for a in _actions_of_kind(actions, kind)] for kind, keys in PRESET_ACTION_INPUTS.items() }, } def _norm_switch(action, key): """A capture switch as the spawn reads it — `bool`, so absent and False are ONE value.""" return bool((action.get("config") or {}).get(key)) def _spawn_presets(rt, inputs, username): """The spawn itself. Takes `preset_inputs(defn)` — never the definition — see that docstring. ⭐⭐ WAVE 31 · T30, THE SECOND HALF: **ONE `user_tables` read for the whole pass, not five.** Every `ut_ensure` below used to take its own copy of the tenant document through `ut_get` → `ut_all` → `rt.get`, which deep-copies unconditionally (ceiling 35.8 MB / ~1.4 s), and the TikTok arm reaches FOUR of them plus the `ut_get` above — measured **7,912 ms** on `4258a93`. Lending one snapshot is the same fix W30-T13 made on the automation READ path (`routes_automation.py::_LentTables`); this is deliberately NOT a third copy of that class, but the `tables=` parameter `retire_automation_stage_fields` in this very module already established, because here every consumer is a function we own and pass to directly. """ target = inputs["target"] if not target: return tag = inputs["flowId"] ig_actions = inputs["actions"]["enrich_instagram"] tt_actions = inputs["actions"]["enrich_tiktok"] try: # ⭐ THE ONE READ. Everything below answers out of this snapshot. tables = ut_all(rt) # ⭐⭐ WAVE 30 · T06 — BOTH DISCOVERY KINDS SPAWN ON SAVE, and TikTok's absence here was # R10 simply not holding for the second platform: `TT_PROFILE_FIELDS` reached `ut_ensure` # at exactly ONE site — inside `run_discover_tiktok` — so the database did not exist until # money had already been spent finding out what was in it. That is the precise pre-R10 # behaviour `verify_automation`'s NC39 forbids on the Instagram side. # ⛔⛔ THE DISCRIMINATOR IS THE **TRIGGER**, NOT THE KIND, AND THE GATE PAID FOR THAT # SENTENCE. Widening this to `kind in DISCOVERY_KINDS` looks equivalent — law 1 makes a # discovery trigger choose its kind — but the implication only runs ONE WAY. A definition # may carry `kind: "discover_instagram"` with **no trigger at all**: law 1 sets the kind # from the trigger and never the reverse, so any direct API create can do it, and one of # this repo's own fixtures does. MEASURED: the kind test spawned a preset database under a # dry-run fixture whose check asserts that no table exists — a red on correct-looking code, # caught only because that check happened to exist. # ⇒ R10 is a rule about what somebody PICKED IN THE PICKER, so it reads the picker's own # answer. `DISCOVERY_TRIGGER_KIND` is that map, and a gate asserts it agrees with law 1. # ⚠ TRACKED because it is the ONE key several `ut_ensure` calls in this pass can share: the # discovery arm and both enrich arms all ensure `target`. A later one must not answer out of # a snapshot an earlier one invalidated — see the `tables=` arguments below. wrote_target = False _dkind = inputs["discoveryKind"] if _dkind: _, _, _spawn_label, _spawn_fields = discovery_facts(_dkind) ut_ensure(rt, inputs["label"] or _spawn_label, _spawn_fields, username, key=target, flow_tag=tag, lock_fields=True, tables=tables) # ⭐⭐ WAVE 31 · T34 (D-178) — AND THE `return` THAT USED TO BE HERE IS GONE. # # ⛔ WHAT IT COST: a DISCOVERY automation could never spawn its child databases. The # capture switches below are the ones that promise `ut_tt_posts` / `ut_tt_comments` / # `ut_tt_post_snapshots` at SAVE, before any money is spent (W30-T10) — and only a # `plain` flow ever reached them, because a discovery flow left this function three # lines earlier. So the exact automation the owner would build for TikTok discovery — # find profiles, then capture their posts — silently got the profile table and nothing # else, and the toggles it showed were promises the save path never kept. # ⚠ IT IS A TRAP, NOT TODAY'S ONLY SYMPTOM: nurilab's child tables are absent for a # CONFIGURATION reason as well, so fixing this alone will not make them appear there. # # ⚠ THE SNAPSHOT IS NOW STALE FOR `target`, and the arms below resolve their profile # binding out of that very table (`bound` reads its `fields`). One re-read, paid only on # a save that actually reached the discovery spawn — correctness before the read count, # and it is one read against the 41 this function used to take. `wrote_target` stays # False precisely BECAUSE we refreshed: it tracks staleness, not "did somebody write". tables = ut_all(rt) enrich_actions = ig_actions table = tables.get(target) or {} named = next((a["profileField"] for a in enrich_actions if a["profileField"]), "") bound = next((str(f.get("key") or "") for f in table.get("fields") or [] if isinstance(f.get("profile"), dict)), "") or named if enrich_actions and bound: ut_ensure(rt, table.get("label") or target, _profile_schema_for(bound), username, key=target, flow_tag=tag, lock_fields=True, tables=tables) wrote_target = True # ⭐ WAVE 30 · T08 — the TikTok half of W25/R10: the columns an `enrich_tiktok` step will # fill appear when the automation is SAVED, not when money is first spent. Resolved # through `profile_field_key(..., source=PROFILE_SOURCE_TT)` so it cannot adopt an # Instagram binding, and through the SAME `_tt_profile_schema_for` the runner's own top-up # uses — R10's rule is that the column set does not change between Save and the first run, # and one shared resolver is what makes that structural instead of a convention. if tt_actions: tt_named = next((a["profileField"] for a in tt_actions if a["profileField"]), "") tt_bound = profile_field_key(table, tt_named, source=PROFILE_SOURCE_TT) if tt_bound: ut_ensure(rt, table.get("label") or target, _tt_profile_schema_for(tt_bound), username, key=target, flow_tag=tag, lock_fields=True, tables=None if wrote_target else tables) # ⭐ WAVE 30 · T10 — and the CHILD databases the step's own switches promise, at # SAVE, before any money. R10's rule is "the columns a step will fill appear when # you press Save"; a `postMetrics` toggle whose database only exists after a paid # run is the same complaint one level up. # ⛔ GATED ON THE SWITCHES, never spawned unconditionally: a database that exists # because somebody looked at a toggle, and then never fills, is the "SECOND, empty # database" the discovery default was rewritten to stop producing. for _flag, _child in (("postMetrics", TT_POSTS_TABLE), ("postMetrics", TT_POST_SNAPSHOTS_TABLE), ("commentMetrics", TT_COMMENTS_TABLE)): if any(a[_flag] for a in tt_actions): # R9 (W31-T32): the child arrives LOCKED, from its own declaration. ut_ensure(rt, TT_TABLE_LABELS[_child], TT_TABLE_FIELDS[_child], username, key=_child, flow_tag=tag, lock_fields=True, tables=tables, record_mode=tt_record_mode(_child)) except Exception as e: # noqa: BLE001 # ⛔ NEVER FAILS THE SAVE. The definition is already committed by the time this runs, so # raising here would answer 500 for an automation that IS stored — the caller would retry # and create a second one. The columns then arrive on the first run, which is precisely # the behaviour this function is an improvement on rather than a replacement for. print(f"[aios-auto] preset spawn deferred to the first run: {type(e).__name__}: {e}") def remove(rt, auto_id): aid = str(auto_id) def _up(cur): cur = cur if isinstance(cur, dict) else {} cur.pop(aid, None) return cur # ⛔ WHY THIS IS `async` AND NOT `sync` (owner report 2026-08-12, wave 30, MEASURED). # Owner, verbatim: *"it takes forever to delete an automation"*. A delete was paying up to # TWO BLOCKING UPLOADS inside the request — this one, plus the whole-document rewrite below, # which fires for every DISCOVERY automation because those always tag the columns they # spawned (deleting one marked all 30 columns of `ut_tt_profile` disabled). Measured 977 ms # for the CHEAPEST case (a `plain` flow, no tagged columns) against a 757 ms `GET /automations` # baseline on the same connection; the discovery case is strictly worse by a full-document # read plus a full-document commit. # # `flush="async"` is not a weakening: `Store.update`'s async branch applies the mutation to # the in-process cache, marks it owned+dirty and schedules a COALESCING worker, and the # class's read-your-writes contract means the very next `GET /automations` already sees the # deletion. This is the same trade `user_tables.add_row` makes for ROW creation — strictly # more valuable data than an automation definition — so a delete is not the place to be # stricter than a row insert. [[sync-write-eats-pending-async-write]] is the hazard in the # OTHER direction (a sync writer discarding a pending async write) and `Store.update`'s dirty # branch already handles it. _store_update(rt, _up, flush="async") # C8 — the fields this flow tagged are DISABLED with the reason on them, never orphaned # silently: the column keeps its values and its config, and says why it stopped. Scanned # first so a delete with no tagged fields costs no store commit. # ⚠ HONEST RESIDUAL, stated rather than smoothed: this scan still costs ONE full read of the # tenant's `user_tables` document on EVERY delete (`rt.get` deep-copies by design), and the # update below reads it a second time. Removing that needs a tagged-field index, which is a # feature and not a wave-tail edit — booked rather than improvised. What it no longer costs # is the part the owner could feel: the blocking network commits. # ⭐ WAVE 31 · T39(a) — D-177(a): ONE WALK, AND THE PLAN CARRIES WHAT IT FOUND. # This used to scan the whole document to answer a boolean (`hit`), then walk the WHOLE # document again inside `_disable` to re-derive the same matches. The scan now records the # `(table, field)` pairs it found and the mutation applies exactly those — the shape # `bind_unbound_fields` in this module already uses, so a delete costs one walk instead of two # and the write touches only the fields it named. # ⛔ HONEST RESIDUAL, STATED RATHER THAN SMOOTHED (R6's second sentence): this still costs ONE # full read of the tenant's `user_tables` document, and `rt.update` below takes its own strict # read. **That second read cannot be removed here** — it belongs to the write, and answering # "which fields did this flow tag?" without a scan needs a tagged-field INDEX, which is a # feature and not a wave-tail edit. It is booked as D-179; what changed is the second WALK, # not the read count. plan = [(tk, str(f.get("key") or "")) for tk, t in ut_all(rt).items() for f in ((t or {}).get("fields") or []) if str((f.get("automation") or {}).get("flowId") or "") == aid] if not plan: return def _disable(cur): cur = cur if isinstance(cur, dict) else {} note = f"its automation was deleted {_stamp()}" for tk, fk in plan: for f in ((cur.get(tk) or {}).get("fields") or []): a = f.get("automation") if f.get("key") == fk and isinstance(a, dict): a["disabled"] = True a["statusNote"] = note return cur # The expensive half of the owner's complaint: a whole-document rewrite, committed while the # person waits. Async for the reason given above — the disable is visible to the next read # immediately; only the upload is deferred, and it coalesces with any other pending write to # the same key instead of racing it. rt.update(UT_STORE_KEY, _disable, flush="async") # --------------------------------------------------------------------------------------------- # RUN STATE — process memory only (see the module header for why this may never be persisted) # --------------------------------------------------------------------------------------------- _RUN_LOCK = threading.RLock() _RUNNING = {} # (tenant, id) -> {'startedAt', 'step'} def running(tenant, auto_id): with _RUN_LOCK: return dict(_RUNNING.get((tenant, str(auto_id))) or {}) or None def _claim(tenant, auto_id): with _RUN_LOCK: if (tenant, str(auto_id)) in _RUNNING: return False _RUNNING[(tenant, str(auto_id))] = {"startedAt": _iso(), "step": "starting"} return True def _step(tenant, auto_id, text): with _RUN_LOCK: cur = _RUNNING.get((tenant, str(auto_id))) if cur is not None: cur["step"] = text def _no_step(_text): """The default `step` sink for a runner nobody gave a live slot to (a gate driving a runner directly). A runner must never REQUIRE the slot — the slot is process state, and the runner is the part a test is allowed to call on its own.""" def _release(tenant, auto_id): with _RUN_LOCK: _RUNNING.pop((tenant, str(auto_id)), None) def _commit_run(rt, auto_id, state, summary, counts, ok, affected=None, steps=None, notes=None): """THE ONE store write a run performs on the `automations` bucket. `steps` is the per-NODE outcome map the canvas paints its dots from. It is recorded here because it is a MEASUREMENT the runner took as it walked — see `graph`, which refuses to invent a node status when this map is absent (every run stored before W19-C has no `steps`, and painting those nodes green from the run's overall state would be a fabrication). ⭐⭐ `notes` CLOSES DEBT D-103 (2026-08-09). A run's `counts` are integers and the comprehension below drops everything else — so the per-record sentence a vendor gave us had literally nowhere to live, and *"1 profile read(s) were blocked"* was the whole of what the product could say. MEASURED on nurilab: the same record blocked on three separate runs and the reason was unrecoverable from the store afterwards, so the diagnosis had to be rebuilt by calling the vendors by hand. The note is the most valuable thing a run produces — a dead handle, a vendor hiccup and an exhausted key are three different actions — and it was the one thing thrown away. ⚠ ON THE RUN, NOT ON THE RECORD. D-103's own prescription: no tenant table gains a column it did not ask for, and W25/R4 already retired writing status STRINGS into people's grids. """ entry = {"ts": _iso(), "ok": bool(ok), "summary": _s(summary, 400), "counts": {k: int(v) for k, v in (counts or {}).items() if isinstance(v, (int, float))}, "steps": {str(k): str(v) for k, v in (steps or {}).items()}, # Bounded on both axes: a 100-profile run must not put 100 sentences into a store # entry that is kept 20 deep per automation. "notes": [_s(n, 300) for n in (notes or []) if str(n or "").strip()][:25], "affected": list(affected or [])[:200]} def _up(cur): cur = cur if isinstance(cur, dict) else {} d = cur.get(str(auto_id)) if d is None: return cur d["status"] = {"state": state, "lastRunAt": entry["ts"], "lastSummary": entry["summary"]} d["runs"] = ([entry] + list(d.get("runs") or []))[:MAX_RUNS] # Wave 22 (airtable-brief rec 6): K consecutive FAILURES auto-pause the automation with # the reason on it — a dead credential must not burn quota (or a paid vendor's records) # 96 times a day while a green toggle sits over a buried error log. Errors only: # `partial` is the honest-progress state and pausing on it would punish honesty. runs = d.get("runs") or [] if state == "error" and len(runs) >= CONSECUTIVE_FAILURE_PAUSE and all( not r.get("ok") for r in runs[:CONSECUTIVE_FAILURE_PAUSE]): sch = d.get("schedule") if isinstance(sch, dict) and sch.get("enabled"): sch["enabled"] = False trg = d.get("trigger") if isinstance(trg, dict): trg["paused"] = True d["statusNote"] = (f"auto-paused after {CONSECUTIVE_FAILURE_PAUSE} consecutive " f"failures — fix the cause, then re-enable: " f"{entry['summary'][:120]}") return cur _store_update(rt, _up, flush="sync") _notify_outcome(rt, auto_id, state, entry) return entry def _notify_outcome(rt, auto_id, state, entry): """⭐⭐ WAVE 27 ITEM 31 / CONTRACT C5 — the bell rings on EVERY outcome, good and bad (I7). Owner: notify on both success and error. `_commit_run` is the ONE place a run's outcome is written, so it is the only place this can go without a second definition of "what happened". ⛔ AMENDMENT A1 MOVED TWO REQUIREMENTS HERE AND THE FIRST IS A CROSS-TENANT DEFECT IF MISSED. 1. **`st=` IS PASSED EXPLICITLY.** `core.alerts.notify`'s `st=None` default resolves to the MODULE-level store, which is tenant #0's (Royal Imports). The automation tick runs on a background thread with no session, so an omitted `st` files nurilab's automation failures in Royal Imports' inbox — D-16's class, silently, forever. 2. **THE OWNER IS THE AUTOMATION'S `createdBy`, NEVER THE RUNNER'S IDENTITY.** The scheduler has no session at all, so the alternative is not "the wrong person" but "nobody", and `notify` returns None on a blank owner — the notification would simply never exist, on exactly the runs nobody was watching. ⚠ AND NO SERVER ADMISSION WAS NEEDED. A1 measured that `routes_alerts._TOPICS` gates only view-alert CREATION; `notify()` writes straight into the notifications bucket and `GET /notifications` filters by nothing, so an `automation` topic already reaches the bell. ⚠ FAILURE HERE IS SWALLOWED. A notification that cannot be filed must not turn a run that SUCCEEDED into one that raised — the run entry is already committed one line above, and the bell is a courtesy on top of it. """ try: defn = (rt.get(STORE_KEY) or {}).get(str(auto_id)) or {} owner = str(defn.get("createdBy") or "").strip() if not owner: return from core import alerts as _alerts _alerts.notify( owner=owner, label=_s(defn.get("name") or "Automation", 80), topic="automation", key=str(auto_id), detail=f"{state}: {entry.get('summary') or ''}"[:300], st=rt) except Exception: # noqa: BLE001 pass # --------------------------------------------------------------------------------------------- # INSTAGRAM (R7, extended by R1) — PUBLIC DATA ONLY, NEVER AN INSTAGRAM LOGIN # --------------------------------------------------------------------------------------------- # ⚠ AMENDED W19-C, vendor swapped W20. This section read "ANONYMOUS PUBLIC ENDPOINTS ONLY" and # that is no longer the whole truth: R1 added a paid VENDOR rung (Bright Data, further down) # which does authenticate — to the vendor. The rail that matters is unchanged and is sharper: # # **NOTHING HERE EVER AUTHENTICATES TO INSTAGRAM.** No login, no password, no session cookie, # no account to get banned, public data only. The vendor key is a key to a SUPPLIER, and the # supplier takes the scraping risk; it is never an Instagram credential and this module must # never be given one. # # The rungs below this line are the ANONYMOUS ladder ($0, no key of any kind). # # ⚠ HONEST STATUSES ARE THE FEATURE. Instagram rate-limits and outright blocks datacenter egress # (the HF Space and any AWS Lambda are both in that class), and its anonymous surface has been # progressively closed for years. So this returns a STATE — ok / partial / blocked / error — and # the cell says which. A pull that quietly wrote zero posts and reported success would be worse # than no automation at all: the table would look maintained and be empty. # # THE LADDER, tried in order with pacing between rungs. Each rung records how it did in `via`, # so a run history answers "what still works anonymously?" from data rather than from memory. PACE_SECONDS = float(os.environ.get("AIOS_IG_PACE_SECONDS") or 2.5) #: ⭐ MAP THE SCHEMA, NOT JUST WHAT WAS POPULATED ON THE PROBE — and the reason is this page's #: own headline. **HISTORY IS UNBUYABLE.** No vendor sells "followers on 1 January"; every number #: is a NOW value, so the series starts the day capture starts and a field we do not capture #: TODAY is permanently lost for today. Dropping a column because it was null on two probe rows #: has exactly the same cost as delaying capture — for that column — and is justified by exactly #: the evidence (n=2) that was judged too thin to close D-25. Adding a column that turns out to #: be usually-empty costs a blank cell; omitting one costs the months before somebody notices. #: Every returned provider field is retained in Source data. Promoted columns make commonly used #: values filterable; the raw source document prevents a new or uncommon field from being lost. #: ⚠ A blank in any of these means NOT READ — never "they have none". The anonymous rungs expose #: almost none of them, which is what the `source` column is for. SNAPSHOT_FIELDS = [ field_def("snapshot_key", "Snapshot"), field_def("influencer_key", "Influencer"), field_def("pulled_at", "Pulled at", "date"), field_def("followers", "Followers", "int"), field_def("following", "Following", "int"), field_def("posts_count", "Post count", "int"), field_def("full_name", "Name"), field_def("bio", "Bio"), field_def("verified", "Verified", "checkbox"), field_def("source", "Read via"), field_def("approx", "Counts are approximate", "checkbox"), # The link in bio — the paid rung's field, and commercially the most useful single string an # influencer row can carry (it is the shop/affiliate destination). field_def("external_url", "Link in bio", "url"), # --- the rest of the vendor's Profiles schema. field_def("ig_id", "Instagram id"), field_def("profile_url", "Profile", "url"), field_def("avg_engagement", "Avg engagement", "pct"), field_def("category", "Category"), field_def("business_category", "Business category"), field_def("is_business", "Business account", "checkbox"), field_def("is_professional", "Professional account", "checkbox"), field_def("is_private", "Private", "checkbox"), field_def("highlights_count", "Story highlight count", "int"), field_def("bio_hashtags", "Bio hashtags"), field_def("pronouns", "Pronouns"), # ⭐ 2026-08-07 — the rest of the vendor's Profiles schema (see `_bd_profile`). The snapshot # row maps the WHOLE schema by design, so these belong here the moment the map reads them; # leaving them out would be the "captured but unrecorded" half of the same loss the note at # the top of this list is about. field_def("profile_name", "Profile name"), field_def("is_joined_recently", "Joined recently", "checkbox"), field_def("has_channel", "Has channel", "checkbox"), field_def("partner_id", "Partner id"), field_def("external_url_title", "Link title"), field_def("fbid", "Facebook id"), field_def("related_accounts", "Related accounts"), field_def("country_code", "Country"), field_def("source_payload", "Source data", "json"), ] POST_FIELDS = [ field_def("shortcode", "Shortcode"), field_def("influencer_key", "Influencer"), field_def("posted_at", "Posted at", "date"), field_def("type", "Type", "select", options=["image", "video", "carousel"]), field_def("caption", "Caption"), field_def("url", "URL", "url"), # A tagged place is useful content context and can be filtered as ordinary text. It is never # presented as the creator's location: a creator can tag a holiday, venue or brand location. field_def("tagged_location", "Tagged location"), # The per-field map below gives operational fields first; this locked JSON document retains # every other value Bright Data supplied, so a vendor schema addition is preserved # instead of being silently discarded while the canonical field model catches up. field_def("source_payload", "Source data", "json"), # Sponsored-post detection — §2c called it one of the fields worth having that the anonymous # ladder cannot reach, and it came back MEASURED-populated (`True`, with the brand attached). field_def("paid_partnership", "Paid partnership", "checkbox"), field_def("partner", "Partner brand"), field_def("hashtags", "Hashtags"), field_def("alt_text", "Alt text"), # ⭐⭐ 2026-08-07 — THE LATEST-VALUE ENGAGEMENT COLUMNS, AND THEY EXIST TO KEEP A ROLLUP AT # ONE HOP. # # "Average views over the last N posts" is naturally TWO hops: profile → posts → each post's # most recent snapshot → `views`. Airtable's rollup is one hop, and growing a second one is a # much bigger build with a much worse failure mode (a rollup over a rollup, invalidated # transitively). So the post row carries its own LATEST value and the rollup reads it # directly. # # ⛔ R3's "ONE STORE FOR ONE SERIES" IS UNTOUCHED, and this is exactly the pattern the profile # row already uses one level up: LATEST on the row (+ `enriched_at` to date it), the SERIES in # `ut_ig_post_snapshots`. These three cells are rewritten on every pull from the snapshot that # was just appended — they are a projection of that store, never a second copy of it, and # deleting them would cost a convenience rather than a measurement. # ⚠ BLANK, NEVER ZERO — and `_ig_zero_is_blank` is what finally ENFORCES it. `postMetrics` is # off by default (it buys one vendor record per post), so on most pulls this stays empty, and # empty means "not read", which is what makes an honest average possible at all. A 0 here # would claim a post nobody watched. ⚠ The rule is scoped to this paid rung: a zero LIKE or # COMMENT count is a real measurement and must survive. # # ⛔⛔ THERE IS NO `views` COLUMN HERE, AND THAT IS A RULING, NOT AN OVERSIGHT (2026-08-08, # instagram-capture.md §4e/§4f — owner-approved after the evidence was bought). # Bright Data's `views` is delivered at ACCOUNT grain: one value across up to 12 distinct # reels, on 18 of 19 creators, through BOTH collection modes, while `likes` varies richly in # the very same rows. Mapping it here is what put one number on many unrelated posts. We # cannot say what it measures at ANY grain (`sriyynntt` returns 0 at 11,102 followers; # `reviewby_ayyaa` returns null at 20k–328k likes), so it is not promoted anywhere — it stays # in Source data, unlabelled, making no claim. Nothing is lost: the payloads are retained # whole, so if the vendor ever populates it per-reel the history is re-derivable. # ⚠ ENUMERATED, so nobody re-opens this hoping for a differently-named field: across all 30 # fields of a Reels row, `views` and `video_play_count` are the ONLY view-shaped keys, and the # Posts dataset carries none at all. `plays` below IS the per-reel equivalent, correctly named # and correctly mapped. It is blank because Meta retired the Plays metric on 2025-04-10 # (folded into a single "Views"), not because we are reading the wrong key. # ⭐⭐ VIEWS IS BACK (2026-08-08), AND FROM A DIFFERENT SOURCE THAN THE ONE THAT WAS RETIRED. # The retired column was fed by Bright Data's account-grain `views`. This one is fed by the # `ig_post_views` CAPABILITY (providers.py), which resolves to Apify's `videoPlayCount` — # measured against a browser-read ground truth to the digit. Same column name, same type, same # rollup; the engine underneath is swappable and the preset database never moved. That is the # owner's schema ruling working exactly as intended. # ⚠ STILL BLANK, NEVER ZERO, and still only on the paid rung. field_def("views", "Views", "int"), field_def("plays", "Plays", "int"), field_def("likes", "Likes", "int"), field_def("comments", "Comments", "int"), # ⭐⭐ 2026-08-09 (owner: *"whatever APIfy has more than BD pls use it and fix the post data # further"*). Two fields the primary source does not return AT ALL, already paid for inside # the same engagement response — so capturing them costs nothing extra. # ⛔ A COLUMN MUST EXIST BEFORE A CELL CAN BE WRITTEN. `user_tables` filters an unknown key on # every write door, so a normaliser that emits `video_duration` without this line writes # nothing and reports success — the wave-28 defect that swallowed nine preset cells. field_def("video_duration", "Video length (s)", "int"), field_def("comments_disabled", "Comments off", "checkbox"), field_def("measured_at", "Engagement read at", "date"), # ⭐ 2026-08-07 — the second half of "spawn relevant Post/Comment database that is LINKED": # a post reaches its own comment rows the same way a profile reaches its posts. Derived from # `shortcode`, so it is correct the moment a comment row exists and needs no maintenance. # ⚠ It resolves to nothing until comment capture is switched on, which is the honest state # for a relation whose far side is empty — the same standing `ut_ig_post_snapshots` has when # `postMetrics` is off. field_def("comments_link", "Comment rows", "link", description="Comment records linked to this post.", link={"table": IG_COMMENTS_TABLE, "on": "shortcode", "from": "shortcode"}), field_def("post_snapshots_link", "Measurement rows", "link", description="Engagement measurements linked to this post.", link={"table": IG_POST_SNAPSHOTS_TABLE, "on": "shortcode", "from": "shortcode"}), field_def("measurements_captured", "Measurements captured", "rollup", rollup={"link": "post_snapshots_link", "fn": "countall"}), ] #: ⭐⭐ 2026-08-07 (owner instruction) — THE COMMENT DATABASE. #: #: Owner: *"an enrichment automation should spawn relevant Post/Comment database that is linked to #: the profile automatically."* So the schema and the LINK ship; what does NOT ship on by default #: is the capture. #: #: ⛔ THIS REVERSES A STANDING RULING (D-22 / R11) AND THE REVERSAL IS DELIBERATE AND BOUNDED. #: Comment capture was refused on two grounds, and BOTH ARE STILL TRUE: #: 1. COST — comments are ~98% of a full-history bill (~$56,000 at 37.5M comments, §2a of #: `instagram-capture.md`). They are the single most expensive thing this product can buy. #: 2. PRIVACY — Bright Data flags `comment_user` and `post_user` as PII, and the append law has #: no erasure path for third parties who never appeared in the tracked set (D-24). A comment #: thread ingests identifiable people who never entered anybody's influencer list. #: ⛔ SO WHAT SHIPS HERE IS THE SCHEMA AND THE LINK. **NOTHING CAPTURES COMMENTS TODAY** — no code #: calls Bright Data's Comments dataset (`gd_ltppn085pokosxh13`), there is no `config.comments` #: switch, and this table stays EMPTY until somebody builds one. Stated in the present tense on #: purpose: an earlier draft of this note described the opt-in switch as if it existed, which is #: exactly the doc that "reads as authority and silently ages" that §0 of `instagram-capture.md` #: is written against. #: ⚠ WHEN IT IS BUILT it should be an OPT-IN defaulted OFF, the same posture as #: `config.postMetrics` and the Bright Data money switch W26/R15 kept — the relation costs #: nothing; the spend and the third-party ingest are a click somebody makes knowingly. #: Embedded comment payloads delivered with a Profile/Post/Reel response are retained because #: they are part of a record already paid for. The separate full Comments dataset remains OFF #: unless `commentMetrics` is explicitly enabled on the enrichment action. COMMENT_FIELDS = [ field_def("comment_key", "Comment"), field_def("shortcode", "Shortcode"), field_def("influencer_key", "Influencer"), # ⭐⭐ OWNER RULING 2026-08-12 — the same one that put `text` on the TikTok comment schema, and # it lands on BOTH networks in the same change on purpose: the two comment tables are read side # by side, and one carrying the content while the other does not is the divergence this repo # keeps paying for. ⛔ TEXT ONLY — `comment_user` / `comment_user_url` are vendor-FLAGGED PII # and stay in `source_payload`, uncolumned. field_def("text", "Comment"), field_def("commented_at", "Commented at", "date"), field_def("likes", "Likes", "int"), field_def("replies", "Replies", "int"), field_def("source_payload", "Source data", "json"), field_def("post_link", "Post", "link", description="Post record linked to this comment.", link={"table": IG_POSTS_TABLE, "on": "shortcode", "from": "shortcode", "single": True}), # Author/text and every provider-specific value stay intact in Source data. The promoted # columns intentionally keep the grid concise while Likes and Replies make comment engagement # directly filterable and rollup-ready. ] POST_SNAPSHOT_FIELDS = [ field_def("post_snapshot_key", "Snapshot"), field_def("shortcode", "Shortcode"), field_def("influencer_key", "Influencer"), field_def("pulled_at", "Pulled at", "date"), field_def("likes", "Likes", "int"), field_def("comments", "Comments", "int"), # ⭐⭐ 2026-08-10 — TWO DENORMALISED POST FACTS, and they are worth having on their own merits # before any rollup argument is made. # # `pulled_at` says when we LOOKED; `posted_at` says when the creator POSTED. A fact table with # only the first can describe a measurement but not its AGE, so the single most useful thing # this series can compute — views at N days old, i.e. velocity — is not expressible over it at # all. `type` is the same shape one column over: "reels only" is the default lens on this data # and without it every question has to go back through `ut_ig_posts` to ask what kind of post # this was. # # ⚠ THE WRITER HAS BOTH IN HAND (`capture_rows` builds the identity row and the measurement row # from the same vendor record), so this costs one dict key each and no extra call. # # ⛔ `options` IS NOT OPTIONAL ON A `select`. A select declaring none is the wave-26 item-24 # shape: the filter panel receives an empty list as an ANSWER, `if ([])` is truthy, and the # control renders dead with nothing to say. Same three values `POST_FIELDS` declares — copied # from it deliberately rather than shared, because these are two different tables' columns that # happen to agree today, and `verify_automation` asserts the agreement. # ⛔⛔ AND THEY DO NOT MAKE "THE LAST 10 REELS" EXPRESSIBLE OVER THIS TABLE. That takes TWO # orderings (newest measurement per post, then newest posts), the rollup bag has one `sortBy`, # and `ut_ig_posts` already performs the first of them. See the verdict in # `.claude/wiki/research/entity-vs-series.md` — these columns are for AGE and KIND, not for # moving the window here. field_def("posted_at", "Posted at", "date"), field_def("type", "Type", "select", options=["image", "video", "carousel"]), # Plays move independently of likes on video, so it is its own series rather than a thing to # derive. Paid rung only; blank means not read. # The series regains Views alongside the latest-value projection on the Post row — one store # for one series (R3) is untouched; this IS that store. field_def("views", "Views", "int"), field_def("plays", "Plays", "int"), field_def("source_payload", "Source data", "json"), field_def("post_link", "Post", "link", description="Post record linked to this measurement.", link={"table": IG_POSTS_TABLE, "on": "shortcode", "from": "shortcode", "single": True}), ] #: ⭐⭐ WAVE 32 · T41 — THE INSTAGRAM TWIN OF `TT_TABLE_FIELDS`, WHICH DID NOT EXIST. #: #: `TT_TABLE_FIELDS`' own note said so and named the consequence: *"The Instagram side has no #: equivalent map, which is exactly why its table names are scattered across the module."* It is a #: map now, for a concrete reason rather than symmetry: A's boot-time delivery sweep (`W32-T07`) has #: to call `ut_ensure` for every locked child of BOTH platforms, and a sweep that can ask TikTok for #: its table set and must hand-type Instagram's is one platform away from the drift this pair of #: maps exists to stop. #: #: ⛔ THESE FOUR LISTS ARE THE *CANONICAL* DECLARATION — the per-tenant BACKLINK fields #: `ensure_ig_graph` appends are NOT here, and must not be. A backlink is derived per profile #: database (`_profile_backlink_field`), so folding it into a module constant would make one #: tenant's relation a global fact. `ensure_ig_graph` reads this map and adds its own backlinks on #: top, which is why that function still owns the graph write and this only owns the schema. IG_TABLE_FIELDS = { IG_SNAPSHOTS_TABLE: SNAPSHOT_FIELDS, IG_POSTS_TABLE: POST_FIELDS, IG_POST_SNAPSHOTS_TABLE: POST_SNAPSHOT_FIELDS, IG_COMMENTS_TABLE: COMMENT_FIELDS, } #: The labels those four wear in the database list. ⚠ LIFTED VERBATIM from `ensure_ig_graph`'s own #: `graph` literal, which is now derived from this map — a renamed label here renames the table #: everywhere rather than leaving two spellings of one database. IG_TABLE_LABELS = { IG_SNAPSHOTS_TABLE: "IG snapshots", IG_POSTS_TABLE: "IG posts", IG_POST_SNAPSHOTS_TABLE: "IG post snapshots", IG_COMMENTS_TABLE: "IG comments", } # --------------------------------------------------------------------------------------------- # ⭐⭐ TIKTOK — THE FIVE `ut_tt_*` SCHEMAS (wave 29 · item 7 · D-9 · rulings R1 + R2) # --------------------------------------------------------------------------------------------- # ⛔ EVERY VENDOR FIELD NAME BELOW IS FROM ONE PROBED SOURCE — `waves/wave29/proto/tiktok-schema.md`, # 40 profile / 43 post / 17 comment fields read live from `GET /datasets/{id}/metadata` for $0.00, # each carrying the vendor's own type, description and `pii` flag. NOTHING HERE IS GUESSED, and the # names live in `connectors_tt.py` (the map), not here (the schema). This file declares what a # COLUMN is; the connector declares what the wire calls it. # # ⭐ THE SAME LAW AS THE INSTAGRAM LISTS ABOVE: map the schema, not just what was populated on the # probe. HISTORY IS UNBUYABLE — no vendor sells "followers on 1 January" — so a field we do not # capture today is permanently lost for today, and an occasionally-empty column costs a blank cell # while a missing one costs the months before somebody notices. # # ⚠ TWO DELIBERATE DIVERGENCES FROM THE INSTAGRAM SCHEMA, both licensed by R2 ("the two schemas may # diverge where the vendors do") and both recorded so neither reads as an omission: # # 1. NO `plays` COLUMN. TikTok returns ONE number, `play_count`, and it is the count TikTok # itself displays under a video — i.e. our `views`. Instagram has two columns because Meta # once had two metrics (`plays` retired 2025-04-10). Writing one vendor number into two of our # columns would manufacture a second measurement that a rollup could average or double-count, # and blank-vs-zero discipline says an unmeasured column must be blank, not a copy. # ⇒ `play_count` -> `views`, and `plays` does not exist on this family. # 2. `tt_id`, NOT `ig_id`. The probe doc flags this: our profile key is literally NAMED `ig_id`. # A TikTok row carrying a column labelled "Instagram id" is a header that lies, and this is a # NEW table family with no stored rows to migrate — so the honest name costs nothing. # # ⚠ THE COLUMNS THAT STAY BLANK ARE NAMED RATHER THAN DROPPED SILENTLY. TikTok has no equivalent of # `category`, `business_category`, `is_professional`, `highlights_count`, `bio_hashtags`, # `pronouns`, `profile_name`, `is_joined_recently`, `has_channel`, `partner_id`, # `external_url_title`, `fbid` or `related_accounts` (profile), nor of `alt_text`, # `comments_disabled`, `paid_partnership` or `partner` (post) — `commerce_info` is a LOCATION, not # a paid-partnership flag, and TikTok's `comment_setting` is profile-level rather than per-post. # Those columns are ABSENT here rather than declared and permanently blank — a column nothing can # ever write is a promise the grid keeps making that the vendor cannot keep. #: ⭐ WAVE 29 (contract C2) — the value the `profile` FLAG carries on a TikTok handle column: the #: twin of `PROFILE_SOURCE_IG`, which is declared further down beside the `PLATFORM_*` vocabulary #: with the note on why those two families of string are NOT the same thing. It sits here rather #: than there because the lists below are evaluated at IMPORT and would raise a NameError otherwise. #: ⛔ THIS CONSTANT IS ONLY HALF THE CONTRACT, AND THE OTHER HALF HAS LANDED (verified 2026-08-12, #: W30-T08): `platform/core/user_tables.py:PROFILE_SOURCES` now reads `('instagram', 'tiktok')`. #: Before it did, `_clean_profile` REFUSED this flag and the handle column of a `ut_tt_profile` #: spawn was dropped — fail-closed and loud by design, because the alternative is a profile #: database whose profile column silently is not one. Left written down rather than deleted: the #: refusal is what a TikTok binding looks like on any deployment where that line is missing. PROFILE_SOURCE_TT = "tiktok" #: The profile row: LATEST values plus `enriched_at`. The SERIES lives in `ut_tt_snapshots` — #: R3's "one store for one series" carries over unchanged, because the reason for it does (no #: vendor sells history, so the append table is the only history there will ever be). TT_PROFILE_FIELDS = [ # `platform` FIRST, exactly as the Instagram preset set has it — and on all five tables here, # not just this one. The IG family carries it on the profile table alone, which is why a TikTok # POST could never have lived beside an Instagram post; declaring it everywhere is what makes a # future cross-platform union view a UNION rather than a guess. tt_field_def("platform", "Platform"), tt_field_def("handle", "Handle", pinned=True, profile={"source": PROFILE_SOURCE_TT}), tt_field_def("full_name", "Name"), tt_field_def("tt_id", "TikTok id"), tt_field_def("profile_url", "Profile", "url"), tt_field_def("bio", "Bio"), tt_field_def("external_url", "Link in bio", "url"), tt_field_def("verified", "Verified", "checkbox"), tt_field_def("is_private", "Private", "checkbox"), tt_field_def("is_business", "Business account", "checkbox"), tt_field_def("followers", "Followers", "int"), tt_field_def("following", "Following", "int"), tt_field_def("posts_count", "Video count", "int"), # TikTok gives THREE engagement rates where Instagram gives one. All three are stored ×100 for # the same reason `avg_engagement` is on the IG side (C1-a): the vendor sends a 0–1 fraction and # our `pct` renderer appends the sign to the stored number, so the raw fraction would print a # 6.6% creator as `0.0%`. tt_field_def("avg_engagement", "Avg engagement", "pct"), tt_field_def("like_engagement", "Like engagement", "pct"), tt_field_def("comment_engagement", "Comment engagement", "pct"), # Total likes RECEIVED across the account's videos — a TikTok-only number with no Instagram # equivalent, and one of the few profile-level engagement facts a vendor gives away. tt_field_def("likes_received", "Likes received", "int"), tt_field_def("country_code", "Country"), tt_field_def("region", "Region"), tt_field_def("predicted_lang", "Language"), # ⚠ ACCOUNT AGE, NOT A MEASUREMENT STAMP. The vendor's `create_time` on a profile is when the # ACCOUNT was created; it is emphatically not "when `followers` was true". TikTok carries no # measurement timestamp at all, exactly like Instagram — which is why the append law below # (`@`) is the only thing that can date a number. tt_field_def("account_created_at", "Account created", "date"), tt_field_def("first_found", "First found", "date"), tt_field_def("last_found", "Last found", "date"), tt_field_def("found_count", "Times found", "int"), tt_field_def("created_by", "Found by"), tt_field_def("enriched_at", "Enriched at", "date"), tt_field_def("source", "Read via"), tt_field_def("source_payload", "Source data", "json"), tt_field_def("profile_snapshots_link", "Measurement rows", "link", description="Profile measurements linked to this account.", link={"table": TT_SNAPSHOTS_TABLE, "on": "influencer_key", "from": "handle"}), tt_field_def("posts_link", "Post rows", "link", description="Post records linked to this account.", link={"table": TT_POSTS_TABLE, "on": "influencer_key", "from": "handle"}), ] #: The profile SERIES. One row per profile per pull, keyed `@` — the append law #: from `instagram-capture.md` §3, unchanged, because its cause is unchanged: the vendor stamps #: nothing, so the only honest date a number can carry is the moment WE read it. TT_SNAPSHOT_FIELDS = [ tt_field_def("platform", "Platform"), tt_field_def("snapshot_key", "Snapshot"), tt_field_def("influencer_key", "Influencer"), tt_field_def("pulled_at", "Pulled at", "date"), tt_field_def("followers", "Followers", "int"), tt_field_def("following", "Following", "int"), tt_field_def("posts_count", "Video count", "int"), tt_field_def("likes_received", "Likes received", "int"), tt_field_def("full_name", "Name"), tt_field_def("bio", "Bio"), tt_field_def("verified", "Verified", "checkbox"), tt_field_def("is_private", "Private", "checkbox"), tt_field_def("is_business", "Business account", "checkbox"), tt_field_def("avg_engagement", "Avg engagement", "pct"), tt_field_def("like_engagement", "Like engagement", "pct"), tt_field_def("comment_engagement", "Comment engagement", "pct"), tt_field_def("external_url", "Link in bio", "url"), tt_field_def("tt_id", "TikTok id"), tt_field_def("profile_url", "Profile", "url"), tt_field_def("country_code", "Country"), tt_field_def("region", "Region"), tt_field_def("predicted_lang", "Language"), tt_field_def("account_created_at", "Account created", "date"), tt_field_def("source", "Read via"), tt_field_def("approx", "Counts are approximate", "checkbox", description="Checked when the counts on this row are rounded, not exact."), tt_field_def("source_payload", "Source data", "json"), ] #: One row per TikTok post. Identity is `shortcode`, and on TikTok that is a 19-digit numeric id — #: ✅ the SAME shape as the Comments dataset's `post_id`, so the comments→posts link joins on #: equality with NO normaliser (measured in the probe doc; Instagram needed one). TT_POST_FIELDS = [ tt_field_def("platform", "Platform"), tt_field_def("shortcode", "Post id"), tt_field_def("influencer_key", "Influencer"), tt_field_def("posted_at", "Posted at", "date"), # ⛔ THE VENDOR'S VOCABULARY IS `"video"` / `"content"` AND OURS HAS NO `"content"`. Declaring # the vendor's word would put an untranslated API token in front of a user; declaring an option # the mapper can emit but the select does not list would fail `_clean_field`. So the OPTIONS # stay this product's words and `connectors_tt.normalize_post` does the translation — the same # posture every other vendor value here takes. `carousel` is listed because a TikTok photo post # carrying more than one image IS one, and the mapper decides from `carousel_images` rather # than from the type token, which cannot express it. tt_field_def("type", "Type", "select", options=["image", "video", "carousel"]), tt_field_def("caption", "Caption"), tt_field_def("url", "URL", "url"), tt_field_def("hashtags", "Hashtags"), tt_field_def("tagged_location", "Commerce location"), # ⛔ `views` AND NO `plays` — see the divergence note at the top of this section. One vendor # number, one column. tt_field_def("views", "Views", "int"), tt_field_def("likes", "Likes", "int"), tt_field_def("comments", "Comments", "int"), tt_field_def("shares", "Shares", "int"), tt_field_def("saves", "Saves", "int"), tt_field_def("video_duration", "Video length (s)", "int"), tt_field_def("measured_at", "Engagement read at", "date", description="When the engagement numbers on this row were read."), tt_field_def("source_payload", "Source data", "json"), tt_field_def("comments_link", "Comment rows", "link", description="Comment records linked to this post.", link={"table": TT_COMMENTS_TABLE, "on": "shortcode", "from": "shortcode"}), tt_field_def("post_snapshots_link", "Measurement rows", "link", description="Engagement measurements linked to this post.", link={"table": TT_POST_SNAPSHOTS_TABLE, "on": "shortcode", "from": "shortcode"}), tt_field_def("measurements_captured", "Measurements captured", "rollup", description="How many measurements this post has.", rollup={"link": "post_snapshots_link", "fn": "countall"}), ] #: The POST series. ⚠ D-117 is live on the Instagram twin — `posted_at`/`type` are denormalised onto #: snapshot rows there and never reconcile with the post row afterwards. That pattern is NOT #: reproduced: this table carries the measurement and its identity, and asks `ut_tt_posts` for what #: kind of post it was. A fact that can disagree with its own dimension table is a fact nobody can #: trust, and the convenience it buys (one fewer hop in a rollup) is not worth a column that can be #: wrong. TT_POST_SNAPSHOT_FIELDS = [ tt_field_def("platform", "Platform"), tt_field_def("post_snapshot_key", "Snapshot"), tt_field_def("shortcode", "Post id"), tt_field_def("influencer_key", "Influencer"), tt_field_def("pulled_at", "Pulled at", "date"), tt_field_def("views", "Views", "int"), tt_field_def("likes", "Likes", "int"), tt_field_def("comments", "Comments", "int"), tt_field_def("shares", "Shares", "int"), tt_field_def("saves", "Saves", "int"), tt_field_def("source_payload", "Source data", "json"), tt_field_def("post_link", "Post", "link", description="Post record linked to this measurement.", link={"table": TT_POSTS_TABLE, "on": "shortcode", "from": "shortcode", "single": True}), ] #: Comments. Same posture as Instagram's: the SCHEMA and the LINK ship, the CAPTURE is opt-in and #: defaults OFF (`commentMetrics`), because comments are the single most expensive thing this #: product can buy and they ingest identifiable third parties who never entered anybody's list #: (D-24). ⚠ TikTok's `replies` is an ARRAY where ours is an INT count — a name collision, resolved #: in the mapper by storing `num_replies` and leaving the array in Source data. TT_COMMENT_FIELDS = [ tt_field_def("platform", "Platform"), tt_field_def("comment_key", "Comment"), tt_field_def("shortcode", "Post id"), tt_field_def("influencer_key", "Influencer"), # ⭐⭐ OWNER RULING 2026-08-12, verbatim: *"why isn't any of the comments column actually HAS # the comments content, fix it."* The text was bought, stored and INVISIBLE — retained whole in # `source_payload` under the 2026-08-07 superseding ruling (`instagram-capture.md` §4b), but # promoted to no column, so a person paying per comment record saw only Likes and Replies. # ⛔ THE COMMENT TEXT ONLY — the commenter's IDENTITY (`commenter_user_name`, `commenter_id`, # `commenter_url`; the vendor FLAGS the first as PII) stays in `source_payload` and gets no # column. The ruling names the comments' CONTENT, and promoting a third party's name and # profile URL into a filterable, exportable column is a different decision that nobody made. tt_field_def("text", "Comment"), tt_field_def("commented_at", "Commented at", "date"), tt_field_def("likes", "Likes", "int"), tt_field_def("replies", "Replies", "int"), tt_field_def("source_payload", "Source data", "json"), tt_field_def("post_link", "Post", "link", description="Post record linked to this comment.", link={"table": TT_POSTS_TABLE, "on": "shortcode", "from": "shortcode", "single": True}), ] #: table key -> the field list that defines it. ⭐ DERIVED CONSUMPTION IS THE POINT: `ut_ensure`, #: the gates and every future runner read THIS rather than naming five constants, so adding a sixth #: `ut_tt_*` table is one entry instead of a sweep. The Instagram side has no equivalent map, which #: is exactly why its table names are scattered across the module. TT_TABLE_FIELDS = { TT_PROFILE_TABLE: TT_PROFILE_FIELDS, TT_SNAPSHOTS_TABLE: TT_SNAPSHOT_FIELDS, TT_POSTS_TABLE: TT_POST_FIELDS, TT_POST_SNAPSHOTS_TABLE: TT_POST_SNAPSHOT_FIELDS, TT_COMMENTS_TABLE: TT_COMMENT_FIELDS, } #: The human labels a spawned `ut_tt_*` table wears in the database list. TT_TABLE_LABELS = { TT_PROFILE_TABLE: "TikTok profiles", TT_SNAPSHOTS_TABLE: "TikTok profile measurements", TT_POSTS_TABLE: "TikTok posts", TT_POST_SNAPSHOTS_TABLE: "TikTok post measurements", TT_COMMENTS_TABLE: "TikTok comments", } #: ⭐⭐ WAVE 31 · OWNER RULING R9 — the `ut_tt_*` children a person may not type records into. #: Owner, verbatim: *"Tiktok database for post and comments and their snapshots should also be a #: locked database with the lock icon, exactly like how instagram"*. #: #: ⛔ DECLARED ONCE BECAUSE THE DEFECT WAS TWO CALL SITES DISAGREEING WITH A THIRD. `ensure_ig_graph` #: passes `record_mode=AUTOMATION_RECORD_MODE` for all four Instagram children; TikTok's two spawn #: sites — `_spawn_presets`' child loop (SAVE) and `_tt_write_tables` (RUN) — each passed #: `lock_fields=True` and no `record_mode` at all. That is the whole bug: not a missing feature, one #: argument missing at two places, which is exactly the shape `_tt_write_tables`' own docstring warns #: about (*"one function, two callers, so the inline and deferred paths cannot answer differently"*). #: A SET plus a resolver means a sixth `ut_tt_*` table joins by declaration, not by remembering. #: #: ⚠ FOUR, NOT THE THREE R9 ENUMERATES, and the fourth is argued rather than assumed. #: `ut_tt_snapshots` is the profile measurement series — the exact twin of `ut_ig_snapshots`, which #: IS locked. R9's own comparison is *"exactly like how instagram"*, and the alternative is to leave #: a machine-append time series that a person can type into, one table away from three that they #: cannot. That is the fix-applied-to-one-platform-and-not-its-twin shape this very wave is closing #: elsewhere (D-177(c)). `ut_tt_profile` is deliberately ABSENT: it is the database a human adds #: handles to, and its Instagram counterpart is not locked either. #: #: ⚠ LOCKED DATABASE, NOT READ-ONLY (DESIGN.md §4): `records_mutable` goes False, so records cannot #: be added/edited/deleted — **adding a FIELD must still work**, and a UI hiding add-field here is a #: defect, not the intent. TT_LOCKED_TABLES = frozenset({TT_SNAPSHOTS_TABLE, TT_POSTS_TABLE, TT_POST_SNAPSHOTS_TABLE, TT_COMMENTS_TABLE}) #: ⭐ THE INSTAGRAM HALF, NAMED. It was only ever the four keys `ensure_ig_graph`'s loop happens to #: iterate, which is a set that exists but cannot be ASKED — so nothing could assert that IG and #: TikTok lock the same shape, and the QA sweep below could not be written at all. IG_LOCKED_TABLES = frozenset({IG_SNAPSHOTS_TABLE, IG_POSTS_TABLE, IG_POST_SNAPSHOTS_TABLE, IG_COMMENTS_TABLE}) #: ⭐⭐ W31 QA — OWNER, VERBATIM (2026-08-13): *"just like Instagram Post database (which is #: locked), only the IG Profile and TT Profile should be editable."* That is this set plus its #: complement: every `ut_ig_*`/`ut_tt_*` child is locked, and `ut_ig_profile` / `ut_tt_profile` are #: the two a person types handles into. Registered into `core.user_tables` at import so the lock is #: a DECLARATION rather than a flag some past automation run happened to stamp — see that module's #: `_LOCKED_RECORD_KEYS` for why the stored flag alone left production unlocked. #: ⛔ THE REGISTRATION ITSELF IS IN `main.py`, NOT HERE, and that is this module's own rule rather #: than a preference: it imports NOTHING from `core` at module level (see `MAX_UT_ROWS` and #: `MACHINE_OWNERS`, both local literals "so this module stays dependency-light for the API's boot #: path"). Adding `import core.user_tables` at line 3570 would put `core` on the import path of #: every process that touches the engine, to run one line. The composition root wires it, and #: `verify_automation` asserts that main.py CARRIES that call — a declaration whose registrar is #: missing is [[flag-shipped-without-its-writer]], which is the defect class this whole fix is in. LOCKED_CHILD_TABLES = IG_LOCKED_TABLES | TT_LOCKED_TABLES def tt_record_mode(table_key): """R9's answer for one `ut_tt_*` key — the `record_mode` its `ut_ensure` must carry. `""` for anything outside the locked set, which is what `ut_ensure` already treats as "say nothing about record mode", so an unlocked table is untouched rather than explicitly opened. """ return AUTOMATION_RECORD_MODE if str(table_key) in TT_LOCKED_TABLES else "" # --------------------------------------------------------------------------------------------- # THE INSTAGRAM CONNECTOR — moved out (wave 27 item 23). See `connectors_ig.py`. # --------------------------------------------------------------------------------------------- # Bright Data, Apify, the corpus routes and the anonymous ladder used to be ~1,900 lines HERE. # They are a CONNECTOR: they know what a vendor's rows look like. Nothing in the automation # runtime needs that, and the runtime is what this file is for. # # ⭐ THE LIST BELOW IS A DEPENDENCY STATEMENT, NOT A CONVENIENCE. It is every name the automation # runtime still reaches for after the split — twenty, down from the seventy-one that used to be # defined here — so `git diff` on this block is how anyone sees the engine growing a new vendor # dependency. Re-exporting them is also REQUIRED rather than tidy: `routes_automation` and # `routes_connectors` call `engine.bd_ready()` on the module object, and this is what keeps that # true without those files having to learn where the wire went. # ⚠ SO THE RE-EXPORT ALSO MEANS `engine.bd_call` STILL RESOLVES, and a gate that only checked # that would be green whether or not anything moved. `verify_automation`'s `section_split` # therefore asserts `__module__` on the whole moved set — it derives the split from the RUNNING # system instead of trusting this import line ([[gate-answers-the-wrong-question]]). # # ⚠ AND IT IS DELIBERATELY HERE, mid-file, rather than at the top. `connectors_ig` reaches back # for the SSRF rail (`fetch`/`fetch_json`/`Refused`) and for `PACE_SECONDS`; it does so lazily, # inside its functions, precisely so the two modules cannot cycle. This import sits BELOW the rail # and below `PACE_SECONDS`, so even if somebody later makes one of those reach-backs a # module-level import, the names it wants already exist and the cycle still resolves. # ⭐⭐ WAVE 30 · T09 (D-128) — TWO IMPORT BLOCKS NOW, AND THE SPLIT BETWEEN THEM IS THE POINT. # `connectors_bd` is the SUPPLIER's wire — it serves both platforms and knows neither. `connectors_ig` # is INSTAGRAM's vocabulary: its dataset ids, its row mappers, its free rungs. Anything that would # have to change to serve a third platform belongs in the second block, not the first. # ⚠ `bd_ready` is re-exported through this module ON PURPOSE — `routes_automation` and # `routes_connectors` both reach `engine.bd_ready()` on the module object, and those files belong to # other sessions. Dropping it here would 500 two surfaces that never mention a vendor. from connectors_bd import ( # noqa: E402 BD_EXCLUDE_MAX, BD_PATH_SNAPSHOT, BD_RECORD_PRICE_SPEC, _bd_deferral, _bd_first_url, _bd_rows, _first, _ig_int, bd_call, bd_filter_rows, bd_filter_start, bd_filter_status, bd_ready, bd_snapshot_progress, depth_refusal, ) from connectors_ig import ( # noqa: E402 BD_DS_COMMENTS, BD_DS_POSTS, BD_DS_PROFILES, BD_DS_REELS, _bd_comment, _bd_post_metrics, _bd_profile, _bd_tagged_location, DEFERRED_MARK, apify_profile, bd_profiles_batch, ig_handle, public_source, pull_profile, select_post_groups, top_up_views, ) # --------------------------------------------------------------------------------------------- # DISCOVERY — sourcing handles we do NOT already know (owner ruling R7, DEBT D-23) # --------------------------------------------------------------------------------------------- # The engine until now only ENRICHED a profile set somebody typed in. This queries the vendor's # **620,000,000-record pre-collected Profiles corpus** with a real query language and returns # handles nobody here has ever seen. MEASURED end-to-end 2026-08-04: `followers 10k–200k AND # biography includes "floral"` returned five real profiles, one of them a verified 20.8k-follower # Georgia/Florida floral-design company — Fisch Floral's actual market. # # ⛔ FIVE PROPERTIES, EACH OF THEM A MEASURED FAILURE MODE RATHER THAN A PREFERENCE: # # 1. **A DIFFERENT ROUTE AND A DIFFERENT NAMESPACE.** `POST /datasets/filter` — ⚠ with NO `/v3/` # segment; `/datasets/v3/filter` is a 404 that once got written down as "the corpus is # unreachable". Its snapshots are `snap_…` and are read at `/datasets/snapshot/…`; the # scraper's are `sd_…` at `/datasets/v3/snapshot/…`, and the two 404 each other. # 2. **`records_limit` IS REQUIRED.** Unbounded queries die `NOT_ENOUGH_FUNDS` (code 104, # `per_set` billing). ⚠ And bounding is NOT sufficient: **scan time tracks PREDICATE BREADTH, # not row count** — a `records_limit:10` query on a bare `followers > 10000` was still # `building` 50 minutes later. So the predicate is capped too, and the runner tolerates a # snapshot that is not ready when it looks. # 3. **IT IS SLOW, AND THE RUN HANDS OFF RATHER THAN HOLDING A THREAD.** MEASURED delivery # latency on the narrow query: **19.6 minutes** (`created` 12:38:00 → `delivery_time` # 12:57:42). Blocking a worker thread for twenty minutes on a free-tier container to poll is # the wrong shape, so a run polls for a short budget and then PERSISTS the snapshot id; the # next run collects it. A `partial` that says "still building, the next run collects it" is # the honest report of exactly what happened. # 4. **PROMOTION IS MANUAL, ALWAYS (R7).** The measured result set was ~40% on-target and # included a hashtag-aggregator account that is not a person. Auto-promoting a candidate into # a wider profile set would silently multiply the enrichment bill AND pollute the # snapshot series with rows nobody chose. This path never decides WHICH rows to enrich. # 5. **SERIAL.** `429 too_many_parallel_jobs` is real and the original probe's most important # test died on it. # # ⚠ THE PRICE IS NOT MEASURED AND MAY NOT BE PRESENTED AS IF IT WERE. The funds gate fires BEFORE # the API returns a number, `price: 0` means "not priced" rather than "free", and # `/customer/balance` answers 403 for our token — so there is no balance to read and no spend to # report. Every estimate below is derived from the PRICING PAGE and says so. #: The table a discovery run writes when the tenant has no Instagram profile database yet — the #: FIRST one, never a second (`discover_default_table` elects an existing one before this is #: reached). Upsert by handle, so re-finding a profile is free. #: #: ⭐ 2026-08-10 — RENAMED FROM `ut_ig_candidates` / "IG candidates" (owner: *"We only need ONE IG #: profile so it's not confusing"*). The old pair was two kinds of wrong at once. The KEY said #: `candidates` while every other table in the graph says what it holds (`ut_ig_posts`, #: `ut_ig_comments`, `ut_ig_snapshots`, `ut_ig_post_snapshots`) — and "candidate" was a concept #: W26/R6 retired when it deleted `tracked`: a row here is a PROFILE, and whether anyone wants it #: is a stage column's business, not the table's name. The LABEL then disagreed with the key on #: any tenant who renamed the table, which is exactly the mismatch the owner hit. #: #: ⚠ SAFE TO MOVE ONLY BECAUSE NOTHING HOLDS THE OLD KEY. Census 2026-08-10 across all three #: tenant stores: royal-imports ABSENT, gtmlab has no tables at all, and nurilab's was deleted the #: same day (0 rows, 0 views, 0 dependent automations). An automation that stored the old key #: explicitly still resolves to it — an explicit target is never rewritten — so a legacy tenant #: would keep working; there simply is not one. #: ⛔ A future rename is NOT this cheap. Once rows exist under a key, moving it is a migration #: touching rows, `link.table`, every stored `targetTable` and the `ut__*` bucket family. DISCOVER_TABLE = "ut_ig_profile" #: The label that key is CREATED with. `ut_ensure` sets a label only on create and never relabels, #: so changing this can rename nothing that already exists. One constant because it was three #: copies of the same string literal, and a default spelled three times is one edit from drifting. DISCOVER_LABEL = "IG profile" #: A hard ceiling on one run's ask — and it is DERIVED, not chosen. A corpus row MEASURED at #: ~35 KB (`file_size` 175,617 for 5 rows), so 1,000 records ≈ 33 MB would have crossed #: `BD_MAX_KB` and come back as a truncated document **after being billed**. 500 leaves real #: headroom. ⚠ If the row size grows, this number is the thing to re-derive. BD_MAX_RECORDS = 500 #: How long ONE run waits on a building snapshot before handing off to the next run (see 3 above). BD_FILTER_WAIT = float(os.environ.get("AIOS_BD_FILTER_WAIT") or 120) BD_FILTER_POLL = float(os.environ.get("AIOS_BD_FILTER_POLL") or 15) #: The operator set, enumerated BY REJECTION — send a bogus one and the API's own validation error #: lists the legal set. The cheapest kind of measurement, and it makes this list a fact. BD_FILTER_OPS = ("=", "!=", "<", "<=", ">", ">=", "in", "not_in", "includes", "not_includes", "array_includes", "not_array_includes", "is_null", "is_not_null") #: Operators that take NO value (everything else requires one). BD_NULLARY_OPS = ("is_null", "is_not_null") #: Fields a predicate may name. MEASURED-accepted (24 of 25 probed; **`country_code` is REJECTED** #: by the API and is therefore absent rather than offered-and-broken). #: ⛔ `email_address`, `phone_number` and `business_email` ARE filterable and are DELIBERATELY #: MISSING. Selecting people BY CONTACT DETAIL across 620M records is the materially heavier #: privacy posture D-24 flags — a different licence question from per-URL enrichment, and one the #: owner has not been asked. A vocabulary that cannot express it cannot be asked for it by #: accident; adding them back is a decision, not a typo. BD_FILTER_FIELDS = ("followers", "following", "posts_count", "avg_engagement", "biography", "category_name", "business_category_name", "is_business_account", "is_professional_account", "is_verified", "account", "full_name", "external_url", "bio_hashtags", "post_hashtags", "profile_url", "related_accounts", "profile_name", "id", "fbid", "highlights_count") #: ⭐ The three the FIND SURFACE leads with, and the reason is measurement rather than taste: #: these are the fields seen carrying values on real CORPUS rows. `category_name` and #: `related_accounts` are filterable and were null/empty on every row we have looked at — a filter #: on an unpopulated field returns nothing and looks exactly like "no such influencers exist". BD_FILTER_LEAD = ("followers", "biography", "avg_engagement") #: ⭐⭐ WAVE 32 · T46 / DEBT D-167 — THE FILTER VOCABULARY HAS A PLATFORM NOW, AND IT COST MONEY #: NOT TO. `BD_FILTER_FIELDS` above was measured against INSTAGRAM in wave 22; waves 29 and 30 hung #: a second platform on the same tuple, so a `discover_tiktok` could be built on any of the **16 #: fields Bright Data's TikTok Profiles dataset does not have** — and `BD_FILTER_LEAD`'s third #: field, `avg_engagement`, is one of the three the Find panel LEADS with and is absent there. #: A search on a field the corpus does not carry does not error: it returns nothing, and reads #: exactly like *"no such creators exist"* after the money has been spent. #: #: ⛔ MEASURED, NOT REASONED — W30-B20, against TikTok's 40 declared dataset fields: **5 of the 21 #: names survive.** They are these. Anything else is refused at the door with the field named, #: which is `clean_predicates`' existing posture (*"refuses rather than coerces … the alternative #: turns 'find me verified accounts in Georgia' into 'find me any account' and bills for the #: difference"*) — now applied per platform rather than per product. #: ⚠ NOT DERIVED FROM `connectors_tt.normalize_profile`, TEMPTING AS THAT IS. That mapper declares #: the vendor's READ names on a returned row; these are the names its FILTER validator accepts, and #: the two vocabularies are not the same thing on either platform (`awg_engagement_rate` reads, #: `avg_engagement` filters). Deriving one from the other would look rigorous and be wrong. TT_FILTER_FIELDS = ("followers", "following", "biography", "is_verified", "id") #: The two of `BD_FILTER_LEAD`'s three that TikTok actually carries. `avg_engagement` is dropped #: for the reason above — leading with a field the corpus cannot answer is worse than leading with #: two. TT_FILTER_LEAD = ("followers", "biography") def filter_fields(kind=""): """The searchable field names for one discovery kind — `(fields, lead)`. ⛔ ONE ACCESSOR, so the ROUTE that publishes the vocabulary and the VALIDATOR that enforces it cannot come to disagree about it. That divergence is D-167 itself: the route served 21 names with no platform dimension and `clean_predicates` validated against the same tuple with no kind parameter, so both halves were consistently wrong together and nothing could notice. """ if str(kind or "") == "discover_tiktok": return TT_FILTER_FIELDS, TT_FILTER_LEAD return BD_FILTER_FIELDS, BD_FILTER_LEAD #: ⛔ C4 (wave 22) — THE TOO-BIG GUARD, and its unit is the PREDICATE, not the row count. #: MEASURED (2026-08-04): `records_limit: 10` over a bare `followers > 10000` was still #: `building` 50 minutes later — scan time tracks PREDICATE BREADTH; bounding the rows does not #: bound the scan. What actually narrows a 620M-row corpus is CONTENT: a text/array match on a #: field that discriminates. Numeric ranges and booleans partition the corpus into slabs the #: scanner still has to walk, so they refine a search and cannot BE one. BD_NARROWING_FIELDS = frozenset({ "biography", "account", "full_name", "profile_name", "category_name", "business_category_name", "external_url", "bio_hashtags", "post_hashtags", "related_accounts", "id", "fbid", "profile_url"}) #: The ops that actually pin content — `!=`/`not_includes` on a text field matches nearly the #: whole corpus, which is breadth wearing a condition's clothes. BD_NARROWING_OPS = frozenset({"=", "in", "includes", "array_includes"}) #: Fields MEASURED carrying values on real corpus rows — D-25's fill-rate table, 50 mixed #: profiles (snapshot `snap_msfrlbmt7qvej8uby`, collected 2026-08-05). The bar is ≥60% #: populated: below it a filter misses more corpus than it matches, and the surface should #: say so. MEASURED AND EXCLUDED: category_name 48%, business_category_name 34%, #: related_accounts 22%, post_hashtags 18%, bio_hashtags 10% — and the PII trio the map never #: carries anyway read 2%/0%/0%, so even the thing R3 forbids would barely have worked. #: ⚠ posts_count was 100% populated on CORPUS rows — the fabricated-zero finding #: (`_bd_posts_count`) is a SCRAPE-path fact and both stay true. BD_POPULATED_FIELDS = frozenset({ "account", "followers", "following", "posts_count", "avg_engagement", "biography", "external_url", "is_business_account", "is_professional_account", "is_verified", "full_name", "profile_name", "highlights_count", "id", "fbid", "profile_url"}) BD_MIN_NARROWING = 1 # ── THE FIELD SCHEMA — what a person sees, and what they are allowed to ask. ────────────────── # ⛔ THE SURFACE USED TO SHOW THE VENDOR'S OWN COLUMN NAMES AND ALL FOURTEEN OPERATORS ON EVERY # ROW. So `is_business_account` offered `>=`, `fbid` sat in the list with no explanation of what # it is, and `not_array_includes` was a thing a customer was expected to reason about. Owner: # *"look at each damn schema and only provide toggles operator that make sense"*. # # ⚠ THE LABEL IS NOW LOAD-BEARING IN BOTH DIRECTIONS. The old comment in `AutomationFind.tsx` # defended raw names on the grounds that a refusal names the field exactly as the row does — and # it was right, which is why `field_label()` is used by the REFUSALS too. Prettifying only the UI # is how you get an error message about `bio_hashtags` on a row labelled "Hashtags in bio". BD_FIELD_LABELS = { "followers": "Followers", "following": "Following", "posts_count": "Post count", "avg_engagement": "Engagement rate", "biography": "Bio", "category_name": "Category", "business_category_name": "Business category", "is_business_account": "Business account", "is_professional_account": "Professional account", "is_verified": "Verified", "account": "Handle", "full_name": "Name", "profile_name": "Profile name", "external_url": "Link in bio", "profile_url": "Profile link", "bio_hashtags": "Hashtags in bio", "post_hashtags": "Hashtags in posts", "related_accounts": "Related accounts", "id": "Instagram ID", "fbid": "Facebook ID", "highlights_count": "Story highlight count", } #: A one-line "what IS this" for the rows nobody can be expected to guess. Absent = self-evident; #: a hint on all 21 rows is a hint on none of them (the wave-24 chip lesson). BD_FIELD_HINTS = { "account": "The @username", "avg_engagement": "Likes and comments as a share of followers", "id": "Instagram's internal number for the account — for matching a list you already have", "fbid": "The linked Facebook id — for matching a list you already have", "profile_name": "The name shown above the bio, when it differs from the account name", "related_accounts": "Accounts Instagram suggests alongside this one", } #: The KIND decides the comparisons offered and the control drawn for the value. BD_FIELD_KINDS = { **{n: "number" for n in ("followers", "following", "posts_count", "avg_engagement", "highlights_count")}, **{n: "boolean" for n in ("is_business_account", "is_professional_account", "is_verified")}, **{n: "tags" for n in ("bio_hashtags", "post_hashtags", "related_accounts")}, **{n: "choice" for n in ("category_name", "business_category_name")}, **{n: "text" for n in ("biography", "account", "full_name", "profile_name", "external_url", "profile_url", "id", "fbid")}, } #: ⛔ ONLY WHAT MAKES SENSE, and the ORDER is the order they are offered in — the first entry of #: each list is what `default_operator` must agree with. BD_OPS_BY_KIND = { # ⚠ `in`/`not_in` are DELIBERATELY ABSENT. "is any of" is what `=` becomes the moment a second # value is typed (see `BD_MULTI_VALUE_OPS`), so offering both would be two controls for one # idea — and the second one is the one written in vendor grammar. "text": ("includes", "not_includes", "=", "is_not_null", "is_null"), "choice": ("=", "!=", "is_not_null", "is_null"), "number": (">=", "<=", "=", ">", "<"), # A yes/no field has exactly one sensible comparison and a two-option value. `!=` on a boolean # is `=` with the other value, written the confusing way. "boolean": ("=",), "tags": ("array_includes", "not_array_includes", "is_not_null", "is_null"), } #: Plain English for every comparison the surface can offer. The vendor's token stays on the wire #: (it is what the API accepts); nobody has to read it. BD_OP_LABELS = { "includes": "contains", "not_includes": "does not contain", "=": "is", "!=": "is not", "in": "is any of", "not_in": "is none of", ">=": "at least", "<=": "at most", ">": "more than", "<": "less than", "array_includes": "has any of", "not_array_includes": "does not have", "is_null": "is empty", "is_not_null": "is not empty", } #: Yes/No, so a boolean is a two-item dropdown instead of a box you type `true` into. BD_BOOLEAN_OPTIONS = [{"value": "true", "label": "Yes"}, {"value": "false", "label": "No"}] #: ⭐ THE OPERATORS THAT MAY CARRY SEVERAL VALUES — owner item 3, "if i want to include many #: keywords like floral, flower, beauty". One condition row holds the list; `bd_filter_start` #: expands it into a NESTED OR group, so it composes with the other conditions instead of forcing #: the whole search to "match any" (which would also drag Followers into the union — the hole the #: OR guard closed the same day). #: #: ⛔ POSITIVE OPERATORS ONLY, and this is a correctness line rather than a scope line. "does not #: contain floral OR does not contain flower" matches nearly every account alive — a negative over #: a list is an AND (De Morgan), and quietly OR-ing it would build a filter that reads like a #: narrowing and behaves like the whole corpus. Negatives stay single-valued until someone needs #: them enough to write the AND branch. BD_MULTI_VALUE_OPS = frozenset({"includes", "=", "array_includes"}) MAX_PREDICATE_VALUES = 12 def field_label(name): """The human name for a searchable field — used by the SURFACE and by every REFUSAL.""" return BD_FIELD_LABELS.get(name) or str(name or "that field") def field_kind(name): return BD_FIELD_KINDS.get(name, "text") def ops_for(name): """The comparisons this field may be asked. A tuple, in offer order.""" return BD_OPS_BY_KIND.get(field_kind(name), BD_OPS_BY_KIND["text"]) #: ⭐ WHAT A **FRESH** CONDITION ON A FIELD STARTS AS — and it lives HERE, beside the narrowing #: law, because the two drifted apart and the drift cost a user their first automation. #: #: ⛔ THE SCAR, IN THREE WAVES. Wave 21 fixed a seeded condition the server refused for having no #: value by making a new condition NULLARY (`is_not_null` — "this field has any value"), and its #: comment called that "narrowing-in-the-right-direction". Wave 22 then wrote `BD_NARROWING_OPS` #: and `is_not_null` is NOT IN IT, which silently made that default a condition the save door #: always refuses. Wave 24 deleted the create wizard, so the Find panel became the ONLY way in — #: and the refusal moved from a corner case to the first thing a new user meets. Three waves, one #: sentence of drift, and nothing red anywhere at any point. #: #: So the rule is now DERIVED and ASSERTED (`verify_automation.py`): a fresh condition on a field #: that CAN narrow must narrow. The client asks for `defaultOperator` per field and never picks #: one itself — a client-side default is a second copy of this table, free to disagree with the #: guard that judges it, which is exactly what happened. BD_ARRAY_FIELDS = frozenset({"bio_hashtags", "post_hashtags", "related_accounts"}) BD_BOOLEAN_FIELDS = frozenset({"is_business_account", "is_professional_account", "is_verified"}) #: Narrowing fields whose values are IDENTIFIERS — a substring match on a handle or an id is a #: worse question than an equality, and both narrow. BD_EXACT_FIELDS = frozenset({"account", "id", "fbid", "profile_url"}) def default_operator(name): """The operator a NEW condition on `name` starts with. Narrowing wherever the field can narrow, so a fresh condition is ONE step (type a value) from saveable rather than two (change the comparison, then type a value) — with the second step being one the surface never told anyone to take. """ kind = field_kind(name) if kind == "tags": return "array_includes" if kind in ("boolean", "choice"): # ⚠ `choice` LANDS HERE AND NOT ON `includes`, which is the arm it used to fall through # to (Category is a narrowing field). `includes` is not in `BD_OPS_BY_KIND["choice"]`, so # the default would have been an operator the same module refuses to offer — the exact # default-versus-guard split this function was written to close, reintroduced one commit # later by adding a kind. Asserted per field in `verify_automation.py`. return "=" if kind == "number": # `>=` cannot narrow the scan and is not pretending to — it is the comparison a person # reaching for Followers means, and the guard's sentence is the honest answer when it is # the ONLY condition present. return ">=" return "=" if name in BD_EXACT_FIELDS else "includes" def predicate_narrows(p): """Does ONE predicate narrow the corpus scan? Content field + content op, nothing else.""" return (str((p or {}).get("name") or "") in BD_NARROWING_FIELDS and str((p or {}).get("operator") or "") in BD_NARROWING_OPS) def narrowing_refusal(preds, operator="and", kind=""): """C4's server law: the sentence when a predicate set does not narrow, else ''. One function so create/patch (via `clean_predicates`) and RUN (legacy stored configs predate the law) refuse in the same words. ⭐ THE JOIN IS PART OF THE LAW, and it was missed for two waves. The guard asked "does ANY predicate narrow?" — a question that is only correct under AND. Under **OR** the result is a UNION, so the query is exactly as broad as its WIDEST branch: `biography includes "florist" OR followers >= 10000` saved clean and asked the vendor for every account over ten thousand followers, which is the measured 50-minute / NOT_ENOUGH_FUNDS shape the guard exists to stop. A money wall that a dropdown three rows above it can walk through is not a wall. So: under OR every branch must narrow; under AND one is enough. An EMPTY set never narrows either way — `all([])` is True, which would have made "no conditions at all" the widest legal query of the lot. """ preds = list(preds or []) if str(operator or "").lower() == "or": if preds and all(predicate_narrows(p) for p in preds): return "" return ("with Match set to any, every condition has to describe the account itself " "(Bio, Handle, Name, a hashtag). Matching any means the results are added " "together, so one follower or yes/no condition widens the whole search. Narrow " "every condition, or set Match to all") if any(predicate_narrows(p) for p in preds): return "" # ⭐ WAVE 32 · T46 (D-167's neighbour) — THE NETWORK'S OWN NAME. This sentence said # "Instagram" to somebody building a TIKTOK search, on the refusal they are most likely to # read, because the string predates the second platform. `discovery_facts` already carries the # network's name for exactly this class of string, so there is no second literal. _net = discovery_facts(kind)[0] if kind else PLATFORM_INSTAGRAM return ("add a condition that describes the account itself — Bio contains, Handle is, a " f"hashtag. Follower counts and yes/no conditions alone match too much of {_net} " "to search") def filter_meta(): """C4's per-field flags for the Find surface: `[{name, populated, narrowing, defaultOperator}]` over the same 21 names `BD_FILTER_FIELDS` offers (the 3 PII fields stay structurally absent, R3). `defaultOperator` rides here rather than being a client constant for the reason written over `default_operator`: the client's own default was `is_not_null` for every field, which the narrowing guard refuses on every field. """ return [{"name": n, "label": field_label(n), "hint": BD_FIELD_HINTS.get(n, ""), "kind": field_kind(n), "populated": n in BD_POPULATED_FIELDS, "narrowing": n in BD_NARROWING_FIELDS, "defaultOperator": default_operator(n), # The comparisons THIS field may be asked, already in plain English and already in # offer order. A client that filtered a global list by field kind would be a second # copy of `BD_OPS_BY_KIND`, and the copy is what goes stale. "operators": [{"value": o, "label": BD_OP_LABELS.get(o, o), "nullary": o in BD_NULLARY_OPS, # `in`/`is any of` and the tag operators take a LIST — the control # has to know that, and deriving it from the token in the client is # the same second-copy mistake one level down. "multi": o in BD_MULTI_VALUE_OPS} for o in ops_for(n)], "options": BD_BOOLEAN_OPTIONS if field_kind(n) == "boolean" else []} for n in BD_FILTER_FIELDS] #: ⭐ WAVE 25 · CONTRACT C1 — THE UNIFIED, CROSS-TENANT INSTAGRAM PRESET SET. #: #: ONE server-owned list. Every tenant gets the same keys and the same labels, which is what #: "unified across tenant" means and what makes the pooled master series joinable at all — two #: tenants calling the same number `followers` and `follower_count` is a pool you cannot query. #: ⛔ THERE IS NO CLIENT COPY OF THIS LIST. C and D read it off the wire (`GET #: /automations/presets`); a client-side copy is the wave-9 silent-drop failure in a new hat. #: #: ⭐ R3 — THESE HOLD **LATEST**, AND NOTHING ELSE. The full time series stays as append rows in #: `ut_ig_snapshots`. ONE STORE FOR ONE SERIES: no per-record `json` history column, no second copy #: of a number that already has a home. #: ⛔ W29-T34 — THIS LINE USED TO ADD *"which the metric fields already read"*, AND THAT WAS WRONG. #: A `metric` field reads the PLATFORM-WIDE MASTER, not this tenant's snapshot table: #: `compute_metric_cells` imports `ig_master` and calls `ig_master.series_for(...)` — a #: cross-tenant pooled repo in a different HF dataset, which is the whole point of C6's pooling and #: which no rollup kind can reach. The metric header below says exactly that, so the module was #: contradicting itself on which store answers the question. #: ⚠ And the practical finding behind the correction, recorded because it is surprising: a live #: census of all four tenants (13 tables / 228 fields / 64 store files) found **ZERO** metric fields #: in existence, while 47 rollups are live. The vocabulary stays by owner ruling; nothing uses it. #: #: ⭐⭐ 2026-08-07 — THIS IS NOW *EVERY* PROFILE FIELD THE VENDOR RETURNS, BY OWNER INSTRUCTION, #: AND THAT REVERSES D-25's FILL-RATE RULE. Owner, verbatim: *"make sure to have ALL Fields #: available to us from Bright Data to be pre-set Fields for us and populated."* #: #: ⛔ THE RULE IT REPLACES IS WRITTEN DOWN HERE RATHER THAN DELETED, because it was a good rule #: and the next reader will otherwise re-derive it and prune these columns back. It said: a #: LATEST-value column costs something a snapshot row does not — it is a column every user sees on #: their grid forever — so the split followed D-25's MEASURED fill-rates (50 mixed profiles, #: snapshot `snap_msfrlbmt7qvej8uby`) and a field earned a column only at ≥60% populated. That is #: why `business_category` (34%), `bio_hashtags` (10%), `post_hashtags` (18%) and `pronouns` were #: captured into the snapshots and were NOT preset columns. #: #: ⚠ THE MEASUREMENT IS UNCHANGED AND STILL WORTH KNOWING — several of these columns WILL be #: mostly blank on the scrape path, and D-25's central finding is why: **the scrape path and the #: corpus path are different data.** `avg_engagement` is 0% populated on scrape and 88% on corpus; #: `category` is 0% on scrape and 70% on corpus. So a blank here is very often "this ROUTE does #: not carry it", not "this account does not have it" — which is exactly what `enriched_at` and #: the blank-never-zero law exist to keep honest. What changed is the ANSWER to "does a #: sometimes-blank column earn a place on the grid", and that was always the owner's call to make. #: Source data retains every field the provider returns. The only separate decision is whether to #: request the paid full Comments dataset (`commentMetrics`, default OFF). #: #: ⭐ WAVE 26 · R5 — THE NETWORK VOCABULARY. A handle is only unique WITHIN a network, so this is #: half of the candidate identity (contract C3: the upsert key is `(platform, handle)`). #: #: ⚠ THE VALUES ARE DISPLAY STRINGS ON PURPOSE. They land in an ordinary `text` cell that a person #: reads, filters and groups by, so `"Instagram"` beats `"ig"` — and the set is small, closed and #: written down here rather than inferred from a runner's module name, because the day a second #: runner spells it `"instagram"` is the day the dedup key stops working and nothing errors: the #: profile simply appears twice. #: ⭐ 2026-08-07 — WHERE A PROFILE HANDLE POINTS, as `core.user_tables.PROFILE_SOURCES` spells it. #: A local literal for the same boot-path reason `MACHINE_OWNERS` and `UT_FIELD_TYPES` are locals #: here, and held in step by a gate rather than an import. ⛔ NOT the same vocabulary as #: `PLATFORM_*` below: this names the FLAG's source (which validator reads the cell), that names #: the NETWORK a row belongs to (half of the identity). They read alike and mean different things, #: which is exactly why both are written down instead of inferred. PROFILE_SOURCE_IG = "instagram" #: ⚠ Its TikTok twin, `PROFILE_SOURCE_TT`, is declared UP with the `ut_tt_*` schemas (wave 29): the #: field lists there are evaluated at import and would not see a constant defined here. PLATFORM_INSTAGRAM = "Instagram" #: ⭐ WAVE 29 — the `ut_tt_*` family stamps this on every row of all five of its tables. (It really #: was written by nothing until D-9 landed; the note that said so is kept in the log, not here.) PLATFORM_TIKTOK = "TikTok" PLATFORM_FACEBOOK = "Facebook" PLATFORMS = (PLATFORM_INSTAGRAM, PLATFORM_TIKTOK, PLATFORM_FACEBOOK) def discovery_facts(kind): """⭐⭐ WAVE 30 · T05 — THE THREE THINGS THAT GENUINELY DIFFER BETWEEN THE DISCOVERY KINDS: the network's NAME, the database a run writes to when the config names none, and that database's label. `(platform, table, label)`. ⛔ WHY A FUNCTION AND NOT A DICT LITERAL: `DISCOVER_TABLE` and `DISCOVER_LABEL` are declared several hundred lines BELOW this point, so a dict evaluated here would NameError at import. A function body resolves at call time and does not care. (`PROFILE_SOURCE_IG`'s note directly above records the same import-order trap costing a constant its natural home.) ⛔ AND WHY IT EXISTS AT ALL. These three facts were written out inline at FIVE call sites — `clean_config`, `graph`'s discovery arm, `graph`'s `det` dict, `_flow_table` and `compose_sentence` — and wave 29 taught four of the five about TikTok by not touching them: they tested the string `"discover_instagram"`, so a stored TikTok automation fell through to a default meant for Instagram, or to no arm at all. Every one of those misses was SILENT and every gate stayed green. `DISCOVERY_KINDS` answers *"is this a corpus search?"*; this answers *"whose?"* — and between them a sixth platform is one tuple entry and one arm here, rather than a hunt through the file for string comparisons somebody has to think to look for. ⚠ DELIBERATELY NOT A PLATFORM REGISTRY. The dataset ids, the field maps and the row mappers stay in the connector modules that own them. This is the presentation-and-default quartet the ENGINE needs, and widening it is how it becomes a second `SOURCES` with a longer name. ⭐ T06 ADDED THE FOURTH ELEMENT, `spawn_fields` — the field list `_presets_after_write` gives the target database on SAVE. It belongs here rather than beside that function because it must equal what the RUN-TIME `ut_ensure` uses, and wave 25's R10 exists precisely to stop those two disagreeing: columns that change between Save and the first run are two different answers to "what does this database look like". ⚠ THE TWO PLATFORMS PASS DIFFERENT-LOOKING LISTS AND THAT IS CORRECT, not an oversight. Instagram spawns `CANDIDATE_FIELDS` (= the preset set PLUS the discovery bookkeeping) because its preset list alone is a subset of what its runner writes; TikTok's `TT_PROFILE_FIELDS` already carries its own bookkeeping columns, so it IS the whole set. Each side is the list its own runner ensures — which is the rule, rather than "both use the one named CANDIDATE". """ if kind == "discover_tiktok": return (PLATFORM_TIKTOK, TT_PROFILE_TABLE, TT_TABLE_LABELS[TT_PROFILE_TABLE], TT_PROFILE_FIELDS) return PLATFORM_INSTAGRAM, DISCOVER_TABLE, DISCOVER_LABEL, CANDIDATE_FIELDS #: ⭐ WAVE 26 · R3 — THE TYPES ARE HONEST NOW, AND THE MIGRATION IS THE PRICE OF THAT. #: #: This block used to say the `text` types were deliberate, and the reasoning was sound as far as #: it went: `ut_ensure` merges by KEY and never re-types an existing column, so re-declaring #: `followers` as `int` gives NEW tables a schema every table already in production does not have #: — one vocabulary with two shapes, the exact drift this list exists to prevent. That argument #: was never wrong; it was an argument for doing the MIGRATION, and the note used it as a reason #: not to. Owner, 2026-08-06: *"Why is everything that instagram field found is in a text format? #: change this."* #: ⛔ SO THE TWO HALVES ARE INSEPARABLE AND SHIP TOGETHER. Declaring a type here without #: `migrate_ig_field_types()` reintroduces exactly the split-schema the old note feared — and it #: would do it silently, because a merged-by-key field list produces no error when two tables #: disagree about what a column IS. #: #: ⚠ `avg_engagement` CARRIES A UNIT CONVERSION, NOT JUST A TYPE (amendment C1-a). The vendor #: sends a 0–1 fraction (MEASURED: 0.0074 / 0.0656 / 0.0014 / 0.0148 / 0.0274 / 0.0091) and this #: product's `pct` renders the stored number with a `%` appended — so storing the raw fraction #: would print every creator in the book as `0.0%`. It is stored ×100 from here on, at the #: mapping and in the migration both. See `_pct100`. PRESET_PROFILE_FIELDS = [ # ⭐ WAVE 26 · R5 — WHICH NETWORK THIS HANDLE IS ON, and it is half of the identity now. # `@inayma` on Instagram and `@inayma` on TikTok are two accounts owned by two different # people as often as not, so the dedup key is (platform, handle) and never handle alone # (C3). ⛔ NOT called `source`: `SNAPSHOT_FIELDS` already spends that word on *which rung # answered* (`field_def("source", "Read via")`), and one word meaning two things across two # tables in one product is the drift this whole list exists to prevent. field_def("platform", "Platform"), # ⭐⭐ 2026-08-07 (owner ruling) — THE HANDLE IS THE PRIMARY COLUMN *AND* THE ENRICH BINDING. # Owner: *"if a user use this automation for instagram scraping, the Unique ID will always be # REPLACED or made as the Handle. in any case the user can add their own record right and edit # those records."* Two declarations, one field, and they are the same sentence read twice: # # `pinned` — the grid's identity column is `find(f => f.pinned) ?? fields[0]`, so WITHOUT # this the primary is an accident of creation order. It is a DECLARATION, not a # reorder: the stored field order is left alone (saved view orders keep working) # and the client pins the column to position 0 wherever it sits. That is why this # fixes tables that already exist without moving a single stored field. # `profile` — C3/R7's flag. It makes `handle` the column `enrich_instagram` BINDS to # (`profile_field_key` step 2), which is what turns "I added a profile to my # database, how do I enrich it?" from an unanswerable question into a step. It # also validates the cell (`@Name`, a pasted instagram.com link and a bare handle # all normalise to the bare handle) and routes a typed value to the SHARED # definition row rather than one user's overlay (C3-A1) — which is precisely what # makes the owner's second clause true: a record you add BY HAND is a record the # engine can read and enrich. # # ⛔ THE TWO KEYS MUST SURVIVE `_clean_field` UNCHANGED or `verify_api` W25-1 goes red. The flag # is also why this field can never be retyped: `_clean_field` REFUSES a profile flag on # anything but `text`. field_def("handle", "Handle", pinned=True, profile={"source": PROFILE_SOURCE_IG}), field_def("profile_url", "Profile", "url"), field_def("full_name", "Name"), field_def("followers", "Followers", "int"), field_def("following", "Following", "int"), field_def("avg_engagement", "Avg engagement", "pct"), field_def("bio", "Bio"), field_def("external_url", "Link in bio", "url"), field_def("verified", "Verified", "checkbox"), field_def("category", "Category"), # --- the enrichment half: what `run_field_instagram` already pulls, kept to the fields # MEASURED ≥60% populated (see the note above). field_def("posts_count", "Post count", "int"), field_def("highlights_count", "Story highlight count", "int"), field_def("is_business", "Business account", "checkbox"), field_def("is_professional", "Professional account", "checkbox"), # ⚠ STAYS `text`. It is a 17-digit opaque identifier, not a quantity — typing it `int` would # invite a grid to sum it, and some of them exceed 2^53 so a JS reader would round it. field_def("ig_id", "Instagram id"), # --- ⭐ 2026-08-07: the rest of the vendor's profile schema, promoted by owner instruction # (see the block comment above for what that reverses). Types are HONEST from birth, which # costs nothing here: no table has ever carried these columns, so there is no stored `text` # definition for the migration to convert — R3's conversion rule applies to the columns that # already exist, and these are new everywhere. field_def("business_category", "Business category"), field_def("is_private", "Private account", "checkbox"), field_def("bio_hashtags", "Bio hashtags"), field_def("pronouns", "Pronouns"), field_def("profile_name", "Profile name"), field_def("is_joined_recently", "Joined recently", "checkbox"), field_def("has_channel", "Has channel", "checkbox"), field_def("partner_id", "Partner id"), field_def("external_url_title", "Link title"), # ⚠ `text` for `ig_id`'s reason, one identifier over: it is a name, not a quantity. field_def("fbid", "Facebook id"), field_def("related_accounts", "Related accounts"), field_def("country_code", "Country"), field_def("source_payload", "Source data", "json"), # ⭐⭐ 2026-08-07 (owner instruction) — THE RELATION AND THE ROLLUPS OVER IT. # # Owner: *"an enrichment automation should spawn relevant Post/Comment database that is # linked to the profile automatically… and this rollup needs to have formula that we can use # to calculate things like average Views over last N posts."* Both halves are these five # columns, and they are PRESETS rather than something a person assembles, because "linked # automatically" is the instruction — a relation you have to wire up by hand is the feature # not existing. # # ⛔ `from` IS NOT DECLARED, ON PURPOSE. `_link_from_key` falls back to the PROFILE-flagged # column and then the pinned one, both of which are `handle` on every preset table — so this # links correctly on a database where the handle column was renamed, and on one where the # flag sits somewhere unexpected. Naming `handle` here would be the hard-coded subject # [[gate-answers-the-wrong-question]] warns about, one layer down. # ⚠ "Post rows", NOT "Post count" — one opens records and one reports the account total. # Two count-like columns with the same label would be unreadable. Caught by the label-collision # gate rather than on screen, which is what that gate is for. field_def("posts_link", "Post rows", "link", description="Post records linked to this profile.", link={"table": IG_POSTS_TABLE, "on": "influencer_key"}), field_def("profile_snapshots_link", "Profile history", "link", description="Profile snapshots linked to this profile.", link={"table": IG_SNAPSHOTS_TABLE, "on": "influencer_key"}), field_def("post_snapshots_link", "Post measurement rows", "link", description="Post engagement measurements linked to this profile.", link={"table": IG_POST_SNAPSHOTS_TABLE, "on": "influencer_key"}), field_def("comments_link", "Comment rows", "link", description="Comment records linked to this profile.", link={"table": IG_COMMENTS_TABLE, "on": "influencer_key"}), # ⚠ THE ROLLUPS READ `ut_ig_posts`' OWN LATEST COLUMNS, which is why those exist — a rollup # is ONE HOP (Airtable's rule and ours), and the engagement SERIES lives one table further # out in `ut_ig_post_snapshots`. # ⚠ 12 IS THIS MEASURE'S OWN WINDOW (`AVG_WINDOW_POSTS`), NOT THE CAPTURE CAP. It used to be # `MAX_POSTS_PER_PULL` and read "the vendor's ceiling" — both halves are now wrong: the cap is 30 # (owner, 2026-08-09) and 30 is not the vendor's limit either. Binding these four to the cap meant # raising it silently redefined every column named `_12`. `sortBy` is mandatory alongside a # `limit`, and this is why — "the last 12" has to name what makes one post later than another. # ⭐ UN-RETIRED 2026-08-08 together with the column it averages. It was retired for four hours # because its input was an account-grain constant; with the input now a real per-reel # measurement the average means what its label says again. field_def("avg_views_12", "Avg views · last 12 posts", "rollup", rollup={"link": "posts_link", "field": "views", "fn": "average", "limit": AVG_WINDOW_POSTS, "sortBy": "posted_at", "sortDir": "desc", "distinctBy": "shortcode"}), field_def("avg_plays_12", "Avg plays - last 12 posts", "rollup", rollup={"link": "posts_link", "field": "plays", "fn": "average", "limit": AVG_WINDOW_POSTS, "sortBy": "posted_at", "sortDir": "desc", "distinctBy": "shortcode"}), field_def("avg_likes_12", "Avg likes · last 12 posts", "rollup", rollup={"link": "posts_link", "field": "likes", "fn": "average", "limit": AVG_WINDOW_POSTS, "sortBy": "posted_at", "sortDir": "desc", "distinctBy": "shortcode"}), field_def("avg_comments_12", "Avg comments · last 12 posts", "rollup", rollup={"link": "posts_link", "field": "comments", "fn": "average", "limit": AVG_WINDOW_POSTS, "sortBy": "posted_at", "sortDir": "desc", "distinctBy": "shortcode"}), # ⭐ THE ONE HONEST POST COUNT WE HAVE. D-82: the vendor's `posts_count` is a FABRICATED ZERO # on the paid rung (49/49 rows measured), so it is discarded and that column reads blank after # a paid enrich. This counts the post rows actually captured — a different number and a true # one, which is why it gets its own column and its own label rather than quietly filling # `posts_count` with something that is not what that field means. field_def("posts_captured", "Posts captured", "rollup", rollup={"link": "posts_link", "fn": "countall", "distinctBy": "shortcode"}), field_def("profile_reads", "Profile reads", "rollup", rollup={"link": "profile_snapshots_link", "fn": "countall", "distinctBy": "snapshot_key"}), field_def("post_measurements_captured", "Post measurements captured", "rollup", rollup={"link": "post_snapshots_link", "fn": "countall", "distinctBy": "post_snapshot_key"}), field_def("comments_captured", "Comments captured", "rollup", rollup={"link": "comments_link", "fn": "countall", "distinctBy": "comment_key"}), # ⭐⭐ WAVE 27 ITEM 16 — WHERE THIS CREATOR ACTUALLY IS, DERIVED, FOR NOTHING. # # ⛔ THE VENDOR DOES NOT SELL THIS. `country_code` was MEASURED `None` on every corpus row we # have ever looked at, and the filter API REFUSES it as a predicate — so residency is the one # thing a buyer most wants and the one thing the profile row cannot answer. The competitor # teardown found the same gap solved by GUESSING from bio words, two buckets deep # ([[janney-ai-teardown]]). # # ⭐ WE ARE ALREADY HOLDING THE ANSWER AND WERE THROWING IT AWAY. Each post carries the place # it was tagged in, and the enrich has ALREADY BOUGHT twelve of them: `tagged_location` on the # post row, plus the vendor's fuller `location`/`location_details` inside the paid # `source_payload` that `_bd_tagged_location` normalises away. The modal city across a # creator's own posts is a far better residency signal than a word in a bio, and it costs a # dict comprehension. **Zero new vendor spend** — this is the whole reason it is a v1. # # ⚠ IT IS A GUESS AND THE COLUMN SAYS SO, IN ITS NAME AND IN ITS NEIGHBOUR. A travel creator # posting from twelve cities gets a low confidence rather than a confident wrong answer, and # `location_confidence` is the number a person filters on before trusting the guess. A single # geotagged post produces NO guess at all — n=1 dressed as a pattern is what # `SEED_MIN_ROWS_SHARING` refuses fifty lines up, and it would read as 100% certain. field_def("location_guess", "Location (guess)"), field_def("location_confidence", "Location confidence", "pct"), # ⭐ R3's stamp. WITHOUT IT THE WHOLE SET IS UNREADABLE: a blank `followers` means "never # enriched" and a stale one means "enriched in March", and no cell on the row can tell them # apart. It is the single field that turns the other fifteen from numbers into measurements. field_def("enriched_at", "Enriched at", "date"), # ⭐ WAVE 26 · R1 — THE LAST-N POST WINDOW, AND IT IS A VIEW RATHER THAN A STORE. # ⛔ READ THE R3 NOTE ABOVE BEFORE CHANGING THIS. "One store for one series" still holds: the # authoritative post record is `ut_ig_posts` (keyed by shortcode, so it ACCUMULATES) and the # authoritative engagement series is `ut_ig_post_snapshots` (append-per-pull, carrying # views/likes/comments at a `pulled_at`). This cell is a DERIVED window over those two, # rewritten each run, so a person reading the profile row can see the recent posts without a # join — and deleting it would cost a convenience, never a measurement. Shape: contract C2. ] #: The keys the preset set owns — derived, so a field added above cannot be forgotten here. PRESET_PROFILE_KEYS = tuple(f["key"] for f in PRESET_PROFILE_FIELDS) #: ⭐ 2026-08-07 — WHICH preset field declares the primary column, and WHICH declares the profile #: flag. DERIVED for the same reason `PRESET_PROFILE_KEYS` is, and the negative control is what #: argued for it: with `"handle"` hard-coded in the migration, stripping the declaration left the #: migration cheerfully stamping a column the product no longer claimed. Moving a declaration now #: moves the migration with it, and REMOVING one stops the migration rather than leaving it to #: enforce a rule nobody declares any more. #: ⚠ Both are `""` when nothing declares them, and every reader treats `""` as "do nothing" — #: never as "field number zero". PRESET_PINNED_KEY = next((f["key"] for f in PRESET_PROFILE_FIELDS if f.get("pinned")), "") PRESET_FLAG_KEY = next((f["key"] for f in PRESET_PROFILE_FIELDS if f.get("profile")), "") #: Discovery's table is the preset set PLUS its own bookkeeping. ⛔ DERIVED, NEVER RE-TYPED: the #: alternative is two lists that describe the same columns and drift one label at a time, which is #: the failure C1 exists to prevent. The five below are facts about the SEARCH (how often we found #: them, who for, and a human's decision) rather than about the profile, so they are discovery's #: and not part of the cross-tenant set. CANDIDATE_FIELDS = [ *PRESET_PROFILE_FIELDS, # ⭐ WAVE 26 · R3 — `first_found` / `last_found` ARE DATES, not ISO strings in a text cell. # They were written by `_iso()`, so the cell read `2026-08-05T14:03:11+07:00` and the owner # called that format "extremely confusing" — correctly: it is a machine timestamp shown to a # person, unsortable as a date and unfilterable by "last 7 days". The STAMP still carries its # offset everywhere it is a time axis (`ut_ig_snapshots.pulled_at` is untouched); what # changed is that "when did we first see this account" is a DAY, and a day is all anybody # asks it for. `migrate_ig_field_types()` converts the stored strings. field_def("found_count", "Times found", "int"), field_def("first_found", "First found", "date"), field_def("last_found", "Last found", "date"), # ⛔ DEBT D-73 — THIS NOTE STATED A RETIRED LAW AS CURRENT FACT, directly above the field it # describes, which is the first place anybody looks up what this column means. It read: *"the # candidate pool is PER-USER — the upsert key is the COMPOUND (handle, created_by), so two # people discovering the same profile each get their own row, their own found_count and their # own review card."* Wave 26 · R4/R5 retired every clause of that. # # ⭐ THE CURRENT LAW: the identity is `(platform, handle)` and the TENANT is the unit. Two # people in one workspace who discover the same profile share ONE row, one `found_count` and # one history — because they are looking at one company's leads, not two private lists. # `created_by` survives as an INFORMATIONAL stamp only: "Found by", answering who saw it # first. It is no part of the dedup key, and a run must never branch on it. # ⚠ Stamped by the RUNNER (the automation's creator), never by `_candidate_row` — unchanged, # and the one clause of the old note that was still true. field_def("created_by", "Found by"), # ⛔ WAVE 26 · R6 — `tracked` WAS HERE AND IS DELETED. Owner, 2026-08-06: *"We already have # a Field called stage to track the progress of the Automation per Record. We don't need # another checkbox for this."* Correct, and it had been true since wave 22 shipped the stage # column: every candidate carried BOTH a `stage_` select saying where it was AND a # boolean saying whether it was kept — two progress fields that could disagree, with no rule # about which one won. The stage field is the one progress column. Nothing replaces this. ] def _profile_backlink_field(table_key, table_label, profile_key): """A deterministic reciprocal link from one canonical IG table to one profile database. The key includes the target table identity, so ten separate Profile databases can all point into the same canonical Posts/Comments/history tables without one relation overwriting the next. The join is derived in both directions and therefore needs no cross-table fan-out write. """ digest = hashlib.sha1(str(table_key).encode("utf-8")).hexdigest()[:10] return field_def( f"profiles_{digest}", f"Profiles - {str(table_label or table_key)[:48]}", "link", description=f"Profile records from {str(table_label or table_key)[:48]} linked to this row.", link={"table": str(table_key), "on": str(profile_key), "from": "influencer_key"}, ) def _profile_schema_for(bound_key): """Preset fields for an existing Profile table without inventing a second identity column.""" out = [] for field in PRESET_PROFILE_FIELDS: item = dict(field) if item.get("key") == PRESET_FLAG_KEY and bound_key != PRESET_FLAG_KEY: item.pop("profile", None) out.append(item) return out def _tt_profile_schema_for(bound_key): """⭐ WAVE 30 · T08 — TikTok preset columns for a database whose profile column is `bound_key`. ⛔ NOT `_profile_schema_for` WITH A LIST ARGUMENT, and the difference is not stylistic. That function walks `PRESET_PROFILE_FIELDS` and only ever removes the flag from `PRESET_FLAG_KEY`, because on the Instagram side the flag lives on exactly one known column. TikTok's binding may be ANY column a person named (`profile_field_key` step 1), so the rule here has to be stated the other way round: **every field that is not the bound one arrives as DATA**, stripped of both the identity flag and `pinned`. ⚠ Otherwise a database whose TikTok column is `creator` would gain a rival `handle` carrying `profile: {source: "tiktok"}` — a SECOND profile column, which `user_tables` refuses at both write doors, so the whole top-up would be rejected and every preset cell would then be dropped for want of a column. One shared helper for the save-time and run-time paths, so those two cannot answer "which columns does a TikTok enrich need" differently. """ out = [] for field in TT_PROFILE_FIELDS: item = dict(field) if str(item.get("key") or "") != str(bound_key or ""): item.pop("profile", None) item.pop("pinned", None) out.append(item) return out def _locked_ig_field(field): """One canonical Instagram field with the immutable preset declaration attached.""" item = dict(field) automation = dict(item.get("automation") or {}) automation["preset"] = True item["automation"] = automation return item def _profile_binding(table): """The declared Profile identity field, falling back only to the canonical handle.""" fields = list((table or {}).get("fields") or []) flagged = next((str(f.get("key") or "") for f in fields if isinstance(f.get("profile"), dict)), "") if flagged: return flagged return PRESET_FLAG_KEY if any(f.get("key") == PRESET_FLAG_KEY for f in fields) else "" def _ig_schema_contract(rt, profile_tables=None): """The complete per-tenant Instagram graph contract, including every Profile backlink. The fixed child tables are shared. Profile tables are discovered structurally, so an older user-named target and the built-in candidate database receive the same preset columns, Links, Rollups, locks, and reciprocal fields. Retired machine keys are the only columns deleted. """ tables = ut_all(rt) # ⭐ WAVE 32 · T42 — `_ig_contract_tables`, NOT `_ig_profile_tables`: this function decides # which databases RECEIVE Instagram's 48 columns, and the owner's rule is that a database gets # them only if it is used for Instagram. See that function for why the detector stays wider. profile_keys = sorted(set(profile_tables if profile_tables is not None else _ig_contract_tables(rt))) backlinks, wanted, drops = [], {}, { IG_SNAPSHOTS_TABLE: {"post_hashtags"}, # ⚠ `views` IS DELIBERATELY ABSENT FROM THIS DROP SET AGAIN. It was dropped earlier today # while it carried Bright Data's account-grain number; it is now declared in POST_FIELDS # and fed by the `ig_post_views` capability. Leaving it here would have made the migration # delete, on every authenticated read, the column the same release just added — the # drop set and the field list are two halves of ONE contract and must move together. } for table_key in profile_keys: table = tables.get(table_key) or {} bound = _profile_binding(table) if not bound: continue label = str(table.get("label") or table_key) backlinks.append(_profile_backlink_field(table_key, label, bound)) schema = _profile_schema_for(bound) # ⭐⭐ 2026-08-10 (owner: *"retire the old hardcoded method and replace the Instagram # database work with correct Rollup and Link fields"*) — THE DERIVED OVERLAY, AND IT # BELONGS HERE RATHER THAN IN A SEPARATE MIGRATION. # # ⛔ A standalone converter would LOSE. `_reconcile_ig_graph_fields` overwrites every # contract key it owns — `rollup` included, by its own note — so a column converted beside # the contract is re-typed back to `int` on the next reconcile, silently, hours later. # Making the CONTRACT itself say "this column is derived" leaves exactly one writer of the # schema instead of two that disagree. # # ⚠ THE SET IS PER TENANT AND IS A PROOF, NOT A LIST. `_derived_profile_columns` converts a # column only where this tenant's own rows show the fold already equals the stored value — # measured, because a named list would have blanked 221 live cells on nurilab alone # (`avg_engagement` and `category`, 62 each). A tenant whose series is thinner converts # fewer columns and loses nothing; as its history fills in, later passes convert more. derived = set(_derived_profile_columns(table, tables.get(IG_SNAPSHOTS_TABLE) or {})) wanted[table_key] = [_as_derived_field(f) if str(f.get("key")) in derived else f for f in schema] # `tracked` was the pre-Stage progress checkbox. Keeping both allows two progress states # to disagree, so it is retired by key just like the old nested Posts JSON. drops[table_key] = {"posts", "post_hashtags", "tracked"} wanted.update({ IG_SNAPSHOTS_TABLE: [*SNAPSHOT_FIELDS, *backlinks], IG_POSTS_TABLE: [*POST_FIELDS, *backlinks], IG_POST_SNAPSHOTS_TABLE: [*POST_SNAPSHOT_FIELDS, *backlinks], IG_COMMENTS_TABLE: [*COMMENT_FIELDS, *backlinks], }) return wanted, drops # Only a unit-changing retype needs a cell conversion. Integer/checkbox/date cells already use # the scalar strings their renderers expect; engagement is the exception because Bright Data's # 0-1 fraction becomes this product's 0-100 percentage. _IG_RETYPE_CONVERTERS = { (IG_SNAPSHOTS_TABLE, "avg_engagement"): _pct100, } def _reconcile_ig_graph_fields(rt, wanted_by_table, drop_by_table=None): """Repair machine-owned IG field declarations in one coalesced store update. `ut_ensure` deliberately merges only missing keys. That is correct for user columns but not sufficient for a canonical relation: a stale `link.table`, rollup function, or type would survive forever. This pass overwrites the contract keys for the fields we own, preserves unrelated/user fields, and removes only explicitly retired preset columns plus their cells. """ drop_by_table = drop_by_table or {} wanted_by_table = { key: [_locked_ig_field(f) for f in fields] for key, fields in wanted_by_table.items() } changed = False def _up(cur): nonlocal changed cur = cur if isinstance(cur, dict) else {} for table_key, wanted_fields in wanted_by_table.items(): table = cur.get(table_key) if table is None: continue wanted = {str(f.get("key")): dict(f) for f in wanted_fields} drops = set(drop_by_table.get(table_key) or ()) fields, seen = [], set() drops = _guarded_drops(table, drops) for stored in table.get("fields") or []: key = str(stored.get("key") or "") if key in drops: changed = True continue desired = wanted.get(key) if desired is None: fields.append(stored) continue # ⭐⭐ 2026-08-09 (owner: *"everything is custom and changeable always"*) — A # COLUMN A HUMAN HAS TAKEN OVER IS LEFT ALONE. Every key below is overwritten # from the shipped contract, `rollup` included, so without this the newly # unlocked "edit a preset rollup" would save, render, recompute and then revert # at the next enrichment run — an edit that looks like it worked and undoes # itself hours later. `user_tables.user_edited` is the ONE reader of the stamp. if _ut().user_edited(stored): fields.append(stored) seen.add(key) continue repaired = dict(stored) for dkey, value in desired.items(): if dkey != "automation": repaired[dkey] = value automation = dict(stored.get("automation") or {}) automation.update(desired.get("automation") or {}) repaired["automation"] = automation # These bags define behaviour, not decoration. If the desired field does not use # one, a stale bag from an earlier type must not remain attached to it. for bag in ("link", "rollup", "profile", "pinned"): if bag not in desired: repaired.pop(bag, None) if desired.get("type") not in ("select", "multiselect"): repaired.pop("options", None) converter = _IG_RETYPE_CONVERTERS.get((table_key, key)) if (converter and stored.get("type") in ("text", "", None) and repaired.get("type") != stored.get("type")): for row in (table.get("rows") or {}).values(): if str(row.get(key) or "").strip(): converted = converter(row.get(key)) if converted: row[key] = converted changed = True if repaired != stored: changed = True fields.append(repaired) seen.add(key) for key, desired in wanted.items(): if key not in seen and not any(str(f.get("key") or "") == key for f in fields): fields.append(desired) changed = True if drops: for row in (table.get("rows") or {}).values(): for key in drops: if key in row: row.pop(key, None) changed = True table["fields"] = fields return cur # Avoid a commit when the declarations are already exact. snapshot = ut_all(rt) needs = False for table_key, wanted_fields in wanted_by_table.items(): if table_key not in snapshot: continue table = snapshot.get(table_key) or {} by_key = {str(f.get("key") or ""): f for f in table.get("fields") or []} # ⚠ THE SAME GUARD, OR THE MIGRATION STOPS BEING IDEMPOTENT. This precheck exists to skip # the commit when nothing would change; a spared `tracked` column left in the drop set # answers "yes, work to do" on every single call, forever, and the schema pass would # rewrite the table on every authenticated read without ever changing a byte. drops = _guarded_drops(table, drop_by_table.get(table_key) or ()) if drops & set(by_key): needs = True break for desired in wanted_fields: stored = by_key.get(str(desired.get("key") or "")) mismatch = stored is None if stored is not None: for key, value in desired.items(): if key == "automation": if any((stored.get("automation") or {}).get(k) != v for k, v in value.items()): mismatch = True break elif stored.get(key) != value: mismatch = True break if not mismatch: stale_bags = {"link", "rollup", "profile", "pinned"} - set(desired) mismatch = any(k in stored for k in stale_bags) if (not mismatch and desired.get("type") not in ("select", "multiselect") and "options" in stored): mismatch = True if mismatch: needs = True break if needs: break if needs: rt.update(UT_STORE_KEY, _up, flush="sync") return changed def ensure_ig_graph(rt, username="automation", flow_tag="", profile_table="", profile_field=""): """Ensure the ONE per-tenant Instagram relational graph and return its canonical keys. Every enrichment path calls this function. Fixed table keys prevent a flow aimed at a new Profile database from spawning `IG posts 2`; deterministic reciprocal link fields connect each Profile database to the same Posts, Comments, profile-history, and post-history stores. """ profile = ut_get(rt, profile_table) if profile_table else None bound = str(profile_field or "").strip() if profile is not None and not bound: bound = next((str(f.get("key") or "") for f in profile.get("fields") or [] if isinstance(f.get("profile"), dict)), "") profile_label = str((profile or {}).get("label") or profile_table) backlink = [_profile_backlink_field(profile_table, profile_label, bound)] \ if profile_table and profile is not None and bound else [] # ⭐ WAVE 32 · T41 — DERIVED from `IG_TABLE_FIELDS`/`IG_TABLE_LABELS`, which used to be this # literal. The backlinks stay HERE because they are per-tenant, per-profile-database facts; the # schema is a module constant. Splitting it that way is what let A's delivery sweep ask this # module for Instagram's child set instead of hand-typing a fifth copy of these four names. graph = {key: (IG_TABLE_LABELS[key], [*fields, *backlink]) for key, fields in IG_TABLE_FIELDS.items()} keys = {} for key, (label, fields) in graph.items(): keys[key] = ut_ensure(rt, label, fields, username, key=key, flow_tag=flow_tag, record_mode=AUTOMATION_RECORD_MODE, lock_fields=True) if profile_table and profile is not None and bound: profile_fields = _profile_schema_for(bound) ut_ensure(rt, profile_label or profile_table, profile_fields, username, key=profile_table, flow_tag=flow_tag, lock_fields=True) # Reconcile the WHOLE tenant graph, not just this run's target. That is what keeps an older # Profile database and IG candidates on the same universal schema while all of them point to # the same four children. Existing flow provenance is merged, never replaced. wanted, drops = _ig_schema_contract(rt) _reconcile_ig_graph_fields(rt, wanted, drops) return keys # ── ⭐ WAVE 26 · THE MIGRATION (owner rulings R3 + R4/R5, contracts C1-a and C3) ─────────────── # # Two changes land on data that already exists, and neither is optional once the field defs move: # # R3 the preset columns take honest types, so the CELLS have to match them — an ISO stamp in a # `date` column and a "20872" string in an `int` column are the split-schema the old # "types are text on purpose" note correctly feared. Re-typing without converting is worse # than not re-typing. # R4 the candidate identity becomes `(platform, handle)`, so rows that exist today as # `(handle, created_by)` DUPLICATES must be merged rather than left to collide. # # ⛔ IDEMPOTENCY IS THE WHOLE DESIGN, and the key is the STORED FIELD TYPE — never a flag, never a # marker column. Run twice, an `avg_engagement` of 0.0074 would go 0.74 then 74, and nothing would # look wrong until somebody read a 74% engagement rate off a creator with 3,000 followers. So a # column is converted only while its STORED definition still says `text`; the moment it says `pct` # the work is done and re-running is a no-op. That makes the migration safe to call on every boot, # which is what it is for. # # ⚠ THE TABLE LIST IS DERIVED, NOT NAMED — D-71's lesson, one wave old and already paid for twice. # `observed_categories` named its two source tables and went silently empty the day wave 25 let a # user point an automation somewhere else. Naming tables here would leave exactly those user-named # databases on the old schema, which is the same bug wearing a migration's clothes. #: A table is an Instagram profile table if it declares `handle` plus a real slice of the preset #: vocabulary. Deliberately structural: it finds `ut_beauty_influencer_leads` without being told. MIGRATE_MIN_PRESET_FIELDS = 3 #: key -> the converter its new type needs. `checkbox` needs none (the writers already emit '1'/''). _MIGRATE_CONVERT = { "first_found": _day, "last_found": _day, "enriched_at": _day, "avg_engagement": _pct100, } #: ⭐⭐ WAVE 30 · T08 — THE COLUMNS THAT CAN ONLY BE TIKTOK'S, DERIVED rather than typed out, so #: that adding a column to either declaration keeps this set correct instead of quietly emptying #: it. Eight today: `tt_id`, `like_engagement`, `comment_engagement`, `likes_received`, `region`, #: `predicted_lang`, `account_created_at`, `source`. TT_ONLY_PROFILE_KEYS = (frozenset(f["key"] for f in TT_PROFILE_FIELDS) - frozenset(PRESET_PROFILE_KEYS) - frozenset(f["key"] for f in CANDIDATE_FIELDS)) def _is_tt_profile_table(tbl): """⭐⭐ WAVE 30 · T08 — is this database a TIKTOK profile table? ⛔ THE DEFECT THIS EXISTS TO CLOSE WAS LIVE AND SHIPPED, and it converted a customer's TikTok database into an Instagram one on the SECOND write to it. MEASURED: a `ut_tt_profile` spawned by W30-T06 comes out correct — `handle` carrying `profile: {source: "tiktok"}` — and one more `ut_ensure` on that table (a re-save, the discovery runner's own ensure, the next run) leaves it carrying `profile: {source: "instagram"}` and the description *"Instagram username without the @ symbol"*. `profile_field_key(..., source="tiktok")` then answers `""` forever, so every TikTok enrich step on that database reports UNBOUND, and every row gets stamped `platform: Instagram` — which is half of W26/R4's `(platform, handle)` identity, so two different people's accounts merge under one key with nothing red anywhere. ⚠ It hid because the migration SKIPS a table that does not exist yet, so the spawn itself is always clean and only the second write is not. A gate that creates and asserts once cannot see it; the assertion has to survive a second `ut_ensure`. ⭐ TWO LEGS, and the second is the one that still works after the first has been eaten: 1. the table SAYS SO — a column declaring `profile: {source: "tiktok"}` is the database's own statement of which network it is about, and it is the same declaration `profile_field_key` reads, so there is one answer to "whose profile table is this"; 2. it declares a column only TikTok has. Needed because leg 1 is exactly what the defect destroys: on a table already corrupted in production, the flag now says Instagram, and a detector resting on it alone would agree with the corruption and keep re-applying it. """ fields = (tbl or {}).get("fields") or [] for f in fields: p = f.get("profile") if isinstance(p, dict) and str(p.get("source") or "") == PROFILE_SOURCE_TT: return True return bool({f.get("key") for f in fields} & TT_ONLY_PROFILE_KEYS) #: ⭐⭐ WAVE 32 · T40 — THE MIRROR OF `TT_ONLY_PROFILE_KEYS`, and it is the subject of owner item 1: #: the profile columns that can only ever be INSTAGRAM's. 26 today. Derived by the same subtraction #: in the other direction, so a column moved between the two declarations changes both sets at once #: rather than leaving one of them quietly asserting a stale fence. #: ⚠ IT IS DERIVED FROM `CANDIDATE_FIELDS`, NOT `PRESET_PROFILE_FIELDS` — discovery's bookkeeping #: columns (`found_count`, `first_found`, `last_found`, `created_by`) are declared on BOTH platforms, #: so subtracting the preset list alone would report four keys as Instagram-only that TikTok's own #: table has carried since wave 29. IG_ONLY_PROFILE_KEYS = (frozenset(f["key"] for f in CANDIDATE_FIELDS) - frozenset(f["key"] for f in TT_PROFILE_FIELDS)) #: ⭐⭐ WAVE 32 · T42 — the preset columns NOBODY TYPES: every `link` and `rollup` in the Instagram #: preset set. Derived from the declaration, so adding a rollup to the contract widens this for #: free. `_carries_ig_presets` uses it as the leg that survives when the `automation.preset` stamp #: is absent — a hand-made creator list has `handle` and `followers`; it does not have `posts_link`. PRESET_MACHINE_ONLY_KEYS = frozenset( f["key"] for f in PRESET_PROFILE_FIELDS if f.get("type") in ("link", "rollup")) def platform_schema(platform): """⭐⭐ WAVE 32 · T41 — ONE ASKABLE DECLARATION OF WHAT A PLATFORM'S DATABASES ARE. Returns, for `PLATFORM_INSTAGRAM` or `PLATFORM_TIKTOK`: {"platform", "profile_table", "profile_label", "profile_fields", "preset_keys", "only_keys", "locked_tables", "children": {key: {"label", "fields", "record_mode"}}} ⛔ **WHY IT IS A FUNCTION AND NOT A DICT LITERAL — the same import-order trap `discovery_facts` records twenty lines from its own declaration.** `CANDIDATE_FIELDS` and `DISCOVER_TABLE` are declared HUNDREDS of lines below `TT_TABLE_FIELDS`, so any module-level dict spanning both platforms NameErrors at import. A function body resolves at call time and does not care. That is not a style preference here; it is the reason two earlier attempts at a platform registry in this file ended up as five inline copies instead. ⛔⛔ **AND IT IMPORTS NOTHING FROM `core`, WHICH IS THE ACTUAL CONTRACT WITH `main.py`.** This module is deliberately dependency-light on the API's boot path (`MAX_UT_ROWS`, `MACHINE_OWNERS` and `LOCKED_CHILD_TABLES` all say so in their own notes), and the REGISTRAR — the thing that turns these names into `core.user_tables` state — lives in the composition root. So this returns plain lists and dicts: a caller can `register_locked_records(...)`, `ut_ensure(...)` or diff it against a live tenant without `core` ever appearing on the engine's import path. `verify_ automation` asserts that, by importing this module in a CLEAN interpreter and checking `core.user_tables` is absent from `sys.modules` afterwards — [[artifact-with-no-importer]] in reverse: a declaration whose registrar cannot reach it is the same defect as a registrar with nothing to register. ⚠ FIELD DICTS ARE COPIED ONE LEVEL. Every existing call site passes the module lists straight into `ut_ensure`, which is safe only because nobody has yet appended to one; a boot sweep looping over both platforms is exactly the caller that would. The copy costs a few hundred dict constructions once per boot and removes the whole question. ⚠ `profile_fields` is the platform's PRESET set as spawned — Instagram's is `CANDIDATE_FIELDS` (the preset set PLUS discovery's bookkeeping), not `PRESET_PROFILE_FIELDS`, because that is what `discovery_facts` actually hands the spawn. Two answers to "Instagram's field set" is the drift this function exists to end, so it gives the one the product uses. """ name = str(platform or "") if name == PLATFORM_TIKTOK: children = {k: {"label": TT_TABLE_LABELS[k], "fields": [dict(f) for f in TT_TABLE_FIELDS[k]], "record_mode": tt_record_mode(k)} for k in sorted(TT_TABLE_FIELDS) if k != TT_PROFILE_TABLE} return { "platform": PLATFORM_TIKTOK, "profile_table": TT_PROFILE_TABLE, "profile_label": TT_TABLE_LABELS[TT_PROFILE_TABLE], "profile_fields": [dict(f) for f in TT_PROFILE_FIELDS], "preset_keys": tuple(f["key"] for f in TT_PROFILE_FIELDS), "only_keys": TT_ONLY_PROFILE_KEYS, "locked_tables": TT_LOCKED_TABLES, "children": children, } if name == PLATFORM_INSTAGRAM: children = {k: {"label": IG_TABLE_LABELS[k], "fields": [dict(f) for f in fields], # ⚠ NOT `tt_record_mode`'s twin by accident: `ensure_ig_graph` passes # `AUTOMATION_RECORD_MODE` for all four unconditionally, and # `IG_LOCKED_TABLES` is exactly those four. Derived from the SET so that # unlocking one there unlocks it here, rather than from the constant. "record_mode": (AUTOMATION_RECORD_MODE if k in IG_LOCKED_TABLES else "")} for k, fields in sorted(IG_TABLE_FIELDS.items())} return { "platform": PLATFORM_INSTAGRAM, "profile_table": DISCOVER_TABLE, "profile_label": DISCOVER_LABEL, "profile_fields": [dict(f) for f in CANDIDATE_FIELDS], "preset_keys": tuple(f["key"] for f in CANDIDATE_FIELDS), "only_keys": IG_ONLY_PROFILE_KEYS, "locked_tables": IG_LOCKED_TABLES, "children": children, } raise ValueError(f"no schema is declared for platform {platform!r}") def definition_platforms(defn): """Which networks ONE automation uses — `{"Instagram"}`, `{"TikTok"}`, both, or empty. ⭐ WAVE 32 · T42. Three ways a definition names a network, and all three count, because the owner's rule is about what a database is USED FOR rather than about how the automation was built: its discovery KIND (a corpus search is a network by construction), its enrich ACTIONS (a `plain` flow that enriches is using that network), and the retired `field_instagram` kind, which is uncreatable but alive on stored definitions (D-65) and still runs Instagram. """ defn = defn or {} kind = str(defn.get("kind") or "") out = set() if kind in DISCOVERY_KINDS: out.add(discovery_facts(kind)[0]) if kind == "field_instagram": out.add(PLATFORM_INSTAGRAM) actions = (defn.get("flow") or {}).get("actions") or [] if _actions_of_kind(actions, "enrich_instagram"): out.add(PLATFORM_INSTAGRAM) if _actions_of_kind(actions, "enrich_tiktok"): out.add(PLATFORM_TIKTOK) return out def automation_platforms_by_table(rt, definitions=None): """⭐⭐ WAVE 32 · T42 (owner item 1, third clause) — WHICH NETWORKS EACH DATABASE IS USED FOR. `{table_key: {"Instagram", "TikTok"}}`, over every stored automation. The owner's words are the whole specification: *"if a user decide to use one database for both Tiktok and Instagram scraping, only then can the pre-set Fields can exist in the same database."* — so the question a schema pass has to be able to ask is *"which networks target THIS database"*, and until now nothing in this module could answer it. ⛔ ASKABLE AND `core`-FREE, for `platform_schema`'s reason one function up: it reads only the automations bucket through `all_definitions`, returns plain sets, and the T41 subprocess probe covers it — a second askable declaration that quietly needed `core` would make the import-purity guarantee mean "true for the things we remembered". ⚠ `definitions` is a LENT list, same contract as `ut_ensure`'s `tables=`: a caller already holding the bucket must not pay for a second deep copy of it. """ # ⚠ `all_definitions` answers a DICT keyed by id, not a list — and iterating it directly hands # every consumer a STRING that looks like a definition until the first `.get`. Normalised here # so a caller lending its own list gets the same treatment. defs = all_definitions(rt) if definitions is None else definitions defs = list(defs.values()) if isinstance(defs, dict) else list(defs or []) out = {} for defn in defs: if not isinstance(defn, dict): continue table = str(_flow_table(defn) or (defn.get("config") or {}).get("targetTable") or "") if not table: continue found = definition_platforms(defn) if found: out.setdefault(table, set()).update(found) return out def platform_schemas(): """Both declarations, in `PLATFORMS` order — what a boot sweep loops over. ⚠ `PLATFORM_FACEBOOK` is in `PLATFORMS` (it is a value the `platform` COLUMN may hold) and has no schema, so this filters rather than raising: a caller sweeping every declared platform must not be broken by a vocabulary entry that names no databases. `platform_schema` still raises for it, which is the right answer to somebody ASKING for a schema that does not exist. """ out = [] for name in PLATFORMS: try: out.append(platform_schema(name)) except ValueError: continue return out def _ig_profile_tables(rt, tables=None): """Every `ut_*` table in this tenant that looks like an Instagram profile table. ``tables`` is the optional lent snapshot (W29-T01): read-only walk, so a caller that already holds the bucket must not pay for a second deep copy of it. ⛔ WAVE 30 · T08 — AND THE TEST IS NOW EXCLUSIVE, because "looks like" was network-blind and a TikTok profile table looks EXACTLY like one: 18 of `ut_tt_profile`'s 30 columns are Instagram preset keys, and the bar here is `handle` plus three. Every caller of this function then treats the match as an Instagram table — `migrate_ig_tables` rewrites its declarations to Instagram's contract, and `discover_default_table` offers it as the default target for an Instagram search. """ out = [] for key, tbl in sorted(((ut_all(rt) if tables is None else tables) or {}).items()): keys = {f.get("key") for f in (tbl or {}).get("fields") or []} if "handle" not in keys or len(keys & set(PRESET_PROFILE_KEYS)) < MIGRATE_MIN_PRESET_FIELDS: continue if _is_tt_profile_table(tbl): continue out.append(key) return out def _ig_contract_tables(rt, tables=None): """⭐⭐ WAVE 32 · T42 — the tables Instagram's preset CONTRACT may be applied to. ⛔⛔ THIS IS A DIFFERENT QUESTION FROM `_ig_profile_tables` AND CONFLATING THEM COST FIVE RED CHECKS. That function answers *"does this database LOOK like an Instagram profile table?"* and two callers need exactly that: `discover_default_table`, which elects an existing table so a new automation does not mint a second empty one — and it necessarily runs BEFORE any automation targets that table, so a targeted-only test makes it elect nothing forever — and the detector's own negative control. THIS function answers *"may we write Instagram's 48 columns into it?"*, which is the owner's rule and a strictly narrower set. One normalizer answering two questions is how a fix in one becomes a silent regression in the other ([[one-question-two-normalizers]]). A table qualifies when it looks like one AND any of: - it is `ut_ig_profile`, Instagram's canonical database by declaration; - it ALREADY carries the contract (`_carries_ig_presets`) — narrowing must never orphan a table that has the columns, or its links and rollups silently stop being repaired; - an INSTAGRAM automation targets it, which is the owner's rule stated in code. ⚠ The automations bucket is read LAZILY, on the first candidate that needs the question asked — a tenant whose tables all already carry the contract pays nothing, on a path this wave is otherwise trying to make faster. """ targets = None out = [] for key in _ig_profile_tables(rt, tables=tables): tbl = (ut_all(rt) if tables is None else tables).get(key) or {} if key != DISCOVER_TABLE and not _carries_ig_presets(tbl): if targets is None: targets = automation_platforms_by_table(rt) if PLATFORM_INSTAGRAM not in targets.get(key, set()): continue out.append(key) return out def _preset_owned(field): """Is this column MACHINE-authored, i.e. may `retract_foreign_presets` delete it? ⛔ MODULE-LEVEL, NOT A CLOSURE, AND THAT IS WHY IT MOVED. It was a nested `_machine` and the negative control written against it came out BLIND: nothing outside the function could break the one guard standing between a boot-time repair and a customer's own column, so the control ended up patching the key SETS instead and proved something else. A guard a control cannot reach is a guard nobody has tested [[gate-negative-control]]. Two legs, the second for tables that predate the stamp: the `automation.preset` mark that `ut_ensure(lock_fields=True)` and `_locked_ig_field` both write, or a key that is a preset LINK/ROLLUP — nobody types `posts_link` by hand. """ automation = (field or {}).get("automation") if isinstance(automation, dict) and automation.get("preset") is True: return True return str((field or {}).get("key") or "") in PRESET_MACHINE_ONLY_KEYS def _filled_count(rows, key): """How many rows carry a non-blank value in `key` — the number that decides REPORT vs DELETE. ⛔ ALSO MODULE-LEVEL FOR ITS CONTROL'S SAKE. This one number is the whole of W30/R6's second sentence inside `retract_foreign_presets`: a foreign column with cells in it is reported, never deleted. A control that cannot make this lie cannot prove the report is doing anything. """ return sum(1 for r in (rows or {}).values() if isinstance(r, dict) and str(r.get(key) or "").strip()) def retract_foreign_presets(rt, log=print): """⭐⭐ WAVE 32 · T46 / DEBT D-152 — REMOVE THE PRESET COLUMNS OF A NETWORK A DATABASE IS NOT USED FOR. The RETRACTION half of `_ig_contract_tables`' recruitment rule, and A's boot sweep (`W32-T07`) calls it once per tenant. ⛔ WHY IT HAS TO EXIST AT ALL, and this is the wave's own thesis one layer down. W30-T08 fixed the DETECTOR, so the corruption cannot recur — and nothing undoes it. MEASURED on a fixture: a `ut_tt_profile` corrupted before that fix keeps ALL 26 Instagram-only columns and its `profile: {source: "instagram"}` flag through `ut_ensure`, `migrate_ig_tables` AND the discovery spawn. Three write paths, zero repairs. *"The corruption cannot recur"* and *"the corruption is gone"* are different claims [[a-migration-that-runs-on-the-next-write]]. Returns `{"tables", "columns", "cells", "flags", "kept"}`. **`kept` is the important one** — see guarantee 4. THE FOUR GUARANTEES, in the order they matter: 1. **Only machine-authored columns go.** A field must be stamped `automation.preset: True` (or be a link/rollup this contract owns) to be eligible. A column a PERSON typed is never touched, whatever it is named — the D-152 row calls this *"a VALUE BACKFILL over live customer rows"*, and that is the line it must not cross. 2. **Only the EXCLUSIVE sets.** `IG_ONLY_PROFILE_KEYS` (26) and `TT_ONLY_PROFILE_KEYS` (8), both derived. A key both platforms declare — `platform`, `handle`, `followers`, the four discovery bookkeeping columns — is shared vocabulary and is never a foreign column. 3. **Idempotent.** A second call finds nothing and writes nothing, so it is safe on every boot, which is what makes it callable from `main.py` rather than being a script somebody has to remember to run (the whole reason D-201 is still open). 4. ⛔ **A FOREIGN COLUMN THAT HOLDS DATA IS REPORTED, NOT DELETED.** W30/R6's second sentence: a limit that cannot be removed is reported with its cause. Deleting a customer's populated cells to make a grid look clean is not a repair, and this is precisely the *"deserves its own supervision"* clause of D-152. `kept` names every such column with its row count, so the caller can print it and a person can decide. ⚠ THE FLAG IS REPAIRED TOO, and it is the half D-152 names explicitly: a TikTok profile table whose `handle` was re-declared `profile: {source: "instagram"}` answers `""` to `profile_field_key(..., source="tiktok")` forever, so every TikTok enrich on it reports UNBOUND. Repaired only where `_is_tt_profile_table` recognises the table by its OWN TikTok-only columns, i.e. by the leg the corruption cannot have eaten. """ stats = {"tables": 0, "columns": 0, "cells": 0, "flags": 0, "kept": []} targets = automation_platforms_by_table(rt) exclusive = {PLATFORM_INSTAGRAM: IG_ONLY_PROFILE_KEYS, PLATFORM_TIKTOK: TT_ONLY_PROFILE_KEYS} def _up(cur): cur = cur if isinstance(cur, dict) else {} for key, tbl in sorted(cur.items()): if not isinstance(tbl, dict): continue fields = tbl.get("fields") or [] keys = {str(f.get("key") or "") for f in fields} # Only PROFILE databases are in scope: `handle` plus a real slice of either preset # vocabulary. A posts or comments table shares no profile columns and is never a # candidate; a database with nothing in common with either is not one either. if "handle" not in keys: continue tt = _is_tt_profile_table(tbl) used = set(targets.get(key) or ()) if not used: # Nothing targets it. Its own declaration is then the only statement of what it is # for — and a table that says TikTok is not an Instagram table, whatever columns a # past migration wrote into it. used = {PLATFORM_TIKTOK} if tt else {PLATFORM_INSTAGRAM} foreign = set() for platform, own in exclusive.items(): if platform not in used: foreign |= set(own) if not foreign: continue rows = tbl.get("rows") or {} drop, touched = [], False for f in fields: fkey = str(f.get("key") or "") if fkey not in foreign or not _preset_owned(f): continue filled = _filled_count(rows, fkey) if filled: # Guarantee 4 — REPORTED, never deleted. stats["kept"].append({"table": key, "column": fkey, "rows": filled}) continue drop.append(fkey) if drop: tbl["fields"] = [f for f in fields if str(f.get("key") or "") not in drop] for r in rows.values(): if not isinstance(r, dict): continue for fkey in drop: if fkey in r: r.pop(fkey, None) stats["cells"] += 1 stats["columns"] += len(drop) touched = True # ⚠ THE FLAG, and only on a table TikTok's own columns still identify. if tt: for f in tbl.get("fields") or []: p = f.get("profile") if isinstance(p, dict) and str(p.get("source") or "") == PROFILE_SOURCE_IG: f["profile"] = {**p, "source": PROFILE_SOURCE_TT} stats["flags"] += 1 touched = True if touched: stats["tables"] += 1 return cur # ⚠ The precheck is the WHOLE mutation, run against a snapshot, so the common case (nothing to # do) spends no commit — the same posture `_reconcile_ig_graph_fields` takes, and the reason # this is safe to call on every boot. probe = copy.deepcopy(ut_all(rt)) _up(probe) if not (stats["columns"] or stats["flags"]): if stats["kept"]: log(f"[aios-auto] retract: {len(stats['kept'])} foreign column(s) KEPT because they " f"hold data — " + ", ".join(f"{k['table']}.{k['column']} ({k['rows']} rows)" for k in stats["kept"])) return stats stats = {"tables": 0, "columns": 0, "cells": 0, "flags": 0, "kept": []} rt.update(UT_STORE_KEY, _up, flush="sync") log(f"[aios-auto] retract: {stats['columns']} foreign column(s), {stats['cells']} cell(s) and " f"{stats['flags']} mis-declared flag(s) across {stats['tables']} database(s)") if stats["kept"]: log(f"[aios-auto] retract: {len(stats['kept'])} foreign column(s) KEPT because they hold " f"data — " + ", ".join(f"{k['table']}.{k['column']} ({k['rows']} rows)" for k in stats["kept"])) return stats def _carries_ig_presets(tbl): """Has this table ALREADY been given Instagram's preset contract? ⭐⭐ WAVE 32 · T42 — THE RECRUIT/REPAIR LINE, AND IT RESTS ON A STAMP THE CODE ALREADY WRITES. `ut_ensure(lock_fields=True)` and `_locked_ig_field` both set `automation: {preset: True}` on every preset column, and `migrate_ig_tables` BACKFILLS it on a table that predates the stamp (MEASURED: an unstamped 44-column fixture comes out 44/44 stamped after one pass). So "already carries the contract" is a declaration to read, not a threshold to invent — and inventing one was the alternative, because `MIGRATE_MIN_PRESET_FIELDS` is 3 and cannot tell a recruited table from a hand-made creator list that happens to have `handle`, `followers` and `bio`. ⛔ WHY THE LINE EXISTS AT ALL (owner item 1's third clause, MEASURED): a hand-made 5-column database that NO automation targets went to **45 columns** the moment an Instagram automation aimed at a DIFFERENT database was SAVED — because `_ig_schema_contract` walks every structurally-matching table in the tenant and hands each one the full 48-column contract. The owner's rule is that the preset fields may live in a database only if it is used for that scraping; recruitment by resemblance is the opposite of that rule. ⚠ NARROWING NEVER DROPS A COLUMN. `_reconcile_ig_graph_fields` deletes only the fixed retired-key `drops` set, so a de-recruited table keeps every field and cell it has — it stops being ADDED to and REPAIRED, which is the whole of the change. Asserted on a fixture rather than reasoned. ⚠ AND THE COST IS PAID ONLY WHEN IT HAS TO BE: `_ig_profile_tables` reads the automations bucket lazily, on the first candidate that is neither stamped nor the canonical `ut_ig_profile` — so a tenant whose tables are all genuinely Instagram's pays nothing on a path this wave is otherwise trying to make faster. ⛔⛔ TWO LEGS, AND THE SECOND IS THE ONE THAT KEEPS THIS FROM BREAKING A LIVE TENANT. The stamp is written by the very passes this predicate now gates, so a table recruited BEFORE the stamp existed and targeted by no surviving automation would be de-recruited and never stamped — silently losing its rollup/link repair, which is the slowest and worst failure available here (MEASURED on a stripped fixture: 44 columns, 0 stamps, 0 repaired). So a table ALSO counts as already-carrying when it declares one of the preset set's LINK or ROLLUP columns. Those are machine-authored by construction — nobody types `posts_link` or `avg_views_12` into a hand-made creator list — so the second leg is derived from the contract itself, not a threshold somebody picked. [[gate-and-nc-must-not-share-a-binding]]'s cousin: a discriminator written by the thing it discriminates needs an independent second leg, exactly as `_is_tt_profile_table` needed one. """ fields = (tbl or {}).get("fields") or [] for f in fields: if str(f.get("key") or "") not in PRESET_PROFILE_KEYS: continue automation = f.get("automation") if isinstance(automation, dict) and automation.get("preset") is True: return True return bool({str(f.get("key") or "") for f in fields} & PRESET_MACHINE_ONLY_KEYS) def discover_default_table(rt, tables=None): """Which database should a discovery automation write to when nobody has said? (2026-08-10) Owner: *"the damn database is supposed to be dynamic for whatever instagram profile is there. We only need ONE IG profile so it's not confusing."* ⛔ THE COMPLAINT WAS ABOUT A SECOND EMPTY DATABASE, NOT A HARDCODED KEY. `ut_beauty_influencer_ leads` appears NOWHERE in this product as a literal — it is a slug `ut_ensure` minted from the label the tenant typed, and `_ig_profile_tables` finds it structurally (that function's own note says so). What IS hardcoded is `DISCOVER_TABLE`, the FALLBACK target — and because it is a constant rather than a question, a tenant that already had an Instagram profile database got a SECOND, empty one the first time a discovery automation was saved without a target. MEASURED on nurilab: `ut_ig_candidates` (0 rows, 44 preset columns) sitting beside `ut_beauty_influencer_leads` (105 rows), both matching the profile predicate, one of them pure confusion. So the default becomes a question asked of the tenant: exactly one profile database (excluding the fallback) -> that one several, exactly one of which holds rows -> the one in use several in use, or none at all -> "" (the caller keeps DISCOVER_TABLE) ⚠ RETURNS "" RATHER THAN GUESSING. Two populated profile databases is a real ambiguity and picking one silently would write a paid discovery run into a database the user did not name — worse than the empty table this exists to prevent. "" means "no opinion", and the caller's existing fallback stands, which is exactly today's behaviour for that case. ⚠ THE FALLBACK IS EXCLUDED FROM ITS OWN ELECTION. `ut_ig_candidates` carries the full preset schema, so it matches `_ig_profile_tables` — without this line a tenant that already has the empty table would keep re-electing it and nothing would ever change. ⭐ WAVE 29 (W29-T01) — ONE READ, NOT TWO. This function read the whole `user_tables` bucket here and `_ig_profile_tables` read it AGAIN one line below: two 35.8 MB-ceiling deep copies to answer one question, on a route the automation surface calls on every open. The election is now computed from a single snapshot, which is also the only way it can be internally consistent — the two reads could disagree with each other under a concurrent write. ``tables`` lets `GET /automations` lend the copy it already holds; the answer is still derived fresh on every call, so there is no memo to invalidate when a database is created or deleted. """ try: tables = (ut_all(rt) if tables is None else tables) or {} found = [k for k in _ig_profile_tables(rt, tables=tables) if k != DISCOVER_TABLE] except Exception: # noqa: BLE001 return "" if len(found) == 1: return found[0] if len(found) > 1: used = [k for k in found if (tables.get(k) or {}).get("rows")] if len(used) == 1: return used[0] return "" def _apply_discover_default(rt, raw): """Fill in a discovery automation's target BEFORE the pure validator invents one. ⛔ IT HAS TO HAPPEN HERE, AND THE REASON IS THE WHOLE DESIGN. `clean_config` is the thing that turns a missing target into `DISCOVER_TABLE` — and it takes no `rt`, by design (it is a pure validator with a two-argument contract every gate and route depends on). Resolving only at the RUN sites would be cosmetic: the literal is written into the STORED config at save time, so `cfg.get("targetTable")` is truthy forever after and no runtime resolver is ever consulted. `create`/`patch` are the one pair that both know the tenant and sit above that validator. ⚠ ONLY WHEN THE CALLER SAID NOTHING. An explicit target — including one a previous save stored — is the user's choice and is never rewritten. """ cfg = raw.get("config") if not isinstance(cfg, dict) or str(cfg.get("targetTable") or "").strip(): return raw default = discover_default_table(rt) if not default: return raw return {**raw, "config": {**cfg, "targetTable": default}} def _vestigial_name_field(fields, rows): """⭐ 2026-08-07 — the born-blank `Name` column, or None. The owner's *"REPLACED"* half. `user_tables.create()` mints EXACTLY ONE column on a hand-made database — `{key:'name', label:'Name', type:'text', default:True}` — and `ut_ensure` then APPENDS the automation's columns after it, never reordering. So on every Instagram database somebody created before pointing an automation at it, column 1 is a `Name` nothing will ever write. ⛔ IT IS ONLY VESTIGIAL IF NOBODY EVER USED IT, and the test is deliberately conservative because this is the one branch in the migration that DESTROYS something. A column carrying any declaration (an automation binding, a metric, a profile flag) is somebody's work; a column with a value in any row is somebody's data. Either way it survives as an ordinary column and merely stops being the locked primary, which the `pinned` half achieves on its own. ⚠ `key == 'name'` IS ALREADY DECISIVE and the rest is belt: a user-created column is minted `custom__` (`buildOverlayField`) and every machine one comes from a `field_def` list, so nothing but `create()`'s default can hold this key. The label and type are checked anyway — a renamed or retyped column is a column someone touched on purpose. """ f = next((f for f in fields if f.get("key") == "name"), None) if f is None or f.get("label") != "Name" or f.get("type") != "text": return None if any(f.get(k) for k in ("automation", "metric", "profile", "measure", "formula")): return None if any(str(r.get("name") or "").strip() for r in rows.values()): return None return f def _merge_candidates(rows): """C3's merge rule, applied to rows now colliding on `(platform, handle)`. Earliest `first_found` wins - latest `last_found` wins - `found_count` SUMS - every other cell takes the first non-blank, preferring the most recently enriched row. ⚠ THE SURVIVING ROW KEEPS THE LOWEST ID. Row ids are referenced by saved views, comments and board cards; minting a new one would orphan all of them, so a merge picks a survivor rather than creating one. """ groups, keepers = {}, {} for rid in sorted(rows, key=lambda r: (len(str(r)), str(r))): r = rows[rid] h = str(r.get("handle") or "").strip() if not h: # ⛔⛔ THESE USED TO BE `continue`d, AND THE OUTPUT REPLACES THE TABLE'S ROWS — so a # row with no handle was SILENTLY DELETED. Owner, 2026-08-07: *"How come when I add a # manual handle in my 'Beauty influencer leads' database, after an automation run, it # gets deleted?"* # # A row you add by hand is born blank and stays blank until you type into it, so the # window is not narrow — it is every row, from creation until the cell is committed. # And before the profile flag shipped (834decd) a typed handle landed in the TYPIST'S # OVERLAY stratum, leaving the definition row's `handle` empty forever: the row was # dropped even after it looked filled in on screen. # # ⚠ THE FUNCTION'S JOB IS TO MERGE DUPLICATE CANDIDATES, and a row with no handle is # not a candidate — it is somebody's row. It cannot collide with anything (there is # nothing to key it on), so it is carried through UNTOUCHED rather than judged. A # merge pass that deletes what it cannot classify is not a merge, it is a filter. keepers[rid] = r continue groups.setdefault((str(r.get("platform") or PLATFORM_INSTAGRAM).strip(), h), []).append((rid, r)) out, merged = dict(keepers), 0 for _key, members in groups.items(): if len(members) == 1: out[members[0][0]] = members[0][1] continue merged += len(members) - 1 # Most-recently-enriched first, so "first non-blank" prefers the freshest measurement. ordered = sorted(members, key=lambda m: str(m[1].get("enriched_at") or ""), reverse=True) keep_id = sorted(rid for rid, _r in members)[0] acc = {} for _rid, r in ordered: for k, v in r.items(): if k not in acc and str(v or "").strip(): acc[k] = v firsts = [str(r.get("first_found") or "").strip() for _i, r in members if r.get("first_found")] lasts = [str(r.get("last_found") or "").strip() for _i, r in members if r.get("last_found")] if firsts: acc["first_found"] = min(firsts) if lasts: acc["last_found"] = max(lasts) total = sum(_ig_int(r.get("found_count")) or 0 for _i, r in members) if total: acc["found_count"] = str(total) out[keep_id] = acc return out, merged #: ⛔ DEBT D-74 + D-81 (wave 27 item 12) — `tracked` WAS DELETED FROM THE VOCABULARY AND NEVER #: FROM THE TABLES. W26/R6 removed the column from `CANDIDATE_FIELDS` and from ~14 engine sites #: *"we already have a Field called stage… we don't need another checkbox for this"* — and #: MEASURED 2026-08-07 on `royal-imports/aios-nurilab-data`, `ut_beauty_influencer_leads` still #: declared it, filled on 1 of 41 rows. A checkbox on the owner's flagship database that nothing #: writes, nothing reads, and anybody can still click. #: #: ⚠ THE DECISION WAS WHICH, NOT WHETHER (D-74 stated both futures): sweep the ticks and lose the #: record of who approved what before the wave, or leave them and let the next hand-made `tracked` #: column silently inherit them. The wave takes the sweep — R6 deleted the concept, so a surviving #: tick is not a decision anybody can act on, and an inert cell that reappears in the next EXPORT #: as a column nobody declared is the worse half. TRACKED_KEY = "tracked" def _retired_tracked_field(fields): """The retired `tracked` column when it is still the MACHINE's own, else None. ⛔ GUARDED THE WAY `_vestigial_name_field` IS, and for the same reason: this is a branch that DESTROYS something. `tracked` was minted `{type: "checkbox"}` by the discovery preset, so a column still declaring a checkbox is the one R6 retired. A column somebody has since RETYPED, or bound a formula/link/rollup/measure to, is their work wearing an old key — it survives as an ordinary column and merely stops being fed by anything, which was already true. ⚠ THE ASYMMETRY WITH THE TWO KEYS BESIDE IT IS DELIBERATE. `posts` and `post_hashtags` are dropped unguarded: both are machine-derived cells with no history of anyone editing them. `tracked` is the one a HUMAN ticked, so it is the one that gets the conservative test. """ f = next((f for f in fields if str(f.get("key") or "") == TRACKED_KEY), None) if f is None or f.get("type") not in ("checkbox", "", None): return None if any(f.get(k) for k in ("formula", "link", "rollup", "measure", "metric")): return None return f def _guarded_drops(table, drops): """`drops` minus any key this table is allowed to keep — today, exactly `tracked` (D-81). ⛔ THE GUARD HAS TO LIVE WHERE THE DROP HAPPENS, and finding that out cost a red check. `tracked` is deleted from TWO places: `migrate_ig_tables`' profile loop, which asks `_retired_tracked_field` whether the column is still the machine's own, and the schema reconciliation, which drops by KEY with no question asked. Guarding only the first was decorative — the schema pass runs immediately afterwards on the same table and deleted the retyped column the guard had just spared. Two paths to one deletion with one of them checked is the shape where a control looks present and enforces nothing ([[defects-that-mask-each-other]]). """ drops = set(drops or ()) if TRACKED_KEY in drops: stored = [f for f in (table or {}).get("fields") or [] if str(f.get("key") or "") == TRACKED_KEY] if stored and _retired_tracked_field(stored) is None: drops.discard(TRACKED_KEY) return drops def _orphan_tracked_rows(fields, rows): """Row ids carrying a `tracked` CELL that no field declares — D-74's actual subject. ⛔ THE COLUMN AND THE CELLS WERE RETIRED SEPARATELY, WHICH IS WHY THIS IS NOT COVERED BY THE BRANCH ABOVE. R6 deleted the field DEFINITION from `CANDIDATE_FIELDS`; a table whose declaration was already dropped keeps every `tracked` key in its row dicts, invisible on every surface that renders from `fields` — and a migration that only looks at `fields` finds nothing to do and `continue`s past the table. Storage, export and any future re-declaration of the key all still see them. """ if any(str(f.get("key") or "") == TRACKED_KEY for f in fields): return [] # the column above owns its own cells return [rid for rid, r in rows.items() if TRACKED_KEY in (r or {})] def backfill_corpus_snapshots(rt, log=print): """The BACKFILL half: every profile row holding a measurement with no series behind it gets its corpus observation. Returns how many were written. ⛔ ONE BUILDER, TWO CALLERS — `run_discover_instagram` for rows arriving now, this for rows that arrived before the forward fix existed. A second row-shape here would be two ideas of what a corpus observation is, and they would disagree on the next promoted column. ⚠ THE CONDITION IS "NO SERIES AT ALL", not "no series for this day", and the narrowness is deliberate. A profile that HAS been enriched already owns real observations; synthesising a corpus row dated `last_found` for it could out-rank the exact read in a `latest` rollup and replace a measured number with a corpus one. The gap this closes is a row with NOTHING behind it, which is the only case where a corpus observation can only improve matters. ⚠ IDEMPOTENT BY THE SNAPSHOT KEY (`@`), never by a flag — so this is safe on every write, which is what `migrate_ig_tables`' placement requires. After one pass the row has a series, so it stops matching and the scan costs a dict comparison. """ snaps = ut_get(rt, IG_SNAPSHOTS_TABLE) if snaps is None: return 0 # no series store — see `append_ig_snapshots` on why we never mint one known = {str((r or {}).get("influencer_key") or "").strip().lower() for r in (snaps.get("rows") or {}).values()} known.discard("") #: A cell counts as a MEASUREMENT if the snapshot schema has somewhere to put it — derived #: from `SNAPSHOT_FIELDS` rather than named, for `_ig_profile_tables`' own reason (D-71: a #: named list goes silently empty the day the schema moves). measured = tuple(fd["key"] for fd in SNAPSHOT_FIELDS if fd["key"] not in _SNAPSHOT_OWN_KEYS and fd.get("type") in ("int", "pct")) incoming = [] for table_key in _ig_profile_tables(rt): for row in ((ut_get(rt, table_key) or {}).get("rows") or {}).values(): handle = str((row or {}).get("handle") or "").strip().lstrip("@").lower() if not handle or handle in known: continue if not any(str((row or {}).get(k) or "").strip() for k in measured): continue # nothing was ever read about this row — there is no observation to record built = corpus_snapshot_row(row) if built: incoming.append(built) known.add(handle) # two profile tables holding one handle write ONE observation return append_ig_snapshots(rt, incoming, log=log) def backfill_post_snapshot_grain(rt, log=print): """Copy `posted_at`/`type` off the POST row onto every measurement missing them. ⛔ STRUCTURAL, NEVER A FLAG: the row matches when the snapshot's cell is blank AND the post's is not, so a pass that has already run matches nothing and a post that genuinely has no date is never re-visited on a promise it cannot keep. Same idempotency discipline as the migration around it. ⚠ IT FILLS BLANKS AND NEVER OVERWRITES. A snapshot's `posted_at` is what the vendor said WHEN THE MEASUREMENT WAS TAKEN; if a later pull disagrees with the post row, the measurement's own record of the moment is the one to keep — a fact table that lets a dimension rewrite its history is not a fact table. """ psnaps = ut_get(rt, IG_POST_SNAPSHOTS_TABLE) posts = ut_get(rt, IG_POSTS_TABLE) if psnaps is None or posts is None: return 0 have = {f.get("key") for f in (psnaps.get("fields") or [])} carried = [k for k in ("posted_at", "type") if k in have] if not carried: return 0 # the columns are not declared yet — `ut_ensure` adds them first by_shortcode = {} for row in (posts.get("rows") or {}).values(): code = str((row or {}).get("shortcode") or "").strip() if code: by_shortcode[code] = row or {} changes = {} for rid, row in (psnaps.get("rows") or {}).items(): post = by_shortcode.get(str((row or {}).get("shortcode") or "").strip()) if not post: continue for key in carried: want = str(post.get(key) or "").strip() if want and not str((row or {}).get(key) or "").strip(): changes.setdefault(str(rid), {})[key] = _s(want, 200) if not changes: return 0 def _up(cur): cur = cur if isinstance(cur, dict) else {} table = cur.get(IG_POST_SNAPSHOTS_TABLE) if table is not None: for rid, values in changes.items(): table.setdefault("rows", {}).setdefault(rid, {}).update(values) return cur rt.update(UT_STORE_KEY, _up, flush="sync") log(f"[ig-migrate] post-measurement grain: {len(changes)} row(s) gained a post date/kind") return len(changes) #: ⭐⭐ 2026-08-10 (owner: *"retire the old hardcoded method and replace the Instagram database work #: with correct Rollup and Link fields"*) — THE PROFILE COLUMNS THAT MAY BECOME ROLLUPS. #: #: A column qualifies when the SERIES has somewhere to put it: `SNAPSHOT_FIELDS` carries the same #: key, so `latest` over `profile_snapshots_link` can re-derive it. That is 27 of the 44 preset #: columns — the other 17 are identity/bookkeeping (`handle`, `platform`, `enriched_at`), already #: relational (the 12 links and rollups), or have no observation behind them at all #: (`location_guess`/`location_confidence`, which WE infer rather than read). #: #: ⛔⛔ AND "DERIVABLE" IS NOT "SAFE", WHICH IS THE WHOLE REASON THIS IS A PREDICATE AND NOT A LIST. #: MEASURED on nurilab before any conversion: a bare `latest` would BLANK 221 filled cells — #: `avg_engagement` and `category` lose 62 each — because the discovery reads that produced those #: values were never recorded as observations (the hole this same wave closed going forward, in its #: historical form: 62 profiles hold exactly ONE observation, the enrichment scrape, dated three #: days AFTER the discovery run whose numbers are on their row). A migration that converts on a #: NAMED LIST would do that damage on any tenant whose series is thinner than the list's author #: assumed. #: #: ⇒ So the migration converts a column only where it can PROVE, on this tenant's own rows, that #: the derived value equals the stored one everywhere. See `_convertible_profile_columns`. #: ⚠ Re-runnable by design: as the series fills in, later passes convert more. A column that is not #: safe today is not refused forever, it is simply not converted yet. IG_DERIVABLE_KEYS = tuple( fd["key"] for fd in PRESET_PROFILE_FIELDS if fd.get("type") not in ("link", "rollup") and fd["key"] not in ("platform", "handle", "enriched_at", "first_found", "last_found", "found_count", "created_by") and fd["key"] in {s["key"] for s in SNAPSHOT_FIELDS}) def _derived_profile_bag(key): """The rollup that replaces the mapper's write of ONE profile column. ⚠ `where: is_not_empty` IS NOT DECORATION. A bare `latest` returns the newest observation's value INCLUDING ITS BLANK, so a cheap pull that did not read the field would erase what an expensive one learned — the exact asymmetry `upsert_rows`' merge exists to prevent on the scalar side. `where` runs BEFORE the ranking (contract C1), so this reads "the newest observation that ACTUALLY READ this field", which is what the materialised cell meant. MEASURED: it rescues 5 of the 12 columns a bare `latest` would damage; the other 7 need the observation itself, which is why the guard below is a proof and not a hope. """ return {"link": "profile_snapshots_link", "field": key, "fn": "latest", "sortBy": "pulled_at", "sortDir": "desc", "where": [{"field": key, "op": "is_not_empty"}], "whereConj": "and"} def _derived_profile_columns(table, snapshots): """Which of `IG_DERIVABLE_KEYS` should be DECLARED as rollups on THIS table. = the ones already converted, PLUS the ones whose derived value equals the stored value on every row. Returns keys. ⛔⛔ IT MUST RE-ASSERT THE ONES ALREADY CONVERTED, and forgetting that is a one-line revert with a several-hour fuse. `_reconcile_ig_graph_fields` overwrites every contract key it owns — `rollup` included, by design — so a converted column that this function stopped naming would be re-typed back to `int`/`text` on the next reconcile, its bag dropped, and its value re-written by the mapper. The edit would look applied and undo itself later, which is exactly [[preset-unlock-needs-a-custody-stamp]]'s shape. An already-converted column is therefore included unconditionally rather than re-proved: its cells ARE the derived values, so a proof would be comparing the fold against itself. ⛔ THE PROOF RUNS PER TENANT, PER COLUMN, AT MIGRATION TIME — never against a list somebody measured on one workspace. `followers` derives exactly on nurilab and could be thin somewhere else; `category` is damaged on nurilab and might be perfect elsewhere. The only honest authority is the data in front of the migration. ⚠ A column with NO stored values anywhere converts too, and that is deliberate rather than an oversight: there is nothing to lose, and leaving it materialised would mean the schema differs between two tenants for no reason a reader could discover. """ fields = {str(f.get("key")): f for f in (table or {}).get("fields") or []} rows = (table or {}).get("rows") or {} by_subject = {} for snap in ((snapshots or {}).get("rows") or {}).values(): key = str((snap or {}).get("influencer_key") or "").strip().lower() if key: by_subject.setdefault(key, []).append(snap or {}) # The engine's own ordering, once per subject — newest first, unrankable rows last. ordered = {} for subject, observations in by_subject.items(): keyed = [(o, _sort_key(o.get("pulled_at"), "date")) for o in observations] rankable = [(o, k) for o, k in keyed if k is not None] rankable.sort(key=lambda ok: ok[1], reverse=True) ordered[subject] = [o for o, _k in rankable] + [o for o, k in keyed if k is None] # ⛔ A rollup needs the LINK it folds. A profile table that has not been through the graph # reconcile has no `profile_snapshots_link`, and declaring one there would mint a column whose # link resolves to nothing — a blank cell wearing a configured column's clothes. if "profile_snapshots_link" not in fields: return [] out = [] for key in IG_DERIVABLE_KEYS: field = fields.get(key) if not field: continue if field.get("type") == "rollup": out.append(key) # already converted — see the note above on why continue declared = str(field.get("type") or "text") safe = True for row in rows.values(): stored = str((row or {}).get(key) or "").strip() window = [o for o in ordered.get( str((row or {}).get("handle") or "").strip().lower(), []) if str(o.get(key) or "").strip() != ""] # the `where` guard, applied here derived = str(_rollup_fold("latest", [o.get(key) for o in window], declared) or "").strip() if stored != derived: safe = False break if safe: out.append(key) return out def _as_derived_field(field): """One preset scalar declaration → the same column declared as a rollup over the series.""" key = str(field.get("key") or "") return {**{k: v for k, v in field.items() if k not in ("agg",)}, "type": "rollup", "rollup": _derived_profile_bag(key)} def migrate_ig_tables(rt, log=print, only=""): """Bring every Instagram Profile and canonical child table onto one current contract. Returns counted changes rather than a reassuring line. Profile cells are converted before their type declarations change; then the graph reconciliation adds any missing preset Links/Rollups, repairs child types, stamps locks, adds all reciprocal backlinks, and deletes only retired machine fields. `only` narrows to a single table, which is how `ut_ensure` calls it: the migration then rides the WRITE PATH, so a table is brought forward immediately before anything appends to it and no caller has to remember to run anything. """ stats = {"tables": 0, "retyped": 0, "converted": 0, "stamped": 0, "merged": 0, # ⭐ 2026-08-07 (owner ruling) — the primary-column half. Counted separately from # `stamped` because they answer different questions: that one is "how many ROWS got a # platform", these are "how many TABLES changed shape". "pinned": 0, "flagged": 0, "droppedName": 0, "droppedRecentPosts": 0, "droppedPostHashtags": 0, "droppedTracked": 0, "droppedTrackedCells": 0, "schema": 0} want = {f["key"]: f["type"] for f in CANDIDATE_FIELDS} # ⛔ `only` NARROWS THE SET, IT DOES NOT BYPASS THE TEST — and skipping that cost six red # checks the first time. `ut_ensure` calls this with the key it is ABOUT TO CREATE, so on a # first run the table does not exist yet: with the detection bypassed it fell through to the # writer and stamped a stub table with empty fields, clobbering the creation that was the whole # point of the call (the owner stamp and the flow tag went with it). A table that does not # exist, or that is not an Instagram profile table, has nothing to migrate. # ⭐ WAVE 32 · T42 — the CONTRACT set, not the detector's. This loop retypes columns, stamps # `platform` and drops retired keys, i.e. it edits the schema — so it answers to the owner's # rule about which databases Instagram may write into, exactly like `_ig_schema_contract` at # the bottom of this function. `_ig_profile_tables` stays the wider question and keeps its own # callers (`discover_default_table`, `backfill_corpus_snapshots`, the detector's NC). detected = set(_ig_contract_tables(rt)) targets = ([only] if only in detected else []) if only else sorted(detected) # ⭐⭐ WAVE 31 · T30 — `only` NOW NARROWS THE WHOLE FUNCTION, AND IT DID NOT. # # ⛔ THE BUG WAS THAT `only` NARROWED THE LOOP ABOVE AND NOTHING ELSE. The tenant-wide passes # at the bottom — two full `_ig_schema_contract` + `_reconcile_ig_graph_fields` rounds and two # backfills — ignore `only` entirely, so `migrate_ig_tables(only=X)` paid a whole-tenant # Instagram reconciliation no matter what X was. And `ut_ensure` calls it that way on EVERY # ensure whose field set intersects `PRESET_PROFILE_KEYS` — which TikTok's profile schema does, # because the two networks share column names. # MEASURED on the wave-30 fixture, `only="ut_tt_profile"`: `_ig_profile_tables` returns `[]`, # `stats` comes back completely empty and ZERO store writes happen — at a cost of **10 # whole-document reads**, each a `Store.get` deep copy (documented ceiling 35.8 MB / ~1.4 s). # Across one save's four `ut_ensure` calls that was **40 of the 41** reads behind owner item 7's # *"it just say Saving... and takes a long time"*. The migration was not slow; it was running at # all. # ⚠ THE COMMENT ONE SCREEN DOWN IS WHY THIS WENT UNSEEN FOR THREE WAVES: *"Both are guarded on # 'is there anything to do' and cost a dict scan in the steady state, which is what makes them # safe on a function that rides every `ut_ensure`."* That guard is on the WRITE. The read was # never guarded, and a scan of a freshly deep-copied 35 MB document is not a dict scan # [[read-a-gates-predicate-for-what-it-excludes]]. # ⛔ SCOPED TO THE PROVEN-EMPTY CASE ONLY, deliberately. When `only` IS a detected profile table # the function is unchanged, tenant-wide passes included — that is the Instagram path every # migration check in `verify_automation` exercises. And a caller passing no `only` still gets # the full sweep. The claim being made here is narrow and checkable: *a table that is not in the # Instagram graph has no Instagram migration*, which is the same thing the `detected` test above # already decided one line earlier and then threw away. # ⛔ AND IT IS NOT A LEND. Lending one snapshot through the rest of this function would be the # obvious next fix and it is WRONG: the second contract pass exists precisely because it must # read what the first pass WROTE (see its own note below), so a read-write-read sequence cannot # answer out of one snapshot without silently un-fixing that idempotency. if only and not targets: return stats for table_key in targets: tbl = ut_get(rt, table_key) or {} fields = [dict(f) for f in tbl.get("fields") or []] rows = {rid: dict(r) for rid, r in (tbl.get("rows") or {}).items()} # ⭐ THE IDEMPOTENCY KEY. Only a column whose STORED type is still the old one is touched, # so the cell conversions below can never run twice on the same value. stale = [f for f in fields if f.get("key") in want and f.get("type") != want[f["key"]] and f.get("type") in ("text", "", None)] stale_keys = {f["key"] for f in stale} has_platform = any(f.get("key") == "platform" for f in fields) needs_stamp = [rid for rid, r in rows.items() if not str(r.get("platform") or "").strip()] # ⭐⭐ 2026-08-07 (owner ruling) — MAKE THE HANDLE THE PRIMARY COLUMN AND THE BINDING. # `ut_ensure` merges by KEY and never updates a field that already exists, so declaring the # two keys on `PRESET_PROFILE_FIELDS` reaches NEW tables only. Every table already in # production needs them stamped here, and that is the whole reason this branch exists. # # ⚠ IDEMPOTENT BY READING THE STORED DECLARATION, never by a marker column — W26's rule, # and it is free here: re-stamping a boolean and a two-key dict is a no-op by nature, which # is exactly why these are safe to run on every write where a value conversion would not be. pin_f = next((f for f in fields if f.get("key") == PRESET_PINNED_KEY), None) \ if PRESET_PINNED_KEY else None flag_f = next((f for f in fields if f.get("key") == PRESET_FLAG_KEY), None) \ if PRESET_FLAG_KEY else None # ⛔ AT MOST ONE PROFILE COLUMN PER TABLE is the law `user_tables` enforces at both write # doors, and a migration is not exempt from it. If somebody has already flagged another # column on this table, stamping the handle too would leave the table in a state its own # validator refuses — and `_profile_of` returns the FIRST match, so the enrich action would # silently bind to whichever came first in the field list. Leave it alone and pin only. other_profile = next((f for f in fields if f is not flag_f and isinstance(f.get("profile"), dict)), None) want_pin = pin_f is not None and pin_f.get("pinned") is not True want_flag = (flag_f is not None and other_profile is None and not isinstance(flag_f.get("profile"), dict)) vestigial = _vestigial_name_field(fields, rows) retired = {f.get("key") for f in fields} & {"posts", "post_hashtags"} if _retired_tracked_field(fields) is not None: retired.add(TRACKED_KEY) orphan_tracked = _orphan_tracked_rows(fields, rows) if (not stale_keys and has_platform and not needs_stamp and not want_pin and not want_flag and vestigial is None and not retired and not orphan_tracked): continue stats["tables"] += 1 if want_pin: pin_f["pinned"] = True stats["pinned"] += 1 if want_flag: flag_f["profile"] = {"source": PROFILE_SOURCE_IG} stats["flagged"] += 1 if vestigial is not None: # ⚠ THE CELLS GO WITH THE COLUMN. A row dict keeping a `name` key whose field no longer # exists is invisible everywhere except the next export, where it reappears as a column # nobody declared. `_vestigial_name_field` has already proven every one of them blank. fields = [f for f in fields if f is not vestigial] for r in rows.values(): r.pop("name", None) stats["droppedName"] += 1 if retired: fields = [f for f in fields if f.get("key") not in retired] for r in rows.values(): for key in retired: r.pop(key, None) stats["droppedRecentPosts"] += int("posts" in retired) stats["droppedPostHashtags"] += int("post_hashtags" in retired) stats["droppedTracked"] += int(TRACKED_KEY in retired) for rid in orphan_tracked: # D-74: a tick whose column was already gone. Counted apart from the column drop # because they answer different questions — "how many tables still declared it" and # "how many rows were still carrying it after the declaration went". rows[rid].pop(TRACKED_KEY, None) stats["droppedTrackedCells"] += 1 for key in stale_keys: conv = _MIGRATE_CONVERT.get(key) if not conv: continue # int/checkbox already store the right text shape for r in rows.values(): if str(r.get(key) or "").strip(): new = conv(r[key]) # "" from a converter means UNREADABLE, and the cell keeps its original text. # A migration that empties a cell is indistinguishable from one that moved it. if new: r[key], stats["converted"] = new, stats["converted"] + 1 for f in stale: f["type"] = want[f["key"]] stats["retyped"] += 1 for rid in needs_stamp: rows[rid]["platform"] = PLATFORM_INSTAGRAM stats["stamped"] += 1 if not has_platform: fields = [dict(f) for f in CANDIDATE_FIELDS if f["key"] == "platform"] + fields rows, merged = _merge_candidates(rows) stats["merged"] += merged def _up(cur, _f=fields, _r=rows): cur = cur or {} t = dict(cur.get(table_key) or {}) t["fields"], t["rows"] = _f, _r cur[table_key] = t return cur rt.update(UT_STORE_KEY, _up) # The profile loop owns value conversions and deduplication. The schema pass owns the # declarations across ALL related databases and is safe only after those conversions, because # the stored old type is the idempotency key for unit-changing values such as engagement. wanted, drops = _ig_schema_contract(rt) stats["schema"] = int(_reconcile_ig_graph_fields(rt, wanted, drops)) # ⛔⛔ A SECOND CONTRACT PASS, AND IT IS NOT BELT-AND-BRACES — WITHOUT IT THIS FUNCTION IS NOT # IDEMPOTENT, which is the one property everything else here is built on. # # The derived overlay (`_derived_profile_columns`) decides by COMPARING the profile cell to the # fold of the series. The pass above REPAIRS THE SERIES — it retypes the child columns and runs # `_IG_RETYPE_CONVERTERS`, which rewrites `avg_engagement` from the vendor's 0-1 fraction to our # 0-100 percent. So on a tenant whose children are stale, the first proof compares a repaired # profile cell against an UNREPAIRED observation, finds them different, and declines to convert # a column that is in fact perfectly derivable. The next call then converts it — and # `verify_automation`'s "a SECOND pass is a no-op" check went red, which is exactly what that # check is for. MEASURED on the wave-26 fixture: 24 columns converted on pass 2, none on pass 1. # # ⚠ The module already knew this shape one level up — the profile loop's own note says the # schema pass "is safe only after those conversions, because the stored old type is the # idempotency key". Same argument, one table further down. # ⚠ CHEAP WHEN THERE IS NOTHING TO DO: on a current tenant the contract recomputes to the same # declarations and `_reconcile_ig_graph_fields` writes nothing. Twice, not looped: the repair is # what moves the data, and it has already run. wanted2, drops2 = _ig_schema_contract(rt) stats["schema"] = int(bool(stats["schema"] | int(_reconcile_ig_graph_fields(rt, wanted2, drops2)))) # ⭐⭐ 2026-08-10 — THE TWO VALUE BACKFILLS, and they run AFTER the schema pass on purpose: # `backfill_post_snapshot_grain` writes into columns that pass has just declared, and running # it first would find nothing to fill and report a truthful zero about the wrong moment. # Both are guarded on "is there anything to do" and cost a dict scan in the steady state, which # is what makes them safe on a function that rides every `ut_ensure`. stats["series"] = backfill_corpus_snapshots(rt, log=log) stats["postGrain"] = backfill_post_snapshot_grain(rt, log=log) if stats["tables"] or stats["schema"] or stats["series"] or stats["postGrain"]: log(f"[ig-migrate] {stats}") return stats # ── WAVE 25 · C6 (owner ruling R5) — SEEDING A SEARCH FROM A VIEW OR COHORT. ────────────────── # "Find me more accounts like the ones in this view." The server reads the seed rows, ranks their # SHARED characteristics, and FILLS IN the discovery conditions — visible and editable, never # hidden (R5). # # ⛔ THE RANKING IS RESTRICTED TWICE, AND BOTH RESTRICTIONS ARE MEASURED RATHER THAN CHOSEN: # 1. R5 restricts it to `BD_POPULATED_FIELDS` — a characteristic extracted from a field the # corpus barely populates produces a filter that returns nothing and looks exactly like # "no such accounts exist" (the trap `BD_FILTER_LEAD` already exists to warn about). # 2. A derived predicate must NARROW, or `narrowing_refusal` refuses to save it. The # intersection of the two sets is MEASURED at exactly eight fields — `account`, `biography`, # `external_url`, `fbid`, `full_name`, `id`, `profile_name`, `profile_url` — and each of them # narrows under its own `default_operator`, asserted in the gate rather than assumed. # # ⭐ AND FOUR OF THOSE EIGHT CANNOT GENERALISE, which is the part worth stating: `account`, `id`, # `fbid` and `profile_url` are IDENTITIES. A filter derived from them re-finds the seed rows and # nothing else — a search that returns what you gave it, at full price. So the ranker reads only # the three that describe a KIND of account: shared words in the bio, a shared domain in the link, # and shared words in the name. SEED_SOURCES = ("view", "cohort") SEED_MAX_ROWS = 200 # bounded and DISCLOSED in `basis.rows` — never a silent [:N] SEED_MAX_PREDICATES = 3 # a filter of ten derived guesses is not a filter SEED_MIN_COVERAGE = 0.5 # a "shared" characteristic under half the rows share is not shared SEED_MIN_ROWS = 2 # one row has nothing to share WITH #: ⛔ AND A COVERAGE BAR ALONE IS NOT ENOUGH, which the gate caught rather than review: at two #: seed rows, `coverage >= 0.5` is satisfied by a word appearing in exactly ONE of them. So every #: bio word in a two-row seed qualified as "shared", and the surface would have offered a #: characteristic derived from a single record as the thing those records have in common. A #: shared characteristic needs at least two rows to be shared BY — n=1 dressed as a pattern is #: the fabrication this module refuses everywhere else. SEED_MIN_ROWS_SHARING = 2 #: Which ROW column feeds which VENDOR field. The row keys are the C1 preset set's; the vendor #: names are `BD_FILTER_FIELDS`'. Two vocabularies for one thing, so the mapping is written down. SEED_FROM_ROW = (("bio", "biography"), ("external_url", "external_url"), ("full_name", "full_name")) #: Words that are shared by everything and therefore discriminate nothing. Deliberately SHORT — #: an aggressive list would silently drop a real signal, and the coverage bar already removes most #: noise. Anything ≤2 characters is dropped by length instead of by enumeration. SEED_STOPWORDS = frozenset({ "the", "and", "for", "with", "you", "your", "our", "are", "not", "all", "any", "from", "this", "that", "have", "has", "was", "com", "www", "http", "https", "out", "new", "more", "who", "how", "その", "инст"} | {"official", "welcome", "contact", "email", "dm", "link"}) def _seed_tokens(text): """One cell → the distinct words worth matching on. Lowercased, de-punctuated, ≥3 chars.""" words = re.findall(r"[a-z0-9]{3,}", str(text or "").lower()) return {w for w in words if w not in SEED_STOPWORDS} def _seed_domain(url): """A link-in-bio → its registrable-ish host, or ''. `linktr.ee/x` and `linktr.ee/y` share a domain and that IS the shared characteristic; the path is the thing that differs.""" raw = str(url or "").strip() if not raw: return "" host = (urlparse(raw if "//" in raw else "https://" + raw).hostname or "").lower() return host[4:] if host.startswith("www.") else host def seed_predicates(rows): """R5: seed rows → `(derived_predicates, basis)`. PURE — no store, no vendor, no network. `basis` is what makes the offer honest rather than magic: how many rows were read, which characteristics were extracted, and **the MEASURED coverage of each**, so a weak signal reads as weak instead of being presented with the same confidence as a strong one. ⛔ EVERY DERIVED PREDICATE IS ASSERTED AGAINST `narrowing_refusal` BEFORE IT IS OFFERED (HARD RULE 11). A suggestion the product's own save door then refuses is the post-W24 default-versus-guard defect rebuilt — and it would be worse here, because the user did not type it: they would be told their own product's suggestion is invalid. ⚠ AN EMPTY ANSWER IS A LEGITIMATE ANSWER. When the seed rows share nothing above the coverage bar, `derived` is `[]` and `basis` says why. An honest nothing beats a filter that looks plausible, returns zero rows, and reads as "no such accounts exist". """ rows = [r for r in (rows or []) if isinstance(r, dict)] total = len(rows) basis = {"rows": total, "fields": [], # R5's second lane, reported as the MEASUREMENT it is rather than sold as a feature. # `related_accounts` is 22% populated corpus-wide (D-25) and is not part of the C1 # preset set, so on our own seed rows it is usually simply absent — which the surface # must SAY, or a lane that adds nothing reads as a lane that failed. "related": {"tried": total, "found": sum(1 for r in rows if str(r.get("related_accounts") or "").strip())}, "note": ""} if total < SEED_MIN_ROWS: basis["note"] = ("a seed needs at least two records — one record has nothing to share " "with anything") return [], basis ranked = [] for row_key, vendor_field in SEED_FROM_ROW: counts = {} for r in rows: vals = ({_seed_domain(r.get(row_key))} if vendor_field == "external_url" else _seed_tokens(r.get(row_key))) for v in vals: if v: counts[v] = counts.get(v, 0) + 1 for value, n in counts.items(): cov = n / total if cov >= SEED_MIN_COVERAGE and n >= SEED_MIN_ROWS_SHARING: ranked.append({"name": vendor_field, "value": value, "coverage": round(cov, 3), "rows": n}) # Strongest first; ties broken by the LONGER value, which is the more specific one. ranked.sort(key=lambda f: (-f["coverage"], -len(f["value"]), f["name"])) derived, seen_fields = [], set() for f in ranked: if len(derived) >= SEED_MAX_PREDICATES: break # One predicate per FIELD: two `biography includes` rows AND-ed together narrow to the # accounts carrying both words, which is a much smaller search than the seed implies. if f["name"] in seen_fields: continue p = {"name": f["name"], "operator": default_operator(f["name"]), "value": f["value"]} if not predicate_narrows(p): continue # cannot be offered; it would be refused at the save door seen_fields.add(f["name"]) derived.append(p) basis["fields"].append({"name": f["name"], "label": field_label(f["name"]), "value": f["value"], "coverage": f["coverage"]}) if derived and narrowing_refusal(derived, "and"): # Belt AND braces: the per-predicate check above should make this unreachable, and it is # asserted in the gate. If the LAW ever changes shape, this fails closed with an honest # empty rather than offering something the save door will reject. basis["note"] = ("the shared characteristics found are not specific enough to search on " "— add a condition of your own") return [], basis if not derived: basis["note"] = basis["note"] or ( "these records share no bio words, link domain or name in common — nothing could be " "derived, so the conditions are yours to write") return derived, basis def seed_rows(rt, table_key, view_id=""): """The rows a seed source selects. `(rows, problem)` — bounded, and the bound is DISCLOSED by `basis.rows` rather than silently applied ([[no-unverifiable-aggregates]]).""" t = ut_get(rt, str(table_key or "")) if t is None: return [], f"{table_key!r} is not a database in this workspace" rows = list((t.get("rows") or {}).values()) if str(view_id or "").strip(): # ⛔ THE SAME RESOLVER `enters_view` USES, not a second one. A seed that selected a # different row set from the view it names would be describing a view nobody has. tree, _fields, problem = view_filter(rt, table_key, view_id) if problem: return [], problem rows = [r for r in rows if lane_match(tree, r)] return rows[:SEED_MAX_ROWS], "" def automation_tables(defn): """Every `ut_*` table THIS automation writes into — its config target plus every `create_record` action's table, deduped, order preserved. ⚠ THE SAME TWO AUTHORITIES `observed_category_tables` READS, narrowed to one definition. Split out rather than parameterised because the two questions are different: that one asks "where could a Category value be, anywhere in this tenant" and is deliberately generous; this one asks "which rows has THIS automation already found" and must not reach into a table it does not write, or a nightly search would start excluding another automation's finds. """ keys, seen = [], set() cfg = (defn or {}).get("config") or {} for raw in [cfg.get("targetTable"), *(((a or {}).get("config") or {}).get("table") for a in walk_actions(((defn or {}).get("flow") or {}).get("actions")))]: k = str(raw or "").strip() if k.startswith(UT_PREFIX) and k not in seen: seen.add(k) keys.append(k) return keys def candidate_key(platform, handle): """The discovery upsert identity: `(platform, handle)`, as one string. ⭐ WAVE 29 — LIFTED OUT OF `run_discover_instagram`, where it was a closure, because a second runner now needs the same identity. Two closures computing "the same" key is how one of them grows a different default for a blank `platform` and the two networks quietly start sharing rows — which is the exact data-loss shape wave 26's R4 added `platform` to prevent. ⚠ A BLANK PLATFORM DEFAULTS TO INSTAGRAM, and that is a FACT about the stored data rather than an assumption: every row written before wave 26 came from the Instagram runner. Blank-keyed rows would fail to match their own re-find and duplicate the whole table on the next run. """ return f"{str(platform or PLATFORM_INSTAGRAM).strip()}\n{str(handle or '').strip()}" def already_found_handles(rt, defn, cap=None): """The handles this automation has already written, for the vendor-side exclusion (item 7). ⭐⭐ THE POINT IS THE BILL, not the row count. Discovery upserts by handle, so re-finding a profile has always been harmless — and never free: the vendor bills for every record it returns, so a nightly search over stable keywords pays again, in full, for the same accounts every night and reports them as `seen_again`. MEASURED shape of the waste: nurilab's beauty scout re-found its whole result set on every run. This is the list that stops it, sent as one `not_in` the vendor evaluates BEFORE billing. ⚠ SCOPED TO THIS AUTOMATION'S OWN TABLES. Excluding handles another automation found would hide profiles this one has never seen — cheaper, and wrong: two scouts with different keywords are two questions, and one must not answer with "somebody already looked at that". """ cap = BD_EXCLUDE_MAX if cap is None else int(cap) out, seen = [], set() for table_key in automation_tables(defn): for row in ((ut_get(rt, table_key) or {}).get("rows") or {}).values(): h = str((row or {}).get("handle") or "").strip().lstrip("@").lower() if h and h not in seen: seen.add(h) out.append(h) if len(out) >= cap: return out return out def observed_category_tables(rt): """Every table a Category value could have been written into, DERIVED from the automations this tenant actually has — the default discovery table and the snapshot table are the FLOOR, not the list. ⭐ Two authorities, because wave 24 and wave 25 each moved where the target is declared: `config.targetTable` (the discovery runner's own write) and any `create_record` action's `config.table` (R2 — authoritative since wave 25, and the field the Builder's Database picker writes). Both are read, deduped, order preserved so the defaults come first. """ keys, seen = [], set() def _add(k): k = str(k or "").strip() if k and k.startswith("ut_") and k not in seen: seen.add(k) keys.append(k) _add(DISCOVER_TABLE) _add("ut_ig_snapshots") try: for defn in (all_definitions(rt) or {}).values(): cfg = (defn or {}).get("config") or {} _add(cfg.get("targetTable")) for act in walk_actions(((defn or {}).get("flow") or {}).get("actions")): _add(((act or {}).get("config") or {}).get("table")) except Exception: # noqa: BLE001 # A definition set that cannot be read costs the DERIVED lanes and keeps the defaults — # the same posture as the master lane below: degrade to less vocabulary, never to an error. pass return keys def observed_categories(rt, limit=60): """⭐ DEBT D-59 — the Category values WE HAVE ACTUALLY SEEN, for a combobox that still accepts free text. D-59's exit condition, verbatim: *"EITHER derive the options from values we have actually seen (the `category` column on `ut_ig_candidates` + the master store, offered as a combobox that still accepts free text), OR buy a sample large enough to enumerate the real vocabulary. **Not**: transcribe a published taxonomy and hope it lines up."* This is the first branch. ⛔ WHY A PUBLISHED TAXONOMY WOULD HAVE BEEN WORSE THAN NO DROPDOWN. A filter on a value the corpus does not use returns zero rows and looks exactly like an honest "no such accounts exist" — so a picker built from Instagram's own category list would mislead precisely when it looked most authoritative. Every option here has been observed on a real row, and each carries its COUNT so a value seen once reads differently from one seen forty times. ⚠ IT STAYS A COMBOBOX. These are the values we have seen, not the values that exist — a control that refused anything else would be a second, quieter version of the same lie. """ seen = {} def _eat(rows, key): for r in rows or []: v = " ".join(str((r or {}).get(key) or "").split())[:60] if v: seen[v] = seen.get(v, 0) + 1 # ⛔ THE TABLES ARE DERIVED, NOT NAMED — and this is a REGRESSION BY OMISSION that shipped # green. The two constants below were the whole list, which was correct until **wave 25 R2 # made the Create record action's `config.table` AUTHORITATIVE**: from that ruling on, a # discovery automation writes wherever the user pointed it, and `ut_ig_candidates` is merely # the DEFAULT nobody keeps. MEASURED on nurilab 2026-08-06 — `ut_ig_candidates` held 0 rows # while the tenant's two real target tables held 20 each, so this function honestly reported # `{options: []}` and the combobox it feeds had nothing to offer. A vocabulary harvester that # names its sources goes quietly empty the moment the product lets a user choose one. for key in observed_category_tables(rt): _eat(((ut_get(rt, key) or {}).get("rows") or {}).values(), "category") try: # ...and the PLATFORM master, which is the whole point of pooling it: a tenant with three # candidates still gets a vocabulary drawn from every profile the platform has captured. import ig_master if ig_master.configured(): # ⚠ `_handle().get(SNAP_BUCKET)` — a dict of rows keyed by id. NOT `_upsert()`, which # returns the upsert FUNCTION; calling that and reading `.get(...)` off it answers # None, so the master lane would have degraded to nothing INSIDE the except below and # this would have shipped as a silent no-op with the gate green. Verified against # `ig_master.series_for`, which reads the same bucket the same way. _eat((ig_master._handle().get(ig_master.SNAP_BUCKET) or {}).values(), "category") except Exception: # noqa: BLE001 # A master that cannot be read costs the tenant its own values and nothing else. pass return [{"value": v, "count": n} for v, n in sorted(seen.items(), key=lambda kv: (-kv[1], kv[0]))[:limit]] def preset_plan(rt, table_key): """C1 / R2b: the preset set diffed against ONE database's CURRENT fields. `{table, fields: [{key, label, type, present}], willUse: [...], willCreate: [...]}` ⛔ COMPOSED HERE, NOT IN THE CLIENT. The owner's ask is a config panel that says which columns an automation will USE and which it will CREATE — and the only thing that can answer it is whatever holds both lists. A client that diffed the preset set against a table's fields would be a second implementation of `ut_ensure`'s merge rule, free to disagree with the merge that actually runs; it would be wrong precisely when the two lists differ, which is the only case anybody is asking about. ⚠ A TABLE THAT DOES NOT EXIST IS NOT AN ERROR — it is the ordinary state at R10's spawn-on-save moment, and the honest answer is "all of them will be created". `present` is a fact about the table, so it reads False for every field rather than the call refusing. ⛔ 2026-08-07 — THE PROJECTION IS FOUR KEYS AND DELIBERATELY DROPS EVERY DECLARATION BAG (`link`, `rollup`, `metric`, `profile`, `pinned`). Measured off the live Space the day the relational pair shipped: `posts_link` and the four rollups arrive here with their bags EMPTY. That is CORRECT for this payload — it answers "which columns will be used vs created", where a name and a type are the whole question — and it is safe today because nothing RENDERS a column from this list: the grid reads full field dicts from `/tables` (`scoped_pool` passes `dict(f)` straight through), and the column itself is spawned server-side by `ut_ensure` from `PRESET_PROFILE_FIELDS`, which carries the bag. ⚠ **It stops being safe the moment somebody previews a rollup from this payload** — they would render "Avg views - last 12 posts" configured by nothing. Read the field off `/tables`, or widen this projection deliberately; do not assume a bag is here because the field has one. """ have = {str(f.get("key")) for f in ((ut_get(rt, str(table_key or "")) or {}).get("fields") or [])} fields = [{"key": f["key"], "label": f["label"], "type": f.get("type") or "text", "present": f["key"] in have} for f in PRESET_PROFILE_FIELDS] return {"table": str(table_key or ""), "exists": ut_get(rt, str(table_key or "")) is not None, "fields": fields, "willUse": [f for f in fields if f["present"]], "willCreate": [f for f in fields if not f["present"]]} def clean_predicates(raw, operator="and", kind=""): """Validate a discovery filter into the vendor's shape. Returns `(predicates, error)`. Refuses rather than coerces: a predicate naming a field the API rejects comes back as a 400 with the field named, because the alternative — dropping it — turns "find me verified accounts in Georgia" into "find me any account" and bills for the difference. ⭐ WAVE 32 · T46 (D-167) — `kind` NARROWS THE VOCABULARY TO THE PLATFORM'S. Default `""` keeps Instagram's 21 names, so every existing caller and every stored Instagram automation is unchanged; a `discover_tiktok` is validated against the 5 names its corpus actually has. The refusal it produces is the same sentence, listing the platform's own fields — which is the difference between a search that says why it cannot be built and one that is built, paid for, and comes back empty. """ fields, _lead = filter_fields(kind) out = [] for p in (raw or [])[:12]: if not isinstance(p, dict): continue name = _s(p.get("name") or p.get("field"), 60).strip() op = _s(p.get("operator") or p.get("op"), 20).strip() if name not in fields: return None, ("that field cannot be searched — the searchable ones are: " + ", ".join(field_label(f) for f in fields)) # ⭐ PER-FIELD, not the global list. `is_business_account >= 3` used to validate cleanly # because every operator was legal on every field; a yes/no column offering "at least" # is a question with no meaning that the vendor is nevertheless asked. allowed = ops_for(name) if op not in allowed: return None, (f"{field_label(name)} cannot be asked {BD_OP_LABELS.get(op, op)!r} — " "it takes: " + ", ".join(BD_OP_LABELS.get(o, o) for o in allowed)) entry = {"name": name, "operator": op} if op not in BD_NULLARY_OPS: raw_v = p.get("value") # ONE value or SEVERAL — several is the "contains any of" shape (owner item 3), and # it is stored as a list rather than as N sibling predicates so the row a person sees # and the row that is stored are the same thing. vals = raw_v if isinstance(raw_v, list) else [raw_v] vals = [v.strip() if isinstance(v, str) else v for v in vals] vals = [v for v in vals if not (v is None or (isinstance(v, str) and not v))] if not vals: return None, f"give {field_label(name)} something to compare against" if len(vals) > 1 and op not in BD_MULTI_VALUE_OPS: return None, (f"{BD_OP_LABELS.get(op, op)!r} takes one value — " f"{field_label(name)} can only be given a list with " + ", ".join(BD_OP_LABELS[o] for o in ops_for(name) if o in BD_MULTI_VALUE_OPS)) if len(vals) > MAX_PREDICATE_VALUES: return None, (f"{field_label(name)} takes at most {MAX_PREDICATE_VALUES} values " "in one condition") if field_kind(name) == "boolean": ok = {o["value"] for o in BD_BOOLEAN_OPTIONS} bad = [v for v in vals if str(v).lower() not in ok] if bad: return None, f"{field_label(name)} is answered Yes or No" # ⛔ A REAL BOOLEAN, NOT THE STRING. MEASURED 2026-08-06: the filter API answers # **400** to `{"name": "is_verified", "operator": "=", "value": "true"}` and # accepts `True` (`snap_msh2rp4kh3v833q0u`). Nobody had ever sent one — the field # was a free-text box until this wave, so the Yes/No dropdown that makes it easy # to ask is also the thing that would have made every boolean condition fail. # The dropdown's option VALUES stay "true"/"false" (a