api/routes_automation.py

"""routes_automation.py — wave-18 item 5 (contract C4-AUTO): the automation surface's API.

The router is thin on purpose: everything that can be wrong about an automation — a malformed
cron, a URL the SSRF rail refuses, a key field that is not one of the mapped columns — is decided
in `automation_engine`, which is a pure-ish module a gate can drive without a server. This file
does auth, shape and status codes.

⛔ THE TICK ENDPOINT IS THE ONE UNAUTHENTICATED ROUTE, AND IT IS FAIL-CLOSED TWICE OVER. It takes
no session (an external cron has no cookie), so it is gated on a shared secret in
`X-AIOS-TICK-TOKEN`; and when `AIOS_AUTOMATION_TICK_TOKEN` is UNSET the route refuses everything
rather than admitting everyone. A "no token configured means no check" default is how an internal
trigger becomes a public one — the same class of mistake as an empty-200 permission answer.
"""
import os
import time

from fastapi import APIRouter, Body, Depends, Header, Request

import automation_engine as engine
import oauth_connect
import routes_oauth
import scope_cache
from deps import Session, err, module_gate, require_session

router = APIRouter(prefix="/api/v1")

# C5 (wave 22): the OAuth connector surface rides INSIDE this router — main.py belongs to no
# session this wave, and this router is already mounted there. `/api/v1` + `/oauth/...`.
router.include_router(routes_oauth.router)
# ⭐ WAVE 23 (C11): the connectors directory rides this router for the same reason the OAuth
# routes do — it is mounted already, so E's page needs no main.py change to reach it. (C still
# owns main.py; this avoids an unnecessary ask across a fence.)
import routes_connectors                                             # noqa: E402
router.include_router(routes_connectors.router)

#: The registry key this surface carries (C-AUTONAV — A adds the row; the gate is live now, so
#: the day the row lands the wall is already the one that was tested).
MODULE = "automation"

_GATE = module_gate(MODULE)


def _wire(defn, tenant):
    """One automation, as the client reads it. `running` is PROCESS state, never store state —
    see the engine header on why a persisted 'running' is a permanent lock."""
    live = engine.running(tenant, defn.get("id"))
    sched = defn.get("schedule") or {}
    nxt = engine.next_fire(sched.get("cron")) if sched.get("enabled") else None
    status = dict(defn.get("status") or {})
    if live:
        status = {**status, "state": "running", "startedAt": live.get("startedAt"),
                  "step": live.get("step")}
    return {
        "id": defn.get("id"), "name": defn.get("name"), "kind": defn.get("kind"),
        "config": defn.get("config") or {}, "schedule": sched, "status": status,
        "runs": list(defn.get("runs") or [])[:engine.MAX_RUNS],
        "created": defn.get("created"), "createdBy": defn.get("createdBy"),
        "nextRunAt": nxt.strftime("%Y-%m-%d %H:%M") if nxt else "",
        "running": bool(live),
        # ⭐ WAVE 27 (contract C6) — DEBT D-70: "a paid search is already outstanding at the
        # vendor". THE MONEY GUARD, and it was INERT for a whole wave because this line was
        # missing: the client declared `awaitingResults` REQUIRED and read it in `runBlock()`,
        # nothing ever sent it, so `undefined` was falsy and the Run button stayed armed.
        #
        # ⛔ WHY IT IS NOT `running`. `running` above is PROCESS state (`engine.running`) and is
        # FALSE for the entire 20–30 MINUTE vendor wait, which IS the hazard window: the ticks
        # are 3 minutes apart, so between them the automation is idle and the button re-arms.
        # MEASURED the expensive way on 2026-08-06 (~$0.025): with no snapshot outstanding a
        # Run-now press returned 200 and started a BRAND-NEW billable corpus search.
        #
        # ⛔ AND NOT A PERSISTED "running" FLAG EITHER — the engine header forbids one (it
        # outlives the process and locks the automation forever). `state.pendingSnapshot` is the
        # field that already exists, is already persisted, is already carried untouched through
        # `clean_definition`, and is the SAME expression `pending_collect_ids()` uses to decide
        # what to collect. One truth, two readers.
        "awaitingResults": bool(str((defn.get("state") or {}).get("pendingSnapshot")
                                    or "").strip()),
        # C3 (wave 22): the trigger config rides whole — the webhook token included, because
        # the person configuring the external caller has to be shown the URL somewhere, and
        # this payload is session-gated behind the same wall as everything else here.
        "trigger": defn.get("trigger") or None,
        "statusNote": defn.get("statusNote") or "",
        # The one-sentence summary (airtable-brief rec 7), composed from the definition so it
        # cannot describe steps the engine does not run.
        "sentence": engine.compose_sentence(defn),
        # THE CANVAS TOPOLOGY (R5) rides with the automation rather than being rebuilt in the
        # client, for the same reason `cronPresets` does: the engine that RUNS the steps is the
        # only thing entitled to say what the steps are.
        "graph": engine.graph(defn),
        # ⭐ WAVE 23 (C4/C5) — THE BUILDER'S OWN STATE, and its absence here was a silent-drop
        # bug caught in review rather than by a gate: `flow` was stored, patchable and validated,
        # but never sent back. B would have saved a flow through PATCH, got a 200, and watched
        # every action vanish on reload — the classic "it didn't save" with nothing red anywhere.
        "flow": defn.get("flow") or {"actions": []},
        # ⭐⭐ WAVE 32 · T45 (owner item 10) — CONFIGURED / UNCONFIGURED, per action, on the wire.
        #
        # ⛔ THE SERVER SAYS IT, not the client, and that is the whole reason it is here: the same
        # `engine.action_needs` answers this label, `engine.run_refusal`'s 400 and `run_now`'s own
        # refusal, so a card cannot read "Configured" over an action the run will refuse. Three
        # readers, one predicate — the alternative is a client-side rule that agrees with the
        # server until somebody adds a key to one of them (`awaitingResults` above is this file's
        # own record of what the other shape costs).
        # ⚠ IT IS A LIST, KEYED BY ACTION ID, and it names the NESTED actions too — an unconfigured
        # step inside an If / then branch is exactly the one a person cannot see.
        # ⚠ COSTS NO STORE READ. `action_needs` is pure over `(kind, config)`; this route is the
        # one W30-T12 took the `user_tables` deep copy off, and a label is not worth putting it
        # back. What needs the target database's schema (an enrich binding resolved by the profile
        # FLAG) stays a run-time refusal — see `ACTION_REQUIRED`'s note.
        "unconfigured": engine.unconfigured_actions(defn),
    }


def _triggers_vocab(session):
    """C3's server-owned trigger list: `[{key, label, ready, needs, planned}]`, keys EXACTLY
    `engine.TRIGGER_KEYS` + `engine.TRIGGER_PLANNED`. `ready:false` + `needs` renders as a
    not-configured state — never a dead control, never a client-side union.

    ⭐ WAVE 23 (R2): the list is the WHOLE Airtable-parity vocabulary, and the two triggers we
    have not built ride it with `planned: true`. That is the honest version of "show all, wire
    eight": the picker paints them faded with a reason instead of a shorter list that quietly
    implies the missing ones do not exist. `clean_trigger` refuses them, so the faded state is
    enforced at the door and not merely in the client's `disabled` attribute.
    """
    tick_on = _tick_state()["enabled"]
    g = oauth_connect.status(session.runtime, session.uname).get("google") or {}
    email_ready = bool(g.get("configured")) and bool(g.get("connected")) \
        and not g.get("reconnect")
    email_needs = "" if email_ready else (
        "connect_gmail" if g.get("configured") else "configure_google")
    per = {
        "manual": (True, ""),
        "schedule": (tick_on, "" if tick_on else "arm_tick"),
        "event_field": (True, ""),
        "record_updated": (True, ""),
        "record_created": (True, ""),
        "enters_view": (True, ""),
        "webhook": (True, ""),
        "email": (email_ready, email_needs),
        "form_submitted": (True, ""),
        # ⭐ WAVE 24 (C-TRIG) — Instagram discovery, now a trigger. Readiness is the vendor key,
        # the same bit `paidReady` carries: with no key the search door is closed and the picker
        # must say so rather than offering a control that silently finds nothing. `clean_trigger`
        # still ACCEPTS it either way — readiness is a deployment fact, not a validity one, which
        # is the same split `email` already makes.
        "ig_profile_match": (engine.bd_ready(), "" if engine.bd_ready() else "configure_brightdata"),
        # ⭐ WAVE 29 (D-9 / R1) — TikTok, live. ⛔ IT NEEDS ITS OWN ROW EVEN THOUGH THE ANSWER IS
        # IDENTICAL, and the reason is the `.get(k, (True, ""))` default below: a trigger this dict
        # forgets is reported READY, so a deployment with no vendor key would offer TikTok search
        # as configured and the search would find nothing. Same key, same readiness bit, stated.
        "tiktok_profile_match": (engine.bd_ready(),
                                 "" if engine.bd_ready() else "configure_brightdata"),
    }
    #: ⭐ WAVE 24 — the server's own one-line description per trigger. C-TYPES: the picker renders
    #: THIS under the option, because a CLIENT paraphrase of a server vocabulary is a second copy
    #: of it, free to drift. Absent = the client shows nothing, never something invented.
    detail = {
        "manual": "It runs only when you press Run now",
        "schedule": "It runs on a repeating schedule",
        "event_field": "A record in the database starts matching a condition you set",
        "record_updated": "Any of the columns you watch is changed",
        "record_created": "A new record is added to the database",
        "enters_view": "A record starts appearing in a saved view",
        "webhook": "Something outside calls this automation's URL",
        "email": "A message arrives in the connected mailbox",
        "form_submitted": "Somebody submits one of this database's forms",
        "ig_profile_match": "Search Instagram for profiles matching your filters, on a schedule",
        "button_clicked": "Somebody presses a button on a record",
        "comment_added": "Somebody comments on a record",
        "web_page_changed": "A page you are watching is different from last time",
        "tiktok_profile_match": "Search TikTok for profiles matching your filters, on a schedule",
    }

    def _taxonomy(k):
        """⭐ WAVE 25 · C2 — the four taxonomy keys, composed from the ENGINE's maps.

        ⛔ `.get(k) or FALLBACK`, NEVER `TRIGGER_GROUP_OF[k]`. The first draft of this indexed the
        map on the reasoning that a default is how a Connector trigger quietly appears under
        Database — and the gate rejected it, correctly, against the incident `per.get(k, ...)`
        eight lines below records: a key added to `TRIGGER_KEYS` without remembering a dict beside
        it raised KeyError and took `GET /automations` down, i.e. the whole surface, which polls
        this every 2.5 s. A mis-grouped row is cosmetic; a 500 is not, and the ranking is not
        close.
        ⚠ THE CLASSIFICATION IS STILL MANDATORY — it is enforced at the GATE (no shipped trigger
        may land in `other`) rather than at the request. Soft here, hard there.
        """
        g = engine.TRIGGER_GROUP_OF.get(k) or engine.TRIGGER_GROUP_FALLBACK
        return {"group": g,
                "groupLabel": engine.TRIGGER_GROUPS[g]["label"],
                "groupOrder": engine.TRIGGER_GROUPS[g]["order"],
                # The SUB-group inside "Connector"; None everywhere else. ⚠ Group by this key,
                # render its label — it is NOT a connector-directory slug (see the engine's note).
                "connector": engine.TRIGGER_CONNECTOR.get(k),
                # D-55: "the cron drives this one". Derived from the engine's schedule set MINUS
                # `manual`, so the client's `CRON_DRIVEN_TRIGGERS` copy can be deleted.
                "schedules": k in engine.TRIGGER_CRON_KEYS}

    out = []
    for k in engine.TRIGGER_KEYS:
        # ⚠ `.get` WITH A DEFAULT, NOT `per[k]`. This loop walks the ENGINE's vocabulary and
        # indexed a hand-maintained dict beside it: adding a key to `TRIGGER_KEYS` without
        # remembering this dict raised KeyError and 500'd `GET /automations` — the payload the
        # whole automation surface polls every 2.5 s — with every gate and `tsc` still green.
        # Defaulting to "ready, needs nothing" is the honest fallback: a trigger the engine
        # offers and this route has no readiness opinion about is simply available.
        ready, needs = per.get(k, (True, ""))
        row = {"key": k, "label": engine.TRIGGER_LABELS[k], "ready": ready, "needs": needs,
               "planned": False, "detail": detail.get(k, ""),
               # A3(3): the connect affordance is SERVER-COMPOSED — the client never maps a
               # `needs` token to a route, so B's CONNECT_PROVIDERS shim deletes itself.
               "connect": None, **_taxonomy(k)}
        if not ready and needs == "connect_gmail":
            row["connect"] = {"provider": "google",
                              "startUrl": "/api/v1/oauth/google/start"}
        out.append(row)
    for k in engine.TRIGGER_PLANNED:
        out.append({"key": k, "label": engine.TRIGGER_LABELS[k], "ready": False,
                    "needs": "coming_soon", "planned": True, "connect": None,
                    "detail": detail.get(k, ""), **_taxonomy(k)})
    return out


def _tick_state():
    """⭐ WAVE 21 (C6 amendment A1) — can a SCHEDULE fire on this deployment?

    TWO independent paths can: the in-process scheduler (`AIOS_AUTOMATIONS=1`,
    `automation_engine.py` module bottom) and an external cron POSTing `/automations/tick`,
    gated on `AIOS_AUTOMATION_TICK_TOKEN` (AWS EventBridge in production). The Step-1 Trigger
    card must be honest in both directions: "schedules won't fire" on a deployment where
    EventBridge demonstrably fires them daily is the exact lie R9 forbids. `external` means
    "the door is OPEN", never "the caller is alive" — the client's copy says so."""
    inproc = os.environ.get("AIOS_AUTOMATIONS") == "1"
    ext = bool(os.environ.get("AIOS_AUTOMATION_TICK_TOKEN"))
    return {"enabled": bool(inproc or ext),
            "source": "in-process" if inproc else ("external" if ext else "")}


#: W29-T01 — the Board retirement ran, per tenant, this process. Same shape and same reasoning as
#: `_C8_MIGRATED` below: a one-way cleanup of data no living code path creates any more, whose
#: cost is a full walk of every row-cell in the tenant and whose result on pass 2..N is always
#: "nothing to do". ⚠ A module-global keyed by tenant is deliberately NOT reset by a table create
#: or delete — a new database cannot contain the legacy stage cells this retires.
_BOARD_RETIRED = set()

#: ⭐⭐ WAVE 30 · T12 — THE PICKER DEFAULT, MEMOISED. `{tenant: (stamp, table_key)}`, the shape
#: `scope_cache` stores.
#:
#: ⛔ WHY THE DERIVED STRING AND NOT THE DOCUMENT. The obvious cache here is the `user_tables`
#: bucket itself, and it is the wrong one: that bucket is up to 35.8 MB per tenant and this box is
#: the HF free tier, so caching it would trade a latency problem for a memory one. What the warm
#: path actually needs is `discover_default_table`'s ANSWER — one short string.
#:
#: ⛔ AND WHY NOT A PLAIN `_BOARD_RETIRED`-STYLE ONCE-PER-PROCESS SET, which would have been less
#: code: the election reads which profile databases exist and which hold rows, and BOTH change
#: while the process lives (somebody creates a database, an automation writes the first row). A
#: once-per-process memo would pin the picker's default to whatever was true at boot and never
#: correct itself — a default that disagrees with the save door, which is exactly the wave-25 R2
#: defect the `"table"` line's own comment records. A TTL bounds the staleness instead.
#:
#: ⚠ `scope_cache` rather than a hand-rolled dict, because it is the house pattern for precisely
#: this (`routes_customers`, `routes_products`, `pages` all use it) and it is stale-while-refresh:
#: once a copy exists NO request blocks on a rebuild. Automation was the one module importing it
#: nowhere, which the wave-30 scout named as the reason every other surface feels fast.
_DISCOVER_DEFAULT = {}
#: 5 minutes — the same order as `apiBridge.ts:CUSTOMERS_FRESH_MS` on the client. Overridable so a
#: gate can pin it rather than sleep.
_DISCOVER_DEFAULT_TTL = float(os.environ.get("AIOS_AUTOMATION_DEFAULT_TTL") or 300)


#: The one store key `_LentTables` intercepts. DERIVED from the engine's own constant rather than
#: written out here: `engine.ut_all` reads the bucket through it, so if that key ever moves, the
#: lend moves with it instead of silently becoming a pass-through that still looks correct.
_UT_STORE_KEY = engine.UT_STORE_KEY


def _LentTables(runtime, tables):
    """⭐⭐ WAVE 30 · T13, NOW W31-C1 — a read-only `st` that serves ONE already-read `user_tables`
    document and passes every other key straight through to the real runtime.

    ⭐⭐ WAVE 31 · C1 — THE BODY IS NOW `core.user_tables.lend`, AND THE CLASS THAT USED TO BE HERE
    IS GONE. W30's version carried its own note that the general fix was unavailable at the time:
    *"adding a `tables=` parameter to `may_open` would mean editing platform/core/user_tables.py,
    which belongs to another lane this wave."* It is session B's lane THIS wave, they built the
    generalisation (`user_tables.lend` + the `_LENDABLE` allow-list) for W31-T10, and this is the
    second caller adopting it rather than becoming a third copy of the shape.

    ⭐ AND THE SWAP FIXES SOMETHING THIS FUNCTION NEVER COVERED. The old class intercepted exactly
    one key, so `may_open`'s LAST branch — `shares.may_see` → `role_for` → `st.get('object_shares')`
    — still took a fresh read PER TABLE. `_LENDABLE` covers both buckets, so a shared database no
    longer costs a second whole-document read per row of the picker. The name is kept because
    `verify_automation`'s NC66 pins this symbol as the thing it replaces to restore the `1 + N`
    state; keeping it a function means that control still has exactly one seam to swap.

    ⛔ THE OBVIOUS FIX IS STILL THE FORBIDDEN ONE. Inlining the creator-or-admin test here would
    remove the N reads and re-create the exact defect wave 20 fixed: this route USED to carry its
    own wider rule (`createdBy in (uname, 'automation', 'scheduler')`), so a non-admin saw a
    database in the picker and was refused the moment they opened it. `may_open` stays THE one
    resolver, unmodified and still called per table; it is simply no longer charged for a document
    the caller is already holding.
    """
    # ⚠ IMPORTED HERE, not at module scope — `core` is deliberately kept out of this module's
    # import-time graph (the same reason `automation_tables` does it locally 300 lines down).
    import core.user_tables as _ut
    return _ut.lend(runtime, **{_UT_STORE_KEY: tables})


def _discover_default(session, tables):
    """The discovery picker's default table for this tenant — WITHOUT a bucket read on a warm call.

    ⚠ `tables` is the document the caller ALREADY holds on a cold call (the retirement pass reads
    one). Passing it through means the cold path elects from the copy it has rather than taking a
    second one, so this is never an extra read — only ever a saved one.
    """
    if not session.runtime.available():
        return ""
    if tables is not None:
        # Cold call: the document is in hand. Elect from it and prime the memo in the same pass.
        value = engine.discover_default_table(session.runtime, tables=tables)
        _DISCOVER_DEFAULT[session.tenant] = (time.time(), value)
        return value
    return scope_cache.get(_DISCOVER_DEFAULT, session.tenant, _DISCOVER_DEFAULT_TTL,
                           lambda: engine.discover_default_table(session.runtime))


def warm_default(rt):
    """⭐⭐ WAVE 30 · T12, THE COLD HALF. Elect the discovery picker's default at BOOT.

    ⛔ THE HALF THE MEMO CANNOT FIX, and it is why this exists rather than being a nicety. The memo
    above makes calls 2..N free; **call 1 is still a full `user_tables` download**, and it lands on
    whoever clicks Automation first after a deploy — the one visitor with no cache anywhere,
    waiting on a document whose documented ceiling is 35.8 MB / ~1.4 s, taken under the store's
    single lock. That is the owner's *"only automation has a loading screen"* on a cold Space, and
    two waves of memoising the warm path could never touch it. Automation was the only module
    importing `scope_cache` nowhere AND the only one absent from `_prewarm`.

    ⭐ IT ELECTS, IT DOES NOT CACHE THE DOCUMENT — deliberately, and this is the whole design.
    `discover_default_table` reads the bucket once here, in the prewarm daemon thread where nobody
    is waiting, and what survives is a DERIVED STRING. Holding the 35.8 MB document resident would
    trade a latency the tenant notices for memory the HF free tier does not have, which is the
    argument `_DISCOVER_DEFAULT`'s own note makes against caching it.

    ⚠ SAME TTL AS THE REQUEST PATH, not a permanent set: the election reads which databases exist
    and which hold rows, and both change while the process lives. Priming by assignment is exactly
    what the cold request path already does one function up, so there is one way this memo is
    filled, not two.

    Returns the elected key (`""` when the store is unavailable), so a caller can log it rather
    than guess whether the warm-up did anything. Called from `main.py:_prewarm` only.
    """
    tenant = str(getattr(rt, "key", "") or "")
    if not tenant or not rt.available():
        return ""
    value = engine.discover_default_table(rt)
    _DISCOVER_DEFAULT[tenant] = (time.time(), value)
    return value


@router.get("/automations")
def list_automations(session: Session = Depends(_GATE)):
    """`{automations: [...], kinds: [...], cronPresets: [...]}` — the rail's whole payload.

    The vocabularies ride WITH the list rather than sitting in a client constant: the cron
    presets and the kind list are the server's, and a client copy of either is a thing that goes
    stale silently (the editor would offer a preset the parser rejects)."""
    # ⭐⭐ WAVE 29 (W29-T01, owner item 5: "Automation takes a while to appear"). THE COMPLAINT WAS
    # THIS ROUTE, and the cost was never the automations: it read the whole `user_tables` bucket
    # THREE TIMES per call — once for the stage-field retirement scan, twice more inside
    # `discover_default_table`. That bucket's documented ceiling is 35.8 MB / ~1.4 s to serialize
    # (`automation_engine.py` header) and `core/store.py` re-serializes on EVERY `.get()`, hit or
    # miss, UNDER THE STORE LOCK — so the reads also serialize behind each other. ~4 s of deep
    # copying to render a rail that shows a name and a toggle.
    #
    # Now: ONE read, lent to both callers. And the retirement SCAN — O(all row-cells in the
    # tenant), which hits its own `if not stage_keys: return` only AFTER walking every row of
    # every table — runs once per tenant per process, mirroring the `_C8_MIGRATED` guard below.
    #
    # ⛔ WHY SKIPPING THE SCAN ON CALLS 2..N IS SAFE, and it is not "because it is idempotent":
    # `all_definitions` STRIPS retired board state on every read (`automation_engine.py`'s
    # `_without_retired_board`), so the persist step is hygiene, not correctness. No client can
    # ever be shown state this skip left behind. Nothing writes new `stage_auto_` cells either —
    # wave 26's R6 deleted the board that made them.
    # ⭐⭐ WAVE 30 · T12 (owner items 4/5, the complaint that has now survived TWO waves).
    # ⛔ W29-T01 GUARDED THE SCAN AND NOT THE READ, and that is the whole of what was left. The
    # line below used to be unconditional — every single request deep-copied the tenant's entire
    # `user_tables` document (documented ceiling 35.8 MB / ~1.4 s, and `core/store.py:Store.get`
    # re-serializes on EVERY `.get()`, hit or miss, UNDER THE STORE LOCK so the copies also queue
    # behind each other) — while on calls 2..N the result was DISCARDED: `tables` had exactly two
    # consumers, the `_BOARD_RETIRED` scan (skipped after call 1) and a picker DEFAULT STRING.
    # A whole-tenant document, per request, to render a rail showing a name and a toggle.
    tables = None
    if session.runtime.available() and session.tenant not in _BOARD_RETIRED:
        tables = engine.ut_all(session.runtime)
        _BOARD_RETIRED.add(session.tenant)
        # Idempotent Board retirement removes only engine-marked stage fields and legacy Board
        # state. User-created Status/Stage columns remain intact.
        engine.retire_automation_board_state(session.runtime, tables=tables)
    defs = engine.all_definitions(session.runtime)
    items = [_wire(d, session.tenant) for _, d in
             sorted(defs.items(), key=lambda kv: (kv[1].get("name") or "").lower())]
    return {"automations": items,
            # ⭐ WAVE 24 — DERIVED from the engine's own `KINDS`, not a hand-written trio. It was
            # three literals that happened to match, which is a second copy of a server
            # vocabulary; R6 has just made two of them uncreatable and `plain` has joined, so a
            # hand list would now be wrong in three ways at once. `creatable` carries R6 onto the
            # wire, so the ruling is a fact the client can read rather than one it must remember.
            "kinds": [{"key": k, "label": engine.KIND_LABELS.get(k, k),
                       "creatable": k not in engine.RETIRED_KINDS}
                      for k in engine.KINDS],
            "cronPresets": engine.CRON_PRESETS,
            # ⚠ A BOOLEAN, NEVER THE KEY. The surface needs to say "the paid rung is not
            # configured" honestly instead of offering a tier that will silently refuse — and
            # that needs exactly one bit. Shipping the key itself to a browser would put a
            # billable secret in every user's devtools.
            "paidReady": engine.bd_ready(),
            # THE SOURCE REGISTRY (D-9's seam), on the wire for the same reason `cronPresets` is:
            # the module that RUNS a source is the only thing entitled to say what it can do, and
            # a client copy of "Instagram can discover, TikTok cannot" goes stale in silence.
            "sources": engine.source_status(),
            # The discovery vocabulary, likewise server-owned: every name here is MEASURED-
            # accepted by the vendor's own validator, so a client that invented one would build a
            # query the API rejects. `lead` is the subset seen carrying VALUES on real rows.
            # ⭐⭐ WAVE 32 · T46 (D-167) — THE VOCABULARY HAS A PLATFORM, AND `byKind` IS ADDITIVE
            # ON PURPOSE. `fields`/`lead` keep INSTAGRAM's 21 and 3, so no stored automation and no
            # client that has not adopted this changes behaviour today; `byKind` carries the per-
            # corpus answer, and the SERVER already refuses a TikTok predicate naming one of the 16
            # fields TikTok's dataset does not have (`clean_predicates(..., kind)`). The door is
            # closed either way — this is what lets the Find panel stop OFFERING them.
            # ⚠ Derived through `engine.filter_fields`, the same accessor the validator uses, so
            # the published vocabulary and the enforced one cannot drift — which is precisely what
            # D-167 was: a route serving 21 names and a validator checking the same 21, both wrong
            # about TikTok together, with nothing able to notice.
            "discoverByKind": {k: {"fields": list(engine.filter_fields(k)[0]),
                                   "lead": list(engine.filter_fields(k)[1])}
                               for k in engine.DISCOVERY_KINDS},
            "discover": {"fields": list(engine.BD_FILTER_FIELDS),
                         "lead": list(engine.BD_FILTER_LEAD),
                         "operators": list(engine.BD_FILTER_OPS),
                         "nullaryOperators": list(engine.BD_NULLARY_OPS),
                         "maxRecords": engine.BD_MAX_RECORDS,
                         # Wave 22 C4 — ADDITIVE: per-field flags for the toggle rows (the 3
                         # PII fields stay structurally absent, R3) + the too-big guard's
                         # numbers, so the surface can say WHY a filter is refused before the
                         # server has to.
                         "filterMeta": engine.filter_meta(),
                         "guard": {"minNarrowing": engine.BD_MIN_NARROWING,
                                   "maxRecords": engine.BD_MAX_RECORDS},
                         # ⛔ D-59's `categoryOptions` IS DELIBERATELY NOT HERE — it is served by
                         # `GET /automations/discover/categories`. It sat on this payload for one
                         # commit and that was a real defect: THIS ROUTE IS POLLED EVERY 2.5 s by
                         # the whole automation surface, and deriving the options reads two user
                         # tables plus the PLATFORM MASTER, which is a different HF repo — i.e. a
                         # network round-trip per poll, per open tab. Caught by the gate's own
                         # output, which started carrying `store:get:master_snapshots` errors from
                         # a suite whose stated contract is that it touches no network.
                         # C6/R5's vocabulary, so the seed picker cannot offer a source the
                         # validator refuses.
                         "seedSources": list(engine.SEED_SOURCES),
                         "seedMaxRows": engine.SEED_MAX_ROWS,
                         # W29-T01: `tables` is the snapshot read once at the top of this handler.
                         # ⚠ IT WAS TAKEN BEFORE THE RETIREMENT WROTE, and that is deliberate and
                         # harmless: retirement only removes `stage_auto_*` FIELDS and pops those
                         # same keys off rows, and the election reads the preset-profile vocabulary
                         # (`handle` + N preset keys) and whether a table has ANY rows. A stage key
                         # is in neither set, and no row is ever deleted — so the pre-write
                         # snapshot and the post-write bucket cannot elect different tables.
                         # 2026-08-10 — the OFFER must name the table the SAVE will actually use.
                         # This was the bare `DISCOVER_TABLE` constant while `create`/`patch` now
                         # resolve a targetless discovery flow to the profile database the tenant
                         # already has, so the picker would have shown `ut_ig_candidates` and the
                         # save would have written somewhere else — a default that disagrees with
                         # itself across two panels, which is the shape wave 25's R2 fixed for
                         # `targetTable` vs the action's `table`.
                         # ⭐ WAVE 30 · T12 — through the per-tenant memo. The ELECTION rule and
                         # everything the comment above says about it are unchanged; what changed
                         # is that a warm request no longer re-reads a 35.8 MB document to
                         # recompute a string that did not move.
                         "table": (_discover_default(session, tables)
                                   or engine.DISCOVER_TABLE)},
            "storeAvailable": bool(session.runtime.available()),
            # Wave 22 C3 — the trigger vocabulary, session-scoped because email readiness is a
            # per-USER fact (the poll runs through the creator's own Gmail connection).
            "triggers": _triggers_vocab(session),
            # ⭐ WAVE 23 C4 (R3) — the ACTION MENU, including what we have not built. Each row
            # carries `ready`, so B paints "Send email" and "Run script" faded with the server's
            # own reason instead of omitting them — the owner asked for Airtable's full menu, and
            # a shorter list would imply those actions do not exist. `clean_actions` REFUSES an
            # unready kind, so the faded state is a wall rather than a styling choice.
            "actionsCatalog": engine.action_catalog(),
            # The builder's own vocabulary: how deep a condition tree may nest, how deep groups
            # may nest, and the ceilings. B reads these instead of hard-coding the same numbers
            # into its "+ Add condition" affordance.
            "flow": {"condOps": list(engine.LANE_OPS),
                     "nullaryCondOps": list(engine.LANE_NULLARY_OPS),
                     "maxCondDepth": engine.MAX_COND_DEPTH,
                     "maxCondChildren": engine.MAX_COND_CHILDREN,
                      "maxGroupDepth": engine.MAX_GROUP_DEPTH,
                      "maxActions": engine.MAX_ACTIONS,
                      # ⭐ W29-T09 (owner item 11) — THE POST-GROUP VOCABULARY, on the wire for the
                      # same reason `cronPresets` is: `clean_post_groups` REFUSES a type it does
                      # not know, so a client that invented one would build a config the save door
                      # rejects with a sentence about a word the person never typed. Both halves
                      # ride — the stored key and the label a person reads — because a client-side
                      # translation of `video` into "Reels" is a second copy of this list.
                      "postTypes": [{"key": t, "label": engine.POST_TYPE_LABELS.get(t, t)}
                                    for t in engine.POST_TYPES],
                      # The ceiling a group's limit is judged against. `clean_post_groups` bounds a
                      # group by the action's OWN `maxPosts`, not by a constant, so the control can
                      # only warn honestly if it reads the same number the validator uses.
                      "maxPostsPerPull": engine.MAX_POSTS_PER_PULL},
            # Wave 21 C6-A1: {"enabled": bool, "source": "in-process"|"external"|""} — see
            # `_tick_state` for why one boolean off AIOS_AUTOMATIONS alone would lie.
            "tick": _tick_state()}


@router.post("/automations/discover/estimate")
def discover_estimate(body: dict = Body(default=None), session: Session = Depends(_GATE)):
    """What would this search cost? Shown BEFORE the run, never after.

    ⚠ THE ANSWER IS AN ESTIMATE AND SAYS SO IN ITS OWN PAYLOAD (`basis: "SPEC"`). Bright Data
    never returns a price before a run — the funds gate fires first and `price: 0` means "not
    priced", not "free" — and this account's token cannot read a balance (`/customer/balance`
    answers 403). A number presented as billed would be the invented measurement this whole
    module refuses to make.
    """
    return engine.discover_estimate((body or {}).get("recordsLimit"))


#: C8's migration ran, per tenant, this process. Once is the point: the sweep only writes when
#: an unbound bag exists, so after the first pass this is a read that finds nothing.
_C8_MIGRATED = set()


def _table_views(session, key):
    """⭐ WAVE 24 (item 8) — the saved views on ONE user table as `[{id, label}]`, or **None**
    when the workspace bucket could not be read at all.

    The client said "This server did not offer a view list" because the payload genuinely had
    no `views` key. C-TYPES: **absent stays ABSENT.** `[]` is a MEASUREMENT ("this database has
    no saved views") and `None` is a STATE ("nobody looked") — collapsing them is precisely how
    an empty picker comes to read as a claim about the database.

    ⛔ SOURCED FROM `core.table_store`, WHICH IS THE READER `view_filter` ALREADY USES — not a
    second one, the same one, listed instead of looked up. The brief said to source it the way
    the grid does; the grid's projection (`aios_grid.views_from_defs`) is the wrong list HERE and
    the difference matters: it INJECTS the system view ("All records") and PROJECTS cohorts as
    views, and neither of those lives in the stored bucket — so `view_filter` answers "view no
    longer exists" for every one of them. A picker built from that projection would offer options
    the `enters_view` trigger cannot resolve, which is a worse bug than the missing key it fixes.
    (An `enters_view` trigger on "All records" would also mean "fire on every record", so its
    absence is correct rather than a gap.)

    ⚠ `consume_corrections=False`: this is a READ for a picker, and the default MUTATES — it
    takes and clears the pending field-correction acknowledgement, so listing views would eat a
    protocol message meant for the grid.
    """
    try:
        import core.table_store as table_store
        # ⛔ THE REACHABILITY PROBE IS NOT REDUNDANT, and leaving it out was a real defect this
        # gate's own NC caught: EVERY `TableStore` accessor wraps its store read in `try/except`
        # and returns `{}` on failure. So a bucket that could not be read is indistinguishable
        # from one with no views — which collapses exactly the ABSENT/EMPTY distinction this
        # function exists to preserve, and the caller would ship `views: []` as a measurement
        # nobody took. The one read that is allowed to raise has to be ours.
        session.runtime.get(f"{key}_table_workspace")
        tops = table_store.make(f"{key}_table_workspace", st=session.runtime)
        own = (tops.workspace(session.uname, consume_corrections=False) or {}).get("views") or {}
        shared = tops.shared_views(session.uname, session.admin) or {}
    except Exception:                                             # noqa: BLE001
        return None
    merged = {**shared, **own}          # a view lives in ONE home; the merge is belt-and-braces
    return sorted(
        ({"id": str(vid), "label": str((v or {}).get("name") or vid)}
         for vid, v in merged.items() if isinstance(v, dict)),
        key=lambda r: r["label"].lower())


@router.get("/automations/tables")
def automation_tables(session: Session = Depends(_GATE)):
    """The blank databases an automation can target, with their fields.

    ⚠ A READ-ONLY MIRROR of C3-UT's `GET /api/v1/tables`, under this router's own prefix so the
    two can never collide. It exists because the automation editor needs the table+field list to
    build a config at all, and D must not block on A's route landing. When C3-UT is live this
    keeps working (same bucket) — it is a duplicate reader, never a second writer.

    ⛔ WAVE 20, item 3 — IT NO LONGER MIRRORS THE WALL, IT CALLS IT. This route re-implemented
    `may_open` and got it WIDER: it also admitted `createdBy in ('automation', 'scheduler')`, so
    a non-admin saw automation-created databases here and was refused the moment they opened,
    edited or deleted one. A duplicate READER is fine; a duplicate WALL is not, because the two
    only disagree in front of a user. One resolver now, and the engine stamps a human owner so
    the merge takes nothing legitimate away (`ut_ensure`, `MACHINE_OWNERS`).
    """
    import core.user_tables as user_tables

    # C8 (wave 22): bind pre-law automation bags to the definitions that write them — once
    # per tenant per process, and a no-op read after the first real pass. A write-on-read,
    # stated out loud: this is the surface whose stale bags mislead (the column picker), so
    # it is where the truth gets repaired.
    if session.tenant not in _C8_MIGRATED and session.runtime.available():
        _C8_MIGRATED.add(session.tenant)
        try:
            engine.bind_unbound_fields(session.runtime)
        except Exception:                                         # noqa: BLE001
            pass

    out = []
    # ⭐⭐ WAVE 30 · T13 — ONE read of the tenant document, LENT to the wall for every table.
    # It was `1 + N` full deep copies (this `ut_all`, then `may_open` → `get` → `all_tables` per
    # table), each of a document with a 35.8 MB / ~370 ms ceiling, all of them queued behind
    # `Store._lock`. `may_open` is unchanged and still asked about every table — see `_LentTables`
    # on why re-implementing the wall here is the one fix that is NOT available.
    _tables_doc = engine.ut_all(session.runtime)
    _lent = _LentTables(session.runtime, _tables_doc)
    for key, t in sorted(_tables_doc.items(),
                         key=lambda kv: (kv[1].get("label") or "").lower()):
        if not user_tables.may_open(key, session.uname, session.admin, st=_lent):
            continue
        row = {"key": key, "label": t.get("label") or key,
               "source": t.get("source") or "Blank",
               "rowCount": len(t.get("rows") or {}),
               "fields": [{"key": f.get("key"), "label": f.get("label"),
                           "type": f.get("type") or "text",
                           "automation": f.get("automation") or None}
                          for f in (t.get("fields") or [])]}
        views = _table_views(session, key)
        if views is not None:
            row["views"] = views          # absent when the bucket did not answer — never []
        out.append(row)
    return {"tables": out}


@router.get("/automations/presets")
def automation_presets(table: str = "", session: Session = Depends(_GATE)):
    """⭐ WAVE 25 · C1 / R2b — the Instagram preset columns, diffed against ONE database.

    `{fields: [{key, label, type, present}], willUse: [...], willCreate: [...]}` — the
    "already in this database" / "will be created" split the owner asked for, so the Create record
    action's configuration can SHOW what pointing it here does before anything is spent.

    ⛔ DECLARED ABOVE `/automations/{auto_id}`, AND THAT IS LOAD-BEARING RATHER THAN TIDY.
    FastAPI matches routes in DECLARATION order, so a literal path registered after a sibling
    path-parameter route is never reached — this handler would simply never run and
    `get_automation` would answer "no automation with that id" for the id `presets`. A 404 with a
    plausible sentence is the worst possible failure here, because it reads as "the endpoint is
    fine, the data is missing". `/automations/tables` sits above the same route for the same
    reason; this follows it rather than inventing a second convention.

    ⚠ WALLED BY `may_open`, LIKE EVERY OTHER TABLE READER. Without it, asking about a table you
    cannot open would answer which of its columns exist — a small disclosure, and exactly the
    duplicate-wall mistake `automation_tables` records at wave 20.
    """
    import core.user_tables as user_tables

    key = str(table or "").strip()
    if key and not user_tables.may_open(key, session.uname, session.admin,
                                        st=session.runtime):
        raise err(404, "unknown_table", "no such database")
    return engine.preset_plan(session.runtime, key)


@router.get("/automations/discover/categories")
def automation_categories(session: Session = Depends(_GATE)):
    """⭐ DEBT D-59 — the Category combobox's options: the values this deployment has ACTUALLY
    SEEN, each with its observed count. `{options: [{value, count}]}`.

    ⛔ ITS OWN ROUTE, ON PURPOSE. This belongs to the Find surface and is asked for when that
    panel opens — it must never ride `GET /automations`, which the whole automation surface polls
    every 2.5 s: deriving these reads two user tables AND the platform master (a different HF
    repo), so on the polled payload it is a network round-trip per tab per poll.

    ⚠ THE CONTROL STAYS A COMBOBOX. These are the values we have seen, not the values that exist.
    D-59 is explicit that transcribing Instagram's published taxonomy would be 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", which misleads precisely when it looks authoritative.
    """
    return {"options": engine.observed_categories(session.runtime)}


@router.post("/automations/seed/derive")
def automation_seed_derive(body: dict = Body(default=None), session: Session = Depends(_GATE)):
    """⭐ WAVE 25 · C6 / R5 — "find me more accounts like the ones in this view".

    Body `{source: "view"|"cohort", table: "<ut_key>", id: "<view id>"}` →
    `{derived: [<predicate>...], basis: {rows, fields:[{name,label,value,coverage}], related, note}}`

    ⛔ IT DERIVES AND RETURNS; IT SAVES NOTHING. R5: the derived conditions are "visible and
    editable, never hidden" — so the client drops them into the ordinary condition rows, where the
    user edits them like anything else, and the ordinary PATCH stores them. A door that both
    derived and saved would make the suggestion feel like a decision.

    ⚠ `basis` IS NOT DECORATION AND MUST BE RENDERED. It carries how many rows were read and the
    MEASURED coverage of each characteristic, which is the difference between "12 of 12 of these
    bios say florist" and "7 of 12 do" — presented identically, the weak one reads as authority.

    ⚠ Declared ABOVE `/automations/{auto_id}` for the reason `/automations/presets` is (FastAPI
    matches in declaration order); that ordering is gated.
    """
    import core.user_tables as user_tables

    body = body or {}
    table = str(body.get("table") or "").strip()
    if not table:
        raise err(400, "no_table", "name the database to read the seed records from")
    if not user_tables.may_open(table, session.uname, session.admin, st=session.runtime):
        raise err(404, "unknown_table", "no such database")
    source = str(body.get("source") or "view").strip().lower()
    if source not in engine.SEED_SOURCES:
        raise err(400, "bad_seed_source",
                  f"a seed comes from one of: {', '.join(engine.SEED_SOURCES)}")
    rows, problem = engine.seed_rows(session.runtime, table, str(body.get("id") or ""))
    if problem:
        raise err(400, "bad_seed", problem)
    derived, basis = engine.seed_predicates(rows)
    return {"source": source, "table": table, "id": str(body.get("id") or ""),
            "derived": derived, "basis": basis}


@router.get("/automations/{auto_id}")
def get_automation(auto_id: str, session: Session = Depends(_GATE)):
    defn = engine.all_definitions(session.runtime).get(str(auto_id))
    if defn is None:
        raise err(404, "unknown_automation", "no automation with that id")
    return {"automation": _wire(defn, session.tenant)}


@router.post("/automations")
def create_automation(body: dict = Body(default=None), session: Session = Depends(_GATE)):
    """Create an automation. ⭐ WAVE 23 (C2): the body may carry
    `target: {mode: "existing"|"new"|"automated", table?, label?}` — the wizard's FIRST question,
    answered before the kind. `new` mints the blank database in the same call, so the automation
    is never saved pointing at a table that does not exist yet."""
    if not session.runtime.available():
        raise err(503, "store_unavailable", "the tenant store is unavailable — nothing was saved")
    defn, error = engine.create(session.runtime, body or {}, username=session.uname)
    if error:
        raise err(400, "invalid_automation", error)
    return {"automation": _wire(defn, session.tenant)}


@router.patch("/automations/{auto_id}")
def patch_automation(auto_id: str, body: dict = Body(default=None),
                     session: Session = Depends(_GATE)):
    if not session.runtime.available():
        raise err(503, "store_unavailable", "the tenant store is unavailable — nothing was saved")
    defn, error = engine.patch(session.runtime, auto_id, body or {}, username=session.uname)
    if error:
        raise err(400 if error != "no such automation" else 404,
                  "invalid_automation" if error != "no such automation" else "unknown_automation",
                  error)
    return {"automation": _wire(defn, session.tenant)}


@router.delete("/automations/{auto_id}")
def delete_automation(auto_id: str, session: Session = Depends(_GATE)):
    """Delete the DEFINITION. ⚠ The database it filled is NOT touched — an automation is the
    thing that writes rows, not the thing that owns them, and deleting a job must never be a way
    to lose data (the same rule the orphan count encodes)."""
    if engine.running(session.tenant, auto_id):
        raise err(409, "automation_running", "it is running — wait for it to finish")
    engine.remove(session.runtime, auto_id)
    return {"deleted": str(auto_id)}


@router.post("/automations/preview")
def preview_source(body: dict = Body(default=None), session: Session = Depends(_GATE)):
    """What does this URL actually offer? The field-map step — never writes anything."""
    body = body or {}
    url = str(body.get("url") or "").strip()
    if not url:
        raise err(400, "no_url", "give a page URL to read")
    try:
        return engine.preview(url, str(body.get("extract") or "table"),
                              int(body.get("tableIndex") or 0))
    except engine.Refused as e:
        raise err(400, "refused_url", str(e))
    except Exception as e:                                        # noqa: BLE001
        raise err(502, "fetch_failed", f"could not read that page — {type(e).__name__}: "
                                       f"{str(e)[:160]}")


@router.post("/automations/{auto_id}/run")
def run_automation(auto_id: str, session: Session = Depends(_GATE)):
    """Start a run on a background thread. 409 when one is already in flight."""
    defn = engine.all_definitions(session.runtime).get(str(auto_id))
    if defn is None:
        raise err(404, "unknown_automation", "no automation with that id")
    # ⛔ WAVE 32 · T45 (owner item 10) — REFUSED, NAMING THE ACTION. `run_now` refuses too, for the
    # tick and the webhook; this one exists so the person who pressed the button reads the reason
    # instead of watching a run start and end with nothing done. The two ask the SAME function, so
    # they cannot come to disagree about what "configured" means.
    refusal = engine.run_refusal(defn)
    if refusal:
        raise err(400, "action_unconfigured", refusal)
    if not engine.run_async(session.runtime, session.tenant, auto_id, username=session.uname):
        raise err(409, "automation_running", "that automation is already running")
    return {"started": str(auto_id), "startedAt": time.strftime("%Y-%m-%dT%H:%M:%S")}


@router.post("/automations/{auto_id}/nodes/{node_id}/toggle")
def toggle_automation_node(auto_id: str, node_id: str, session: Session = Depends(_GATE)):
    """Flip one step on the canvas. The SERVER decides what a node's switch means (see
    `engine.NODE_TOGGLES`) — the client only reports which node was clicked.

    A node with no switch answers 400 with the sentence saying why, rather than silently doing
    nothing: a control that appears to work and does not is worse than one that refuses.
    """
    if not session.runtime.available():
        raise err(503, "store_unavailable", "the tenant store is unavailable — nothing was saved")
    if engine.running(session.tenant, auto_id):
        raise err(409, "automation_running", "it is running — wait for it to finish")
    defn, error = engine.toggle_node(session.runtime, auto_id, node_id)
    if error:
        raise err(404 if error == "no such automation" else 400,
                  "unknown_automation" if error == "no such automation" else "node_not_toggleable",
                  error)
    return {"automation": _wire(defn, session.tenant)}


@router.post("/automations/{auto_id}/hook/{token}")
async def automation_hook(auto_id: str, token: str, request: Request):
    """C3's webhook trigger — the tick-endpoint pattern one level down: unauthenticated BY
    DESIGN (the external caller has no cookie), gated on a per-automation token minted when the
    trigger was configured, constant-time compared. The tenant is FOUND by the (id, token)
    pair — a wrong token answers 403 for every tenant, so the route confirms nothing about
    which slugs exist.

    ⭐ WAVE 24 (D-41) — THE BODY IS READ DEFENSIVELY AND IS NEVER A REASON TO REFUSE.
    ⛔ It is deliberately NOT declared as `body: dict = Body(...)`, which is the obvious way to
    write this and would be a live regression: FastAPI would then VALIDATE the payload, so an
    existing caller posting text, form-encoding, an empty body or slightly malformed JSON would
    start getting a 422 from a door that has accepted anything since wave 22. A webhook sender is
    somebody else's system; we do not get to change what it must send in order to fire a flow.
    An unreadable body simply maps nothing — the flow still fires, exactly as it did before.

    `run_in_threadpool` keeps the store I/O off the event loop: `hook_fire` walks every tenant
    and may commit a row, and this handler had to become `async` only to read the request body.
    """
    from starlette.concurrency import run_in_threadpool
    from harness import runtime as _rt

    try:
        payload_in = await request.json()
    except Exception:                                             # noqa: BLE001
        payload_in = None

    def _fire():
        last = (404, {"error": "unknown_automation",
                      "message": "no automation with that id and token"})
        for slug in _rt.known_tenants():
            try:
                rt = _rt.get_runtime(slug)
            except Exception:                                     # noqa: BLE001
                continue
            status, payload = engine.hook_fire(rt, slug, auto_id, token, body=payload_in)
            if status == 200:
                return 200, payload
            if status != 404:
                last = (status, payload)
        return last

    status, payload = await run_in_threadpool(_fire)
    if status == 200:
        return payload
    raise err(status, str(payload.get("error") or "refused"),
              str(payload.get("message") or "refused"))


@router.post("/automations/tick")
def tick(request: Request, x_aios_tick_token: str = Header(default="")):
    """Fire every due schedule, for every tenant. The durable-cron entry point (R5).

    Unauthenticated BY DESIGN and gated on a shared secret instead — an EventBridge rule has no
    cookie. Refuses when the secret is not configured (see the module header)."""
    want = os.environ.get("AIOS_AUTOMATION_TICK_TOKEN") or ""
    if not want:
        raise err(403, "tick_disabled",
                  "AIOS_AUTOMATION_TICK_TOKEN is not configured — the tick endpoint is closed")
    got = x_aios_tick_token or request.headers.get("X-AIOS-TICK-TOKEN") or ""
    if got != want:
        raise err(403, "bad_tick_token", "that token is not valid for this deployment")
    started = engine.tick_all()
    return {"started": started, "at": time.strftime("%Y-%m-%dT%H:%M:%S")}


@router.post("/automations/ig/purge")
def purge_ig_subject(body: dict = Body(default=None), session: Session = Depends(_GATE)):
    """D-24 (wave 22): the right-to-erasure door for ONE Instagram subject — walks the
    tenant's four `ut_ig_*` tables AND the platform master (R2 pooled a copy there, so a purge
    that skipped it would not be erasure). Admin-only: erasure is a compliance act, not a
    grid gesture. Answers the per-table removal counts — the one aggregate whose drill is the
    rows' ABSENCE."""
    if not session.admin:
        raise err(403, "admin_only", "erasing a subject is an admin action")
    if not session.runtime.available():
        raise err(503, "store_unavailable", "the tenant store is unavailable — nothing was "
                                            "purged")
    handle = str((body or {}).get("handle") or "").strip()
    if not handle:
        raise err(400, "no_handle", "name the Instagram handle to erase")
    counts = engine.purge_subject(session.runtime, handle)
    return {"handle": handle.lstrip("@").lower(), "removed": counts,
            "total": sum(counts.values())}


@router.get("/automations/metrics/{table_key}/{field_key}/{row_id}/rows")
def metric_drill(table_key: str, field_key: str, row_id: str,
                 session: Session = Depends(_GATE)):
    """C7's drill: the EXACT master snapshot rows behind one metric cell
    ([[no-unverifiable-aggregates]]) — recomputed on ask with the same function that filled
    the cell, so the drill can never disagree with the number by construction."""
    import core.user_tables as user_tables

    if not user_tables.may_open(table_key, session.uname, session.admin, st=session.runtime):
        raise err(404, "unknown_table", "no such database")
    t = engine.ut_get(session.runtime, table_key) or {}
    fdef = next((f for f in (t.get("fields") or []) if f.get("key") == field_key), None)
    if not fdef or not isinstance(fdef.get("metric"), dict):
        raise err(404, "not_a_metric", "that column is not a metric field")
    row = (t.get("rows") or {}).get(str(row_id))
    if row is None:
        raise err(404, "unknown_row", "that record is not in the database")
    import ig_master
    url_field = next((f.get("key") for f in (t.get("fields") or [])
                      if f.get("type") == "url"), "")
    handle = engine._table_handle(row, url_field)
    bag = fdef["metric"]
    series = ig_master.series_for({handle}) if handle else {}
    value, rows = engine.metric_value(series.get(handle), bag.get("measure"),
                                      bag.get("window"), bag.get("agg") or "")
    return {"table": table_key, "field": field_key, "rowId": str(row_id), "handle": handle,
            "measure": bag.get("measure"), "window": bag.get("window"),
            "value": value, "rows": rows[:200],
            "note": "" if value is not None else
            "no master data answers this window — the cell is honestly blank"}


@router.get("/automations/{auto_id}/rows")
def run_rows(auto_id: str, session: Session = Depends(_GATE)):
    """The rows the LAST run touched — the drill-down behind a run's counts.

    Every count in this product drills to the exact rows behind it ([[no-unverifiable-aggregates]]);
    a run history that said "412 updated" and could not show which would be the thing that rule
    exists to forbid.
    """
    defn = engine.all_definitions(session.runtime).get(str(auto_id))
    if defn is None:
        raise err(404, "unknown_automation", "no automation with that id")
    last = (defn.get("runs") or [{}])[0]
    table_key = (defn.get("config") or {}).get("targetTable") or ""
    t = engine.ut_get(session.runtime, table_key) or {}
    rows = t.get("rows") or {}
    ids = [str(i) for i in (last.get("affected") or [])]
    return {"table": table_key, "label": t.get("label") or table_key,
            "fields": [{"key": f.get("key"), "label": f.get("label")}
                       for f in (t.get("fields") or [])],
            "rows": [{"id": i, **(rows.get(i) or {})} for i in ids if i in rows],
            "truncated": len(ids) >= 200}


# --- the in-process scheduler ------------------------------------------------------------
# ⚠ OPT-IN (`AIOS_AUTOMATIONS=1`), and that is an AMENDMENT to the wave brief's "default-on",
# made on a measurement: `verify_api.py` runs for 196 s, i.e. longer than three tick intervals,
# and `tick_all` reaches `runtime.get_runtime()` — which builds tenants and moves the LRU cache
# that verify_api's own isolation checks read. Default-on would put a background thread inside
# the subject of another session's gate. `AIOS_PREWARM=1` at `main.py:353` is the same decision
# for the same reason, so this follows it rather than inventing a second convention.
#
# The loop also SLEEPS FIRST (`engine.scheduler_loop`) — defence in depth, proven in
# verify_automation.py section T rather than assumed.
#
# ⛔ THE DEPLOY MUST SET `AIOS_AUTOMATIONS=1` (with `AIOS_AUTOMATION_TICK_TOKEN`) or schedules
# only ever fire from the external cron POSTing /automations/tick. Both paths work; neither is
# implicit. Booked in the session-D mailbox.
engine.start_scheduler()

# C3 (wave 22): register the trigger listener onto the platform's row-event seam. THIS module
# is where the registration belongs — it is the one place that already imports both sides, so
# neither the engine nor platform/core grows a dependency on the other. Idempotent: a reimport
# must not double-fire every trigger.
import core.user_tables as _ut_hooks  # noqa: E402

if engine.grid_hook not in _ut_hooks.ROW_HOOKS:
    _ut_hooks.ROW_HOOKS.append(engine.grid_hook)

# ⭐⭐ W31 QA — DECLARE THE MACHINE-OWNED CHILD DATABASES, for the same reason and in the same
# place as the hook above: this module already imports both sides, so neither the engine nor
# `platform/core` grows a dependency on the other. Idempotent by construction (a set).
#
# ⛔ THE OWNER FOUND WHAT THIS FIXES, IN PRODUCTION, AFTER THE TICKET READ GREEN. W31-T32 locked
# the TikTok children by stamping `recordMode` at the two SPAWN sites, and proved the stamp
# arrives "on the next write". That is true and it is not the `done-when`: every TikTok database
# already sitting in a tenant kept offering "+ New record" until somebody re-ran a TikTok
# automation, and nobody had. Owner, verbatim (2026-08-13): *"the databases for Tiktok do not have
# the small lock icon as I asked"* — and the rule, restated: *"just like Instagram Post database
# (which is locked), only the IG Profile and TT Profile should be editable."*
# ⚠ A DECLARATION NEEDS NO WRITE, so it is true for EVERY tenant the moment the API boots — no
# migration, no boot ordering, no per-tenant walk, and nothing that a stale store can undo. The
# stored flag still locks a table nobody declares; the two are OR'd.
_ut_hooks.register_locked_records(engine.LOCKED_CHILD_TABLES)