Deploy AIOS web (React glide grid + FastAPI slice)

RELEASES.json

@@ -1,5 +1,5 @@
 {
- "current": "b859a84",
+ "current": "139bf21",
  "releases": [
   {
    "version": "v14",

VERSION

@@ -1 +1 @@
-b859a84
+139bf21

api/automation_engine.py

@@ -23,8 +23,9 @@ the skill version takes a URL from a developer on a CLI; this takes one from a r
 """
 from __future__ import annotations
 
-import datetime as _dt
-import hmac
+import datetime as _dt
+import hashlib
+import hmac
 import ipaddress
 import json
 import os
@@ -92,7 +93,12 @@ IG_TABLE_PREFIX = "ut_ig_"
 #: have inherited a 200,000-row ceiling, and with it the measured 35.8 MB single-bucket
 #: serialisation cost, purely by an accident of naming. Naming the append tables is the version
 #: that stays true when the next `ut_ig_*` table is not one.
-APPEND_TABLES = frozenset({"ut_ig_snapshots", "ut_ig_post_snapshots"})
+IG_SNAPSHOTS_TABLE = "ut_ig_snapshots"
+IG_POSTS_TABLE = "ut_ig_posts"
+IG_POST_SNAPSHOTS_TABLE = "ut_ig_post_snapshots"
+IG_COMMENTS_TABLE = "ut_ig_comments"
+AUTOMATION_RECORD_MODE = "automation"
+APPEND_TABLES = frozenset({IG_SNAPSHOTS_TABLE, IG_POST_SNAPSHOTS_TABLE})
 
 #: ⭐ WAVE 24 (owner ruling R6) — `plain` IS WHAT AN AUTOMATION IS NOW, and it is the DEFAULT.
 #: The create wizard is deleted, so nobody picks a kind any more: a new automation is a trigger
@@ -622,7 +628,7 @@ def next_fire(expr, now=None, lookahead_days=400):
 # THE UPSERT — pure, so the arithmetic is testable without a store or a network
 # ---------------------------------------------------------------------------------------------
 
-def upsert_rows(existing, incoming, key_field, cap=None):
+def upsert_rows(existing, incoming, key_field, cap=None):
     """Merge scraped rows into a user table's rows BY KEY. Returns `(rows, counts)`.
 
     THE RULE THAT MATTERS: **an orphan is COUNTED, NEVER DELETED.** A row that has stopped
@@ -691,7 +697,41 @@ def upsert_rows(existing, incoming, key_field, cap=None):
     counts["duplicates"] = incoming_dupes + len(dupe_ids)
     counts["orphans"] = sum(
         1 for kv, rid in by_key.items() if kv not in seen_keys and rid not in dupe_ids)
-    return rows, counts
+    return rows, counts
+
+
+def dedupe_canonical_rows(existing, key_field, newest_by=""):
+    """Collapse duplicate logical rows while preserving the lowest stable row id.
+
+    Canonical entity tables use this before every upsert. Snapshot tables deliberately do not:
+    repeated shortcodes there are new timestamped observations, not duplicate posts. Values are
+    taken newest-first and then filled from older rows, so a sparse fresh projection does not
+    erase a field an earlier row knew.
+    """
+    rows = {str(k): dict(v or {}) for k, v in (existing or {}).items()}
+    groups = {}
+    for rid, row in rows.items():
+        identity = str(row.get(key_field) or "").strip()
+        if identity:
+            groups.setdefault(identity, []).append((rid, row))
+    removed = 0
+    for members in groups.values():
+        if len(members) < 2:
+            continue
+        keep = min((rid for rid, _row in members), key=lambda r: (not r.isdigit(), int(r) if r.isdigit() else r))
+        ordered = sorted(members, key=lambda item: str(item[1].get(newest_by) or ""), reverse=True) \
+            if newest_by else members
+        merged = {}
+        for _rid, row in ordered:
+            for key, value in row.items():
+                if key not in merged or str(merged.get(key) or "").strip() == "":
+                    merged[key] = value
+        rows[keep] = merged
+        for rid, _row in members:
+            if rid != keep:
+                rows.pop(rid, None)
+                removed += 1
+    return rows, removed
 
 
 # ---------------------------------------------------------------------------------------------
@@ -797,7 +837,8 @@ def ut_key_for(label, key=None):
 MACHINE_OWNERS = ("automation", "scheduler")
 
 
-def ut_ensure(rt, label, fields, username="automation", key=None, flow_tag=""):
+def ut_ensure(rt, label, fields, username="automation", key=None, flow_tag="",
+              record_mode="", lock_fields=False):
     """Create the table if it is missing; return its key. Idempotent — a re-run of an automation
     that owns a table must not spawn `ut_x_2`, so the key is DERIVED from the label (or given)
     and an existing table with that key is adopted, not duplicated.
@@ -838,21 +879,34 @@ def ut_ensure(rt, label, fields, username="automation", key=None, flow_tag=""):
             # A migration that cannot run must not stop the automation from writing its rows.
             # The old schema still reads; a refused write loses the pull we just paid for.
             pass
-    wanted = [({**f, "automation": {"flowId": str(flow_tag)}}
-               if flow_tag and not f.get("automation") else dict(f))
-              for f in (fields or [])]
+    wanted = []
+    for raw_field in (fields or []):
+        field = dict(raw_field)
+        if flow_tag or lock_fields:
+            automation = dict(field.get("automation") or {})
+            if flow_tag:
+                automation.setdefault("flowId", str(flow_tag))
+            if lock_fields:
+                automation["preset"] = True
+            field["automation"] = automation
+        wanted.append(field)
     created = _iso()
     human = username if username and username not in MACHINE_OWNERS else ""
 
     # ⚠ SKIP THE WRITE WHEN NOTHING WOULD CHANGE. Without this, every re-run spends a store
     # commit re-writing an identical definition — against a 20 s flush floor and a 256/hr repo
     # budget, an idempotent helper that always writes is the same defect as a per-row insert.
-    have = ut_get(rt, key)
-    if (have is not None
-            and not [f for f in wanted
-                     if f.get("key") not in {g.get("key") for g in (have.get("fields") or [])}]
-            and not (human and (have.get("createdBy") or "") in MACHINE_OWNERS)):
-        return key
+    have = ut_get(rt, key)
+    have_fields = {str(f.get("key") or ""): f for f in ((have or {}).get("fields") or [])}
+    missing = [f for f in wanted if f.get("key") not in have_fields]
+    missing_locks = [f for f in wanted
+                     if lock_fields and f.get("key") in have_fields
+                     and (not isinstance(have_fields[f.get("key")].get("automation"), dict)
+                          or have_fields[f.get("key")]["automation"].get("preset") is not True)]
+    if (have is not None and not missing and not missing_locks
+            and not (record_mode and have.get("recordMode") != record_mode)
+            and not (human and (have.get("createdBy") or "") in MACHINE_OWNERS)):
+        return key
 
     def _up(cur):
         cur = cur if isinstance(cur, dict) else {}
@@ -860,21 +914,34 @@ def ut_ensure(rt, label, fields, username="automation", key=None, flow_tag=""):
         if t is None:
             if len(cur) >= MAX_UT_TABLES:
                 return cur
-            cur[key] = {"key": key, "label": str(label)[:60], "source": "Automation",
-                        "createdBy": username, "created": created,
-                        "fields": wanted, "rows": {}}
-            return cur
+            cur[key] = {"key": key, "label": str(label)[:60], "source": "Automation",
+                        "createdBy": username, "created": created,
+                        "fields": wanted, "rows": {}}
+            if record_mode:
+                cur[key]["recordMode"] = record_mode
+            return cur
         have = {f.get("key") for f in (t.get("fields") or [])}
-        for f in wanted:
-            if f.get("key") not in have:
-                t.setdefault("fields", []).append(f)
-                have.add(f.get("key"))
+        for f in wanted:
+            if f.get("key") not in have:
+                t.setdefault("fields", []).append(f)
+                have.add(f.get("key"))
+            elif lock_fields:
+                stored = next((g for g in (t.get("fields") or [])
+                               if g.get("key") == f.get("key")), None)
+                if stored is not None:
+                    automation = dict(stored.get("automation") or {})
+                    if flow_tag:
+                        automation.setdefault("flowId", str(flow_tag))
+                    automation["preset"] = True
+                    stored["automation"] = automation
         # ADOPTION: a machine name is not an owner. It never overwrites a human one.
         if human and (t.get("createdBy") or "") in MACHINE_OWNERS:
             t["createdBy"] = human
         # An automation's table SAYS an automation owns it — the nav badge reads from this.
-        t["source"] = t.get("source") or "Automation"
-        return cur
+        t["source"] = t.get("source") or "Automation"
+        if record_mode:
+            t["recordMode"] = record_mode
+        return cur
 
     rt.update(UT_STORE_KEY, _up, flush="sync")
     return key
@@ -1759,15 +1826,26 @@ def _presets_after_write(rt, defn, username):
     arrives later. The run-time path reports that condition LOUDLY (`ut_missing`, D-11), so the
     honest failure still has exactly one home.
     """
-    if (defn.get("trigger") or {}).get("key") != "ig_profile_match":
-        return
-    cfg = defn.get("config") or {}
-    target = str(cfg.get("targetTable") or "")
-    if not target:
-        return
-    try:
-        ut_ensure(rt, cfg.get("targetLabel") or "IG candidates", CANDIDATE_FIELDS, username,
-                  key=target, flow_tag=str(defn.get("id") or ""))
+    cfg = defn.get("config") or {}
+    target = str(_flow_table(defn) or cfg.get("targetTable") or "")
+    if not target:
+        return
+    try:
+        if (defn.get("trigger") or {}).get("key") == "ig_profile_match":
+            ut_ensure(rt, cfg.get("targetLabel") or "IG candidates", CANDIDATE_FIELDS, username,
+                      key=target, flow_tag=str(defn.get("id") or ""), lock_fields=True)
+            return
+        enrich_actions = _actions_of_kind(((defn.get("flow") or {}).get("actions") or []),
+                                          "enrich_instagram")
+        table = ut_get(rt, target) or {}
+        named = next((str((a.get("config") or {}).get("profileField") or "").strip()
+                      for a in enrich_actions
+                      if str((a.get("config") or {}).get("profileField") or "").strip()), "")
+        bound = next((str(f.get("key") or "") for f in table.get("fields") or []
+                      if isinstance(f.get("profile"), dict)), "") or named
+        if enrich_actions and bound:
+            ut_ensure(rt, table.get("label") or target, _profile_schema_for(bound), username,
+                      key=target, flow_tag=str(defn.get("id") or ""), lock_fields=True)
     except Exception as e:                                            # noqa: BLE001
         # ⛔ NEVER FAILS THE SAVE. The definition is already committed by the time this runs, so
         # raising here would answer 500 for an automation that IS stored — the caller would retry
@@ -1945,8 +2023,7 @@ SNAPSHOT_FIELDS = [
     field_def("is_professional", "Professional account"),
     field_def("is_private", "Private"),
     field_def("highlights_count", "Highlights"),
-    field_def("bio_hashtags", "Bio hashtags"),
-    field_def("post_hashtags", "Post hashtags"),
+    field_def("bio_hashtags", "Bio hashtags"),
     field_def("pronouns", "Pronouns"),
     # ⭐ 2026-08-07 — the rest of the vendor's Profiles schema (see `_bd_profile`). The snapshot
     # row maps the WHOLE schema by design, so these belong here the moment the map reads them;
@@ -1988,8 +2065,12 @@ POST_FIELDS = [
     # ⚠ BLANK, NEVER ZERO. `postMetrics` is off by default (it buys one vendor record per post),
     # so on most pulls these stay empty — and empty means "not read", which is what makes an
     # honest average possible at all. A 0 here would claim a post nobody watched.
-    field_def("views", "Views", "int"),
-    field_def("likes", "Likes", "int"),
+    field_def("views", "Views", "int"),
+    # Bright Data's Reels schema exposes BOTH views and video play count. They are not aliases:
+    # its own example returns 388 views and 1,890 plays for one Reel. Keeping two columns avoids
+    # a plausible-looking "Views" value that is actually repeat plays.
+    field_def("plays", "Plays", "int"),
+    field_def("likes", "Likes", "int"),
     field_def("comments", "Comments", "int"),
     field_def("measured_at", "Engagement read at", "date"),
     # ⭐ 2026-08-07 — the second half of "spawn relevant Post/Comment database that is LINKED":
@@ -1998,8 +2079,12 @@ POST_FIELDS = [
     # ⚠ It resolves to nothing until comment capture is switched on, which is the honest state
     # for a relation whose far side is empty — the same standing `ut_ig_post_snapshots` has when
     # `postMetrics` is off.
-    field_def("comments_link", "Comment rows", "link",
-              link={"table": "ut_ig_comments", "on": "shortcode", "from": "shortcode"}),
+    field_def("comments_link", "Comment rows", "link",
+              link={"table": IG_COMMENTS_TABLE, "on": "shortcode", "from": "shortcode"}),
+    field_def("post_snapshots_link", "Measurement rows", "link",
+              link={"table": IG_POST_SNAPSHOTS_TABLE, "on": "shortcode", "from": "shortcode"}),
+    field_def("measurements_captured", "Measurements captured", "rollup",
+              rollup={"link": "post_snapshots_link", "fn": "countall"}),
 ]
 
 #: ⭐⭐ 2026-08-07 (owner instruction) — THE COMMENT DATABASE.
@@ -2031,21 +2116,29 @@ COMMENT_FIELDS = [
     field_def("comment_key", "Comment"), field_def("shortcode", "Shortcode"),
     field_def("influencer_key", "Influencer"),
     field_def("commented_at", "Commented at", "date"),
-    field_def("likes", "Likes", "int"),
-    field_def("replies", "Replies", "int"),
+    field_def("likes", "Likes", "int"),
+    field_def("replies", "Replies", "int"),
+    field_def("post_link", "Post", "link",
+              link={"table": IG_POSTS_TABLE, "on": "shortcode", "from": "shortcode",
+                    "single": True}),
     # ⚠ NO AUTHOR COLUMN AND NO TEXT COLUMN, and their absence is the ruling rather than an
     # oversight. `comment_user` is vendor-flagged PII and the text is the payload D-22 priced.
     # What is left is the SHAPE of a thread — how many, how recent, how engaged — which is the
     # part that informs an influencer decision without ingesting a stranger.
 ]
 POST_SNAPSHOT_FIELDS = [
-    field_def("post_snapshot_key", "Snapshot"), field_def("shortcode", "Shortcode"),
-    field_def("pulled_at", "Pulled at"), field_def("likes", "Likes"),
-    field_def("comments", "Comments"),
+    field_def("post_snapshot_key", "Snapshot"), field_def("shortcode", "Shortcode"),
+    field_def("influencer_key", "Influencer"),
+    field_def("pulled_at", "Pulled at"), field_def("likes", "Likes"),
+    field_def("comments", "Comments"),
     # Plays/views move independently of likes on video, so it is its own series rather than a
     # thing to derive. Paid rung only; blank means not read.
-    field_def("views", "Views"),
-]
+    field_def("views", "Views", "int"),
+    field_def("plays", "Plays", "int"),
+    field_def("post_link", "Post", "link",
+              link={"table": IG_POSTS_TABLE, "on": "shortcode", "from": "shortcode",
+                    "single": True}),
+]
 
 
 def ig_handle(url):
@@ -2191,7 +2284,8 @@ def _loose_count(txt):
 BD_BASE_DEFAULT = "https://api.brightdata.com"
 BD_DS_PROFILES = "gd_l1vikfch901nx3by4"       # Instagram – Profiles. 36 fields, 620M records
 BD_DS_POSTS = "gd_lk5ns7kz21pck8jpis"         # Instagram – Posts. 43 fields
-BD_PATH_SCRAPE = "/datasets/v3/scrape"        # SYNC: rows come back inline. param: dataset_id
+BD_DS_REELS = "gd_lyclm20il4r5helnj"           # Instagram Reels: views and play counts
+BD_PATH_SCRAPE = "/datasets/v3/scrape"        # SYNC: rows come back inline. param: dataset_id
 BD_PATH_TRIGGER = "/datasets/v3/trigger"      # ASYNC: -> {"snapshot_id": "sd_…"}
 BD_PATH_SNAPSHOT = "/datasets/v3/snapshot"    # /<sd_id>?format=json -> the rows
 BD_PATH_FILTER = "/datasets/filter"           # ⚠ NO /v3/ — the CORPUS query (discovery)
@@ -2501,8 +2595,7 @@ def _bd_profile(node, handle):
         "is_professional": _bd_flag(node, "is_professional_account"),
         "is_private": _bd_flag(node, "is_private"),
         "highlights_count": _ig_int(_first(node, "highlights_count")),
-        "bio_hashtags": _bd_list(node, "bio_hashtags"),
-        "post_hashtags": _bd_list(node, "post_hashtags"),
+        "bio_hashtags": _bd_list(node, "bio_hashtags"),
         "pronouns": str(_first(node, "pronouns", default="") or ""),
         # ⭐ 2026-08-07 (owner instruction: *"have ALL Fields available to us from Bright Data to
         # be pre-set Fields for us and populated"*) — THE REST OF THE PROFILES SCHEMA.
@@ -2558,8 +2651,13 @@ def _bd_post_identity(p):
     code = ig_shortcode(p.get("url")) if isinstance(p, dict) else ""
     if not code:
         return None
-    out = {"shortcode": code, "url": f"https://www.instagram.com/p/{code}/",
-           "type": _bd_type(p.get("content_type"), p.get("type"))}
+    kind = _bd_type(p.get("content_type"), p.get("type"))
+    # The URL route is part of Bright Data's endpoint contract. A Reel sent through the Posts
+    # scraper can return likes/comments while omitting its view fields; `/reel/{shortcode}` is
+    # the input documented for the Reels scraper.
+    route = "reel" if kind == "video" else "p"
+    out = {"shortcode": code, "url": f"https://www.instagram.com/{route}/{code}/",
+           "type": kind}
     caption = _first(p, "caption", "description", default="")
     if isinstance(caption, dict):
         caption = _first(caption, "text", default="")
@@ -2599,8 +2697,11 @@ def _bd_post_metrics(row):
         "caption": str(_first(row, "description", "caption", default="") or ""),
         "likes": _ig_int(_first(row, "likes", "like_count")),
         "comments": _ig_int(_first(row, "num_comments", "comment_count", "comments")),
-        "views": _ig_int(_first(row, "video_play_count", "video_view_count",
-                                "engagement_score_view", "play_count")),
+        # Reels expose views and plays separately (and the counts can differ materially). Keep
+        # them separate so a repeat-play count never wears the "Views" label.
+        "views": _ig_int(_first(row, "views", "view_count", "video_view_count",
+                                "engagement_score_view")),
+        "plays": _ig_int(_first(row, "video_play_count", "play_count")),
         # Sponsored-post detection. MEASURED populated (`True`, with the brand alongside) — and
         # it is the one field here that answers a commercial question the counts cannot.
         "paid_partnership": _bd_flag(row, "is_paid_partnership"),
@@ -2829,15 +2930,31 @@ def pull_profile_bd(url, max_posts=DEFAULT_POSTS_PER_PULL, post_metrics=False, l
         if got:
             posts.append(got)
 
-    if post_metrics and posts:
-        time.sleep(min(PACE_SECONDS, 1.0))    # the vendor is paid, but it is still someone's API
-        rows2, note2 = bd_scrape(BD_DS_POSTS, [p["url"] for p in posts])
-        if note2:
-            # The engagement half refusing does NOT lose the identity half — the profile and the
-            # post rows still land, and the run says which part was not readable.
-            note = f"post metrics unavailable: {note2}"
-        else:
-            by_code = {}
+    if post_metrics and posts:
+        time.sleep(min(PACE_SECONDS, 1.0))    # the vendor is paid, but it is still someone's API
+        # Posts and Reels are different Bright Data endpoints. Sending every permalink to the
+        # Posts endpoint was why Inayma's five video rows got likes/comments but no views: the
+        # documented Reels response is where `views` and `video_play_count` live. Split the batch
+        # without duplicating any record, so enabling post metrics still buys one row per post.
+        ordinary = [p for p in posts if p.get("type") != "video"]
+        videos = [p for p in posts if p.get("type") == "video"]
+        rows2, metric_notes = [], []
+        if ordinary:
+            got_rows, got_note = bd_scrape(BD_DS_POSTS, [p["url"] for p in ordinary])
+            rows2.extend(got_rows)
+            if got_note:
+                metric_notes.append(f"posts: {got_note}")
+        if videos:
+            got_rows, got_note = bd_scrape(BD_DS_REELS, [p["url"] for p in videos])
+            rows2.extend(got_rows)
+            if got_note:
+                metric_notes.append(f"reels: {got_note}")
+        if metric_notes and not rows2:
+            # The engagement half refusing does NOT lose the identity half — the profile and the
+            # post rows still land, and the run says which part was not readable.
+            note = f"post metrics unavailable: {'; '.join(metric_notes)}"
+        else:
+            by_code = {}
             for r in rows2:
                 got = _bd_post_metrics(r)
                 if got:
@@ -2848,8 +2965,10 @@ def pull_profile_bd(url, max_posts=DEFAULT_POSTS_PER_PULL, post_metrics=False, l
                     # The metrics row is RICHER (it knows posted_at); merge it over the identity
                     # row, dropping keys it could not read so a blank never overwrites a value.
                     p.update({k: v for k, v in extra.items() if v not in (None, "")})
-            if not by_code:
-                note = "post metrics were requested but the Posts dataset returned no rows"
+            if not by_code:
+                note = "post metrics were requested but the Posts dataset returned no rows"
+            elif metric_notes:
+                note = f"some post metrics were unavailable: {'; '.join(metric_notes)}"
 
     # IDENTITY WITHOUT MEDIA IS STILL `partial`, on the paid rung too. The rule does not soften
     # because we are paying: a run that wrote a follower count and no posts must not paint green
@@ -3300,7 +3419,6 @@ PRESET_PROFILE_FIELDS = [
     field_def("business_category", "Business category"),
     field_def("is_private", "Private account", "checkbox"),
     field_def("bio_hashtags", "Bio hashtags"),
-    field_def("post_hashtags", "Post hashtags"),
     field_def("pronouns", "Pronouns"),
     field_def("profile_name", "Profile name"),
     field_def("is_joined_recently", "Joined recently", "checkbox"),
@@ -3329,7 +3447,13 @@ PRESET_PROFILE_FIELDS = [
     # identically-named columns is unreadable. Caught by the preset set's own label-collision
     # gate rather than on screen, which is what that gate is for.
     field_def("posts_link", "Post rows", "link",
-              link={"table": "ut_ig_posts", "on": "influencer_key"}),
+              link={"table": IG_POSTS_TABLE, "on": "influencer_key"}),
+    field_def("profile_snapshots_link", "Profile history", "link",
+              link={"table": IG_SNAPSHOTS_TABLE, "on": "influencer_key"}),
+    field_def("post_snapshots_link", "Post measurement rows", "link",
+              link={"table": IG_POST_SNAPSHOTS_TABLE, "on": "influencer_key"}),
+    field_def("comments_link", "Comment rows", "link",
+              link={"table": IG_COMMENTS_TABLE, "on": "influencer_key"}),
     # ⚠ THE ROLLUPS READ `ut_ig_posts`' OWN LATEST COLUMNS, which is why those exist — a rollup
     # is ONE HOP (Airtable's rule and ours), and the engagement SERIES lives one table further
     # out in `ut_ig_post_snapshots`.
@@ -3339,20 +3463,36 @@ PRESET_PROFILE_FIELDS = [
     # has to name what makes one post later than another.
     field_def("avg_views_12", "Avg views · last 12 posts", "rollup",
               rollup={"link": "posts_link", "field": "views", "fn": "average",
-                      "limit": MAX_POSTS_PER_PULL, "sortBy": "posted_at", "sortDir": "desc"}),
+                      "limit": MAX_POSTS_PER_PULL, "sortBy": "posted_at", "sortDir": "desc",
+                      "distinctBy": "shortcode"}),
+    field_def("avg_plays_12", "Avg plays - last 12 posts", "rollup",
+              rollup={"link": "posts_link", "field": "plays", "fn": "average",
+                      "limit": MAX_POSTS_PER_PULL, "sortBy": "posted_at", "sortDir": "desc",
+                      "distinctBy": "shortcode"}),
     field_def("avg_likes_12", "Avg likes · last 12 posts", "rollup",
               rollup={"link": "posts_link", "field": "likes", "fn": "average",
-                      "limit": MAX_POSTS_PER_PULL, "sortBy": "posted_at", "sortDir": "desc"}),
+                      "limit": MAX_POSTS_PER_PULL, "sortBy": "posted_at", "sortDir": "desc",
+                      "distinctBy": "shortcode"}),
     field_def("avg_comments_12", "Avg comments · last 12 posts", "rollup",
               rollup={"link": "posts_link", "field": "comments", "fn": "average",
-                      "limit": MAX_POSTS_PER_PULL, "sortBy": "posted_at", "sortDir": "desc"}),
+                      "limit": MAX_POSTS_PER_PULL, "sortBy": "posted_at", "sortDir": "desc",
+                      "distinctBy": "shortcode"}),
     # ⭐ THE ONE HONEST POST COUNT WE HAVE. D-82: the vendor's `posts_count` is a FABRICATED ZERO
     # on the paid rung (49/49 rows measured), so it is discarded and that column reads blank after
     # a paid enrich. This counts the post rows actually captured — a different number and a true
     # one, which is why it gets its own column and its own label rather than quietly filling
     # `posts_count` with something that is not what that field means.
     field_def("posts_captured", "Posts captured", "rollup",
-              rollup={"link": "posts_link", "fn": "countall"}),
+              rollup={"link": "posts_link", "fn": "countall", "distinctBy": "shortcode"}),
+    field_def("profile_reads", "Profile reads", "rollup",
+              rollup={"link": "profile_snapshots_link", "fn": "countall",
+                      "distinctBy": "snapshot_key"}),
+    field_def("post_measurements_captured", "Post measurements captured", "rollup",
+              rollup={"link": "post_snapshots_link", "fn": "countall",
+                      "distinctBy": "post_snapshot_key"}),
+    field_def("comments_captured", "Comments captured", "rollup",
+              rollup={"link": "comments_link", "fn": "countall",
+                      "distinctBy": "comment_key"}),
     # ⭐ R3's stamp. WITHOUT IT THE WHOLE SET IS UNREADABLE: a blank `followers` means "never
     # enriched" and a stale one means "enriched in March", and no cell on the row can tell them
     # apart. It is the single field that turns the other fifteen from numbers into measurements.
@@ -3364,8 +3504,7 @@ PRESET_PROFILE_FIELDS = [
     # views/likes/comments at a `pulled_at`). This cell is a DERIVED window over those two,
     # rewritten each run, so a person reading the profile row can see the recent posts without a
     # join — and deleting it would cost a convenience, never a measurement. Shape: contract C2.
-    field_def("posts", "Recent posts", "json"),
-]
+]
 #: The keys the preset set owns — derived, so a field added above cannot be forgotten here.
 PRESET_PROFILE_KEYS = tuple(f["key"] for f in PRESET_PROFILE_FIELDS)
 
@@ -3385,7 +3524,7 @@ PRESET_FLAG_KEY = next((f["key"] for f in PRESET_PROFILE_FIELDS if f.get("profil
 #: the failure C1 exists to prevent. The five below are facts about the SEARCH (how often we found
 #: them, who for, and a human's decision) rather than about the profile, so they are discovery's
 #: and not part of the cross-tenant set.
-CANDIDATE_FIELDS = [
+CANDIDATE_FIELDS = [
     *PRESET_PROFILE_FIELDS,
     # ⭐ WAVE 26 · R3 — `first_found` / `last_found` ARE DATES, not ISO strings in a text cell.
     # They were written by `_iso()`, so the cell read `2026-08-05T14:03:11+07:00` and the owner
@@ -3408,7 +3547,153 @@ CANDIDATE_FIELDS = [
     # column: every candidate carried BOTH a `stage_<auto>` select saying where it was AND a
     # boolean saying whether it was kept — two progress fields that could disagree, with no rule
     # about which one won. The stage field is the one progress column. Nothing replaces this.
-]
+]
+
+
+def _profile_backlink_field(table_key, table_label, profile_key):
+    """A deterministic reciprocal link from one canonical IG table to one profile database.
+
+    The key includes the target table identity, so ten separate Profile databases can all point
+    into the same canonical Posts/Comments/history tables without one relation overwriting the
+    next. The join is derived in both directions and therefore needs no cross-table fan-out write.
+    """
+    digest = hashlib.sha1(str(table_key).encode("utf-8")).hexdigest()[:10]
+    return field_def(
+        f"profiles_{digest}", f"Profiles - {str(table_label or table_key)[:48]}", "link",
+        link={"table": str(table_key), "on": str(profile_key), "from": "influencer_key"},
+    )
+
+
+def _profile_schema_for(bound_key):
+    """Preset fields for an existing Profile table without inventing a second identity column."""
+    out = []
+    for field in PRESET_PROFILE_FIELDS:
+        item = dict(field)
+        if item.get("key") == PRESET_FLAG_KEY and bound_key != PRESET_FLAG_KEY:
+            item.pop("profile", None)
+            item.pop("pinned", None)
+        out.append(item)
+    return out
+
+
+def _reconcile_ig_graph_fields(rt, wanted_by_table, drop_by_table=None):
+    """Repair machine-owned IG field declarations in one coalesced store update.
+
+    `ut_ensure` deliberately merges only missing keys. That is correct for user columns but not
+    sufficient for a canonical relation: a stale `link.table`, rollup function, or type would
+    survive forever. This pass overwrites the contract keys for the fields we own, preserves
+    unrelated/user fields, and removes only explicitly retired preset columns plus their cells.
+    """
+    drop_by_table = drop_by_table or {}
+    changed = False
+
+    def _up(cur):
+        nonlocal changed
+        cur = cur if isinstance(cur, dict) else {}
+        for table_key, wanted_fields in wanted_by_table.items():
+            table = cur.get(table_key)
+            if table is None:
+                continue
+            wanted = {str(f.get("key")): dict(f) for f in wanted_fields}
+            drops = set(drop_by_table.get(table_key) or ())
+            fields, seen = [], set()
+            for stored in table.get("fields") or []:
+                key = str(stored.get("key") or "")
+                if key in drops:
+                    changed = True
+                    continue
+                desired = wanted.get(key)
+                if desired is None:
+                    fields.append(stored)
+                    continue
+                repaired = dict(stored)
+                repaired.update(desired)
+                # These bags define behaviour, not decoration. If the desired field does not use
+                # one, a stale bag from an earlier type must not remain attached to it.
+                for bag in ("link", "rollup"):
+                    if bag not in desired:
+                        repaired.pop(bag, None)
+                if repaired != stored:
+                    changed = True
+                fields.append(repaired)
+                seen.add(key)
+            for key, desired in wanted.items():
+                if key not in seen and not any(str(f.get("key") or "") == key for f in fields):
+                    fields.append(desired)
+                    changed = True
+            if drops:
+                for row in (table.get("rows") or {}).values():
+                    for key in drops:
+                        if key in row:
+                            row.pop(key, None)
+                            changed = True
+            table["fields"] = fields
+        return cur
+
+    # Avoid a commit when the declarations are already exact.
+    snapshot = ut_all(rt)
+    needs = False
+    for table_key, wanted_fields in wanted_by_table.items():
+        table = snapshot.get(table_key) or {}
+        by_key = {str(f.get("key") or ""): f for f in table.get("fields") or []}
+        drops = set(drop_by_table.get(table_key) or ())
+        if drops & set(by_key):
+            needs = True
+            break
+        for desired in wanted_fields:
+            stored = by_key.get(str(desired.get("key") or ""))
+            if stored is None or any(stored.get(k) != v for k, v in desired.items()):
+                needs = True
+                break
+        if needs:
+            break
+    if needs:
+        rt.update(UT_STORE_KEY, _up, flush="sync")
+    return changed
+
+
+def ensure_ig_graph(rt, username="automation", flow_tag="", profile_table="", profile_field=""):
+    """Ensure the ONE per-tenant Instagram relational graph and return its canonical keys.
+
+    Every enrichment path calls this function. Fixed table keys prevent a flow aimed at a new
+    Profile database from spawning `IG posts 2`; deterministic reciprocal link fields connect
+    each Profile database to the same Posts, Comments, profile-history, and post-history stores.
+    """
+    profile = ut_get(rt, profile_table) if profile_table else None
+    bound = str(profile_field or "").strip()
+    if profile is not None and not bound:
+        bound = next((str(f.get("key") or "") for f in profile.get("fields") or []
+                      if isinstance(f.get("profile"), dict)), "")
+    profile_label = str((profile or {}).get("label") or profile_table)
+    backlink = [_profile_backlink_field(profile_table, profile_label, bound)] \
+        if profile_table and profile is not None and bound else []
+
+    graph = {
+        IG_SNAPSHOTS_TABLE: ("IG snapshots", [*SNAPSHOT_FIELDS, *backlink]),
+        IG_POSTS_TABLE: ("IG posts", [*POST_FIELDS, *backlink]),
+        IG_POST_SNAPSHOTS_TABLE: ("IG post snapshots", [*POST_SNAPSHOT_FIELDS, *backlink]),
+        IG_COMMENTS_TABLE: ("IG comments", [*COMMENT_FIELDS, *backlink]),
+    }
+    keys = {}
+    for key, (label, fields) in graph.items():
+        keys[key] = ut_ensure(rt, label, fields, username, key=key, flow_tag=flow_tag,
+                              record_mode=AUTOMATION_RECORD_MODE, lock_fields=True)
+
+    # `ut_ensure` stamps newly created fields with the first flow that introduced them. Do not
+    # put that tag into the reconciliation template: these canonical tables are shared by many
+    # Profile databases, and otherwise every run would rewrite their provenance from flow A to
+    # flow B and back again.
+    wanted = {key: [dict(f) for f in fields]
+              for key, (_label, fields) in graph.items()}
+    drops = {IG_SNAPSHOTS_TABLE: {"post_hashtags"}}
+    if profile_table and profile is not None and bound:
+        profile_fields = _profile_schema_for(bound)
+        ut_ensure(rt, profile_label or profile_table, profile_fields, username,
+                  key=profile_table, flow_tag=flow_tag, lock_fields=True)
+        wanted[profile_table] = [dict(f) for f in profile_fields]
+        drops[profile_table] = {"posts", "post_hashtags"}
+    _reconcile_ig_graph_fields(rt, wanted, drops)
+    return keys
 
 
 # ── ⭐ WAVE 26 · THE MIGRATION (owner rulings R3 + R4/R5, contracts C1-a and C3) ───────────────
@@ -3558,7 +3843,8 @@ def migrate_ig_tables(rt, log=print, only=""):
              # ⭐ 2026-08-07 (owner ruling) — the primary-column half. Counted separately from
              # `stamped` because they answer different questions: that one is "how many ROWS got a
              # platform", these are "how many TABLES changed shape".
-             "pinned": 0, "flagged": 0, "droppedName": 0}
+             "pinned": 0, "flagged": 0, "droppedName": 0,
+             "droppedRecentPosts": 0, "droppedPostHashtags": 0}
     want = {f["key"]: f["type"] for f in CANDIDATE_FIELDS}
     # ⛔ `only` NARROWS THE SET, IT DOES NOT BYPASS THE TEST — and skipping that cost six red
     # checks the first time. `ut_ensure` calls this with the key it is ABOUT TO CREATE, so on a
@@ -3600,12 +3886,13 @@ def migrate_ig_tables(rt, log=print, only=""):
         other_profile = next((f for f in fields if f is not flag_f
                               and isinstance(f.get("profile"), dict)), None)
         want_pin = pin_f is not None and pin_f.get("pinned") is not True
-        want_flag = (flag_f is not None and other_profile is None
-                     and not isinstance(flag_f.get("profile"), dict))
-        vestigial = _vestigial_name_field(fields, rows)
-        if (not stale_keys and has_platform and not needs_stamp
-                and not want_pin and not want_flag and vestigial is None):
-            continue
+        want_flag = (flag_f is not None and other_profile is None
+                     and not isinstance(flag_f.get("profile"), dict))
+        vestigial = _vestigial_name_field(fields, rows)
+        retired = {f.get("key") for f in fields} & {"posts", "post_hashtags"}
+        if (not stale_keys and has_platform and not needs_stamp
+                and not want_pin and not want_flag and vestigial is None and not retired):
+            continue
         stats["tables"] += 1
         if want_pin:
             pin_f["pinned"] = True
@@ -3613,14 +3900,21 @@ def migrate_ig_tables(rt, log=print, only=""):
         if want_flag:
             flag_f["profile"] = {"source": PROFILE_SOURCE_IG}
             stats["flagged"] += 1
-        if vestigial is not None:
+        if vestigial is not None:
             # ⚠ THE CELLS GO WITH THE COLUMN. A row dict keeping a `name` key whose field no longer
             # exists is invisible everywhere except the next export, where it reappears as a column
             # nobody declared. `_vestigial_name_field` has already proven every one of them blank.
             fields = [f for f in fields if f is not vestigial]
             for r in rows.values():
                 r.pop("name", None)
-            stats["droppedName"] += 1
+            stats["droppedName"] += 1
+        if retired:
+            fields = [f for f in fields if f.get("key") not in retired]
+            for r in rows.values():
+                for key in retired:
+                    r.pop(key, None)
+            stats["droppedRecentPosts"] += int("posts" in retired)
+            stats["droppedPostHashtags"] += int("post_hashtags" in retired)
 
         for key in stale_keys:
             conv = _MIGRATE_CONVERT.get(key)
@@ -4598,7 +4892,7 @@ def capture_rows(res, pulled):
                      ("alt_text", 400)):
             if p.get(k):
                 ident[k] = _s(p.get(k), n)
-        metrics = {k: p.get(k) for k in ("likes", "comments", "views")}
+        metrics = {k: p.get(k) for k in ("likes", "comments", "views", "plays")}
         # ⭐⭐ 2026-08-07 — THE LATEST ENGAGEMENT VALUES, ONTO THE POST ROW ITSELF.
         #
         # This is what keeps a rollup at ONE HOP (Airtable's rule and ours): without it, "average
@@ -4624,11 +4918,12 @@ def capture_rows(res, pulled):
             ident["measured_at"] = _day(pulled)
         idents.append(ident)
         if any(v is not None for v in metrics.values()):
-            metrics_rows.append({
-                "post_snapshot_key": f"{p['shortcode']}@{pulled}",
-                "shortcode": p["shortcode"], "pulled_at": pulled,
-                "likes": _s(metrics["likes"]), "comments": _s(metrics["comments"]),
-                "views": _s(metrics["views"])})
+            metrics_rows.append({
+                "post_snapshot_key": f"{p['shortcode']}@{pulled}",
+                "shortcode": p["shortcode"], "influencer_key": prof.get("username"),
+                "pulled_at": pulled,
+                "likes": _s(metrics["likes"]), "comments": _s(metrics["comments"]),
+                "views": _s(metrics["views"]), "plays": _s(metrics["plays"])})
     return snap_row, idents, metrics_rows
 
 
@@ -4651,7 +4946,7 @@ PRESET_FROM_PROFILE = {
     # `PRESET_PROFILE_KEYS` entry must either be written by this map or be explicitly declared
     # as written elsewhere (`platform`, `enriched_at`, `posts` are stamped by `preset_cells`).
     "business_category": "business_category", "is_private": "is_private",
-    "bio_hashtags": "bio_hashtags", "post_hashtags": "post_hashtags",
+    "bio_hashtags": "bio_hashtags",
     "pronouns": "pronouns", "profile_name": "profile_name",
     "is_joined_recently": "is_joined_recently", "has_channel": "has_channel",
     "partner_id": "partner_id", "external_url_title": "external_url_title",
@@ -4666,95 +4961,15 @@ PRESET_FROM_PROFILE = {
 #: ⚠ The five relational columns are written by `compute_relation_cells` on the tick, NOT by an
 #: enrichment run — which is the whole point of them: they stay true when the LINKED table
 #: changes, and a pull that touched no profile still updates a profile's post count.
-PRESET_WRITTEN_ELSEWHERE = ("platform", "handle", "enriched_at", "posts",
-                            "posts_link", "avg_views_12", "avg_likes_12", "avg_comments_12",
-                            "posts_captured")
-
-
-#: The keys one post carries into the window. Deliberately SHORTER than `POST_FIELDS`: this is a
-#: readable summary on a profile row, not a second copy of the post record. `ut_ig_posts` keeps
-#: everything, including the fields nobody reads at a glance (`alt_text`, `hashtags`, `partner`).
-_WINDOW_POST_KEYS = ("shortcode", "url", "posted_at", "type", "caption")
-#: The engagement half, present only when the PAID rung bought it (R2). ABSENT, never zero.
-_WINDOW_METRIC_KEYS = ("views", "likes", "comments")
-#: A caption is the one unbounded field here, and 32 KB is the cell ceiling. Bounded per post so
-#: the window cannot approach it: 12 posts x 280 chars is ~4 KB, comfortably inside `MAX_JSON_CELL`.
-WINDOW_CAPTION_CHARS = 280
-#: ⛔ THE FIELD LAYER'S CEILING, READ LAZILY — AND THE LAZINESS IS NOT STYLE.
-#:
-#: `core.user_tables.MAX_JSON_CELL` is what actually refuses an oversized cell, so reading it beats
-#: restating 32 KB here (a local copy drifts the day that constant moves, and the engine would then
-#: build a document the door rejects).
-#: ⚠ BUT IMPORTING IT AT MODULE LEVEL CHANGED BEHAVIOUR SOMEWHERE ELSE ENTIRELY. Pulling
-#: `core.user_tables` in at import time initialises the store layer earlier than this module used
-#: to, and a gate asserting "the upsert ran THREE times for the WHOLE run" went to FOUR: the
-#: platform master's write had been failing before its first upsert and now failed after it. Same
-#: outcome, one more call against a 256-commits/hr budget — caught by a COUNT, invisible to every
-#: behavioural check. A top-level import is a side effect; this one buys a constant.
-def _max_json_cell():
-    try:
-        from core.user_tables import MAX_JSON_CELL
-        return MAX_JSON_CELL
-    except Exception:                                             # noqa: BLE001
-        return 32 * 1024
-
-
-def posts_window(res, pulled):
-    """⭐ WAVE 26 · R1 / contract C2 — the last-N posts, as a DERIVED json cell.
+PRESET_WRITTEN_ELSEWHERE = (
+    "platform", "handle", "enriched_at",
+    "posts_link", "profile_snapshots_link", "post_snapshots_link", "comments_link",
+    "avg_views_12", "avg_plays_12", "avg_likes_12", "avg_comments_12",
+    "posts_captured", "profile_reads", "post_measurements_captured", "comments_captured",
+)
 
-    ⛔ THIS IS A VIEW, AND R3 IS UNTOUCHED. The authoritative post record is `ut_ig_posts` (keyed
-    by shortcode, so it ACCUMULATES across runs) and the authoritative engagement series is
-    `ut_ig_post_snapshots` (one appended row per pull, carrying views/likes/comments at a
-    `pulled_at`). Nothing here is a second home for either: the cell is rewritten every run, and
-    deleting it would cost a convenience rather than a measurement. That is exactly what makes it
-    compatible with *"one store for one series"* — the owner asked for the posts to be visible ON
-    the profile row AND to keep accumulating, and those are two different jobs.
 
-    ⚠ AN ABSENT METRIC IS AN ABSENT KEY, NEVER A ZERO. `postMetrics` is off by default (R2 — it
-    buys one extra vendor record per post), so most windows carry identity and no engagement. A
-    `views: 0` would claim we measured a post nobody watched; the key simply is not there, and the
-    client renders nothing for it. This module's blank-never-zero law, at one more seam.
-    """
-    posts = [p for p in ((res or {}).get("posts") or []) if isinstance(p, dict)]
-    if not posts:
-        return ""
-    out = []
-    for p in posts:
-        row = {}
-        for k in _WINDOW_POST_KEYS:
-            v = p.get(k)
-            if v is None or str(v).strip() == "":
-                continue
-            row[k] = _s(v, WINDOW_CAPTION_CHARS) if k == "caption" else _s(v, 300)
-        for k in _WINDOW_METRIC_KEYS:
-            v = p.get(k)
-            if v is None or str(v).strip() == "":
-                continue                      # NOT read -> not a key. Never 0.
-            n = _ig_int(v)
-            if n is not None:
-                row[k] = n
-        if row:
-            out.append(row)
-    if not out:
-        return ""
-    # Newest first, so "the last 10 posts" reads top-down. A post with no date sorts last rather
-    # than first: an unknown date is not a recent one.
-    out.sort(key=lambda r: str(r.get("posted_at") or ""), reverse=True)
-    doc = {"n": len(out), "metrics": any(k in r for r in out for k in _WINDOW_METRIC_KEYS),
-           "as_of": _day(pulled), "posts": out}
-    text = json.dumps(doc, ensure_ascii=False, separators=(",", ":"))
-    # ⛔ THE CELL CEILING IS THE FIELD LAYER'S, and blowing it would have the write REFUSED — so
-    # the window sheds posts until it fits rather than losing the whole cell. Shedding is safe
-    # precisely because this is a view: `ut_ig_posts` still holds every one of them.
-    ceiling = _max_json_cell()
-    while len(text.encode("utf-8")) > ceiling and len(doc["posts"]) > 1:
-        doc["posts"] = doc["posts"][:-1]
-        doc["n"] = len(doc["posts"])
-        text = json.dumps(doc, ensure_ascii=False, separators=(",", ":"))
-    return text
-
-
-def preset_cells(res, pulled):
+def preset_cells(res, pulled):
     """C4/R3: one pull → the LATEST-value cells written onto the enriched record.
 
     ⛔ `enriched_at` IS ALWAYS WRITTEN when a pull succeeded, and it is the field that makes the
@@ -4776,9 +4991,6 @@ def preset_cells(res, pulled):
     # arrived with no platform at all — and a blank half of the dedup key is how one account
     # becomes two rows.
     cells["platform"] = PLATFORM_INSTAGRAM
-    win = posts_window(res, pulled)
-    if win:
-        cells["posts"] = win
     # R3: `enriched_at` is a `date` column now. It answers "how stale is this number", which is a
     # question in days; the full stamp keeps its precision on the snapshot series.
     cells["enriched_at"] = _day(pulled)
@@ -4816,21 +5028,13 @@ def run_field_instagram(rt, defn, username="automation", log=print, step=_no_ste
     # the relational tables the pull lands in (R7): every row timestamped for time-range filters
     if dry:
         # The Write node is off: resolve the keys, create nothing. (`ut_ensure` writes.)
-        snap_key, post_key, ps_key = ("ut_ig_snapshots", "ut_ig_posts", "ut_ig_post_snapshots")
-    else:
-        snap_key = ut_ensure(rt, "IG snapshots", SNAPSHOT_FIELDS, username,
-                             key="ut_ig_snapshots", flow_tag=str(defn.get("id") or ""))
-        post_key = ut_ensure(rt, "IG posts", POST_FIELDS, username, key="ut_ig_posts",
-                             flow_tag=str(defn.get("id") or ""))
-        ps_key = ut_ensure(rt, "IG post snapshots", POST_SNAPSHOT_FIELDS, username,
-                           key="ut_ig_post_snapshots", flow_tag=str(defn.get("id") or ""))
-        # ⭐ 2026-08-07 — the COMMENT database, spawned beside the other three so a post's
-        # `comments_link` names a table that exists rather than one that might. It stays EMPTY
-        # until comment capture is switched on — the same honest state `ut_ig_post_snapshots`
-        # holds while `postMetrics` is off, which is why it is ensured on the same line rather
-        # than being made conditional.
-        ut_ensure(rt, "IG comments", COMMENT_FIELDS, username, key="ut_ig_comments",
-                  flow_tag=str(defn.get("id") or ""))
+        snap_key, post_key, ps_key = (IG_SNAPSHOTS_TABLE, IG_POSTS_TABLE,
+                                      IG_POST_SNAPSHOTS_TABLE)
+    else:
+        graph = ensure_ig_graph(rt, username, str(defn.get("id") or ""),
+                                profile_table=table_key)
+        snap_key, post_key, ps_key = (graph[IG_SNAPSHOTS_TABLE], graph[IG_POSTS_TABLE],
+                                      graph[IG_POST_SNAPSHOTS_TABLE])
     missing = [] if dry else ut_missing(rt, snap_key, post_key, ps_key)
     snaps = dict((ut_get(rt, snap_key) or {}).get("rows") or {})
     posts = dict((ut_get(rt, post_key) or {}).get("rows") or {})
@@ -4902,7 +5106,9 @@ def run_field_instagram(rt, defn, username="automation", log=print, step=_no_ste
 
     # --- THE THREE UPSERTS. Once each, over the whole run's accumulated rows.
     snaps, c_snap = upsert_rows(snaps, in_snaps, "snapshot_key", cap=row_cap(snap_key))
-    posts, c_post = upsert_rows(posts, in_posts, "shortcode", cap=row_cap(post_key))
+    posts, collapsed_posts = dedupe_canonical_rows(posts, "shortcode", newest_by="measured_at")
+    posts, c_post = upsert_rows(posts, in_posts, "shortcode", cap=row_cap(post_key))
+    c_post["duplicates"] += collapsed_posts
     psnaps, c_ps = upsert_rows(psnaps, in_psnaps, "post_snapshot_key", cap=row_cap(ps_key))
     counts["new_posts"] = c_post["inserted"]
     capped = [(snap_key, c_snap["capped"]), (post_key, c_post["capped"]),
@@ -4936,14 +5142,15 @@ def run_field_instagram(rt, defn, username="automation", log=print, step=_no_ste
                     tt.setdefault("rows", {}).setdefault(str(rid), {})[fkey] = val
                 for rid, vals in stage_writes.items():
                     tt.setdefault("rows", {}).setdefault(str(rid), {}).update(vals)
-            for k, rws in ((snap_key, snaps), (post_key, posts), (ps_key, psnaps)):
-                tgt = cur.get(k)
-                if tgt is not None:
-                    tgt["rows"] = rws
-            return cur
-
-        rt.update(UT_STORE_KEY, _up, flush="sync")  # ONE coalesced update for all four tables
-        if lanes:
+            for k, rws in ((snap_key, snaps), (post_key, posts), (ps_key, psnaps)):
+                tgt = cur.get(k)
+                if tgt is not None:
+                    tgt["rows"] = rws
+            _refresh_relations_inplace(cur, log=log)
+            return cur
+
+        rt.update(UT_STORE_KEY, _up, flush="sync")  # ONE coalesced update for all four tables
+        if lanes:
             ensure_stage_field(rt, table_key, defn, username)
 
     # --- C6 (R2): WRITE-THROUGH to the platform master. Three postures, never conflated:
@@ -5191,10 +5398,10 @@ def run_discover_instagram(rt, defn, username="automation", log=print, step=_no_
     # down the rows path and parses as one unusable row instead of an error.
     counts["dropped"] = max(0, len(rows) - len(incoming))
     label = cfg.get("targetLabel") or "IG candidates"
-    table_key = (ut_key_for(label, cfg.get("targetTable") or DISCOVER_TABLE) if dry
-                 else ut_ensure(rt, label, CANDIDATE_FIELDS, username,
-                                key=cfg.get("targetTable") or DISCOVER_TABLE,
-                                flow_tag=str(defn.get("id") or "")))
+    table_key = (ut_key_for(label, cfg.get("targetTable") or DISCOVER_TABLE) if dry
+                 else ut_ensure(rt, label, CANDIDATE_FIELDS, username,
+                                key=cfg.get("targetTable") or DISCOVER_TABLE,
+                                flow_tag=str(defn.get("id") or ""), lock_fields=True))
     existing = dict((ut_get(rt, table_key) or {}).get("rows") or {})
     # --- ⭐ WAVE 26 · C3 / owner ruling R4 — THE UPSERT KEY IS `(platform, handle)`.
     #
@@ -6873,7 +7080,7 @@ def enrich_selection(rt, table_key, cfg, profile_key, today=None):
     return picked, "; ".join(notes)
 
 
-def _has_action(actions, kind):
+def _has_action(actions, kind):
     """Does this flow contain `kind` ANYWHERE, forks included?
 
     ⛔ FORKS ARE THE WHOLE REASON THIS IS A FUNCTION. A group's children live under
@@ -6890,7 +7097,20 @@ def _has_action(actions, kind):
         for br in ((a.get("config") or {}).get("branches") or []):
             if _has_action((br or {}).get("actions") or [], kind):
                 return True
-    return False
+    return False
+
+
+def _actions_of_kind(actions, kind):
+    """Every action of `kind`, including actions nested inside If branches."""
+    out = []
+    for action in actions or []:
+        if not isinstance(action, dict):
+            continue
+        if action.get("kind") == kind:
+            out.append(action)
+        for branch in ((action.get("config") or {}).get("branches") or []):
+            out.extend(_actions_of_kind((branch or {}).get("actions") or [], kind))
+    return out
 
 
 def apply_actions(rt, defn, table_key, row_ids, username="automation", log=print):
@@ -6960,20 +7180,33 @@ def apply_actions(rt, defn, table_key, row_ids, username="automation", log=print
     # step at it, and (b) add a SECOND profile column to a table that already flagged a different
     # one, which `user_tables` forbids at both write doors. A table with no profile column is an
     # UNBOUND enrich: it must stay unbound and say so, which is a different defect (D-79) with its
-    # own honest refusal, not something to paper over by inventing the binding.
-    _tbl_now = ut_get(rt, table) or {}
-    _bound = next((f for f in (_tbl_now.get("fields") or [])
-                   if isinstance(f.get("profile"), dict)), None)
-    if _bound and _has_action(actions, "enrich_instagram"):
+    # own honest refusal, not something to paper over by inventing the binding.
+    _tbl_now = ut_get(rt, table) or {}
+    _enrich_actions = _actions_of_kind(actions, "enrich_instagram")
+    _bound = next((f for f in (_tbl_now.get("fields") or [])
+                   if isinstance(f.get("profile"), dict)), None)
+    if _bound is None:
+        _named = next((str((a.get("config") or {}).get("profileField") or "").strip()
+                       for a in _enrich_actions
+                       if str((a.get("config") or {}).get("profileField") or "").strip()), "")
+        _bound = next((f for f in (_tbl_now.get("fields") or [])
+                       if str(f.get("key") or "") == _named), None)
+    if _bound and _enrich_actions:
         # ⚠ AND THE DECLARATIONS ARE STRIPPED FROM ANYTHING NEW. The binding already exists —
         # `_bound` is it — so a preset arriving now must carry DATA, never a second identity: a
         # table whose profile column is `ig_handle` would otherwise gain a rival `handle`.
         topup = [({k: v for k, v in f.items() if k not in ("profile", "pinned")}
                   if f.get("key") != _bound.get("key") else dict(f))
                  for f in PRESET_PROFILE_FIELDS]
-        try:
-            ut_ensure(rt, _tbl_now.get("label") or table, topup, username, key=table,
-                      flow_tag=str(defn.get("id") or ""))
+        try:
+            if any(not bool((a.get("config") or {}).get("dryRun")) for a in _enrich_actions):
+                ensure_ig_graph(rt, username, str(defn.get("id") or ""),
+                                profile_table=table, profile_field=str(_bound.get("key") or ""))
+            else:
+                # Dry run keeps its original no-create contract; only the already-established
+                # Profile table is topped up, and no canonical related database is spawned.
+                ut_ensure(rt, _tbl_now.get("label") or table, topup, username, key=table,
+                          flow_tag=str(defn.get("id") or ""), lock_fields=True)
         except Exception as exc:                                      # noqa: BLE001
             # A schema top-up that cannot run must not stop the enrichment: the cells that DO
             # have columns still land, which is strictly better than the pull being thrown away.
@@ -7256,32 +7489,30 @@ def _enrich_flush(rt, defn, username, acc, log):
         if acc.get("dry"):
             out["enrichDryRun"] = acc["profiles"]
         return out
-    tag = str(defn.get("id") or "")
-    snap_key = ut_ensure(rt, "IG snapshots", SNAPSHOT_FIELDS, username,
-                         key="ut_ig_snapshots", flow_tag=tag)
-    post_key = ut_ensure(rt, "IG posts", POST_FIELDS, username, key="ut_ig_posts", flow_tag=tag)
-    ps_key = ut_ensure(rt, "IG post snapshots", POST_SNAPSHOT_FIELDS, username,
-                       key="ut_ig_post_snapshots", flow_tag=tag)
-    # ⭐ 2026-08-07 — the COMMENT database. Spawned on BOTH enrich paths, not just one: the two
-    # writers already ensure the same three tables independently, and a fourth that only one of
-    # them knows about is how a database exists for automations created after Tuesday.
-    ut_ensure(rt, "IG comments", COMMENT_FIELDS, username, key="ut_ig_comments", flow_tag=tag)
+    tag = str(defn.get("id") or "")
+    profile_table = str(_flow_table(defn) or "")
+    graph = ensure_ig_graph(rt, username, tag, profile_table=profile_table)
+    snap_key, post_key, ps_key = (graph[IG_SNAPSHOTS_TABLE], graph[IG_POSTS_TABLE],
+                                  graph[IG_POST_SNAPSHOTS_TABLE])
     missing = ut_missing(rt, snap_key, post_key, ps_key)
     snaps, c_snap = upsert_rows(dict((ut_get(rt, snap_key) or {}).get("rows") or {}),
                                 acc["snaps"], "snapshot_key", cap=row_cap(snap_key))
-    posts, c_post = upsert_rows(dict((ut_get(rt, post_key) or {}).get("rows") or {}),
-                                acc["posts"], "shortcode", cap=row_cap(post_key))
+    old_posts, collapsed_posts = dedupe_canonical_rows(
+        dict((ut_get(rt, post_key) or {}).get("rows") or {}), "shortcode", newest_by="measured_at")
+    posts, c_post = upsert_rows(old_posts, acc["posts"], "shortcode", cap=row_cap(post_key))
+    c_post["duplicates"] += collapsed_posts
     psnaps, c_ps = upsert_rows(dict((ut_get(rt, ps_key) or {}).get("rows") or {}),
                                acc["psnaps"], "post_snapshot_key", cap=row_cap(ps_key))
 
     def _up(cur):
         cur = cur if isinstance(cur, dict) else {}
-        for k, rws in ((snap_key, snaps), (post_key, posts), (ps_key, psnaps)):
-            if cur.get(k) is not None:
-                cur[k]["rows"] = rws
-        return cur
-
-    rt.update(UT_STORE_KEY, _up, flush="sync")     # ONE coalesced update for all three tables
+        for k, rws in ((snap_key, snaps), (post_key, posts), (ps_key, psnaps)):
+            if cur.get(k) is not None:
+                cur[k]["rows"] = rws
+        _refresh_relations_inplace(cur, log=log)
+        return cur
+
+    rt.update(UT_STORE_KEY, _up, flush="sync")     # ONE coalesced update for all three tables
     # LOUD, never silent (D-11): a full append table means the SERIES has stopped growing, which
     # is the failure a chart cannot show you.
     capped = c_snap["capped"] + c_post["capped"] + c_ps["capped"]
@@ -8067,7 +8298,7 @@ def _ut():
 
     Pulling this module in at import time initialises the store layer earlier than
     `automation_engine` used to, and the last time that happened it moved a store-commit COUNT
-    from three to four on an unrelated gate (see `_max_json_cell`'s note). The relational pass
+    from three to four on an unrelated gate. The relational pass
     needs three constants and two predicates from the field layer; it does not need to change
     when this module is imported.
     """
@@ -8121,7 +8352,7 @@ def _sort_key(value, ftype):
     return (raw.lower(),)
 
 
-def _rollup_fold(fn, values):
+def _rollup_fold(fn, values):
     """`values` (raw cells, in the order the window kept them) → the aggregate, as a STRING.
 
     Returns `""` for "nothing to aggregate", NEVER `0`. ⛔ That distinction is this module's
@@ -8131,8 +8362,13 @@ def _rollup_fold(fn, values):
     ⚠ The ONE exception is the count family, where zero IS the answer — "how many linked records"
     over an empty set is genuinely 0, not unknown.
     """
-    if fn == "countall":
-        return str(len(values))
+    if fn == "countall":
+        return str(len(values))
+    if fn == "latest":
+        # Ordering belongs to the rollup bag (`sortBy` is mandatory for this function). Preserve
+        # a blank on the newest row rather than reaching backwards and presenting an older value
+        # as current.
+        return "" if not values or values[0] is None else str(values[0])[:ROLLUP_MAX_CHARS]
     if fn == "counta":
         return str(len([v for v in values if str(v or "").strip() != ""]))
     if fn == "count":
@@ -8187,7 +8423,7 @@ def _link_from_key(fields, bag):
     return str(pin.get("key") or "") if pin else ""
 
 
-def _linked_rows_by_join(linked, on_key):
+def _linked_rows_by_join(linked, on_key):
     """`{join value (lower-cased) -> [(row_id, row)]}` over one linked table, built ONCE.
 
     ⚠ Lower-cased because the join values this exists for are Instagram handles, which the
@@ -8199,16 +8435,46 @@ def _linked_rows_by_join(linked, on_key):
         k = str((row or {}).get(on_key) or "").strip().lower()
         if k:
             idx.setdefault(k, []).append((str(rid), row or {}))
-    return idx
-
-
-def compute_relation_cells(rt, table_key):
+    return idx
+
+
+def _rollup_condition_matches(row, condition, field_types):
+    """Evaluate one Airtable-style linked-record condition against a candidate row."""
+    field = str((condition or {}).get("field") or "")
+    op = str((condition or {}).get("op") or "")
+    raw = (row or {}).get(field)
+    text = str(raw or "").strip()
+    if op == "is_empty":
+        return text == ""
+    if op == "is_not_empty":
+        return text != ""
+    wanted = str((condition or {}).get("value") or "").strip()
+    if op == "contains":
+        return wanted.casefold() in text.casefold()
+    if op == "not_contains":
+        return wanted.casefold() not in text.casefold()
+    if op in ("eq", "neq"):
+        left_num, right_num = _lane_num(text), _lane_num(wanted)
+        equal = (left_num == right_num if left_num is not None and right_num is not None
+                 else text.casefold() == wanted.casefold())
+        return equal if op == "eq" else not equal
+    ftype = (field_types or {}).get(field, "text")
+    left = _sort_key(text, ftype)
+    right = _sort_key(wanted, ftype)
+    if left is None or right is None:
+        return False
+    return ((op == "gt" and left > right) or (op == "gte" and left >= right)
+            or (op == "lt" and left < right) or (op == "lte" and left <= right))
+
+
+def compute_relation_cells(rt, table_key, tables=None):
     """Recompute every DERIVED LINK cell and every ROLLUP cell on ONE table. Returns rows touched.
 
     Zero store reads when the table declares neither kind — the same cheap-by-construction shape
     `compute_metric_cells` has, so walking every table on a tick costs a dict scan per table.
     """
-    t = ut_get(rt, table_key)
+    store = tables if tables is not None else ut_all(rt)
+    t = (store or {}).get(table_key)
     fields = list((t or {}).get("fields") or [])
     links = [f for f in fields if _ut().is_derived_link(f)]
     rollups = [f for f in fields if isinstance(f.get("rollup"), dict)]
@@ -8225,10 +8491,23 @@ def compute_relation_cells(rt, table_key):
         if not lk_key or lk_key in resolved or not isinstance((f or {}).get("link"), dict):
             continue
         bag = f["link"]
-        linked = ut_get(rt, str(bag.get("table") or "")) or {}
+        linked = (store or {}).get(str(bag.get("table") or "")) or {}
         linked_types[lk_key] = {str(lf.get("key")): str(lf.get("type") or "text")
                                 for lf in (linked.get("fields") or [])}
-        if bag.get("on"):
+        if bag.get("inverse"):
+            # Airtable's reciprocal side: this row is linked to every SOURCE row whose ordinary
+            # link cell contains this row id. The source cell remains the one relationship truth.
+            source_rows = linked.get("rows") or {}
+            inverse_key = str(bag.get("inverse") or "")
+            inverse_index = {}
+            for source_id, source_row in source_rows.items():
+                for target_id in [part.strip() for part in
+                                  str((source_row or {}).get(inverse_key) or "").split(",")]:
+                    if target_id:
+                        inverse_index.setdefault(target_id, []).append(
+                            (str(source_id), source_row or {}))
+            resolved[lk_key] = {str(rid): inverse_index.get(str(rid), []) for rid in rows}
+        elif bag.get("on"):
             idx = _linked_rows_by_join(linked, str(bag["on"]))
             from_key = _link_from_key(fields, bag)
             resolved[lk_key] = {
@@ -8261,15 +8540,25 @@ def compute_relation_cells(rt, table_key):
             want = ",".join(i for i, _r in hits[:_ut().LINK_MAX_IDS])
             if str(row.get(fk, "")) != want:
                 changes.setdefault(rid, {})[fk] = want
-        for f in rollups:
+        for f in rollups:
             fk, bag = str(f["key"]), f["rollup"]
             lk_key = str(bag.get("link") or "")
-            hits = list((resolved.get(lk_key) or {}).get(rid) or [])
+            hits = list((resolved.get(lk_key) or {}).get(rid) or [])
             # ⚠ A rollup whose link field does not exist (renamed, deleted) resolves to NOTHING
             # and therefore to a blank cell — never to a stale number. A column that keeps
             # printing yesterday's answer after its input is gone is the worst of the options.
-            sort_by = str(bag.get("sortBy") or "")
-            if sort_by:
+            conditions = list(bag.get("conditions") or [])
+            if conditions:
+                matches = lambda pair: [
+                    _rollup_condition_matches(pair[1], condition,
+                                               linked_types.get(lk_key) or {})
+                    for condition in conditions]
+                if str(bag.get("conditionConj") or "and") == "or":
+                    hits = [pair for pair in hits if any(matches(pair))]
+                else:
+                    hits = [pair for pair in hits if all(matches(pair))]
+            sort_by = str(bag.get("sortBy") or "")
+            if sort_by:
                 ftype = (linked_types.get(lk_key) or {}).get(sort_by, "text")
                 # ⛔ PARTITION, THEN SORT. A row whose sort cell is blank or unparseable is not
                 # rankable, and it must land at the END whichever direction is asked for — which
@@ -8280,8 +8569,21 @@ def compute_relation_cells(rt, table_key):
                 rankable = [(p, k) for p, k in keyed if k is not None]
                 rankable.sort(key=lambda pk: pk[1],
                               reverse=str(bag.get("sortDir") or "desc") == "desc")
-                hits = [p for p, _k in rankable] + [p for p, k in keyed if k is None]
-            limit = int(bag.get("limit") or 0)
+                hits = [p for p, _k in rankable] + [p for p, k in keyed if k is None]
+            distinct_by = str(bag.get("distinctBy") or "")
+            if distinct_by:
+                seen, unique = set(), []
+                for pair in hits:
+                    identity = str(pair[1].get(distinct_by) or "").strip().lower()
+                    # A blank is not an identity. Keep it rather than collapsing every unknown
+                    # record into one synthetic duplicate.
+                    if identity and identity in seen:
+                        continue
+                    if identity:
+                        seen.add(identity)
+                    unique.append(pair)
+                hits = unique
+            limit = int(bag.get("limit") or 0)
             if limit:
                 hits = hits[:limit]
             src = str(bag.get("field") or "")
@@ -8289,10 +8591,20 @@ def compute_relation_cells(rt, table_key):
                                 [r.get(src) for _i, r in hits] if src else [1] * len(hits))
             if str(row.get(fk, "")) != want:
                 changes.setdefault(rid, {})[fk] = want
-    if not changes:
-        return 0
-
-    def _up(cur):
+    if not changes:
+        return 0
+
+    # A run that just wrote linked rows passes its in-flight user_tables bucket here so the
+    # relation refresh joins the SAME coalesced commit. The standalone/tick path below keeps
+    # the public helper's old persist-on-change behaviour.
+    if tables is not None:
+        tt = tables.get(table_key)
+        if tt is not None:
+            for r, vals in changes.items():
+                tt.setdefault("rows", {}).setdefault(r, {}).update(vals)
+        return len(changes)
+
+    def _up(cur):
         cur = cur if isinstance(cur, dict) else {}
         tt = cur.get(table_key)
         if tt is not None:
@@ -8304,28 +8616,44 @@ def compute_relation_cells(rt, table_key):
     return len(changes)
 
 
-def refresh_relations(rt, log=print):
+def _refresh_relations_inplace(tables, log=print):
+    """Refresh every relation against one mutable user_tables bucket; perform no store write."""
+    touched = 0
+    for tk, table in list((tables or {}).items()):
+        fields = (table or {}).get("fields") or []
+        if not any(f.get("type") == "rollup" or _ut().is_derived_link(f) for f in fields):
+            continue
+        try:
+            touched += compute_relation_cells(None, tk, tables=tables)
+        except Exception as e:                                    # noqa: BLE001
+            log(f"[aios-auto] relation refresh {tk} failed: {type(e).__name__}: {e}")
+    return touched
+
+
+def refresh_relations(rt, log=print):
     """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.
     """
-    touched = 0
-    # ⚠ FILTER OFF THE DICT ALREADY IN HAND, exactly as `refresh_metrics` does. Calling
-    # `compute_relation_cells` for every table would make it re-`ut_get` each one just to
-    # discover it declares neither kind — so the "costs one dict scan" claim above is only true
-    # with this line, and without it the sibling two functions down does strictly less work for
-    # the same job.
-    for tk, t in ut_all(rt).items():
-        fs = (t or {}).get("fields") or []
-        if not any(f.get("type") == "rollup" or _ut().is_derived_link(f) for f in fs):
-            continue
-        try:
-            touched += compute_relation_cells(rt, tk)
-        except Exception as e:                                    # noqa: BLE001
-            log(f"[aios-auto] relation refresh {tk} failed: {type(e).__name__}: {e}")
-    return touched
+    snapshot = {
+        str(key): {**(table or {}),
+                   "rows": {str(rid): dict(row or {})
+                            for rid, row in ((table or {}).get("rows") or {}).items()}}
+        for key, table in (ut_all(rt) or {}).items()
+    }
+    if not _refresh_relations_inplace(snapshot, log=log):
+        return 0
+    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="sync")
+    return actual[0]
 
 
 def refresh_metrics(rt, today=None, log=print):

api/routes_tables.py

@@ -36,6 +36,17 @@ def _ops(session, table_key):
     return table_store.make(f"{table_key}_table_workspace", st=session.runtime)
 
 
+def _refresh_relations(session):
+    """Refresh reciprocal Links and Rollups after a human record/schema mutation."""
+    import automation_engine as engine
+    try:
+        engine.refresh_relations(session.runtime, log=lambda *_args: None)
+    except Exception as exc:                                      # noqa: BLE001
+        # The source write already landed. A later automation/tick will repair materialised
+        # cells; never answer 503 and invite the browser to repeat a successful mutation.
+        print(f"[tables] relation refresh deferred: {type(exc).__name__}: {exc}")
+
+
 def _defn_or_refuse(session, table_key):
     """The per-table wall: 404 for a key that does not exist, 403 for one this session may not
     open. 404-before-403 leaks nothing useful — ut keys are guessable slugs, and 'exists but
@@ -49,6 +60,16 @@ def _defn_or_refuse(session, table_key):
     return defn
 
 
+def _records_or_refuse(session, table_key):
+    """The human record-write wall for a database the automation engine owns."""
+    defn = _defn_or_refuse(session, table_key)
+    if not _ut().records_mutable(table_key, st=session.runtime):
+        raise err(403, "records_read_only",
+                  "records in this automation-owned database are read-only — add Instagram "
+                  "handles in a Profile database and let enrichment populate this database")
+    return defn
+
+
 #: Tables whose ROWS ARE PER-USER (D-39, wave 22 C6's view half): a non-admin sees the rows
 #: they created plus any row nobody owns; an admin sees the whole pool.
 #:
@@ -174,6 +195,7 @@ def list_tables(session: Session = Depends(require_session)):
             continue
         out.append({"key": key, "label": ut_label(t, key, meta),
                     "source": t.get("source") or "Blank",
+                    "recordsMutable": ut.records_mutable(key, st=session.runtime),
                     "createdBy": t.get("createdBy") or "",
                     "created": t.get("created") or "",
                     "fields": [dict(f) for f in (t.get("fields") or [])],
@@ -341,15 +363,13 @@ def table_rows(table_key: str, session: Session = Depends(require_session)):
     g = ut_assembly(session, table_key)
     merged = {str(r["pid"]): {k: v for k, v in r.items() if k != "pid"}
               for r in g["rows_src"]}
-    for pid, cells in (g["ws"].get("overlays") or {}).items():
-        if isinstance(cells, dict):
-            merged.setdefault(str(pid), {}).update(cells)
     rows = aios_grid.rows_from_pool(
         g["rows_src"], g["fields"], merged, derived=g["derived"])
     return {"fields": g["fields"], "rows": rows, "today": g["today"],
             "pulled_at": time.strftime("%Y-%m-%d %H:%M"),
             "identity": {"pid": "pid"},
-            "scope": {"table": table_key, "rowCount": len(rows)}}
+            "scope": {"table": table_key, "rowCount": len(rows)},
+            "recordsMutable": _ut().records_mutable(table_key, st=session.runtime)}
 
 
 @router.post("/tables/{table_key}/rows", status_code=201)
@@ -361,7 +381,7 @@ def add_row(table_key: str, body: dict = Body(default=None),
     that requested `rid: 7` and got 12 because 7 had been re-used must find that out from the
     response rather than assume; the client re-anchors on what came back.
     """
-    _defn_or_refuse(session, table_key)
+    _records_or_refuse(session, table_key)
     ut = _ut()
     values = (body or {}).get("values") or {}
     if not isinstance(values, dict):
@@ -385,6 +405,7 @@ def add_row(table_key: str, body: dict = Body(default=None),
                           f"profile link (instagram.com/name)")
         raise err(400, "refused",
                   f"row refused — the table may be at its {ut.MAX_ROWS}-row cap")
+    _refresh_relations(session)
     return {"rid": rid, "pid": int(rid)}
 
 
@@ -411,6 +432,13 @@ def _field_or_refuse(session, table_key, fkey=""):
                   "owns its own columns")
     if fkey and not ut.may_edit_field(table_key, fkey, session.uname, session.admin,
                                       st=session.runtime):
+        field = next((f for f in (defn.get("fields") or [])
+                      if f.get("key") == str(fkey)), None)
+        if isinstance((field or {}).get("automation"), dict) \
+                and field["automation"].get("preset") is True:
+            raise err(403, "preset_field_locked",
+                      "Instagram pre-set fields are locked; you may sort, filter or hide this "
+                      "column, and add your own columns separately")
         raise err(403, "forbidden", "that column can only be changed by the database's creator "
                                     "or an admin")
     if not fkey and not (session.admin or defn.get("createdBy") == session.uname):
@@ -434,6 +462,10 @@ def add_field(table_key: str, body: dict = Body(default=None),
         # re-implements it (a second copy of the rule is how two doors start disagreeing).
         raise err(400, "refused",
                   _refusal_sentence(ut, session, body or {}, table_key=table_key))
+    if field.get("type") == "link":
+        synced = ut.sync_reciprocal_link(table_key, field["key"], st=session.runtime)
+        field = synced.get("field") or field
+        _refresh_relations(session)
     return {"field": field}
 
 
@@ -516,6 +548,9 @@ def patch_field(table_key: str, fkey: str, body: dict = Body(default=None),
         # one thing that was never wrong (the D-46 lesson, one door over).
         raise err(400, "refused",
                   _refusal_sentence(ut, session, body, table_key=table_key, fkey=fkey))
+    synced = ut.sync_reciprocal_link(table_key, fkey, st=session.runtime)
+    field = synced.get("field") or field
+    _refresh_relations(session)
     out = {"field": field}
     if migrated is not None:
         out["migrated"] = migrated
@@ -528,18 +563,20 @@ def delete_field(table_key: str, fkey: str, session: Session = Depends(require_s
     if not _ut().delete_field(table_key, fkey, st=session.runtime):
         raise err(400, "refused",
                   "that column could not be removed — a database must keep at least one")
+    _refresh_relations(session)
     return {"deleted": fkey}
 
 
 @router.delete("/tables/{table_key}/rows/{rid}")
 def delete_row(table_key: str, rid: str, session: Session = Depends(require_session)):
-    _defn_or_refuse(session, table_key)
+    _records_or_refuse(session, table_key)
     try:
         ok = _ut().delete_row(table_key, rid, st=session.runtime)
     except Exception:
         raise err(503, "store_unavailable", "the delete did not land — try again")
     if not ok:
         raise err(400, "refused", "rows can only be deleted from user-created databases")
+    _refresh_relations(session)
     return {"ok": True}
 
 
@@ -554,6 +591,7 @@ def patch_row(table_key: str, pid: int, body: dict = Body(default=None),
     updates = dict(body or {})
     if not updates:
         raise err(400, "empty_patch", "no fields to update")
+    _records_or_refuse(session, table_key)
     g = ut_assembly(session, table_key, consume_corrections=False)
     if pid not in g["pids"]:
         raise err(403, "out_of_scope", "that row is not in this database")
@@ -576,8 +614,6 @@ def patch_row(table_key: str, pid: int, body: dict = Body(default=None),
     # rows envelope uses (`table_rows`): definition under, overlay over.
     stored = dict(((_ut().get(table_key, st=session.runtime) or {}).get("rows") or {})
                   .get(str(pid)) or {})
-    stored.update((grid_events.table_workspace(ctx, allowed_pids=None)
-                   .get("overlays") or {}).get(str(pid)) or {})
     accepted = {k: stored.get(k) for k in updates if k in stored}
 
     def _took(k):
@@ -615,4 +651,5 @@ def patch_row(table_key: str, pid: int, body: dict = Body(default=None),
     cleared = sorted(k for k in also if k in _ut().PROFILE_PRESET_KEYS)
     if cleared:
         out["cleared"] = cleared
+    _refresh_relations(session)
     return out

platform/core/grid_events.py

@@ -1755,6 +1755,19 @@ def handle_one(event, ctx):
                     refused = True
                     continue
                 _auto = (_fdef or {}).get('automation')
+                # An ordinary Link is a shared database relationship. Validate its selected
+                # target ids and write it through to the definition row; a personal overlay
+                # would be invisible to reciprocal links and Rollups.
+                if (_fdef or {}).get('type') == 'link':
+                    _linked = _ut_mod.patch_link_cell(
+                        ctx.scope_key, pid, key, value, st=ctx.table.st)
+                    if _linked is None:
+                        refused = True
+                        continue
+                    if _linked != str(value):
+                        refused = True
+                    row_events.append((key, _linked))
+                    continue
                 if isinstance(_auto, dict) and _auto.get('stageField'):
                     _row = ((_tdef or {}).get('rows') or {}).get(str(pid)) or {}
                     _cur = str(_row.get(key) or '').strip()
@@ -1846,7 +1859,7 @@ def handle_one(event, ctx):
                 # `flowId` and not merely `automation`: a user-configured `automation` COLUMN
                 # carries its own gear bag (kind/source/urlField/settings) and no flowId, and it
                 # is not this ruling's subject.
-                if isinstance(_auto, dict) and _auto.get('flowId'):
+                if isinstance(_auto, dict) and (_auto.get('flowId') or _auto.get('preset')):
                     refused = True
                     continue
             # ⭐ WAVE 23 (C7) — the json wall, BEFORE the generic 10 000-char truncation, which
@@ -1904,7 +1917,13 @@ def handle_one(event, ctx):
                     clean = normalized
             updates[key] = clean
         if updates:
-            if _store_of(ctx).available():
+            if str(ctx.scope_key or '').startswith('ut_') and ctx.table is not None:
+                # User-table cells are collaborative record data. Views/layout remain personal,
+                # but record values live in the definition rows so Links, Rollups, automations,
+                # and every viewer observe one truth.
+                import core.user_tables as _ut_values
+                _ut_values.patch_cells(ctx.scope_key, pid, updates, st=ctx.table.st)
+            elif _store_of(ctx).available():
                 _tops(ctx).patch_overlay(uname, pid, updates)
             else:
                 _session_ready()

platform/core/user_tables.py

@@ -32,6 +32,7 @@ the Streamlit host, unchanged). The API passes the session's `TenantRuntime`, wh
 apply the tenant prefix / repo binding, so a Nurilab table lands in Nurilab's store.
 """
 import datetime as _dt
+import hashlib
 import re
 
 import core.store as store
@@ -45,9 +46,21 @@ MAX_FIELDS = 60
 BLANK_SOURCE = 'Blank'
 #: automation-created tables carry their maker instead (wave 18 C4-AUTO)
 AUTOMATION_SOURCE = 'Automation'
+AUTOMATION_RECORD_MODE = 'automation'
 #: every user table's key is prefixed, so it can never collide with a registry module key
 KEY_PREFIX = 'ut_'
 
+
+def records_mutable(table_key, st=None):
+    """May a human add/edit/delete records in this database?
+
+    Ordinary and Profile databases default open. Automation-owned child datasets opt out with a
+    table-level mode; engine writers use their direct coalesced path and are intentionally not
+    routed through this human-door predicate.
+    """
+    table = get(table_key, st) or {}
+    return bool(table) and table.get('recordMode') != AUTOMATION_RECORD_MODE
+
 #: The field types a user table may declare. ⚠ Kept a SUBSET of
 #: `aios_grid.CUSTOM_FIELD_TYPES` (gated in verify_api's W18-UT section). `formula` and
 #: `created_time` stay out (client-computed / row-datum kinds — a base column of either would
@@ -120,13 +133,17 @@ PROFILE_PRESET_KEYS = (
     # the profile facts themselves (CANDIDATE_FIELDS + the enrichment run_field_instagram pulls)
     'full_name', 'followers', 'following', 'posts_count', 'avg_engagement', 'bio',
     'external_url', 'verified', 'category', 'business_category', 'is_business',
-    'is_professional', 'is_private', 'highlights_count', 'bio_hashtags', 'post_hashtags',
+    'is_professional', 'is_private', 'highlights_count', 'bio_hashtags',
     'pronouns', 'ig_id', 'profile_url',
     # ⭐ 2026-08-07 — the rest of the Bright Data profile schema, promoted to preset columns by
     # owner instruction. They are profile FACTS like the nineteen above, so blanking the handle
     # clears them for the same reason (R6): they describe an account this row no longer names.
     'profile_name', 'is_joined_recently', 'has_channel', 'partner_id', 'external_url_title',
     'fbid', 'related_accounts', 'country_code',
+    # relational projections and summaries owned by the Instagram graph
+    'posts_link', 'profile_snapshots_link', 'post_snapshots_link', 'comments_link',
+    'avg_views_12', 'avg_plays_12', 'avg_likes_12', 'avg_comments_12',
+    'posts_captured', 'profile_reads', 'post_measurements_captured', 'comments_captured',
     # R3's LATEST-value stamp. The full series stays in `ut_ig_snapshots` — one store for one
     # series — which is why clearing here can never be a history delete.
     'enriched_at',
@@ -458,7 +475,7 @@ def add_row(table_key, values=None, username=None, st=None, rid=None):
     only when that id is FREE — an undo can never overwrite a row somebody has since created in
     the gap, and it never invents a non-numeric id, because `scoped_pool` reads row ids as ints.
     """
-    if not is_user_table(table_key, st):
+    if not is_user_table(table_key, st) or not records_mutable(table_key, st):
         return None                            # ⚠ never on a connector-backed table
     defn = get(table_key, st) or {}
     rows = defn.get('rows') or {}
@@ -601,7 +618,7 @@ def _clean_metric(raw):
 # So a link cell holds a COMMA-JOINED list of linked row ids — the shape `multiselect` already
 # uses for a set, so the grouping/filter/copy paths already know what to do with it.
 #
-# ⭐ TWO MODES, ONE KIND, AND `on` IS THE DISCRIMINATOR:
+# ⭐ THREE MODES, ONE KIND:
 #   `on` DECLARED  -> a DERIVED link. The linked rows are those whose `on` column equals this
 #                    row's `from` column. Machine-maintained, read-only, recomputed on the same
 #                    pass that already recomputes `metric` cells. THIS IS THE INSTAGRAM CASE, and
@@ -609,13 +626,10 @@ def _clean_metric(raw):
 #                    relation and is rewritten by the engine on every pull, so a second
 #                    user-editable copy of the same relation could only ever drift away from it.
 #   `on` ABSENT    -> an ORDINARY link. The user picks records; the cell is editable; `single` is
-#                    Airtable's `prefersSingleRecordLink`.
-#
-# ⚠ NO `inverse` (Airtable's symmetric back-link field) THIS WAVE, and it is an omission rather
-# than an oversight: maintaining one means every cell write fans out into a second table's rows
-# through `grid_events`' emit guard — the cross-table write-amplification class D-40 was
-# re-deferred BY RULING over. A derived link gets symmetry free (both sides re-derive from the
-# same join), so the gap only touches ordinary links, where a back-link is a convenience.
+#                    Airtable's `prefersSingleRecordLink`. Its value is the source of truth.
+#   `inverse`       -> the COMPUTED reciprocal of one ordinary source link. It never fans writes
+#                    into target rows: the relation pass derives it from the source cells, so the
+#                    same shared-row truth powers both sides and every dependent Rollup.
 
 #: How many linked ids one link cell may carry. Row ids are short numeric strings, so 500 ids is
 #: ~3 KB — comfortably inside a text cell, and `MAX_ROWS` bounds the absolute worst case anyway.
@@ -626,9 +640,13 @@ LINK_MAX_IDS = 500
 #: this product is a scalar — they could only ever return their own input, so offering them would
 #: mint columns that compute nothing. `arrayslice`'s real use ("just the first N") is what `limit`
 #: does below, honestly and by declared rank.
-ROLLUP_FNS = ('sum', 'average', 'min', 'max', 'count', 'counta', 'countall',
+ROLLUP_FNS = ('sum', 'average', 'min', 'max', 'latest', 'count', 'counta', 'countall',
               'and', 'or', 'xor', 'concatenate', 'arrayjoin', 'arraycompact', 'arrayunique')
 ROLLUP_SORT_DIRS = ('asc', 'desc')
+ROLLUP_CONDITION_OPS = ('eq', 'neq', 'contains', 'not_contains', 'is_empty', 'is_not_empty',
+                        'gt', 'gte', 'lt', 'lte')
+ROLLUP_CONDITION_CONJ = ('and', 'or')
+ROLLUP_MAX_CONDITIONS = 20
 #: The ceiling on `limit`. Not a performance bound — `MAX_ROWS` is that — but a refusal to let a
 #: column claim a window bigger than a table can hold.
 ROLLUP_MAX_LIMIT = MAX_ROWS
@@ -659,7 +677,15 @@ def _clean_link(raw):
     out = {'table': table}
     on = _field_key(raw.get('on'))
     frm = _field_key(raw.get('from'))
-    if on:
+    inverse = _field_key(raw.get('inverse'))
+    reciprocal = _field_key(raw.get('reciprocal'))
+    if inverse:
+        # Airtable's reciprocal field: rows in `table` whose ordinary `inverse` link includes
+        # this row. It is computed, so `on`/`from` cannot simultaneously configure another join.
+        if on or frm:
+            return None
+        out['inverse'] = inverse
+    elif on:
         out['on'] = on
         # `from` is OPTIONAL and resolved at compute time (the profile-flagged column, then the
         # pinned one) — which is what makes an Instagram database link up with no configuration
@@ -671,6 +697,8 @@ def _clean_link(raw):
         # than dropped: a bag half of which is silently ignored is [[wrong-parent-not-broken-control]]
         # with the control still on screen.
         return None
+    if reciprocal:
+        out['reciprocal'] = reciprocal
     if raw.get('single') is True:
         out['single'] = True
     return out
@@ -706,10 +734,11 @@ def _clean_rollup(raw):
     except (TypeError, ValueError):
         return None
     sort_by = _field_key(raw.get('sortBy'))
+    distinct_by = _field_key(raw.get('distinctBy'))
     sort_dir = str(raw.get('sortDir') or '').strip().lower()
     if limit < 0 or limit > ROLLUP_MAX_LIMIT:
         return None
-    if limit and not sort_by:
+    if (limit or fn == 'latest') and not sort_by:
         return None
     if sort_dir and sort_dir not in ROLLUP_SORT_DIRS:
         return None
@@ -723,6 +752,35 @@ def _clean_rollup(raw):
             out['limit'] = limit
     elif sort_dir:
         return None
+    if distinct_by:
+        # A rollup may cross a table that already contains duplicate logical records. Keep the
+        # first row after ranking for each identity so "last 12 posts" means twelve POSTS, while
+        # a snapshot table can still keep every timestamped observation by omitting this option.
+        out['distinctBy'] = distinct_by
+    conditions = []
+    raw_conditions = raw.get('conditions') or []
+    if not isinstance(raw_conditions, list) or len(raw_conditions) > ROLLUP_MAX_CONDITIONS:
+        return None
+    for condition in raw_conditions:
+        if not isinstance(condition, dict):
+            return None
+        condition_field = _field_key(condition.get('field'))
+        op = str(condition.get('op') or '').strip().lower()
+        if not condition_field or op not in ROLLUP_CONDITION_OPS:
+            return None
+        item = {'field': condition_field, 'op': op}
+        if op not in ('is_empty', 'is_not_empty'):
+            value = condition.get('value')
+            if not isinstance(value, (str, int, float, bool)):
+                return None
+            item['value'] = str(value)[:1000]
+        conditions.append(item)
+    if conditions:
+        conj = str(raw.get('conditionConj') or 'and').strip().lower()
+        if conj not in ROLLUP_CONDITION_CONJ:
+            return None
+        out['conditions'] = conditions
+        out['conditionConj'] = conj
     return out
 
 
@@ -819,7 +877,7 @@ def is_derived_link(field):
     silently keeps taking writes after the other two stop.
     """
     lk = (field or {}).get('link')
-    return isinstance(lk, dict) and bool(lk.get('on'))
+    return isinstance(lk, dict) and bool(lk.get('on') or lk.get('inverse'))
 
 
 def is_computed_cell(field):
@@ -866,10 +924,18 @@ def clean_machine_fields(fields):
 def may_edit_field(table_key, fkey, viewer, is_admin=False, st=None):
     """May `viewer` change THIS column's definition? Creator/admin always; others only when the
     field itself says `editRole: 'everyone'`. Fail-closed on an unknown field."""
+    table = get(table_key, st) or {}
+    field = next((f for f in (table.get('fields') or []) if f.get('key') == str(fkey)), None)
+    # Instagram's pre-set schema is product contract, not tenant configuration. Even an admin
+    # may sort/filter/hide it, but cannot rename, retype, duplicate or delete it. Ordinary fields
+    # on the same database remain user-owned, including user-created Links and Rollups.
+    if isinstance((field or {}).get('automation'), dict) \
+            and field['automation'].get('preset') is True:
+        return False
     if may_open(table_key, viewer, is_admin, st) and (
-            bool(is_admin) or (get(table_key, st) or {}).get('createdBy') == viewer):
+            bool(is_admin) or table.get('createdBy') == viewer):
         return True
-    for f in ((get(table_key, st) or {}).get('fields') or []):
+    for f in (table.get('fields') or []):
         if f.get('key') == str(fkey):
             return f.get('editRole') == 'everyone'
     return False
@@ -922,6 +988,66 @@ def add_field(table_key, raw, st=None):
     return field
 
 
+def _reciprocal_link_key(table_key, field_key):
+    digest = hashlib.sha1(f'{table_key}:{field_key}'.encode('utf-8')).hexdigest()[:10]
+    return f'linked_{digest}'
+
+
+def sync_reciprocal_link(table_key, field_key, st=None):
+    """Create/repair Airtable's reciprocal link field for one ordinary link.
+
+    The source cell stores picked target row ids. The reciprocal is a computed inverse link on
+    the target database; it lists source rows that include the current target id. Retargeting or
+    retyping the source removes the obsolete inverse in the same store update.
+    """
+    table_key, field_key = str(table_key), str(field_key)
+    result = {'field': None, 'reciprocal': None}
+
+    def _sync(cur):
+        cur = cur if isinstance(cur, dict) else {}
+        source = cur.get(table_key) or {}
+        source_field = next((f for f in (source.get('fields') or [])
+                             if f.get('key') == field_key), None)
+        # Remove every old inverse for this source identity first. This makes retargeting and
+        # deleting deterministic instead of leaving a live-looking backlink on the old table.
+        for candidate in cur.values():
+            if not isinstance(candidate, dict):
+                continue
+            candidate['fields'] = [f for f in (candidate.get('fields') or [])
+                                   if not (isinstance(f.get('link'), dict)
+                                           and f['link'].get('table') == table_key
+                                           and f['link'].get('inverse') == field_key)]
+        if not source_field or source_field.get('type') != 'link':
+            return cur
+        bag = dict(source_field.get('link') or {})
+        if bag.get('on') or bag.get('inverse'):
+            return cur
+        target_key = str(bag.get('table') or '')
+        target = cur.get(target_key)
+        if target is None:
+            return cur
+        reciprocal_key = _reciprocal_link_key(table_key, field_key)
+        reciprocal = _clean_field({
+            'key': reciprocal_key,
+            'label': str(source.get('label') or table_key)[:80],
+            'type': 'link',
+            'link': {'table': table_key, 'inverse': field_key,
+                     'reciprocal': field_key},
+            'editRole': 'admins',
+        })
+        if reciprocal is None:
+            return cur
+        target.setdefault('fields', []).append(reciprocal)
+        bag['reciprocal'] = reciprocal_key
+        source_field['link'] = bag
+        result['field'] = dict(source_field)
+        result['reciprocal'] = dict(reciprocal)
+        return cur
+
+    _st(st).update(STORE_KEY, _sync, flush='sync')
+    return result
+
+
 def patch_field(table_key, fkey, raw, st=None):
     """Edit one column's definition IN PLACE. Returns the stored field, or None if refused.
 
@@ -973,11 +1099,39 @@ def delete_field(table_key, fkey, st=None):
     fields = (get(table_key, st) or {}).get('fields') or []
     if len(fields) <= 1 or not any(f.get('key') == str(fkey) for f in fields):
         return False
+    doomed = next((f for f in fields if f.get('key') == str(fkey)), {})
+    doomed_link = doomed.get('link') if isinstance(doomed.get('link'), dict) else {}
+    inverse_source = None
+    if doomed_link.get('inverse'):
+        source_table = str(doomed_link.get('table') or '')
+        source_field = str(doomed_link.get('inverse') or '')
+        source = get(source_table, st) or {}
+        if len(source.get('fields') or []) <= 1:
+            return False
+        inverse_source = (source_table, source_field)
 
     def _drop(cur):
         t = cur.get(str(table_key))
         if t is not None:
             t['fields'] = [f for f in (t.get('fields') or []) if f.get('key') != str(fkey)]
+        if inverse_source:
+            source_table, source_field = inverse_source
+            source = cur.get(source_table)
+            if isinstance(source, dict):
+                source['fields'] = [f for f in (source.get('fields') or [])
+                                    if f.get('key') != source_field]
+        # An ordinary link owns its reciprocal field. Delete the reciprocal in the same schema
+        # update so no target database can retain a live-looking backlink to a missing source.
+        for candidate in cur.values():
+            if not isinstance(candidate, dict):
+                continue
+            candidate['fields'] = [f for f in (candidate.get('fields') or [])
+                                   if not (isinstance(f.get('link'), dict)
+                                           and ((f['link'].get('table') == str(table_key)
+                                                 and f['link'].get('inverse') == str(fkey))
+                                                or (inverse_source
+                                                    and f['link'].get('table') == inverse_source[0]
+                                                    and f['link'].get('inverse') == inverse_source[1])))]
         return cur
 
     _st(st).update(STORE_KEY, _drop, flush='sync')
@@ -1075,6 +1229,50 @@ def patch_cells(table_key, row_id, values, st=None):
     return True
 
 
+def patch_link_cell(table_key, row_id, fkey, value, st=None):
+    """Persist one user-picked Link cell in the shared row and return its canonical id string.
+
+    A Link is a relationship in the database schema, not one user's visual overlay. Persisting
+    it here makes the reciprocal field and every Rollup see the same source of truth. Derived
+    joins and inverse fields are engine-owned and are refused by this door.
+    """
+    table_key, row_id, fkey = str(table_key), str(row_id), str(fkey)
+    if not is_user_table(table_key, st) or not records_mutable(table_key, st):
+        return None
+    table = get(table_key, st) or {}
+    if row_id not in (table.get('rows') or {}):
+        return None
+    field = next((f for f in (table.get('fields') or []) if f.get('key') == fkey), None)
+    bag = (field or {}).get('link')
+    if (field or {}).get('type') != 'link' or not isinstance(bag, dict) \
+            or bag.get('on') or bag.get('inverse'):
+        return None
+    target = get(str(bag.get('table') or ''), st) or {}
+    valid = set((target.get('rows') or {}).keys())
+    raw_ids = value if isinstance(value, (list, tuple, set)) else str(value or '').split(',')
+    picked, seen = [], set()
+    for raw_id in raw_ids:
+        rid = str(raw_id).strip()
+        if not rid or rid in seen:
+            continue
+        if rid not in valid or len(picked) >= LINK_MAX_IDS:
+            return None
+        seen.add(rid)
+        picked.append(rid)
+        if bag.get('single'):
+            break
+    canonical = ','.join(picked)
+
+    def _set(cur):
+        current = cur.get(table_key)
+        if current is not None and row_id in (current.get('rows') or {}):
+            current['rows'][row_id][fkey] = canonical
+        return cur
+
+    _st(st).update(STORE_KEY, _set, flush='sync')
+    return canonical
+
+
 def patch_profile_cell(table_key, row_id, fkey, value, st=None):
     """Write a PROFILE cell — and, when it is blanked, clear that row's preset cells IN THE SAME
     WRITE (wave 25, contract C3 + owner ruling R6). Returns
@@ -1139,7 +1337,7 @@ def patch_profile_cell(table_key, row_id, fkey, value, st=None):
 
 
 def delete_row(table_key, row_id, st=None):
-    if not is_user_table(table_key, st):
+    if not is_user_table(table_key, st) or not records_mutable(table_key, st):
         return False
 
     def _drop(cur):

web/src/customer-grid/ColumnMenu.tsx

@@ -5,7 +5,7 @@ import { FieldTypeIcon, MenuLabel } from "./icons";
 import { FieldSelectButton } from "./FieldSelect";
 import { CREATABLE_TYPES, choiceOptions, choiceRenames, directionLabel, isMachineOwned,
   isProfileField, parseOptions, ratingMax, ROLLUP_FN_LABELS, ROLLUP_FNS } from "./types";
-import type { Field, FieldFormat, FieldScope, FieldType, Measure, RollupFn,
+import type { Field, FieldFormat, FieldScope, FieldType, Measure, RollupCondition, RollupFn,
   Viewer } from "./types";
 import type { LinkTarget } from "./apiBridge";
 import type { WindowSpec } from "./windows";
@@ -63,6 +63,9 @@ interface FieldConfigExtra {
   rollup?: {
     link: string; field?: string; fn: string;
     limit?: number; sortBy?: string; sortDir?: "asc" | "desc";
+    distinctBy?: string;
+    conditions?: RollupCondition[];
+    conditionConj?: "and" | "or";
   };
 }
 
@@ -83,6 +86,8 @@ interface ColumnMenuProps {
    */
   linkTargets?: LinkTarget[];
   locked: boolean;
+  /** Product-owned preset fields may still drive view actions, but their schema is immutable. */
+  schemaLocked?: boolean;
   /** Wave-5 item 1 — who is looking. Gates the permissions entry (creator-or-admin). */
   viewer?: Viewer;
   /** Wave-5 item 3 — what the CURRENT VIEW does with this field, so the conditional
@@ -748,6 +753,12 @@ function ExtraTypeEditor({
   onRollupLimit,
   rollupSortBy = "",
   onRollupSortBy,
+  rollupDistinctBy = "",
+  onRollupDistinctBy,
+  rollupConditions = [],
+  onRollupConditions,
+  rollupConditionConj = "and",
+  onRollupConditionConj,
 }: {
   kind: FieldType | "measure";
   fields: Field[];
@@ -782,6 +793,12 @@ function ExtraTypeEditor({
   onRollupLimit?: (v: number) => void;
   rollupSortBy?: string;
   onRollupSortBy?: (v: string) => void;
+  rollupDistinctBy?: string;
+  onRollupDistinctBy?: (v: string) => void;
+  rollupConditions?: RollupCondition[];
+  onRollupConditions?: (v: RollupCondition[]) => void;
+  rollupConditionConj?: "and" | "or";
+  onRollupConditionConj?: (v: "and" | "or") => void;
 }) {
   if (kind === "formula") {
     // 2026-07-31 (owner item 2): other FORMULA fields are referencable now — evaluation is
@@ -1028,6 +1045,96 @@ function ExtraTypeEditor({
             </select>
           </label>
         ) : null}
+        <div className="cg-rollup-conditions">
+          <div className="cg-rollup-conditions__head">
+            <span>Conditions</span>
+            <button
+              type="button"
+              className="cg-btn"
+              disabled={targetFields.length === 0 || rollupConditions.length >= 20}
+              onClick={() => onRollupConditions?.([
+                ...rollupConditions,
+                { field: targetFields[0]?.key ?? "", op: "eq", value: "" },
+              ])}
+            >
+              Add condition
+            </button>
+          </div>
+          {rollupConditions.length > 1 ? (
+            <select
+              className="cg-input"
+              aria-label="Match all or any rollup conditions"
+              value={rollupConditionConj}
+              onChange={(event) =>
+                onRollupConditionConj?.(event.target.value as "and" | "or")
+              }
+            >
+              <option value="and">All conditions must match</option>
+              <option value="or">Any condition may match</option>
+            </select>
+          ) : null}
+          {rollupConditions.map((condition, index) => (
+            <div className="cg-rollup-condition" key={`${index}-${condition.field}`}>
+              <select
+                className="cg-input"
+                aria-label={`Condition ${index + 1} column`}
+                value={condition.field}
+                onChange={(event) => onRollupConditions?.(rollupConditions.map((item, i) =>
+                  i === index ? { ...item, field: event.target.value } : item
+                ))}
+              >
+                {targetFields.map((targetField) => (
+                  <option key={targetField.key} value={targetField.key}>{targetField.label}</option>
+                ))}
+              </select>
+              <select
+                className="cg-input"
+                aria-label={`Condition ${index + 1} operator`}
+                value={condition.op}
+                onChange={(event) => onRollupConditions?.(rollupConditions.map((item, i) =>
+                  i === index
+                    ? { ...item, op: event.target.value as RollupCondition["op"] }
+                    : item
+                ))}
+              >
+                <option value="eq">is</option>
+                <option value="neq">is not</option>
+                <option value="contains">contains</option>
+                <option value="not_contains">does not contain</option>
+                <option value="is_empty">is empty</option>
+                <option value="is_not_empty">is not empty</option>
+                <option value="gt">is greater than</option>
+                <option value="gte">is at least</option>
+                <option value="lt">is less than</option>
+                <option value="lte">is at most</option>
+              </select>
+              {!(["is_empty", "is_not_empty"] as string[]).includes(condition.op) ? (
+                <input
+                  className="cg-input"
+                  aria-label={`Condition ${index + 1} value`}
+                  value={condition.value ?? ""}
+                  onChange={(event) => onRollupConditions?.(rollupConditions.map((item, i) =>
+                    i === index ? { ...item, value: event.target.value } : item
+                  ))}
+                />
+              ) : null}
+              <button
+                type="button"
+                className="cg-icon-btn"
+                aria-label={`Remove condition ${index + 1}`}
+                onClick={() => onRollupConditions?.(
+                  rollupConditions.filter((_item, i) => i !== index)
+                )}
+              >
+                ×
+              </button>
+            </div>
+          ))}
+          <div className="cg-field-hint">
+            Like Airtable, these conditions select linked records for this Rollup; view filters
+            do not change its result.
+          </div>
+        </div>
         <label>
           <span>Across</span>
           <select
@@ -1058,6 +1165,21 @@ function ExtraTypeEditor({
             />
           </label>
         ) : null}
+        <label>
+          <span>Deduplicate by</span>
+          <select
+            id={`${idPrefix}-rollup-distinct`}
+            className="cg-input"
+            value={rollupDistinctBy}
+            aria-label="Deduplicate linked records by"
+            onChange={(event) => onRollupDistinctBy?.(event.target.value)}
+          >
+            <option value="">Keep every linked row</option>
+            {targetFields.map((f) => (
+              <option key={f.key} value={f.key}>One row per {f.label}</option>
+            ))}
+          </select>
+        </label>
         {/* ⛔ THE RULE STATED AT THE CONTROL, because the server refuses the pair and a refusal
             the user meets after pressing Create is a refusal they had no way to avoid. "The
             last N" with no declared order is not a measurement — it is whichever N rows happen
@@ -1067,6 +1189,12 @@ function ExtraTypeEditor({
             ? `The ${rollupLimit || 1} most recent linked records, newest first.`
             : "Every linked record. Choose an order above to summarise just the most recent few."}
         </div>
+        {rollupDistinctBy ? (
+          <div className="cg-field-hint">
+            Duplicate linked rows with the same identity count once; the newest row wins when an
+            order is selected.
+          </div>
+        ) : null}
         <div className="cg-field-hint">
           Computed for you and refreshed as the linked records change — the cell cannot be typed
           into.
@@ -1083,6 +1211,7 @@ export default function ColumnMenu({
   fields,
   linkTargets = [],
   locked,
+  schemaLocked = false,
   viewer,
   sortedDir,
   isFiltered,
@@ -1118,7 +1247,7 @@ export default function ColumnMenu({
   userOptions = [],
   measures = [],
 }: ColumnMenuProps) {
-  const [pane, setPane] = useState<MenuPane>(initialPane ?? "menu");
+  const [pane, setPane] = useState<MenuPane>(schemaLocked ? "menu" : initialPane ?? "menu");
   const [note, setNote] = useState(field.note ?? "");
   const [position, setPosition] = useState<CreatePosition | null>(initialPosition ?? null);
   /** Delete is DESTRUCTIVE (a custom field's stored values go with it) — first click arms,
@@ -1159,6 +1288,9 @@ export default function ColumnMenu({
   /** 0 = every linked record (Airtable's only behaviour). Non-zero needs a sort — see below. */
   const [rollupLimit, setRollupLimit] = useState(0);
   const [rollupSortBy, setRollupSortBy] = useState("");
+  const [rollupDistinctBy, setRollupDistinctBy] = useState("");
+  const [rollupConditions, setRollupConditions] = useState<RollupCondition[]>([]);
+  const [rollupConditionConj, setRollupConditionConj] = useState<"and" | "or">("and");
   const [swapTo, setSwapTo] = useState("");
   /** The "New field" half of Change-field keeps its own name/options — a half-typed insert
    *  form must not leak into a swap and vice versa. */
@@ -1290,7 +1422,12 @@ export default function ColumnMenu({
     // absent on the next read. The same "permanently blank column" argument as the two rules
     // above, one step worse — the column does not exist at all.
     (kind !== "link" || linkTable !== "") &&
-    (kind !== "rollup" || (rollupLink !== "" && (rollupFn === "countall" || rollupField !== "")));
+    (kind !== "rollup" || (
+      rollupLink !== "" &&
+      (rollupFn === "countall" || rollupField !== "") &&
+      // `latest` without an order is store order wearing a deterministic name.
+      (rollupFn !== "latest" || rollupSortBy !== "")
+    ));
 
   const extraFor = (
     t: CreateKind,
@@ -1323,7 +1460,11 @@ export default function ColumnMenu({
           // with no declared order, so sending one would turn a valid-looking form into a 400.
           ...(rollupSortBy
             ? { sortBy: rollupSortBy, sortDir: "desc" as const,
-                ...(rollupLimit ? { limit: rollupLimit } : {}) }
+                limit: rollupLimit || 1 }
+            : {}),
+          ...(rollupDistinctBy ? { distinctBy: rollupDistinctBy } : {}),
+          ...(rollupConditions.length
+            ? { conditions: rollupConditions, conditionConj: rollupConditionConj }
             : {}),
         },
       };
@@ -1671,7 +1812,13 @@ export default function ColumnMenu({
             linkSingle={linkSingle}
             onLinkSingle={setLinkSingle}
             rollupLink={rollupLink}
-            onRollupLink={setRollupLink}
+            onRollupLink={(value) => {
+              setRollupLink(value);
+              setRollupField("");
+              setRollupSortBy("");
+              setRollupDistinctBy("");
+              setRollupConditions([]);
+            }}
             rollupField={rollupField}
             onRollupField={setRollupField}
             rollupFn={rollupFn}
@@ -1680,6 +1827,12 @@ export default function ColumnMenu({
             onRollupLimit={setRollupLimit}
             rollupSortBy={rollupSortBy}
             onRollupSortBy={setRollupSortBy}
+            rollupDistinctBy={rollupDistinctBy}
+            onRollupDistinctBy={setRollupDistinctBy}
+            rollupConditions={rollupConditions}
+            onRollupConditions={setRollupConditions}
+            rollupConditionConj={rollupConditionConj}
+            onRollupConditionConj={setRollupConditionConj}
           />
           {needsOptions(kind) && (
             <OptionsEditor
@@ -2307,7 +2460,7 @@ export default function ColumnMenu({
           is one line that STATES the period and opens that pane, because the window is the
           first thing you want to know about a metric column and the menu had become the only
           place it was written down. A row, not a control: the menu pane edits nothing inline. */}
-      {canPeriod && (
+      {canPeriod && !schemaLocked && (
         <button
           type="button"
           className="cg-column-periodline"
@@ -2329,17 +2482,19 @@ export default function ColumnMenu({
             the one window for the field's name, type (with per-type choices/stars/formula)
             and the Change-field control. Offered on EVERY field — what a read-only field
             cannot change, the pane says honestly instead of hiding the door. */}
-        <button type="button" data-overlay-autofocus onClick={() => {
-          setRenameDraft(field.label);
-          setPane("edit");
-        }}>
-          <MenuLabel icon="rename" text="Edit field" />
-        </button>
+        {!schemaLocked && (
+          <button type="button" data-overlay-autofocus onClick={() => {
+            setRenameDraft(field.label);
+            setPane("edit");
+          }}>
+            <MenuLabel icon="rename" text="Edit field" />
+          </button>
+        )}
 
         <div className="cg-menu-sep" role="separator" aria-hidden />
 
         {/* Group 1 — Duplicate · Insert left · Insert right (+ Add at end, same family). */}
-        {onDuplicate && (
+        {!schemaLocked && onDuplicate && (
           <button type="button" onClick={() => { onDuplicate(); onClose(); }}>
             <MenuLabel icon="duplicate" text="Duplicate field" />
           </button>
@@ -2359,15 +2514,17 @@ export default function ColumnMenu({
         {/* Group 2 — the rest of the def-editing family: description · permissions · format.
             Rename and Change field live INSIDE Edit field now (owner item 8) — their old
             rows are gone, not duplicated. */}
-        <button type="button" onClick={() => setPane("note")}>
-          <MenuLabel icon="description" text="Edit field description" />
-        </button>
-        {canPermissions && (
+        {!schemaLocked && (
+          <button type="button" onClick={() => setPane("note")}>
+            <MenuLabel icon="description" text="Edit field description" />
+          </button>
+        )}
+        {!schemaLocked && canPermissions && (
           <button type="button" onClick={() => setPane("permissions")}>
             <MenuLabel icon="permissions" text="Edit field permissions" />
           </button>
         )}
-        {formatKind && (
+        {!schemaLocked && formatKind && (
           <button type="button" onClick={() => setPane("format")}>
             <MenuLabel icon="format" text="Field format" />
           </button>
@@ -2447,7 +2604,7 @@ export default function ColumnMenu({
         >
           <MenuLabel icon="hide" text="Hide field" />
         </button>
-        {onDelete && (
+        {!schemaLocked && onDelete && (
           <button
             type="button"
             className="is-danger"

web/src/customer-grid/CustomerGrid.tsx

@@ -93,13 +93,14 @@ import {
 } from "./clipboard";
 import { cellTipText, expandButtonRect, GROUP_HEADER_FONT, GROUP_LABEL_PAD, headerMarkLayout,
   headerMarkSizes, tipLeft } from "./overlayPlacement";
-import { AnchoredOverlay } from "./OverlaySurface";
+import { AnchoredOverlay, BodyPortal, useOverlayLayer } from "./OverlaySurface";
 import type { AnchorRect } from "./OverlaySurface";
 import { StarIcon } from "./Stars";
 import { ALL_VIEW_ID, MAX_CALENDAR_METRICS, allViewName,
   MAX_FROZEN, choiceOptions, choiceVocabulary, clampFrozenCount, cleanDisplay, formulaOf,
   topicForScope,
   isDateFamilyType, isFilterGroup, isGroupableField, isMachineOwned, isMachineWritten,
+  isDerivedLink,
   isNumericFieldType,
   isPickType, mayEditField,
   isModeFrozen, isUndeletableView, mayEditView, mayToggleViewLock,
@@ -495,18 +496,136 @@ function buildOverlayField(
  *  transition whole (the fold froze, then SNAPPED — the exact "static" the owner named). The
  *  props surface is one stable string, so memo makes chrome state changes free; the grid still
  *  re-renders for its own state (edits, resize) and remounts on route change via `key`. */
-function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
+interface CustomerGridProps {
+  scope?: SurfaceScope;
+  /** Read-only Grid view embedded in a linked-record modal. It keeps the standard
+   * filter/sort/search toolbar while withdrawing schema and row mutations. */
+  embedded?: boolean;
+  /** Exact linked pids to project from the target database. */
+  embeddedRecordIds?: readonly number[];
+  /** Selection mode used by an editable ordinary-Link modal. */
+  embeddedSelectable?: boolean;
+  embeddedSelectedIds?: readonly number[];
+  onEmbeddedSelectionChange?: (recordIds: number[]) => void;
+}
+
+interface LinkGridModalProps {
+  label: string;
+  table: SurfaceScope;
+  recordIds: readonly number[];
+  editable?: boolean;
+  single?: boolean;
+  onSave?: (recordIds: number[]) => void;
+  onClose: () => void;
+}
+
+function LinkGridModal({ label, table, recordIds, editable = false, single = false,
+  onSave, onClose }: LinkGridModalProps) {
+  const panelRef = useRef<HTMLDivElement>(null);
+  const [selectedIds, setSelectedIds] = useState<number[]>(() => [...recordIds]);
+  const changeSelectedIds = useCallback((ids: number[]) => {
+    const next = single ? ids.slice(-1) : ids;
+    setSelectedIds((current) =>
+      current.join(",") === next.join(",") ? current : next
+    );
+  }, [single]);
+  useOverlayLayer({
+    panelRef,
+    onDismiss: onClose,
+    dismissOnOutside: true,
+    initialFocus: "[data-overlay-autofocus]",
+    trapFocus: true,
+  });
+  return (
+    <BodyPortal>
+      <div className="cg-record-backdrop">
+        <section
+          className="cg-link-grid-modal"
+          ref={panelRef}
+          role="dialog"
+          aria-modal="true"
+          aria-label={label}
+          data-overlay-kind="linked-record-grid"
+          tabIndex={-1}
+        >
+          <header className="cg-link-grid-head">
+            <div>
+              <div className="cg-link-grid-title">{label}</div>
+              <div className="cg-link-grid-sub">
+                {(editable ? selectedIds.length : recordIds.length).toLocaleString()} linked{
+                  single ? " (one allowed)" : ""
+                } {(editable ? selectedIds.length : recordIds.length) === 1 ? "record" : "records"}
+              </div>
+            </div>
+            <button
+              type="button"
+              className="cg-icon-btn"
+              aria-label="Close linked records"
+              data-overlay-autofocus
+              onClick={onClose}
+            >
+              ×
+            </button>
+          </header>
+          <div className="cg-link-grid-body">
+            <CustomerGrid
+              scope={table}
+              embedded
+              embeddedRecordIds={editable ? undefined : recordIds}
+              embeddedSelectable={editable}
+              embeddedSelectedIds={selectedIds}
+              onEmbeddedSelectionChange={changeSelectedIds}
+            />
+          </div>
+          {editable ? (
+            <footer className="cg-link-grid-foot">
+              <button type="button" className="cg-btn" onClick={onClose}>Cancel</button>
+              <button
+                type="button"
+                className="cg-btn cg-btn--primary"
+                onClick={() => { onSave?.(selectedIds); onClose(); }}
+              >
+                Save links
+              </button>
+            </footer>
+          ) : null}
+        </section>
+      </div>
+    </BodyPortal>
+  );
+}
+
+function CustomerGrid({
+  scope = "customer",
+  embedded = false,
+  embeddedRecordIds,
+  embeddedSelectable = false,
+  embeddedSelectedIds = [],
+  onEmbeddedSelectionChange,
+}: CustomerGridProps = {}) {
   // Wave 16 C-TOPIC: which TABLE this tree is drawing, derived from the one scope prop.
   const topic = topicForScope(scope);
   const {
     fields: payloadFields,
-    rawRows,
+    rawRows: fetchedRows,
     payload,
     loading,
     overlayEdits,
     setOverlayEdits,
     patchOverlay,
-  } = useCustomerData(scope);
+  } = useCustomerData(scope, {
+    bindSurface: !embedded,
+    includeWorkspace: !embedded,
+    writable: !embedded,
+  });
+  const embeddedIdsKey = embeddedRecordIds?.join(",") ?? "";
+  const rawRows = useMemo(() => {
+    if (!embeddedRecordIds) return fetchedRows;
+    const wanted = new Set(embeddedRecordIds);
+    return fetchedRows.filter((row) => wanted.has(row.pid));
+    // The scalar key keeps this stable when a caller reconstructs the same id list.
+    // eslint-disable-next-line react-hooks/exhaustive-deps
+  }, [fetchedRows, embeddedIdsKey]);
 
   const [fields, setFields] = useState<Field[]>([]);
   const [views, setViews] = useState<SavedView[]>([]);
@@ -551,6 +670,14 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
     setJsonAt(null);
     requestAnimationFrame(() => gridRef.current?.focus());
   }, []);
+  /** A relation opens as the target database's real Grid view, projected to the linked ids.
+   * This is intentionally distinct from JsonViewer: filters, sorts, search, column visibility,
+   * and the standard cell renderers all remain available inside the large modal. */
+  const [linkAt, setLinkAt] = useState<{ pid: number; fieldKey: string } | null>(null);
+  const closeLink = useCallback(() => {
+    setLinkAt(null);
+    requestAnimationFrame(() => gridRef.current?.focus());
+  }, []);
   /** WAVE 21 item 11 (R10) — is "Select records from a list" open? Opened from the view
    *  rail's "…" and closed by the dialog; the SELECTION it produces outlives it. */
   const [selectFromFile, setSelectFromFile] = useState(false);
@@ -672,7 +799,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
   // fills only missing objects and keeps the standalone path useful.
   useEffect(() => {
     if (!payload || payloadFields.length === 0 || initializedKey.current === storageKey) return;
-    const local = readLocal(storageKey);
+    const local = embedded ? null : readLocal(storageKey);
     // Item 3c: the def half of the no-blip layer. A RECENT local stamp beats a
     // lagged host echo (rename survives, retype holds, a delete stays deleted);
     // a caught-up echo returns host objects byte-identical (see optimism.ts).
@@ -688,7 +815,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
       Date.now(),
       payload.workspace != null
     );
-    const hostViews = payload.workspace?.views ?? [];
+    const hostViews = embedded ? [] : payload.workspace?.views ?? [];
     const byId = new Map<string, SavedView>();
     byId.set(ALL_VIEW_ID, allRecordsView(initialFields, scope));
     // D-19 — the HOST'S LIST DECIDES WHICH VIEWS EXIST. A local copy the host no longer
@@ -765,7 +892,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
      * the config the client is actually filtering with, and sending anything else would ask the
      * host to resolve a question nobody on screen is asking.
      */
-    if (reemit.size > 0) {
+    if (!embedded && reemit.size > 0) {
       const stamped = { ...(local?.reemitted ?? {}) };
       for (const view of initialViews) {
         const key = reemit.get(view.id);
@@ -780,10 +907,10 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
     } else {
       reemittedRef.current = { ...(local?.reemitted ?? {}) };
     }
-  }, [payload, payloadFields, storageKey]);
+  }, [payload, payloadFields, storageKey, embedded, scope]);
 
   useEffect(() => {
-    if (!workspaceReady) return;
+    if (!workspaceReady || embedded) return;
     writeLocal(storageKey, {
       fields,
       views,
@@ -800,7 +927,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
       // drop a view this browser created seconds ago.
       viewWrites: pruneTombstones(viewWritesRef.current, Date.now()),
     });
-  }, [workspaceReady, storageKey, fields, views, activeViewId]);
+  }, [workspaceReady, storageKey, fields, views, activeViewId, embedded]);
 
   /**
    * ⭐ THE LIVE WORKSPACE (owner report, 2026-08-04) — what has APPEARED since we mounted.
@@ -825,7 +952,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
    */
   const hostWorkspaceViews = payload?.workspace?.views;
   useEffect(() => {
-    if (!workspaceReady) return;
+    if (!workspaceReady || embedded) return;
     const now = Date.now();
     const nextFields = adoptNewFields(
       fields, payloadFields, fieldStampsRef.current.deleted, now
@@ -835,7 +962,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
       adoptNewViews(current, hostWorkspaceViews, viewTombstonesRef.current, now,
                     (config) => normalizeConfig(config, nextFields))
     );
-  }, [workspaceReady, hostWorkspaceViews, payloadFields, fields]);
+  }, [workspaceReady, hostWorkspaceViews, payloadFields, fields, embedded]);
 
   /** Item 3c — stamp a def write / a delete. Pruned at every touch so the persisted blob
    *  stays a recent window, never an archive. */
@@ -896,7 +1023,9 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
    * this resolves, the appended row does not exist yet, so "bottom" would land on the last OLD
    * row.
    */
-  const isUserTable = scope.startsWith("ut_");
+  const isUserTable = !embedded && scope.startsWith("ut_");
+  const recordsMutable = payload?.recordsMutable !== false;
+  const canMutateRecords = isUserTable && recordsMutable;
   /**
    * ⭐ 2026-08-07 — the databases a `link` column may point at, fetched when the column menu
    * OPENS rather than on every render of the grid.
@@ -921,7 +1050,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
     };
   }, [columnMenu, linkTargets.length]);
   const appendRow = useCallback(async (): Promise<undefined> => {
-    if (!isUserTable) return undefined;
+    if (!canMutateRecords) return undefined;
     const made = await addTableRow(scope);
     // `undefined` either way, never glide's "bottom": glide would move the selection to the
     // last row it currently knows about, which is the row BEFORE the one just created. The
@@ -936,7 +1065,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
     });
     signal(ROWS_STALE_EVENT);
     return undefined;
-  }, [isUserTable, scope]);
+  }, [canMutateRecords, scope]);
 
   /* wave20 item 2 — measure key + window -> the columns that display it. Built ONCE per field
      list and handed to every consumer of "which column is this rule about", so the tint
@@ -946,7 +1075,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
 
   // Airtable behavior: configuration changes to the active view autosave.
   useEffect(() => {
-    if (!workspaceReady) return;
+    if (!workspaceReady || embedded) return;
     const active = views.find((view) => view.id === activeViewId);
     if (!active || sameConfig(active.config, config)) {
       setSaveState("saved");
@@ -968,7 +1097,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
     return () => {
       if (saveTimer.current !== null) window.clearTimeout(saveTimer.current);
     };
-  }, [workspaceReady, activeViewId, config, views]);
+  }, [workspaceReady, activeViewId, config, views, embedded]);
 
   // Numeric dimensions force a glide relayout when either the component frame
   // or Streamlit's main column changes width (notably sidebar collapse).
@@ -999,7 +1128,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
   const mode = tableMode(payload?.counts);
   const serverWindowed = mode === "server-windowed";
   // Owner items 4+6 — the Cohort page's grid renders without the Views sidebar.
-  const hideViews = payload?.workspace?.hideViews === true;
+  const hideViews = embedded || payload?.workspace?.hideViews === true;
   // Wave-6 item 10 — how this view displays. A WINDOWED table is always the grid: list/
   // calendar/kanban compute over the whole matched set, and one page is not it (CG-3's rule,
   // the same reason grouping is off there).
@@ -1066,8 +1195,8 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
   // permissions vs the viewer, fail-closed on restricted fields when the viewer is unknown.
   const viewer = payload?.viewer;
   const canEditField = useCallback(
-    (f: Field): boolean => mayEditField(f, viewer),
-    [viewer]
+    (f: Field): boolean => !embedded && recordsMutable && mayEditField(f, viewer),
+    [embedded, recordsMutable, viewer]
   );
   const unresolvedCount = useMemo(
     () => unresolvedConditions(config.filters, { cohortSets, today }),
@@ -1301,8 +1430,20 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
     useGridSelection(
       displayRows,
       displayPidToIndex,
-      visibleCols.length
+      visibleCols.length,
+      embeddedSelectable ? embeddedSelectedIds : []
     );
+  const embeddedSelectedKey = embeddedSelectedIds.join(",");
+  const selectedPidKey = [...selectedPids].join(",");
+  useEffect(() => {
+    if (!embeddedSelectable) return;
+    if (selectedPidKey !== embeddedSelectedKey)
+      selectPids(embeddedSelectedIds as number[], "replace");
+  }, [embeddedSelectable, embeddedSelectedKey, embeddedSelectedIds, selectedPidKey, selectPids]);
+  useEffect(() => {
+    if (!embeddedSelectable) return;
+    onEmbeddedSelectionChange?.([...selectedPids]);
+  }, [embeddedSelectable, onEmbeddedSelectionChange, selectedPids]);
   /* ════════════════════════ owner item 16 / R4 / C-UNDO ════════════════════════
      THE RECORDING LAYER. `undoStack.ts` owns the stack and every inverse; this owns the one
      thing it cannot: reading the value a cell held BEFORE the write, which only exists at the
@@ -1519,7 +1660,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
    */
   const deleteRecords = useCallback(
     async (pids: number[]): Promise<boolean> => {
-      if (!isUserTable || !pids.length) return false;
+      if (!canMutateRecords || !pids.length) return false;
       // The row as it stands NOW, straight off the rendered records — the same values the
       // reader can see, so a restore puts back what they watched disappear.
       const byPid = new Map<number, Row>();
@@ -1639,7 +1780,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
       );
       return true;
     },
-    [isUserTable, displayRows, fields, scope, clearSelection, delArmed]
+    [canMutateRecords, displayRows, fields, scope, clearSelection, delArmed]
   );
 
   const onGridDelete = useCallback(
@@ -1665,7 +1806,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
        */
       const dragged = !!sel.current
         && (sel.current.range.width > 1 || sel.current.range.height > 1);
-      if (isUserTable && sel.rows.length > 0 && !dragged) {
+      if (canMutateRecords && sel.rows.length > 0 && !dragged) {
         const pids: number[] = [];
         for (const rowIndex of sel.rows) {
           const vr = displayRows[rowIndex];
@@ -1708,7 +1849,8 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
       // fields (`onCellEdited`'s own `canEditField`), it just does not arrive as one entry.
       return false;
     },
-    [displayRows, visibleCols, fieldByKey, canEditField, patchManyAndRecord]
+    [canMutateRecords, deleteRecords, displayRows, visibleCols, fieldByKey, canEditField,
+     patchManyAndRecord]
   );
 
   /**
@@ -1940,7 +2082,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
     // (`.cg-grid-void` is `pointer-events: none`), which is the worst version of this bug:
     // the gate went green on an affordance nobody could see. Found by READING THE SCREENSHOT
     // ([[ui-invisible-to-assertions]], [[finalize-visual-review-sop]]).
-    if (isUserTable)
+    if (canMutateRecords)
       rowsPx += typeof rowHeight === "number" ? rowHeight : rowHeight(displayRows.length);
 
     /* Fit is tested against the client box the OTHER axis's scrollbar leaves behind — the same
@@ -1963,7 +2105,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
       below: rowsPx < clientH ? rowsPx : null,
       right: colsPx < clientW ? colsPx : null,
     };
-  }, [visibleCols, displayRows, rowHeight, gridSize, isUserTable]);
+  }, [visibleCols, displayRows, rowHeight, gridSize, canMutateRecords]);
   /* ═══ end W18-B VOID (geometry) ═══ */
 
   // The record drawer resolves positions against what the MODE paints: the display slice for
@@ -2163,6 +2305,17 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
         });
         return;
       }
+      // A linked-record cell is a doorway to the target database, not an opaque id list.
+      // The large modal mounts the same Grid surface over exactly these pids, so its standard
+      // search, Filters, Sort, Fields, and column menus keep working for every database kind.
+      if (row.kind === "data" && field?.type === "link" && field.link?.table) {
+        if (field.link.table.startsWith("ut_")) {
+          event.preventDefault();
+          setActiveCell(cell[0], cell[1]);
+          setLinkAt({ pid: row.record.pid, fieldKey: field.key });
+          return;
+        }
+      }
       // ⭐ Wave-23 C7 — a JSON cell opens the big viewer. It is the ONLY door: the cell carries
       // `allowOverlay:false`, because glide's overlay is a one-line box and one keystroke in the
       // wrong place inside a 32 KB document turns a well-formed payload into an unparseable one,
@@ -2318,6 +2471,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
 
   const openHeaderMenu = useCallback(
     (column: number, bounds: Rectangle) => {
+      if (embedded) return;
       const definition = visibleCols[column];
       if (!definition?.id) return;
       setColumnMenu({
@@ -2332,7 +2486,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
         },
       });
     },
-    [visibleCols]
+    [embedded, visibleCols]
   );
   const onHeaderClicked = useCallback(
     (column: number, event: HeaderClickedEventArgs) => {
@@ -3956,7 +4110,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
   const menuVisIndex = menuField ? visibleKeys.indexOf(menuField.key) : -1;
   const menuPinnedTo = menuVisIndex >= 0 && frozenN > 1 && menuVisIndex + 1 === frozenN;
   // Item 10 — the toolbar's mode switcher (grid-only on windowed tables, see displayMode).
-  const modeControl = serverWindowed ? undefined : (
+  const modeControl = serverWindowed || embedded ? undefined : (
     <ModeSwitch
       mode={displayMode}
       onMode={setDisplayMode}
@@ -3998,6 +4152,18 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
     ?? (jsonField && jsonRow ? jsonRow[jsonField.key] : "")
     ?? ""
   );
+  const linkField = linkAt ? fieldByKey.get(linkAt.fieldKey) : undefined;
+  const linkRow = linkAt ? rawRows.find((row) => row.pid === linkAt.pid) : undefined;
+  const linkValue = String(
+    (linkAt && overlayEdits[linkAt.pid]?.[linkAt.fieldKey])
+    ?? (linkField && linkRow ? linkRow[linkField.key] : "")
+    ?? ""
+  );
+  const linkedRecordIds = [...new Set(
+    linkValue.split(",").map((part) => Number(part.trim())).filter(
+      (pid) => Number.isInteger(pid) && pid > 0
+    )
+  )];
   const pickerField = picker ? fieldByKey.get(picker.fieldKey) : undefined;
   // A `user` field's choices come from the HOST's real user list, a `select`'s from its own
   // definition — so an assignee is always someone who can log in, and a status is always one of
@@ -4313,7 +4479,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
               height={gridSize.height}
               customRenderers={[ratingCellRenderer, userCellRenderer, imageCellRenderer]}
               headerIcons={HEADER_ICONS}
-              rightElement={
+              rightElement={embedded ? undefined : (
                 // Wave-6 item 4 — the "+" of the header row: the create-field form,
                 // insert-at-end.
                 //
@@ -4350,14 +4516,14 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
                 >
                   +
                 </button>
-              }
+              )}
               rightElementProps={{ sticky: false, fill: false }}
               // Owner item 4 (R8) — the trailing "+" row, USER DATABASES ONLY. Passing
               // `onRowAppended` is what makes glide paint the ghost row at all, so the
               // affordance exists exactly where a POST can succeed. See `appendRow`.
-              onRowAppended={isUserTable ? appendRow : undefined}
+              onRowAppended={canMutateRecords ? appendRow : undefined}
               trailingRowOptions={
-                isUserTable
+                canMutateRecords
                   ? {
                       // The hint sits in the identity column (targetColumn 0), which is the
                       // one a new record is named in — the same cell the cursor lands on.
@@ -4555,7 +4721,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
 
               Rendered only in `grid` mode: list / calendar / kanban / map have their own
               open affordances and glide is not mounted to report bounds for them. */}
-          {displayMode === "grid" && expandAt && (
+          {!embedded && displayMode === "grid" && expandAt && (
             <button
               type="button"
               className="cg-row-expand"
@@ -4615,7 +4781,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
               broke this gate's own coordinate math before it broke a user). It states the
               count and offers "Add to cohort" over exactly the checked pids — the same guarded
               add_to_list event the view menu uses, so the host treats both alike. */}
-          {selectedPids.size > 0 && !serverWindowed &&
+          {!embedded && selectedPids.size > 0 && !serverWindowed &&
             (displayMode === "grid" || displayMode === "map") && (
             <div className="cg-selbar">
               <span className="cg-selbar-count">
@@ -4627,7 +4793,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
                   ⚠ ONLY on a `ut_` database: the customer and product grids are projections of
                   Odoo, where a "delete" would be a write to somebody else's system of record —
                   the server refuses it, and a button that must be refused is worse than none. */}
-              {isUserTable && (
+              {canMutateRecords && (
                 <button
                   type="button"
                   /* ARMED is a VISIBLE state, not just a changed handler: the button that is
@@ -4744,6 +4910,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
           // one-line picker, so it is left out until somebody asks for it.
           linkTargets={linkTargets.filter((t) => t.key !== scope)}
           locked={menuField.key === lockedKey}
+          schemaLocked={menuField.automation?.preset === true}
           viewer={viewer}
           sortedDir={menuSortedDir}
           isFiltered={menuIsFiltered}
@@ -5102,6 +5269,21 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
           onClose={closeJson}
         />
       )}
+      {linkField?.type === "link" && linkField.link?.table.startsWith("ut_") && linkAt && (
+        <LinkGridModal
+          label={linkField.label}
+          table={linkField.link.table as SurfaceScope}
+          recordIds={linkedRecordIds}
+          editable={canEditField(linkField) && !isDerivedLink(linkField)}
+          single={linkField.link.single === true}
+          onSave={(ids) => patchAndRecord(
+            linkAt.pid,
+            { [linkField.key]: ids.join(",") },
+            "linked records"
+          )}
+          onClose={closeLink}
+        />
+      )}
       {pickerField && picker && (
         <AnchoredOverlay
           anchor={picker.anchor}
@@ -5299,7 +5481,7 @@ function CustomerGrid({ scope = "customer" }: { scope?: SurfaceScope } = {}) {
         </AnchoredOverlay>
       )}
 
-      {detailRecord && detailPid !== null && (
+      {!embedded && detailRecord && detailPid !== null && (
         <RecordDetail
           fields={fields}
           record={detailRecord}

web/src/customer-grid/RecordDetail.css

@@ -32,6 +32,72 @@
   outline: none;
 }
 
+/* Universal linked-record viewer: the target is the real Grid surface, not a JSON-shaped
+   approximation. It is deliberately wider than the record detail so filtering and multi-column
+   sorting have the same working room they do on the database page. */
+.cg-link-grid-modal {
+  width: min(1320px, calc(100vw - 48px));
+  height: min(840px, calc(100vh - 48px));
+  min-height: 460px;
+  display: flex;
+  flex-direction: column;
+  overflow: hidden;
+  border: 1px solid color-mix(in srgb, var(--lp-line) 82%, var(--lp-muted));
+  border-radius: 12px;
+  background: var(--lp-surface);
+  box-shadow:
+    0 30px 80px rgba(20, 30, 43, 0.24),
+    0 4px 18px rgba(20, 30, 43, 0.12);
+  color: var(--lp-ink);
+  font-family: Inter, ui-sans-serif, system-ui, -apple-system, "Segoe UI", sans-serif;
+  font-size: var(--lp-fs-xs);
+  animation: cg-record-open 150ms cubic-bezier(0.2, 0, 0, 1);
+}
+
+.cg-link-grid-modal:focus { outline: none; }
+
+.cg-link-grid-head {
+  flex: 0 0 54px;
+  min-height: 54px;
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  gap: 16px;
+  padding: 0 14px 0 18px;
+  border-bottom: 1px solid var(--lp-line);
+  background: var(--lp-surface);
+}
+
+.cg-link-grid-title {
+  font-size: var(--lp-fs-sm);
+  font-weight: 650;
+  line-height: 1.2;
+}
+
+.cg-link-grid-sub {
+  margin-top: 2px;
+  color: var(--lp-muted);
+  font-size: var(--lp-fs-3xs);
+}
+
+.cg-link-grid-body {
+  flex: 1 1 auto;
+  min-width: 0;
+  min-height: 0;
+  background: var(--lp-surface);
+}
+
+.cg-link-grid-foot {
+  flex: 0 0 52px;
+  display: flex;
+  align-items: center;
+  justify-content: flex-end;
+  gap: 8px;
+  padding: 8px 14px;
+  border-top: 1px solid var(--lp-line);
+  background: var(--lp-surface);
+}
+
 @keyframes cg-record-open {
   from {
     opacity: 0;

web/src/customer-grid/apiBridge.ts

@@ -44,8 +44,9 @@ import {
   API_V1,
   CREDENTIALS,
   DATA_ERROR_EVENT,
-  DERIVED_CELLS_EVENT,
-  TOAST_EVENT,
+  DERIVED_CELLS_EVENT,
+  ROWS_STALE_EVENT,
+  TOAST_EVENT,
   WORKSPACE_STALE_EVENT,
   UNAUTHORIZED_EVENT,
   signal,
@@ -257,12 +258,20 @@ export async function patchTopicRow(
         : [];
       const blanks: Partial<Row> = {};
       for (const k of cleared) blanks[k] = "";
-      if (accepted && typeof accepted === "object") {
-        const row = cached?.payload.rows.find((r) => r.pid === pid);
-        if (row) Object.assign(row, accepted, blanks);
-        onAccepted?.({ ...(accepted as Partial<Row>), ...blanks }, cleared);
-      }
-    }
+      if (accepted && typeof accepted === "object") {
+        const row = cached?.payload.rows.find((r) => r.pid === pid);
+        if (row) Object.assign(row, accepted, blanks);
+        onAccepted?.({ ...(accepted as Partial<Row>), ...blanks }, cleared);
+      }
+      // A user-table edit may change a reciprocal Link or a Rollup in another database. Their
+      // row payloads share no cache key with this table, so invalidate the user-table family and
+      // let the active grid re-read the server's materialised relationship cells.
+      if (topic.rowsPath.startsWith("tables/")) {
+        for (const key of [...rowsCache.keys()])
+          if (key.startsWith("tables/")) rowsCache.delete(key);
+        signal(ROWS_STALE_EVENT);
+      }
+    }
     return res.ok;
   } catch {
     return false;

web/src/customer-grid/types.ts

@@ -1309,8 +1309,10 @@ export interface Field {
     urlField?: string;
     /** kind-specific knobs. Scalars only — the host clamps keys and value lengths. */
     settings?: Record<string, string | number | boolean>;
-    /** the automation definition that owns this column (engine-stamped; the machine marker). */
-    flowId?: string;
+    /** the automation definition that owns this column (engine-stamped; the machine marker). */
+    flowId?: string;
+    /** true when the field is part of a product-owned preset schema and cannot be redefined */
+    preset?: boolean;
     /** true on the `stage_<autoId>` column an automation board walks its records through. */
     stageField?: boolean;
   };
@@ -1432,8 +1434,12 @@ export interface Field {
     table: string;
     /** the column IN THE LINKED TABLE that holds this row's key — presence = derived */
     on?: string;
-    /** the column on THIS table supplying the join value; defaults to the profile/pinned one */
-    from?: string;
+    /** the column on THIS table supplying the join value; defaults to the profile/pinned one */
+    from?: string;
+    /** source link field key when this is Airtable's computed reciprocal side */
+    inverse?: string;
+    /** reciprocal field key on the target table when this is the user-picked source side */
+    reciprocal?: string;
     /** Airtable's `prefersSingleRecordLink` */
     single?: boolean;
   };
@@ -1453,11 +1459,27 @@ export interface Field {
     /** the column in the linked table to aggregate; absent only for `countall` */
     field?: string;
     fn: RollupFn;
-    limit?: number;
-    sortBy?: string;
-    sortDir?: "asc" | "desc";
-  };
-}
+    limit?: number;
+    sortBy?: string;
+    sortDir?: "asc" | "desc";
+    /** collapse duplicate logical records after ranking, before applying the window */
+    distinctBy?: string;
+    /** optional Airtable-style conditions applied to linked records before aggregation */
+    conditions?: RollupCondition[];
+    conditionConj?: "and" | "or";
+  };
+}
+
+export const ROLLUP_CONDITION_OPS = [
+  "eq", "neq", "contains", "not_contains", "is_empty", "is_not_empty",
+  "gt", "gte", "lt", "lte",
+] as const;
+export type RollupConditionOp = (typeof ROLLUP_CONDITION_OPS)[number];
+export interface RollupCondition {
+  field: string;
+  op: RollupConditionOp;
+  value?: string;
+}
 
 /**
  * Airtable's 17 rollup functions, minus the two that are meaningless over a scalar cell.
@@ -1469,7 +1491,7 @@ export interface Field {
  * the same discipline every other cross-language vocabulary here uses.
  */
 export const ROLLUP_FNS = [
-  "sum", "average", "min", "max", "count", "counta", "countall",
+  "sum", "average", "min", "max", "latest", "count", "counta", "countall",
   "and", "or", "xor", "concatenate", "arrayjoin", "arraycompact", "arrayunique",
 ] as const;
 export type RollupFn = (typeof ROLLUP_FNS)[number];
@@ -1480,7 +1502,8 @@ export const ROLLUP_NUMERIC_FNS: ReadonlySet<string> =
 
 /** Human wording for each function, so a column menu never shows a bare identifier. */
 export const ROLLUP_FN_LABELS: Record<RollupFn, string> = {
-  sum: "Sum", average: "Average", min: "Minimum", max: "Maximum",
+  sum: "Sum", average: "Average", min: "Minimum", max: "Maximum",
+  latest: "Latest value",
   count: "Count of numbers", counta: "Count of filled cells", countall: "Count of records",
   and: "All are true", or: "Any is true", xor: "Exactly one is true",
   concatenate: "Join (no separator)", arrayjoin: "Join with commas",
@@ -1494,9 +1517,9 @@ export const ROLLUP_FN_LABELS: Record<RollupFn, string> = {
  * is rewritten on every refresh pass, so accepting a typed value would not merely be wrong — it
  * would appear to work and then silently revert, which is worse than refusing it.
  */
-export function isDerivedLink(field: Field): boolean {
-  return field.type === "link" && !!field.link?.on;
-}
+export function isDerivedLink(field: Field): boolean {
+  return field.type === "link" && !!(field.link?.on || field.link?.inverse);
+}
 
 /**
  * Is this the table's PROFILE column (C3/R7)?
@@ -2832,9 +2855,11 @@ export interface ScopeCounts {
   windowed: boolean;
 }
 
-export interface CustomersPayload {
-  fields: Field[];
-  rows: Row[];
+export interface CustomersPayload {
+  fields: Field[];
+  rows: Row[];
+  /** false on automation-owned child databases whose records are populated only by the engine */
+  recordsMutable?: boolean;
   workspace?: GridWorkspace;
   /** Present only for server-windowed tables. See ScopeCounts. */
   counts?: ScopeCounts;

web/src/customer-grid/useCustomerData.ts

@@ -158,12 +158,29 @@ export interface CustomerData {
   patchOverlay: (pid: number, updates: Partial<Row>) => void;
 }
 
-export function useCustomerData(scope: SurfaceScope = "customer"): CustomerData {
+interface CustomerDataOptions {
+  /** A linked-record grid is nested inside its parent grid. It must not replace the
+   * module-level write scope that the parent still owns. */
+  bindSurface?: boolean;
+  /** Linked-record grids deliberately open on a clean Grid view, not on the target
+   * database's last saved view. The rows payload already carries its field contract. */
+  includeWorkspace?: boolean;
+  /** Defence in depth for read-only nested grids. */
+  writable?: boolean;
+}
+
+export function useCustomerData(
+  scope: SurfaceScope = "customer",
+  options: CustomerDataOptions = {},
+): CustomerData {
+  const bindSurface = options.bindSurface !== false;
+  const includeWorkspace = options.includeWorkspace !== false;
+  const writable = options.writable !== false;
   // ⛔ SET BEFORE THE FIRST FETCH, not in an effect. Every event the grid emits carries this
   // scope, and an effect would run AFTER the first render — so a write triggered by that render
   // would be posted as `customer` from the Cohort surface. Module scope + a synchronous set is
   // what keeps the read and the write talking about the same surface.
-  setSurfaceScope(scope);
+  if (bindSurface) setSurfaceScope(scope);
   // Wave 16 C-TOPIC: the topic is DERIVED from the scope (cohort = the customer topic), so
   // the rows fetch, the patch route and the workspace scope can never disagree about which
   // table this is.
@@ -257,12 +274,14 @@ export function useCustomerData(scope: SurfaceScope = "customer"): CustomerData
     loudScope.current = scope;
 
     async function load() {
-      const [payload, workspace] = await Promise.all([
-        fetchTopicRows(topic),
+      const [payload, workspace] = await Promise.all([
+        fetchTopicRows(topic),
         // A workspace that cannot be read at first paint is not "a table with no saved
         // views" — it is a table whose schema we do not have, and the grid used to fill
         // that hole from a localStorage bucket shared with every other surface.
-        fetchWorkspace(scope, { announceFailure: firstPaint }),
+        includeWorkspace
+          ? fetchWorkspace(scope, { announceFailure: firstPaint })
+          : Promise.resolve(null),
       ]);
       if (cancelled || !payload) return;
       setData(workspace ? withWorkspace(payload, workspace) : payload);
@@ -277,8 +296,9 @@ export function useCustomerData(scope: SurfaceScope = "customer"): CustomerData
     // nothing and cost the expensive call. The measure channel rides the workspace
     // (see `withWorkspace`), so a NEW measure column's values arrive on this cheap
     // re-read too — no pool refetch.
-    async function reread() {
-      const ws = await fetchWorkspace(scope);
+    async function reread() {
+      if (!includeWorkspace) return;
+      const ws = await fetchWorkspace(scope);
       if (cancelled || !ws) return;      // absent stays absent — never blank a live panel
       setData((prev) => (prev ? withWorkspace(prev, ws) : prev));
     }
@@ -332,13 +352,14 @@ export function useCustomerData(scope: SurfaceScope = "customer"): CustomerData
     // `topic` is a stable module constant per scope (topicForScope — ut topics are memoized
     // per key for exactly this dependency), so this cannot loop. `rowsBump` re-runs it on the
     // rows-stale signal only.
-  }, [scope, topic, rowsBump]);
+  }, [scope, topic, rowsBump, includeWorkspace]);
 
   // Optimistic overlay write: apply the edit immediately (so the grid cell and
   // the detail panel reflect it at once), snapshot the pre-edit entry, then
   // PATCH. On a non-ok response, restore the snapshot so the UI never shows a
   // value the server rejected.
-  const patchOverlay = useCallback((pid: number, updates: Partial<Row>) => {
+  const patchOverlay = useCallback((pid: number, updates: Partial<Row>) => {
+    if (!writable) return;
     let prevEntry: Partial<Row> | undefined;
     setOverlayEdits((prev) => {
       prevEntry = prev[pid];
@@ -403,7 +424,7 @@ export function useCustomerData(scope: SurfaceScope = "customer"): CustomerData
         return next;
       });
     });
-  }, [topic]);
+  }, [topic, writable]);
 
   return {
     fields: data?.fields ?? [],

web/src/customer-grid/useGridSelection.ts

@@ -42,9 +42,12 @@ export interface GridSelectionApi {
 export function useGridSelection(
   visibleRows: VisibleRow[],
   pidToIndex: Map<number, number>,
-  colCount: number
+  colCount: number,
+  initialSelectedPids: readonly number[] = []
 ): GridSelectionApi {
-  const [selectedPids, setSelectedPids] = useState<Set<number>>(new Set());
+  const [selectedPids, setSelectedPids] = useState<Set<number>>(
+    () => new Set(initialSelectedPids)
+  );
   const [current, setCurrent] = useState<GridSelection["current"] | undefined>(
     undefined
   );

web/src/index.css

@@ -1435,13 +1435,49 @@ body {
   cursor: pointer;
 }
 .cg-option-add:hover { text-decoration: underline; }
-.cg-field-hint {
+.cg-field-hint {
   display: block;
   font-size: var(--lp-fs-3xs);
   line-height: 1.45;
   color: #8a8a96;
   padding: 2px 0 4px;
-}
+}
+
+.cg-rollup-conditions {
+  display: grid;
+  gap: 8px;
+}
+
+.cg-rollup-conditions__head {
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  gap: 10px;
+  color: var(--cg-text);
+  font-size: var(--lp-fs-2xs);
+  font-weight: 650;
+}
+
+.cg-rollup-conditions__head .cg-btn {
+  min-height: 28px;
+  padding: 4px 9px;
+}
+
+.cg-rollup-condition {
+  display: grid;
+  grid-template-columns: minmax(0, 1fr) minmax(0, 1fr) 30px;
+  align-items: center;
+  gap: 6px;
+}
+
+.cg-rollup-condition > input {
+  grid-column: 1 / span 2;
+}
+
+.cg-rollup-condition .cg-icon-btn {
+  width: 30px;
+  height: 30px;
+}
 /* why a control is unavailable, in the place the control would have been. An empty popover
    reads as broken; a sentence reads as a decision (CG-3). */
 .cg-pop-note {