Deploy AIOS web (React glide grid + FastAPI slice)

api/automation_control.py

@@ -0,0 +1,214 @@
+"""AWS automation control-plane client and exact schedule compiler.
+
+Production deliberately has no global tick.  The API tells a small signed AWS control
+function exactly which schedules one automation needs; direct events start one worker
+immediately.  With no enabled definitions the scheduler group is empty and ECS has no task.
+
+This module has no AWS dependency.  The web process talks to the control function over its
+Function URL using an HMAC over ``timestamp + newline + body``.  Local/test deployments with no
+control URL keep the historical in-process behavior; production Postgres fails closed if the
+control plane is missing.
+"""
+from __future__ import annotations
+
+import datetime as dt
+import hashlib
+import hmac
+import json
+import os
+import re
+import time
+import uuid
+
+import requests
+
+
+class ControlPlaneError(RuntimeError):
+    pass
+
+
+_FIELD_RANGES = ((0, 59), (0, 23), (1, 31), (1, 12), (0, 6))
+_DOW_NAMES = ("SUN", "MON", "TUE", "WED", "THU", "FRI", "SAT")
+_MANAGED_RECURRING = ("schedule", "email")
+
+
+def configured() -> bool:
+    return bool(str(os.environ.get("AIOS_AUTOMATION_CONTROL_URL") or "").strip()
+                and str(os.environ.get("AIOS_AUTOMATION_CONTROL_TOKEN") or "").strip())
+
+
+def _production_requires_control() -> bool:
+    if str(os.environ.get("STORE_BACKEND") or "").lower() != "pg":
+        return False
+    try:
+        from core import data_binding
+        return data_binding.is_production_deployment()
+    except Exception:
+        return False
+
+
+def _parse_part(spec: str, lo: int, hi: int) -> set[int]:
+    out: set[int] = set()
+    for raw in str(spec).split(","):
+        part = raw.strip()
+        if not part:
+            raise ValueError("empty cron field")
+        step = 1
+        if "/" in part:
+            part, step_raw = part.split("/", 1)
+            step = int(step_raw)
+            if step < 1:
+                raise ValueError("cron step must be positive")
+        if part in ("*", "?"):
+            first, last = lo, hi
+        elif "-" in part:
+            first_raw, last_raw = part.split("-", 1)
+            first, last = int(first_raw), int(last_raw)
+        else:
+            first = last = int(part)
+        if first < lo or last > hi or first > last:
+            raise ValueError(f"cron value {part!r} is outside {lo}..{hi}")
+        out.update(range(first, last + 1, step))
+    return out
+
+
+def _field(values: set[int], lo: int, hi: int, names=None) -> str:
+    if values == set(range(lo, hi + 1)):
+        return "*"
+    ordered = sorted(values)
+    return ",".join(str(names[v] if names else v) for v in ordered)
+
+
+def aws_cron_expressions(expr: str) -> list[str]:
+    """Translate one validated POSIX five-field cron into one or two AWS cron expressions.
+
+    POSIX uses OR when both day-of-month and weekday are restricted.  AWS requires one of those
+    fields to be ``?``, so that case becomes two schedules.  The durable worker job id includes
+    scheduled time and deduplicates the overlap when both arms match.
+    """
+    parts = str(expr or "").split()
+    if len(parts) != 5:
+        raise ValueError("a cron schedule has exactly five fields")
+    vals = [_parse_part(p, lo, hi) for p, (lo, hi) in zip(parts, _FIELD_RANGES)]
+    minute = _field(vals[0], 0, 59)
+    hour = _field(vals[1], 0, 23)
+    dom = _field(vals[2], 1, 31)
+    month = _field(vals[3], 1, 12)
+    dow = _field(vals[4], 0, 6, _DOW_NAMES)
+    dom_restricted = parts[2] not in ("*", "?")
+    dow_restricted = parts[4] not in ("*", "?")
+    if dom_restricted and dow_restricted:
+        return [f"cron({minute} {hour} {dom} {month} ? *)",
+                f"cron({minute} {hour} ? {month} {dow} *)"]
+    if dom_restricted:
+        return [f"cron({minute} {hour} {dom} {month} ? *)"]
+    if dow_restricted:
+        return [f"cron({minute} {hour} ? {month} {dow} *)"]
+    return [f"cron({minute} {hour} * {month} ? *)"]
+
+
+def generation(defn: dict) -> str:
+    relevant = {k: defn.get(k) for k in ("id", "kind", "config", "schedule", "trigger", "flow")}
+    body = json.dumps(relevant, sort_keys=True, separators=(",", ":"), default=str).encode()
+    return hashlib.sha256(body).hexdigest()[:20]
+
+
+def schedule_scope(tenant: str, auto_id: str) -> str:
+    digest = hashlib.sha256(f"{tenant}\0{auto_id}".encode()).hexdigest()[:20]
+    return f"a-{digest}"
+
+
+def _base_job(tenant: str, defn: dict, mode: str) -> dict:
+    return {"tenant": str(tenant), "automationId": str(defn.get("id") or ""),
+            "generation": generation(defn), "mode": mode,
+            "scheduledTime": "<aws.scheduler.scheduled-time>"}
+
+
+def recurring_specs(tenant: str, defn: dict) -> list[dict]:
+    scope = schedule_scope(tenant, defn.get("id"))
+    specs: list[dict] = []
+    schedule = defn.get("schedule") or {}
+    if schedule.get("enabled"):
+        for index, expression in enumerate(aws_cron_expressions(schedule.get("cron") or "")):
+            specs.append({"name": f"{scope}-schedule-{index}", "mode": "schedule",
+                          "expression": expression, "timezone": "UTC",
+                          "job": _base_job(tenant, defn, "schedule")})
+    trigger = defn.get("trigger") or {}
+    if (trigger.get("key") == "email" and trigger.get("enabled", True)
+            and not trigger.get("paused") and trigger.get("configured", True)):
+        specs.append({"name": f"{scope}-email-0", "mode": "email",
+                      "expression": "rate(15 minutes)", "timezone": "UTC",
+                      "job": _base_job(tenant, defn, "email")})
+    return specs
+
+
+def pending_specs(tenant: str, defn: dict, when: dt.datetime | None = None) -> list[dict]:
+    from automation_engine import pending_collect_ids
+
+    if str(defn.get("id") or "") not in pending_collect_ids(
+            None, definitions={defn["id"]: defn}):
+        return []
+    when = when or (dt.datetime.now(dt.timezone.utc) + dt.timedelta(minutes=15))
+    when = when.astimezone(dt.timezone.utc).replace(microsecond=0)
+    expression = f"at({when.strftime('%Y-%m-%dT%H:%M:%S')})"
+    scope = schedule_scope(tenant, defn.get("id"))
+    return [{"name": f"{scope}-pending-0", "mode": "pending", "expression": expression,
+             "timezone": "UTC", "oneTime": True, "job": _base_job(tenant, defn, "pending")}]
+
+
+def _signed_post(payload: dict) -> dict:
+    url = str(os.environ.get("AIOS_AUTOMATION_CONTROL_URL") or "").strip()
+    token = str(os.environ.get("AIOS_AUTOMATION_CONTROL_TOKEN") or "").strip()
+    if not url or not token:
+        if _production_requires_control():
+            raise ControlPlaneError("production automation control plane is not configured")
+        return {"local": True}
+    body = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
+    stamp = str(int(time.time()))
+    signature = hmac.new(token.encode(), stamp.encode() + b"\n" + body,
+                         hashlib.sha256).hexdigest()
+    try:
+        response = requests.post(url, data=body, timeout=15,
+                                 headers={"Content-Type": "application/json",
+                                          "X-AIOS-Timestamp": stamp,
+                                          "X-AIOS-Signature": signature})
+    except Exception as exc:
+        raise ControlPlaneError(f"automation control request failed: {type(exc).__name__}") from exc
+    if response.status_code >= 300:
+        raise ControlPlaneError(f"automation control answered HTTP {response.status_code}")
+    try:
+        return response.json()
+    except Exception as exc:
+        raise ControlPlaneError("automation control returned invalid JSON") from exc
+
+
+def sync_definition(tenant: str, defn: dict) -> dict:
+    return _signed_post({"action": "sync", "tenant": str(tenant),
+                         "automationId": str(defn.get("id") or ""),
+                         "managedModes": list(_MANAGED_RECURRING),
+                         "schedules": recurring_specs(tenant, defn)})
+
+
+def sync_pending(tenant: str, defn: dict, when: dt.datetime | None = None) -> dict:
+    return _signed_post({"action": "sync", "tenant": str(tenant),
+                         "automationId": str(defn.get("id") or ""),
+                         "managedModes": ["pending"],
+                         "schedules": pending_specs(tenant, defn, when=when)})
+
+
+def delete_definition(tenant: str, auto_id: str) -> dict:
+    return _signed_post({"action": "sync", "tenant": str(tenant),
+                         "automationId": str(auto_id),
+                         "managedModes": ["schedule", "email", "pending"], "schedules": []})
+
+
+def dispatch(tenant: str, auto_id: str, *, mode="manual", username="automation",
+             rows=None, request_id=None) -> dict:
+    job = {"tenant": str(tenant), "automationId": str(auto_id), "mode": str(mode),
+           "username": str(username or "automation"), "rows": list(rows or [])[:100],
+           "requestId": str(request_id or uuid.uuid4())}
+    return _signed_post({"action": "dispatch", "job": job})
+
+
+def safe_schedule_name(name: str) -> bool:
+    return bool(re.fullmatch(r"a-[0-9a-f]{20}-(schedule-[01]|email-0|pending-0)", str(name)))

api/automation_engine.py

@@ -956,10 +956,13 @@ def disable_for_table(rt, table_key, note="target database deleted"):
         for aid, d in (cur or {}).items():
             if isinstance(d, dict) and (d.get("config") or {}).get("targetTable") == key:
                 sch = d.get("schedule")
-                if not isinstance(sch, dict):
-                    sch = d["schedule"] = {}
-                sch["enabled"] = False
-                d["statusNote"] = note
+                if not isinstance(sch, dict):
+                    sch = d["schedule"] = {}
+                sch["enabled"] = False
+                trg = d.get("trigger")
+                if isinstance(trg, dict):
+                    trg["paused"] = True
+                d["statusNote"] = note
                 touched.append(str(aid))
         return cur
 
@@ -14490,11 +14493,26 @@ def run_now(rt, tenant, auto_id, username="automation", log=print, rows=None):
         _release(tenant, auto_id)
 
 
-def run_async(rt, tenant, auto_id, username="automation", log=print, rows=None):
-    """Start a run on a background thread. False when one is already in flight (→ 409)."""
-    if running(tenant, auto_id):
-        return False
-    th = threading.Thread(target=run_now, args=(rt, tenant, auto_id, username, log, rows),
+def run_async(rt, tenant, auto_id, username="automation", log=print, rows=None):
+    """Dispatch one run without keeping production work resident in the web process.
+
+    Production uses the signed AWS control plane and one Fargate task. The worker sets
+    ``AIOS_AUTOMATION_INLINE=1`` so a nested trigger finishes before that finite task exits.
+    Local development retains the historical daemon thread when no control plane is configured.
+    """
+    if running(tenant, auto_id):
+        return False
+    if os.environ.get("AIOS_AUTOMATION_INLINE") == "1":
+        return run_now(rt, tenant, auto_id, username=username, log=log, rows=rows) is not None
+    try:
+        import automation_control as _control
+        if _control.configured():
+            mode = "event" if username in MACHINE_OWNERS else "manual"
+            _control.dispatch(tenant, auto_id, mode=mode, username=username, rows=rows)
+            return True
+    except ImportError:
+        pass
+    th = threading.Thread(target=run_now, args=(rt, tenant, auto_id, username, log, rows),
                           daemon=True, name=f"automation-{tenant}-{auto_id}")
     th.start()
     return True
@@ -14747,12 +14765,16 @@ def _scheduler_tenants(runtime_module, now=None):
 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"))
+    try:
+        import automation_control as _control
+        external = _control.configured()
+    except Exception:
+        external = False
     # 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 "")
+    source = "exact-aws" if external else ("in-process" if in_process else "")
     with _TICK_CACHE_LOCK, _DERIVED_TICK_LOCKS_GUARD:
         return {
             "enabled": bool(external or in_process),

api/automation_worker.py

@@ -0,0 +1,220 @@
+"""One finite AIOS automation job for ECS/Fargate.
+
+The container has no loop and no service.  It claims one durable Postgres job, holds a tenant +
+automation advisory lock, performs only the requested work, reconciles follow-up schedules, and
+exits.  At-least-once delivery is therefore safe and idle AWS compute is literally zero tasks.
+"""
+from __future__ import annotations
+
+import datetime as dt
+import hashlib
+import json
+import os
+import sys
+import traceback
+
+
+def _load_job() -> dict:
+    raw = str(os.environ.get("AIOS_AUTOMATION_JOB") or "").strip()
+    if not raw:
+        raise RuntimeError("AIOS_AUTOMATION_JOB is required")
+    job = json.loads(raw)
+    if not isinstance(job, dict):
+        raise RuntimeError("AIOS_AUTOMATION_JOB must be an object")
+    tenant = str(job.get("tenant") or "").strip().lower()
+    mode = str(job.get("mode") or "").strip().lower()
+    if not tenant or mode not in {"schedule", "email", "pending", "manual", "event", "probe"}:
+        raise RuntimeError("job tenant/mode is invalid")
+    job["tenant"], job["mode"] = tenant, mode
+    job["automationId"] = str(job.get("automationId") or "")
+    return job
+
+
+def _job_id(job: dict) -> str:
+    supplied = str(job.get("requestId") or job.get("jobId") or "").strip()
+    if supplied:
+        return supplied[:180]
+    identity = "\0".join(str(job.get(k) or "") for k in
+                          ("tenant", "automationId", "mode", "generation", "scheduledTime"))
+    return "scheduled-" + hashlib.sha256(identity.encode()).hexdigest()
+
+
+def _scheduled_for(job: dict):
+    raw = str(job.get("scheduledTime") or "").strip()
+    if not raw or raw.startswith("<aws.scheduler."):
+        return None
+    try:
+        return dt.datetime.fromisoformat(raw.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+
+
+def _finish(con, job_id: str, status: str, detail: dict):
+    with con.cursor() as cur:
+        cur.execute("""
+            UPDATE control.automation_jobs
+               SET status=%s, finished_at=now(), detail=detail || %s::jsonb
+             WHERE job_id=%s
+        """, (status, json.dumps(detail, default=str), job_id))
+    con.commit()
+
+
+def _claim(job: dict):
+    import psycopg
+    from core import store_pg
+
+    store_pg.check_write(job["tenant"], operation="automation worker job claim")
+    url = str(os.environ.get("DATABASE_URL") or "").strip()
+    if not url:
+        raise RuntimeError("DATABASE_URL is required")
+    con = psycopg.connect(url)
+    lock_name = f"aios-automation:{job['tenant']}:{job['automationId'] or job['mode']}"
+    with con.cursor() as cur:
+        cur.execute("SELECT pg_try_advisory_lock(hashtextextended(%s, 0))", (lock_name,))
+        if not cur.fetchone()[0]:
+            con.close()
+            return None, "concurrent"
+        jid = _job_id(job)
+        cur.execute("""
+            INSERT INTO control.automation_jobs
+                (job_id, tenant_slug, automation_id, trigger_kind, generation, scheduled_for,
+                 status, attempts, started_at, detail)
+            VALUES (%s,%s,%s,%s,%s,%s,'running',1,now(),'{}'::jsonb)
+            ON CONFLICT (job_id) DO NOTHING
+        """, (jid, job["tenant"], job["automationId"], job["mode"],
+              str(job.get("generation") or ""), _scheduled_for(job)))
+        inserted = cur.rowcount == 1
+        cur.execute("SELECT status, attempts, detail FROM control.automation_jobs WHERE job_id=%s",
+                    (jid,))
+        status, attempts, detail = cur.fetchone()
+        if status in ("completed", "skipped"):
+            con.commit()
+            con.close()
+            return None, status
+        cur.execute("""
+            UPDATE control.automation_jobs
+               SET status='running', attempts=%s, started_at=now(), finished_at=NULL
+             WHERE job_id=%s
+        """, (max(1, int(attempts) + (0 if inserted else 1)), jid))
+    con.commit()
+    return con, dict(detail or {})
+
+
+def _definition_refusal(job: dict, defn: dict | None, engine, control) -> str:
+    mode = job["mode"]
+    if mode == "probe":
+        return ""
+    if not defn:
+        return "automation no longer exists"
+    expected = str(job.get("generation") or "")
+    if expected and expected != control.generation(defn):
+        return "stale schedule generation"
+    if mode == "schedule" and not (defn.get("schedule") or {}).get("enabled"):
+        return "schedule is disabled"
+    trigger = defn.get("trigger") or {}
+    if mode == "email" and not (trigger.get("key") == "email"
+                                and trigger.get("enabled", True)
+                                and not trigger.get("paused")
+                                and trigger.get("configured", True)):
+        return "email trigger is disabled or stale"
+    if mode == "pending" and str(defn.get("id")) not in engine.pending_collect_ids(
+            None, definitions={str(defn.get("id")): defn}):
+        return "paid vendor result is no longer pending"
+    return ""
+
+
+def _run_business(job: dict, rt, defn: dict, engine):
+    mode = job["mode"]
+    auto_id = job["automationId"]
+    if mode == "probe":
+        return {"probe": "ok"}, False
+    if mode == "email":
+        result = engine.email_poll(rt, job["tenant"], auto_id, defn)
+        return {"email": "fired" if result is not None else "idle"}, result is not None
+    entry = engine.run_now(rt, job["tenant"], auto_id,
+                           username=str(job.get("username") or "automation"),
+                           rows=list(job.get("rows") or [])[:100] or None)
+    if entry is None:
+        return {"run": "not-started"}, False
+    return {"run": entry}, True
+
+
+def main() -> int:
+    job = _load_job()
+    os.environ["AIOS_TENANT"] = job["tenant"]
+    os.environ["AIOS_AUTOMATION_INLINE"] = "1"
+    os.environ.setdefault("STORE_BACKEND", "pg")
+    os.environ.setdefault("SAFE_MODE", "1")
+
+    # Imports happen after tenant/identity env is final.  Nothing imports FastAPI or starts an
+    # application server in this worker image.
+    import automation_control as control
+    import automation_engine as engine
+    from harness import runtime
+
+    con, prior = _claim(job)
+    if con is None:
+        print(json.dumps({"jobId": _job_id(job), "result": prior}, sort_keys=True))
+        return 0
+    jid = _job_id(job)
+    try:
+        rt = runtime.get_runtime(job["tenant"])
+        defs = engine.all_definitions(rt)
+        defn = defs.get(job["automationId"])
+        refusal = _definition_refusal(job, defn, engine, control)
+        if refusal:
+            _finish(con, jid, "skipped", {"reason": refusal})
+            print(json.dumps({"jobId": jid, "result": "skipped", "reason": refusal},
+                             sort_keys=True))
+            return 0
+
+        detail = dict(prior or {})
+        did_work = bool(detail.get("businessComplete"))
+        if not detail.get("businessComplete"):
+            outcome, did_work = _run_business(job, rt, defn, engine)
+            detail.update(outcome)
+            detail["businessComplete"] = True
+            with con.cursor() as cur:
+                cur.execute("UPDATE control.automation_jobs SET detail=%s::jsonb WHERE job_id=%s",
+                            (json.dumps(detail, default=str), jid))
+            con.commit()
+
+        # Derived cells are refreshed only after a real run, never on an idle email check and
+        # never because a clock woke up.  These functions persist only when values changed.
+        if did_work and not detail.get("derivedComplete"):
+            engine.refresh_relations(rt)
+            engine._refresh_source_rollups(rt, job["tenant"])
+            detail["derivedComplete"] = True
+            with con.cursor() as cur:
+                cur.execute("UPDATE control.automation_jobs SET detail=%s::jsonb WHERE job_id=%s",
+                            (json.dumps(detail, default=str), jid))
+            con.commit()
+
+        if job["mode"] != "probe":
+            current = engine.all_definitions(rt).get(job["automationId"])
+            if current:
+                control.sync_definition(job["tenant"], current)
+                control.sync_pending(job["tenant"], current)
+            else:
+                control.delete_definition(job["tenant"], job["automationId"])
+        _finish(con, jid, "completed", {"controlSynced": True})
+        print(json.dumps({"jobId": jid, "result": "completed", "didWork": did_work},
+                         sort_keys=True))
+        return 0
+    except Exception as exc:
+        try:
+            _finish(con, jid, "failed", {"error": type(exc).__name__,
+                                         "message": str(exc)[:500]})
+        except Exception:
+            pass
+        traceback.print_exc()
+        return 1
+    finally:
+        try:
+            con.close()
+        except Exception:
+            pass
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

api/routes_automation.py

@@ -18,8 +18,9 @@ from datetime import datetime, timezone
 
 from fastapi import APIRouter, Body, Depends, Header, Request
 
-import ai_review
-import automation_engine as engine
+import ai_review
+import automation_control
+import automation_engine as engine
 import oauth_connect
 import routes_oauth
 import scope_cache
@@ -51,7 +52,17 @@ import routes_connectors                                             # noqa: E40
 #: the day the row lands the wall is already the one that was tested).
 MODULE = "automation"
 
-_GATE = module_gate(MODULE)
+_GATE = module_gate(MODULE)
+
+
+def _sync_schedule_or_503(tenant, defn):
+    """Apply the exact native schedules after a definition write, never silently drift."""
+    try:
+        return automation_control.sync_definition(tenant, defn)
+    except automation_control.ControlPlaneError as exc:
+        raise err(503, "automation_control_unavailable",
+                  "the definition was saved, but its AWS schedule could not be reconciled. "
+                  f"Nothing was reported as armed: {exc}")
 
 
 def _wire(defn, tenant, rt=None):
@@ -561,9 +572,12 @@ def _triggers_vocab(session):
 
 
 def _tick_state():
-    """Can a schedule fire here? PostgreSQL production has one external scheduler only."""
+    """Can this deployment reconcile exact AWS schedules and dispatch finite workers?"""
+    if automation_control.configured():
+        return {"enabled": True, "source": "exact-aws"}
     status = engine.scheduler_status()
-    return {"enabled": bool(status.get("enabled")), "source": status.get("source") or ""}
+    return {"enabled": bool(status.get("source") == "in-process"),
+            "source": "in-process" if status.get("source") == "in-process" else ""}
 
 
 #: W29-T01 — the Board retirement ran, per tenant, this process. Same shape and same reasoning as
@@ -1567,9 +1581,10 @@ def create_automation(body: dict = Body(default=None), session: Session = Depend
     _notes = []
     defn, error = engine.create(session.runtime, body or {}, username=session.uname,
                                 notes=_notes)
-    if error:
-        raise err(400, "invalid_automation", error)
-    return {"automation": _wire(defn, session.tenant, rt=session.runtime), "notes": _notes}
+    if error:
+        raise err(400, "invalid_automation", error)
+    _sync_schedule_or_503(session.tenant, defn)
+    return {"automation": _wire(defn, session.tenant, rt=session.runtime), "notes": _notes}
 
 
 @router.patch("/automations/{auto_id}")
@@ -1602,10 +1617,11 @@ def patch_automation(auto_id: str, body: dict = Body(default=None),
         raise err(400 if error != "no such automation" else 404,
                   "invalid_automation" if error != "no such automation" else "unknown_automation",
                   error)
-    # An action the user removed takes its annotation with it, or the mark outlives its subject
-    # and refuses an id nobody can see.
-    _prune_marks(session, auto_id)
-    return {"automation": _wire(defn, session.tenant, rt=session.runtime), "notes": _notes}
+    # An action the user removed takes its annotation with it, or the mark outlives its subject
+    # and refuses an id nobody can see.
+    _prune_marks(session, auto_id)
+    _sync_schedule_or_503(session.tenant, defn)
+    return {"automation": _wire(defn, session.tenant, rt=session.runtime), "notes": _notes}
 
 
 @router.delete("/automations/{auto_id}")
@@ -1642,11 +1658,17 @@ def delete_automation(auto_id: str, session: Session = Depends(_GATE)):
     _sys = str(_defn.get("system") or "").strip()
     if _sys:
         raise err(409, "system_agent", engine.system_agent_refusal(_sys))
-    engine.remove(session.runtime, auto_id)
+    engine.remove(session.runtime, auto_id)
     # R9's other half: the user MAY delete an agent-authored step, and deleting the whole
     # automation takes its annotations with it rather than stranding them in the document.
-    _prune_marks(session, auto_id)
-    return {"deleted": str(auto_id)}
+    _prune_marks(session, auto_id)
+    try:
+        automation_control.delete_definition(session.tenant, auto_id)
+    except automation_control.ControlPlaneError as exc:
+        raise err(503, "automation_control_unavailable",
+                  "the automation was deleted, but its AWS schedules could not be removed: "
+                  f"{exc}")
+    return {"deleted": str(auto_id)}
 
 
 @router.post("/automations/preview")
@@ -1679,8 +1701,13 @@ def run_automation(auto_id: str, session: Session = Depends(_GATE)):
     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")
+    try:
+        started = engine.run_async(session.runtime, session.tenant, auto_id,
+                                   username=session.uname)
+    except automation_control.ControlPlaneError as exc:
+        raise err(503, "automation_control_unavailable", str(exc))
+    if not started:
+        raise err(409, "automation_running", "that automation is already running")
     return {"started": str(auto_id), "startedAt": time.strftime("%Y-%m-%dT%H:%M:%S")}
 
 
@@ -1697,11 +1724,12 @@ def toggle_automation_node(auto_id: str, node_id: str, session: Session = Depend
     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, rt=session.runtime)}
+    if error:
+        raise err(404 if error == "no such automation" else 400,
+                  "unknown_automation" if error == "no such automation" else "node_not_toggleable",
+                  error)
+    _sync_schedule_or_503(session.tenant, defn)
+    return {"automation": _wire(defn, session.tenant, rt=session.runtime)}
 
 
 @router.post("/automations/{auto_id}/hook/{token}")
@@ -1759,16 +1787,9 @@ def tick(request: Request, x_aios_tick_token: str = Header(default="")):
 
     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"),
-            "scheduler": engine.scheduler_status()}
+    raise err(410, "tick_retired",
+              "the global automation tick is retired. Enabled automations own exact AWS "
+              "schedules and direct events dispatch scale-to-zero workers")
 
 
 @router.post("/automations/ig/purge")

api/routes_tables.py

@@ -1352,11 +1352,21 @@ def delete_table(table_key: str, session: Session = Depends(require_session)):
     defn = _defn_or_refuse(session, table_key)
     if not (session.admin or defn.get("createdBy") == session.uname):
         raise err(403, "forbidden", "only the database's creator or an admin can delete it")
-    try:
-        import automation_engine as engine
-        engine.disable_for_table(session.runtime, table_key)
-    except Exception:
-        pass
+    try:
+        import automation_control
+        import automation_engine as engine
+        touched = engine.disable_for_table(session.runtime, table_key)
+        defs = engine.all_definitions(session.runtime)
+        for auto_id in touched:
+            automation_control.sync_definition(session.tenant, defs[auto_id])
+    except automation_control.ControlPlaneError as exc:
+        raise err(503, "automation_control_unavailable",
+                  "the database is still present, but its automations could not be disarmed in "
+                  f"AWS: {exc}")
+    except Exception as exc:
+        raise err(503, "automation_disable_failed",
+                  f"the database is still present because its automations could not be disabled: "
+                  f"{type(exc).__name__}")
     try:
         _ut().delete(table_key, st=session.runtime)
     except Exception:

platform/core/data_binding.py

@@ -59,6 +59,11 @@ import os
 #: agree — a constant two features share is a constant that drifts ([[constant-two-features-share]]).
 PRODUCTION_SPACES = frozenset({'fsanyoto/runloopable'})
 
+# Provisioning pins this exact identity into the scale-to-zero ECS task definition.  It is the
+# only non-Space process allowed to write production tenant schemas; unknown identities remain
+# refused by the same default-deny guard.
+PRODUCTION_WORKERS = frozenset({'aws:aios-automation-worker'})
+
 #: The stores that hold REAL customer business data. Enumerated 2026-08-18 from the live control
 #: plane (`royal-imports/cfo-os-data::tenants.json`), not guessed.
 #:
@@ -80,7 +85,8 @@ PRODUCTION_DEFAULT_STORE = 'royal-imports/cfo-os-data'
 #: The env vars a Hugging Face Space container sets to describe itself, in the order they are
 #: trusted. `SPACE_ID` is the canonical one; the pair is the documented fallback. Read as a LIST so
 #: `describe()` can report every one of them and what it held.
-_IDENTITY_VARS = ('SPACE_ID', 'SPACE_AUTHOR_NAME', 'SPACE_REPO_NAME', 'SPACE_HOST')
+_IDENTITY_VARS = ('AIOS_DEPLOYMENT_ID', 'SPACE_ID', 'SPACE_AUTHOR_NAME',
+                  'SPACE_REPO_NAME', 'SPACE_HOST')
 
 #: The env a NON-production deployment uses to name the stores it may write, beyond `OS_DATA_REPO`.
 #: Comma-separated.
@@ -121,6 +127,9 @@ def deployment_id():
     platform inside the container, so it cannot be forgotten, mistyped, or left behind by a Space
     that changed role without being redeployed.
     """
+    explicit = _env('AIOS_DEPLOYMENT_ID')
+    if explicit:
+        return explicit
     sid = _env('SPACE_ID')
     if sid:
         return sid
@@ -132,7 +141,7 @@ def deployment_id():
 
 def is_production_deployment():
     """True only on the deployment that owns production data. Unknown ⇒ False (fail closed)."""
-    return deployment_id() in PRODUCTION_SPACES
+    return deployment_id() in (PRODUCTION_SPACES | PRODUCTION_WORKERS)
 
 
 def is_space():

platform/harness/pg/schema.sql

@@ -116,6 +116,30 @@ CREATE TABLE IF NOT EXISTS control.audit (
 CREATE INDEX IF NOT EXISTS audit_at_idx ON control.audit (at DESC);
 CREATE INDEX IF NOT EXISTS audit_tenant_idx ON control.audit (tenant_slug, at DESC);
 
+-- Durable idempotency and audit for the scale-to-zero automation worker.  EventBridge and Step
+-- Functions are at-least-once transports; this row is what makes a retried scheduled minute one
+-- business run instead of two.  No scheduler heartbeat is stored here -- rows exist only when
+-- actual work was dispatched.
+CREATE TABLE IF NOT EXISTS control.automation_jobs (
+    job_id          text        PRIMARY KEY,
+    tenant_slug     text        NOT NULL REFERENCES control.tenants(slug) ON DELETE CASCADE,
+    automation_id   text        NOT NULL,
+    trigger_kind    text        NOT NULL,
+    generation      text        NOT NULL DEFAULT '',
+    scheduled_for   timestamptz,
+    status          text        NOT NULL CHECK (status IN ('running', 'completed', 'skipped',
+                                                           'failed')),
+    attempts        integer     NOT NULL DEFAULT 0,
+    created_at      timestamptz NOT NULL DEFAULT now(),
+    started_at      timestamptz,
+    finished_at     timestamptz,
+    detail          jsonb       NOT NULL DEFAULT '{}'::jsonb
+);
+CREATE INDEX IF NOT EXISTS automation_jobs_tenant_at_idx
+    ON control.automation_jobs (tenant_slug, created_at DESC);
+CREATE INDEX IF NOT EXISTS automation_jobs_open_idx
+    ON control.automation_jobs (status, created_at) WHERE status IN ('running', 'failed');
+
 -- ─────────────────────────────────────────────────────────────────────────────────────────────
 -- PER-TENANT: one schema `t_<slug>` per tenant, holding what the HF store holds today.
 --