Deploy AIOS web (React glide grid + FastAPI slice)

api/automation_engine.py

@@ -2469,11 +2469,19 @@ def retire_automation_board_state(rt, tables=None):
 #: ⭐⭐ WAVE 31 · T35 (D-134) — the definitions memo, and the two things that make it safe.
 #: `tenant -> (monotonic_at, defs)`. Read ONLY through `all_definitions(..., cached=True)`, which
 #: exactly one caller uses (`grid_hook`).
-_DEFS_MEMO = {}
+_DEFS_MEMO = {}
 #: Deliberately SHORT. This exists to collapse one burst of row events into one read, not to be a
 #: cache — an import of 20,000 rows arrives in far less than this, and a stale trigger for two
 #: seconds is bounded and recoverable where a stale one for a minute is a mystery.
-_DEFS_TTL = 2.0
+_DEFS_TTL = 2.0
+
+# The durable external scheduler has a different read pattern from a row-event burst. A live
+# Space is one process / one replica today, and every automation-definition write in this module
+# passes through `_store_update`, so the scheduler can retain definitions until a REAL write
+# invalidates them. This is the cost boundary for Neon: an idle 15-minute wake-up must not turn
+# into 96 reads/day of the same tenant JSON value merely because a timer fired.
+_TICK_DEFS_MEMO = {}
+_TICK_CACHE_LOCK = threading.RLock()
 
 
 def _store_update(rt, fn, flush="sync", keeps_defs=False):
@@ -2501,12 +2509,14 @@ def _store_update(rt, fn, flush="sync", keeps_defs=False):
     trigger set for another tenant. The memo holds at most a handful of entries and the correct,
     boring thing costs nothing.
     """
-    if not keeps_defs:
-        _DEFS_MEMO.clear()
-    return rt.update(STORE_KEY, fn, flush=flush)
+    if not keeps_defs:
+        _DEFS_MEMO.clear()
+        with _TICK_CACHE_LOCK:
+            _TICK_DEFS_MEMO.clear()
+    return rt.update(STORE_KEY, fn, flush=flush)
 
 
-def all_definitions(rt, cached=False):
+def all_definitions(rt, cached=False):
     """Every automation definition for this tenant.
 
     ⭐⭐ `cached=True` IS D-134's FIX, AND IT IS OPT-IN FOR A REASON. `grid_hook` runs ONCE PER ROW
@@ -2536,7 +2546,33 @@ def all_definitions(rt, cached=False):
     out = {str(aid): _without_retired_board(defn)[0] for aid, defn in raw.items()}
     if cached:
         _DEFS_MEMO[str(getattr(rt, "key", "") or "")] = (time.monotonic(), out)
-    return out
+    return out
+
+
+def tick_definitions(rt):
+    """Definitions for the external scheduler, cached until `_store_update` changes them.
+
+    Live has one replica, all product writes use `_store_update`, and a restart naturally drops
+    this process cache. The existing two-second memo remains dedicated to row-event bursts.
+    """
+    key = str(getattr(rt, "key", "") or "")
+    with _TICK_CACHE_LOCK:
+        hit = _TICK_DEFS_MEMO.get(key)
+        if hit is not None:
+            return hit
+    fresh = all_definitions(rt)
+    with _TICK_CACHE_LOCK:
+        return _TICK_DEFS_MEMO.setdefault(key, fresh)
+
+
+def invalidate_tick_cache(tenant=None):
+    """Drop scheduler control-plane state after an out-of-process tenant change."""
+    with _TICK_CACHE_LOCK:
+        if tenant is None:
+            _TICK_DEFS_MEMO.clear()
+            _TICK_TENANT_CACHE.clear()
+        else:
+            _TICK_DEFS_MEMO.pop(str(tenant), None)
 
 
 def _new_id(existing):
@@ -3086,9 +3122,13 @@ def _no_step(_text):
     is the part a test is allowed to call on its own."""
 
 
-def _release(tenant, auto_id):
-    with _RUN_LOCK:
-        _RUNNING.pop((tenant, str(auto_id)), None)
+def _release(tenant, auto_id):
+    with _RUN_LOCK:
+        _RUNNING.pop((tenant, str(auto_id)), None)
+    # A completed/manual/event run may have changed rows that feed a metric, link or source
+    # rollup. Mark the derived lane dirty; the next durable tick will refresh it once. Merely
+    # waking the scheduler does not mark anything dirty and therefore does not poll Neon.
+    _mark_derived_dirty(tenant)
 
 
 def _commit_run(rt, auto_id, state, summary, counts, ok, affected=None, steps=None, notes=None):
@@ -12626,11 +12666,13 @@ def _table_handle(row, url_field):
                or "").strip().lower()
 
 
-def compute_metric_cells(rt, table_key, today=None):
+def compute_metric_cells(rt, table_key, today=None, tables=None, persist=False):
     """Recompute every metric cell on ONE table from the master series. One coalesced write,
     only when something actually changed (the flush-ceiling law); zero reads when the table
     has no metric fields or the master is off. Returns the number of rows touched."""
-    t = ut_get(rt, table_key)
+    owned = tables is not None
+    blob = tables if owned else None
+    t = ((blob or {}).get(str(table_key)) if owned else ut_get(rt, table_key))
     mfields = metric_fields_of_table(t)
     if not mfields:
         return 0
@@ -12664,8 +12706,13 @@ def compute_metric_cells(rt, table_key, today=None):
                 tt.setdefault("rows", {}).setdefault(rid, {}).update(vals)
         return cur
 
-    rt.update(UT_STORE_KEY, _up, flush="sync")
-    return len(changes)
+    if owned:
+        _up(blob)
+        if persist:
+            rt.update(UT_STORE_KEY, _up, flush="sync")
+        return len(changes)
+    rt.update(UT_STORE_KEY, _up, flush="sync")
+    return len(changes)
 
 
 # ── ⭐⭐ THE RELATIONAL PASS (2026-08-07) — derived LINK cells and ROLLUP cells ────────────────
@@ -13204,14 +13251,28 @@ def _refresh_relations_inplace(tables, log=print):
     return touched
 
 
-def refresh_relations(rt, log=print):
+def refresh_relations(rt, log=print, tables=None, persist=True):
     """The tick half of the relational pass — the twin of `refresh_metrics`.
 
     ⚠ Runs for EVERY table, because a link can point anywhere: a rollup on table A goes stale
     when table B gains a row, and A has no way to know that happened. Cheap by construction — a
     table declaring neither kind costs one dict scan.
     """
-    snapshot = {
+    if tables is not None:
+        local = _refresh_relations_inplace(tables, log=log)
+        if not local or not persist:
+            return local
+        actual = [0]
+
+        def _up(cur):
+            cur = cur if isinstance(cur, dict) else {}
+            actual[0] = _refresh_relations_inplace(cur, log=log)
+            return cur
+
+        rt.update(UT_STORE_KEY, _up, flush="async")
+        return actual[0]
+
+    snapshot = {
         str(key): {**(table or {}),
                    "rows": {str(rid): dict(row or {})
                             for rid, row in ((table or {}).get("rows") or {}).items()}}
@@ -13244,15 +13305,17 @@ def refresh_relations(rt, log=print):
     return actual[0]
 
 
-def refresh_metrics(rt, today=None, log=print):
+def refresh_metrics(rt, today=None, log=print, tables=None, persist=True):
     """The tick half of the C7 amendment: `today` advances at tick cadence, so a date-window
     metric can never go staler than one tick while a scheduler exists. Cheap by construction —
     a table without metric fields costs a dict scan and nothing else."""
-    touched = 0
-    for tk, t in ut_all(rt).items():
-        if metric_fields_of_table(t):
-            try:
-                touched += compute_metric_cells(rt, tk, today=today)
+    snapshot = tables if tables is not None else ut_all(rt)
+    touched = 0
+    for tk, t in snapshot.items():
+        if metric_fields_of_table(t):
+            try:
+                touched += compute_metric_cells(rt, tk, today=today, tables=snapshot,
+                                                persist=persist)
             except Exception as e:                                # noqa: BLE001
                 log(f"[aios-auto] metric refresh {tk} failed: {type(e).__name__}: {e}")
     return touched
@@ -14441,7 +14504,7 @@ def run_async(rt, tenant, auto_id, username="automation", log=print, rows=None):
 # THE TICK + the in-process scheduler
 # ---------------------------------------------------------------------------------------------
 
-def pending_collect_ids(rt):
+def pending_collect_ids(rt, definitions=None):
     """⭐ 2026-08-06 — automations holding a snapshot the vendor is still building.
 
     ⛔ THE DEFECT THIS CLOSES, reported live: *"why is the SMALL 10 records test search taking
@@ -14458,7 +14521,8 @@ def pending_collect_ids(rt):
     make an unscheduled automation's paid result depend on someone remembering to press a button.
     """
     out = []
-    for aid, d in all_definitions(rt).items():
+    definitions = all_definitions(rt) if definitions is None else definitions
+    for aid, d in definitions.items():
         if not isinstance(d, dict):
             continue
         if (d.get("trigger") or {}).get("paused"):
@@ -14487,81 +14551,240 @@ def pending_collect_ids(rt):
     return sorted(out)
 
 
-def due_ids(rt, now=None):
-    """Which of this tenant's automations a tick at `now` should start. Pure over the store."""
-    return sorted(aid for aid, d in all_definitions(rt).items() if is_due(d, now))
+def due_ids(rt, now=None, definitions=None):
+    """Which of this tenant's automations a tick at `now` should start. Pure over the store."""
+    definitions = all_definitions(rt) if definitions is None else definitions
+    return sorted(aid for aid, d in definitions.items() if is_due(d, now))
 
 
-def tick(rt, tenant, now=None, log=print):
+def tick(rt, tenant, now=None, log=print):
     """Fire every due automation for ONE tenant + poll every email trigger (C3). Returns the
     ids started. The email polls are bounded and fail-quiet per automation — one broken
     mailbox connection must not stop the tenant's schedules."""
-    started = []
+    started = []
+    definitions = tick_definitions(rt)
     # ⭐ COLLECT-FIRST (2026-08-06). A snapshot the vendor has finished building is a result the
     # tenant has already been charged for; it is collected whether or not this automation is on a
     # schedule. `_claim` makes the union safe — an id in both lists starts once.
-    for aid in dict.fromkeys(list(pending_collect_ids(rt)) + list(due_ids(rt, now))):
+    for aid in dict.fromkeys(list(pending_collect_ids(rt, definitions=definitions))
+                             + list(due_ids(rt, now, definitions=definitions))):
         if run_async(rt, tenant, aid, username="scheduler", log=log):
             started.append(aid)
-    for aid, d in all_definitions(rt).items():
+    for aid, d in definitions.items():
         if (d.get("trigger") or {}).get("key") == "email":
             try:
                 if email_poll(rt, tenant, aid, d, log=log) is not None:
                     started.append(aid)
             except Exception as e:                                # noqa: BLE001
                 log(f"[aios-auto] email poll {tenant}/{aid} failed: {type(e).__name__}: {e}")
-    # C7's amendment: date-window metrics advance with the tick, so `today` is never staler
-    # than one tick while a scheduler exists. A tenant without metric fields pays a dict scan.
-    try:
-        refresh_metrics(rt, log=log)
-    except Exception as e:                                        # noqa: BLE001
-        log(f"[aios-auto] metric refresh {tenant} failed: {type(e).__name__}: {e}")
-    # ⭐ 2026-08-07 — the relational pass, on the same tick and for the same reason. A rollup on
-    # table A goes stale when table B gains a row, and A cannot know that happened; the tick is
-    # the only place that sees both. ⚠ SEPARATE try/except from the metrics above deliberately —
-    # one pass failing must not silently cancel the other, which a shared block would do.
-    try:
-        refresh_relations(rt, log=log)
-    except Exception as e:                                        # noqa: BLE001
-        log(f"[aios-auto] relation refresh {tenant} failed: {type(e).__name__}: {e}")
-    # ⭐ THE READ-THROUGH ROLLUPS, on the same tick and for a stronger version of the same reason.
-    # A linked rollup goes stale when the LINKED TABLE gains a row; a source-backed one goes stale
-    # when ODOO does — and its window moves on its own besides (a `ytd` column is wrong on 1
-    # January without anybody writing anything). The tick is the only place that sees either.
-    #
-    # ⛔ ITS OWN try/except, like the two passes above, and the reason is the same one stated
-    # there: this pass REFUSES loudly on a truncated group set, and a shared block would let that
-    # honest refusal silently cancel the relational pass that had already succeeded.
-    #
-    # ⚠ IT WRITES NOTHING WHEN IT REFUSES — see `rollup_sql`'s header. A half-applied rollup mixes
-    # two vintages of one column and looks completely normal, which is why the refusal is total.
-    try:
-        import rollup_sql
-        for _tk in list((rt.get(UT_STORE_KEY) or {})):
-            if rollup_sql.source_fields((rt.get(UT_STORE_KEY) or {}).get(_tk) or {}):
-                # ⚠ `today` is DEFAULTED INSIDE `compute`, not passed: this scope has no
-                # such local, and `refresh_metrics` above resolves it the same way.
-                n = rollup_sql.compute(rt, _tk)
-                if n:
-                    log(f"[aios-auto] source rollups {tenant}/{_tk}: {n}")
-    except Exception as e:                                        # noqa: BLE001
-        log(f"[aios-auto] source rollup {tenant} failed: {type(e).__name__}: {e}")
+    # Derived cells are maintenance, not part of the scheduler's HTTP acknowledgement. The
+    # background lane reads the large workspace at most once per UTC day, plus once after a real
+    # automation run marks it dirty. An idle wake-up therefore stays rows-free, and Lambda never
+    # times out waiting for a 40 MB tenant document to cross the public network.
+    _start_derived_refresh(rt, tenant, now=now, log=log)
     if started:
         log(f"[aios-auto] tick {tenant}: started {', '.join(started)}")
-    return started
-
-
-def tick_all(now=None, log=print):
+    return started
+
+
+# Postgres-only memo for the expensive derived half of the tick. The marker combines the
+# durable workspace revision with the UTC calendar day: unchanged minutes do no full-document
+# read, while midnight and a real workspace write each trigger one correct refresh.
+_DERIVED_TICK_STATE = {}
+_DERIVED_TICK_LOCKS = {}
+_DERIVED_TICK_LOCKS_GUARD = threading.Lock()
+_DERIVED_DIRTY_GENERATION = {}
+_DERIVED_RUNNING = {}
+_DERIVED_WORK_SEMAPHORE = threading.Semaphore(1)
+_TICK_TENANT_CACHE = {}
+_SCHEDULER_METRICS = {
+    "ticks": 0,
+    "lastTickAt": "",
+    "lastTenantCount": 0,
+    "workspaceRefreshes": 0,
+    "lastWorkspaceRefreshAt": "",
+}
+
+
+def _derived_tick_lock(tenant):
+    key = str(tenant)
+    with _DERIVED_TICK_LOCKS_GUARD:
+        return _DERIVED_TICK_LOCKS.setdefault(key, threading.Lock())
+
+
+def _mark_derived_dirty(tenant):
+    """Record a real automation run without reading the workspace it may have changed."""
+    key = str(tenant)
+    with _DERIVED_TICK_LOCKS_GUARD:
+        _DERIVED_DIRTY_GENERATION[key] = _DERIVED_DIRTY_GENERATION.get(key, 0) + 1
+
+
+def _derived_refresh_due(rt, tenant, now=None, state=None, backend=None):
+    actual_backend = backend if backend is not None else os.environ.get("STORE_BACKEND", "hf")
+    if str(actual_backend).strip().lower() != "pg":
+        return True, None
+    stamp = now or _dt.datetime.now(_dt.timezone.utc)
+    if isinstance(stamp, _dt.datetime):
+        stamp = (stamp.replace(tzinfo=_dt.timezone.utc) if stamp.tzinfo is None
+                 else stamp.astimezone(_dt.timezone.utc))
+    day = stamp.date().isoformat() if hasattr(stamp, "date") else str(stamp)[:10]
+    memo = _DERIVED_TICK_STATE if state is None else state
+    previous = memo.get(str(tenant))
+    with _DERIVED_TICK_LOCKS_GUARD:
+        dirty = bool(_DERIVED_DIRTY_GENERATION.get(str(tenant), 0)) if state is None else False
+    # The old order queried `store_kv.rev` every 15 minutes and therefore woke a suspended Neon
+    # compute even when nothing had changed. Human row routes already coalesce relation refreshes,
+    # Odoo rebuilds refresh their own cells, and automation completion marks this lane dirty. So a
+    # same-day clean marker is sufficient evidence to do literally no database work on an idle tick.
+    if (not dirty and isinstance(previous, (tuple, list)) and len(previous) > 1
+            and str(previous[1]) == day):
+        return False, tuple(previous)
+    try:
+        rev = (rt.revision(UT_STORE_KEY) or {}).get("token")
+        if not rev:
+            return True, None
+    except Exception:
+        return True, None
+    marker = (str(rev), day)
+    return memo.get(str(tenant)) != marker, marker
+
+
+def _refresh_source_rollups(rt, tenant, log=print, tables=None):
+    import rollup_sql
+    snapshot = tables if tables is not None else (rt.get(UT_STORE_KEY) or {})
+    for table_key, table in list(snapshot.items()):
+        if rollup_sql.source_fields(table or {}):
+            n = rollup_sql.compute(rt, table_key, tables=snapshot, persist=True)
+            if n:
+                log(f"[aios-auto] source rollups {tenant}/{table_key}: {n}")
+
+
+def _derived_refresh_worker(rt, tenant, marker, generation, log=print):
+    """Run one coalesced derived pass away from the external tick's HTTP response."""
+    key = str(tenant)
+    ok = True
+    # One rows-bearing pass at a time across the whole process. Four tenants currently total
+    # about 67 MB of workspace JSON; serialising that maintenance avoids a restart turning into
+    # four simultaneous Neon result streams and a compute autoscale spike.
+    with _DERIVED_WORK_SEMAPHORE, _derived_tick_lock(key):
+        try:
+            tables = dict(rt.get(UT_STORE_KEY) or {})
+        except Exception as e:                                    # noqa: BLE001
+            tables = None
+            ok = False
+            log(f"[aios-auto] derived workspace snapshot {key} failed: "
+                f"{type(e).__name__}: {e}")
+        if tables is not None:
+            for label, fn in (
+                    ("metric", lambda: refresh_metrics(rt, log=log, tables=tables)),
+                    ("relation", lambda: refresh_relations(rt, log=log, tables=tables)),
+                    ("source rollup", lambda: _refresh_source_rollups(
+                        rt, key, log=log, tables=tables))):
+                try:
+                    fn()
+                except Exception as e:                            # noqa: BLE001
+                    ok = False
+                    log(f"[aios-auto] {label} refresh {key} failed: "
+                        f"{type(e).__name__}: {e}")
+    with _DERIVED_TICK_LOCKS_GUARD:
+        _DERIVED_RUNNING.pop(key, None)
+        if ok and marker is not None:
+            _DERIVED_TICK_STATE[key] = marker
+            # Do not erase a write that landed after this worker took its snapshot. Its newer
+            # generation remains dirty and the next external tick repairs it.
+            if _DERIVED_DIRTY_GENERATION.get(key, 0) == generation:
+                _DERIVED_DIRTY_GENERATION.pop(key, None)
+            _SCHEDULER_METRICS["workspaceRefreshes"] += 1
+            _SCHEDULER_METRICS["lastWorkspaceRefreshAt"] = _iso()
+
+
+def _start_derived_refresh(rt, tenant, now=None, log=print):
+    """Start at most one rows-bearing maintenance worker for a tenant. Returns whether started."""
+    due, marker = _derived_refresh_due(rt, tenant, now=now)
+    if not due:
+        return False
+    key = str(tenant)
+    with _DERIVED_TICK_LOCKS_GUARD:
+        if key in _DERIVED_RUNNING:
+            return False
+        generation = _DERIVED_DIRTY_GENERATION.get(key, 0)
+        _DERIVED_RUNNING[key] = generation
+    thread = threading.Thread(
+        target=_derived_refresh_worker,
+        args=(rt, key, marker, generation, log),
+        daemon=True,
+        name=f"automation-derived-{key}")
+    thread.start()
+    return True
+
+
+def _pg_tick_refusal(tenant_slug, backend=None, write_refusal=None):
+    actual_backend = backend if backend is not None else os.environ.get("STORE_BACKEND", "hf")
+    if str(actual_backend).strip().lower() != "pg":
+        return None
+    if write_refusal is None:
+        import core.store_pg as _pg
+        write_refusal = _pg.write_refusal
+    return write_refusal(str(tenant_slug))
+
+
+def _scheduler_tenants(runtime_module, now=None):
+    """Known tenants, refreshed once per UTC day rather than on every external wake-up."""
+    stamp = now or _dt.datetime.now(_dt.timezone.utc)
+    if isinstance(stamp, _dt.datetime):
+        stamp = (stamp.replace(tzinfo=_dt.timezone.utc) if stamp.tzinfo is None
+                 else stamp.astimezone(_dt.timezone.utc))
+    day = stamp.date().isoformat() if hasattr(stamp, "date") else str(stamp)[:10]
+    with _TICK_CACHE_LOCK:
+        if _TICK_TENANT_CACHE.get("day") == day:
+            return list(_TICK_TENANT_CACHE.get("slugs") or [])
+    slugs = list(runtime_module.known_tenants())
+    with _TICK_CACHE_LOCK:
+        _TICK_TENANT_CACHE.update({"day": day, "slugs": tuple(slugs)})
+    return slugs
+
+
+def scheduler_status():
+    """Safe process-local observability for the operator plane and tick response."""
+    backend = str(os.environ.get("STORE_BACKEND") or "hf").lower()
+    external = bool(os.environ.get("AIOS_AUTOMATION_TICK_TOKEN"))
+    # A thread object surviving a test/config transition must never advertise a forbidden Pg
+    # scheduler. Production's source of truth is the backend invariant, not stale process state.
+    in_process = bool(backend != "pg" and _SCHEDULER[0] is not None
+                      and _SCHEDULER[0].is_alive())
+    source = "external" if external else ("in-process" if in_process else "")
+    with _TICK_CACHE_LOCK, _DERIVED_TICK_LOCKS_GUARD:
+        return {
+            "enabled": bool(external or in_process),
+            "source": source,
+            "backend": backend,
+            "definitionCaches": len(_TICK_DEFS_MEMO),
+            "tenantCacheDay": str(_TICK_TENANT_CACHE.get("day") or ""),
+            "derivedRunning": len(_DERIVED_RUNNING),
+            **dict(_SCHEDULER_METRICS),
+        }
+
+
+def tick_all(now=None, log=print):
     """Every registered tenant. Fail-quiet per tenant: one tenant's broken store must not stop
     the others' schedules."""
-    from harness import runtime as _rt
-    out = {}
-    for slug in _rt.known_tenants():
-        try:
-            out[slug] = tick(_rt.get_runtime(slug), slug, now=now, log=log)
-        except Exception as e:                                    # noqa: BLE001
-            log(f"[aios-auto] tick {slug} skipped: {type(e).__name__}: {e}")
-    return out
+    from harness import runtime as _rt
+    out = {}
+    slugs = _scheduler_tenants(_rt, now=now)
+    for slug in slugs:
+        try:
+            refusal = _pg_tick_refusal(slug)
+            if refusal:
+                log(f"[aios-auto] tick {slug} skipped before runtime read: {refusal}")
+                continue
+            out[slug] = tick(_rt.get_runtime(slug), slug, now=now, log=log)
+        except Exception as e:                                    # noqa: BLE001
+            log(f"[aios-auto] tick {slug} skipped: {type(e).__name__}: {e}")
+    with _TICK_CACHE_LOCK:
+        _SCHEDULER_METRICS["ticks"] += 1
+        _SCHEDULER_METRICS["lastTickAt"] = _iso()
+        _SCHEDULER_METRICS["lastTenantCount"] = len(slugs)
+    return out
 
 
 #: How often the in-process scheduler wakes. A minute is the cron resolution; anything finer
@@ -14585,7 +14808,7 @@ def scheduler_loop(log=print):
             log(f"[aios-auto] scheduler tick failed: {type(e).__name__}: {e}")
 
 
-def start_scheduler(log=print):
+def start_scheduler(log=print):
     """Start the loop once per process. **OPT-IN: `AIOS_AUTOMATIONS=1`.**
 
     ⚠ AMENDED 2026-08-03 (D), and the reason is a measurement rather than a preference. The wave
@@ -14602,9 +14825,15 @@ def start_scheduler(log=print):
     external EventBridge tick (R5) does not depend on it either way, since that POSTs the
     endpoint rather than riding this thread.
     """
-    if os.environ.get("AIOS_AUTOMATIONS") != "1":
-        return False
-    if _SCHEDULER[0] is not None:
+    if os.environ.get("AIOS_AUTOMATIONS") != "1":
+        return False
+    # PostgreSQL production has one durable wake-up: EventBridge -> Lambda -> protected endpoint.
+    # Refuse the old resident minute loop even if a stale secret is reintroduced; two schedulers
+    # were the measured cause of the Neon transfer incident this guard closes.
+    if str(os.environ.get("STORE_BACKEND") or "hf").strip().lower() == "pg":
+        log("[aios-auto] in-process scheduler refused on PostgreSQL; use the external tick")
+        return False
+    if _SCHEDULER[0] is not None:
         return False
     th = threading.Thread(target=scheduler_loop, kwargs={"log": log},
                           daemon=True, name="automation-scheduler")

api/rollup_sql.py

@@ -125,8 +125,8 @@ def _today():
     return _dt.date.today().strftime("%Y-%m-%d")
 
 
-def compute(rt, table_key, today=None, tables=None):
-    """Write every source-backed rollup cell on `table_key`. Returns `{field_key: cells_written}`.
+def compute(rt, table_key, today=None, tables=None, persist=False):
+    """Write changed source-backed rollup cells on `table_key`.
 
     `tables` (a live `user_tables` dict) is the gate's injection point — the same shape
     `automation_engine.compute_relation_cells` takes, so this can be proven against a fixture
@@ -167,13 +167,25 @@ def compute(rt, table_key, today=None, tables=None):
                 hit = values.get(join)
                 if hit is None and join.endswith(".0"):
                     hit = values.get(join[:-2])
-                row[fkey] = "" if hit is None else str(hit)
-                n += 1
+                value = "" if hit is None else str(hit)
+                if row.get(fkey) != value:
+                    row[fkey] = value
+                    n += 1
             written[fkey] = n
         return cur
 
     if owned:
         _apply(blob)
-    else:
+        changed = {k: n for k, n in written.items() if n}
+        if not changed or not persist:
+            return changed
+        written.clear()
         rt.update(ut.STORE_KEY, _apply, flush="async")
-    return written
+        return {k: n for k, n in written.items() if n}
+
+    _apply(blob)
+    if not any(written.values()):
+        return {}
+    written.clear()
+    rt.update(ut.STORE_KEY, _apply, flush="async")
+    return {k: n for k, n in written.items() if n}

api/routes_automation.py

@@ -560,19 +560,10 @@ def _triggers_vocab(session):
     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 "")}
+def _tick_state():
+    """Can a schedule fire here? PostgreSQL production has one external scheduler only."""
+    status = engine.scheduler_status()
+    return {"enabled": bool(status.get("enabled")), "source": status.get("source") or ""}
 
 
 #: W29-T01 — the Board retirement ran, per tenant, this process. Same shape and same reasoning as
@@ -1773,10 +1764,11 @@ def tick(request: Request, x_aios_tick_token: str = Header(default="")):
         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")}
+    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"),
+            "scheduler": engine.scheduler_status()}
 
 
 @router.post("/automations/ig/purge")

api/routes_platform_admin.py

@@ -568,6 +568,7 @@ def platform_connectors(tenant: str = "", session: Session = Depends(padmin_gate
 @router.get("/automations")
 def platform_automations(tenant: str = "", session: Session = Depends(padmin_gate)):
     """Every tenant's automations, their schedules and run history, plus the fleet cost model."""
+    import automation_engine as engine
     out, errors, retained = [], {}, 20
     for t, rt in _scan(tenant):
         slug = t["slug"]
@@ -580,11 +581,13 @@ def platform_automations(tenant: str = "", session: Session = Depends(padmin_gat
             continue
         retained = max_runs
         out += rows
+    scheduler = engine.scheduler_status()
     return {"automations": out, "count": len(out),
             "enabled": sum(1 for a in out if a["enabled"]),
             "historyRetained": retained,
             "cost": _automation_cost(out), "errors": errors,
-            "tickEnabled": os.environ.get("AIOS_AUTOMATIONS") == "1"}
+            "tickEnabled": bool(scheduler.get("enabled")),
+            "scheduler": scheduler}
 
 
 @router.get("/aws")

platform/harness/runtime.py

@@ -179,6 +179,10 @@ class TenantRuntime:
         """
         return self._store().get_projection(self.store_key(name), drop=drop)
 
+    def revision(self, name):
+        """Return the tiny durable change token for this tenant's addressed bucket."""
+        return self._store().revision(self.store_key(name))
+
     def put(self, name, data):
         return self._store().put(self.store_key(name), data)