| """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") |
|
|
| |
| |
| router.include_router(routes_oauth.router) |
| |
| |
| |
| import routes_connectors |
| router.include_router(routes_connectors.router) |
|
|
| |
| |
| 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), |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| "awaitingResults": bool(str((defn.get("state") or {}).get("pendingSnapshot") |
| or "").strip()), |
| |
| |
| |
| "trigger": defn.get("trigger") or None, |
| "statusNote": defn.get("statusNote") or "", |
| |
| |
| "sentence": engine.compose_sentence(defn), |
| |
| |
| |
| "graph": engine.graph(defn), |
| |
| |
| |
| |
| "flow": defn.get("flow") or {"actions": []}, |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| "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, ""), |
| |
| |
| |
| |
| |
| "ig_profile_match": (engine.bd_ready(), "" if engine.bd_ready() else "configure_brightdata"), |
| |
| |
| |
| |
| "tiktok_profile_match": (engine.bd_ready(), |
| "" if engine.bd_ready() else "configure_brightdata"), |
| } |
| |
| |
| |
| 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"], |
| |
| |
| "connector": engine.TRIGGER_CONNECTOR.get(k), |
| |
| |
| "schedules": k in engine.TRIGGER_CRON_KEYS} |
|
|
| out = [] |
| for k in engine.TRIGGER_KEYS: |
| |
| |
| |
| |
| |
| |
| ready, needs = per.get(k, (True, "")) |
| row = {"key": k, "label": engine.TRIGGER_LABELS[k], "ready": ready, "needs": needs, |
| "planned": False, "detail": detail.get(k, ""), |
| |
| |
| "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 "")} |
|
|
|
|
| |
| |
| |
| |
| |
| _BOARD_RETIRED = set() |
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| _DISCOVER_DEFAULT = {} |
| |
| |
| _DISCOVER_DEFAULT_TTL = float(os.environ.get("AIOS_AUTOMATION_DEFAULT_TTL") or 300) |
|
|
|
|
| |
| |
| |
| _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. |
| """ |
| |
| |
| 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: |
| |
| 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).""" |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| tables = None |
| if session.runtime.available() and session.tenant not in _BOARD_RETIRED: |
| tables = engine.ut_all(session.runtime) |
| _BOARD_RETIRED.add(session.tenant) |
| |
| |
| 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, |
| |
| |
| |
| |
| |
| "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, |
| |
| |
| |
| |
| "paidReady": engine.bd_ready(), |
| |
| |
| |
| "sources": engine.source_status(), |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| "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, |
| |
| |
| |
| |
| "filterMeta": engine.filter_meta(), |
| "guard": {"minNarrowing": engine.BD_MIN_NARROWING, |
| "maxRecords": engine.BD_MAX_RECORDS}, |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| "seedSources": list(engine.SEED_SOURCES), |
| "seedMaxRows": engine.SEED_MAX_ROWS, |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| "table": (_discover_default(session, tables) |
| or engine.DISCOVER_TABLE)}, |
| "storeAvailable": bool(session.runtime.available()), |
| |
| |
| "triggers": _triggers_vocab(session), |
| |
| |
| |
| |
| |
| "actionsCatalog": engine.action_catalog(), |
| |
| |
| |
| "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, |
| |
| |
| |
| |
| |
| |
| "postTypes": [{"key": t, "label": engine.POST_TYPE_LABELS.get(t, t)} |
| for t in engine.POST_TYPES], |
| |
| |
| |
| "maxPostsPerPull": engine.MAX_POSTS_PER_PULL}, |
| |
| |
| "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_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 |
| |
| |
| |
| |
| |
| |
| 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: |
| return None |
| merged = {**shared, **own} |
| 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 |
|
|
| |
| |
| |
| |
| 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: |
| pass |
|
|
| out = [] |
| |
| |
| |
| |
| |
| _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 |
| 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: |
| 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") |
| |
| |
| |
| |
| 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: |
| 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: |
| 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} |
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| engine.start_scheduler() |
|
|
| |
| |
| |
| |
| import core.user_tables as _ut_hooks |
|
|
| if engine.grid_hook not in _ut_hooks.ROW_HOOKS: |
| _ut_hooks.ROW_HOOKS.append(engine.grid_hook) |
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| _ut_hooks.register_locked_records(engine.LOCKED_CHILD_TABLES) |
|
|