Add A2APP — make agent writes to Living UIs verifiable

app/data/action/living_ui_actions.py

@@ -191,10 +191,13 @@ async def living_ui_scaffold(input_data: dict) -> dict:
 @action(
     name="living_ui_notify_ready",
     description=(
-        "Launch, verify, and serve a Living UI project. "
-        "Call this after building the Living UI code. "
-        "This action installs dependencies, runs tests, starts the backend and frontend, "
-        "and notifies the browser. Returns test errors if anything fails."
+        "Launch or RELAUNCH a Living UI project: installs dependencies, runs the "
+        "validation gate, restarts backend and frontend, notifies the browser. "
+        "Call this ONLY after CREATING or CHANGING the app's CODE (migrations, "
+        "hooks, frontend). An app that is already running does NOT need it — "
+        "adding, editing or deleting DATA never requires a relaunch, and calling "
+        "it then rebuilds and restarts a live app for no reason. "
+        "Returns test errors if anything fails."
     ),
     default=False,
     mode="CLI",
@@ -329,9 +332,13 @@ async def living_ui_notify_ready(input_data: dict) -> dict:
         "UI project: a real browser (headless) drives the app "
         "feature-by-feature against reference/requirements.md. A clean "
         "verdict announces the app to the user — the ONLY way a Living UI "
-        "build completes. Observed defects return the failure report: fix, "
+        "BUILD completes. Observed defects return the failure report: fix, "
         "relaunch with living_ui_notify_ready, then call this again. "
-        "Requires the app to be running (living_ui_notify_ready first)."
+        "Requires the app to be running (living_ui_notify_ready first). "
+        "ONLY after building or modifying the app's CODE. NEVER after a data "
+        "change: it drives a real browser and CLICKS through the UI, including "
+        "buttons that create records, so running it against an app holding the "
+        "user's data can alter that data."
     ),
     default=False,
     mode="CLI",

app/living_ui/agent_view.py

@@ -0,0 +1,184 @@
+# -*- coding: utf-8 -*-
+"""
+What the agent and the user each SEE of a Living UI.
+
+Two jobs, both about presentation rather than mechanism:
+
+1. `schema_block()` — the app's data model, inlined into the agent's prompt.
+   Advisory pointers do not work on weak models: across two recorded incidents
+   the agent ignored "Read LIVING_UI.md", never ran `lui ops`, and guessed
+   collection names instead (`items`, `tasks`). It cannot ignore what is
+   already in its context.
+
+2. `humanise_write()` — one plain sentence describing what a write actually
+   did, built from the stored record. The user should never read
+   `cards.create [kapp872i5etufxb] due_date='2026-07-31 00:00:00.000Z'`.
+
+Both read the app's own A2APP `describe` surface, so neither can drift from
+what the app actually is.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+import urllib.request
+from datetime import datetime
+from typing import Any, Dict, Optional
+
+try:
+    from app.logger import logger
+except Exception:  # pragma: no cover
+    import logging
+
+    logger = logging.getLogger(__name__)
+
+# describe is cheap but not free, and it is fetched on every user message.
+# A few minutes of staleness is harmless: the app validates writes itself, so
+# a stale block can only cost a retry, never a bad write.
+_CACHE: Dict[str, tuple] = {}
+_TTL_SECONDS = 300
+_TIMEOUT_SECONDS = 2.0
+
+_SKIP_FIELDS = {"id", "collectionId", "collectionName", "created", "updated"}
+
+
+def _describe(base_url: str) -> Optional[Dict[str, Any]]:
+    """Fetch (and cache) the app's data model. None when the app is down."""
+    cached = _CACHE.get(base_url)
+    if cached is not None and time.time() - cached[0] < _TTL_SECONDS:
+        return cached[1]
+    try:
+        request = urllib.request.Request(
+            f"{base_url}/api/_a2app/describe", headers={"User-Agent": "CraftBot"}
+        )
+        with urllib.request.urlopen(request, timeout=_TIMEOUT_SECONDS) as response:
+            data = json.loads(response.read().decode("utf-8"))
+        _CACHE[base_url] = (time.time(), data)
+        return data
+    except Exception as e:
+        logger.debug(f"[AGENT_VIEW] describe unavailable at {base_url}: {e}")
+        _CACHE[base_url] = (time.time(), None)
+        return None
+
+
+def _type_label(spec: Dict[str, Any]) -> str:
+    """Render a field's type the way the agent needs to see it — including the
+    enum's actual values, whose absence caused a rejected write."""
+    kind = str(spec.get("type", "string"))
+    if kind == "enum" and spec.get("values"):
+        return "one of " + "|".join(str(v) for v in spec["values"])
+    if kind in ("ref", "list<ref>") and spec.get("entity"):
+        arrow = "->" if kind == "ref" else "->[]"
+        return f"{arrow}{spec['entity']}"
+    if spec.get("format"):
+        return str(spec["format"])
+    return kind
+
+
+def schema_block(base_url: str, max_chars: int = 2000) -> Optional[str]:
+    """The data model, compact enough to sit in every prompt.
+
+    Read-only and server-managed fields are omitted: the agent cannot write
+    them, so naming them only invites it to try.
+    """
+    described = _describe(base_url)
+    if not described:
+        return None
+    entities = described.get("entities") or {}
+    if not entities:
+        return None
+
+    lines = []
+    for name, entity in entities.items():
+        fields = []
+        for field_name, spec in (entity.get("fields") or {}).items():
+            if spec.get("readOnly"):
+                continue
+            star = "*" if spec.get("required") else ""
+            fields.append(f"{field_name}({_type_label(spec)}){star}")
+        if fields:
+            lines.append(f"  {name}: {' '.join(fields)}")
+
+    block = "\n".join(lines)
+    if len(block) > max_chars:  # very large apps: names only, still better than nothing
+        block = "\n".join(f"  {n}: {len((e.get('fields') or {}))} fields" for n, e in entities.items())
+    return block
+
+
+def _resolve_ref(base_url: str, entity: str, record_id: str) -> Optional[str]:
+    """A referenced record's human label, so the user reads 'To Do' not an id."""
+    described = _describe(base_url)
+    if not described:
+        return None
+    target = (described.get("entities") or {}).get(entity) or {}
+    label_field = target.get("label")
+    if not label_field:
+        return None
+    try:
+        url = f"{base_url}/api/collections/{entity}/records/{record_id}"
+        with urllib.request.urlopen(url, timeout=_TIMEOUT_SECONDS) as response:
+            record = json.loads(response.read().decode("utf-8"))
+        value = record.get(label_field)
+        return str(value) if value else None
+    except Exception:
+        return None
+
+
+def _humanise_date(value: str) -> str:
+    """'2026-07-31 00:00:00.000Z' -> 'Fri 31 Jul'. Times are kept when present."""
+    text = str(value).strip()
+    try:
+        stamp = datetime.fromisoformat(text.replace("Z", "+00:00").replace(" ", "T", 1))
+    except Exception:
+        return text[:10] or text
+    if stamp.hour == 0 and stamp.minute == 0:
+        return stamp.strftime("%a %-d %b")
+    return stamp.strftime("%a %-d %b %H:%M")
+
+
+_VERBS = {"create": "Added", "update": "Updated", "delete": "Removed"}
+
+
+def humanise_write(base_url: str, collection: str, op: str, record: Dict[str, Any]) -> str:
+    """One sentence a person can read, built from what was actually stored.
+
+    Example: Added "Eat chicken" to To Do — due Fri 31 Jul, priority medium
+    """
+    described = _describe(base_url)
+    entities = (described or {}).get("entities") or {}
+    entity = entities.get(collection) or {}
+    specs = entity.get("fields") or {}
+    label_field = entity.get("label")
+
+    name = record.get(label_field) if label_field else None
+    verb = _VERBS.get(op, "Changed")
+    subject = f'"{name}"' if name else f"a {collection.rstrip('s')}"
+
+    into = ""
+    details = []
+    for key, value in record.items():
+        if key in _SKIP_FIELDS or key == label_field:
+            continue
+        if value in ("", None, [], {}, False, 0):
+            continue
+        spec = specs.get(key) or {}
+        kind = str(spec.get("type", ""))
+
+        if kind == "ref" and spec.get("entity"):
+            resolved = _resolve_ref(base_url, str(spec["entity"]), str(value))
+            if resolved and not into:
+                into = f" to {resolved}"          # the containing thing reads best inline
+                continue
+            details.append(f"{key.replace('_', ' ')} {resolved or value}")
+        elif kind == "datetime":
+            details.append(f"{key.replace('_', ' ').replace(' date', '')} {_humanise_date(value)}")
+        elif kind in ("json", "binary", "list<ref>"):
+            continue                              # nothing a person wants to read
+        else:
+            details.append(f"{key.replace('_', ' ')} {value}")
+
+    sentence = f"{verb} {subject}{into}"
+    if details:
+        sentence += " — " + ", ".join(details[:4])
+    return sentence

app/living_ui/integration_bridge.py

@@ -12,7 +12,8 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Optional
+from pathlib import Path
+from typing import TYPE_CHECKING, Dict, Optional
 
 from aiohttp import web
 import httpx
@@ -38,7 +39,9 @@ class IntegrationBridge:
 
     def __init__(self, manager: "LivingUIManager"):
         self._manager = manager
-        self._http_client = httpx.AsyncClient(timeout=30, follow_redirects=True)
+        # follow_redirects is OFF for the proxy: an allowed host could 302 to an
+        # attacker-controlled host and the injected credentials would follow it.
+        self._http_client = httpx.AsyncClient(timeout=30, follow_redirects=False)
 
     def register_routes(self, app: web.Application) -> None:
         """Register integration bridge routes on the aiohttp app."""
@@ -110,6 +113,39 @@ async def _handle_proxy(self, request: web.Request) -> web.Response:
                 {"error": "Missing required fields: integration, url"}, status=400
             )
 
+        # Gate 1 — capability. An app may only use integrations its manifest
+        # declares. Without this, any app that can reach the bridge can use
+        # every credential the user has connected: a kanban board could send
+        # mail as them. The declaration is the hook the Phase 5 consent flow
+        # attaches to; until then it is at least an explicit, reviewable list.
+        granted, why = self._project_grants(project_id, integration)
+        if not granted:
+            logger.warning(
+                f"[INTEGRATION_BRIDGE] BLOCKED (capability) project={project_id} "
+                f"integration={integration!r}: {why}"
+            )
+            return web.json_response(
+                {"error": f"This app is not permitted to use '{integration}': {why}"},
+                status=403,
+            )
+
+        # Gate 2 — destination. Without it this endpoint is a credential
+        # exfiltration primitive: `url` is caller-controlled and the user's real
+        # OAuth token is injected into whatever host is named, so one line in a
+        # third-party app's pb_hooks could ship a Gmail token anywhere. Fails
+        # CLOSED — an integration with no entry cannot be proxied at all.
+        allowed, resolved = self._resolve_destination(integration, url)
+        if not allowed:
+            logger.warning(
+                f"[INTEGRATION_BRIDGE] BLOCKED proxy from project={project_id} "
+                f"integration={integration!r} url={url!r}: {resolved}"
+            )
+            return web.json_response(
+                {"error": f"Destination not permitted for '{integration}': {resolved}"},
+                status=403,
+            )
+        url = resolved
+
         # Get auth headers from platform client
         auth_headers = self._get_auth_headers(integration)
         if auth_headers is None:
@@ -256,6 +292,108 @@ def _validate_token(self, request: web.Request) -> Optional[str]:
         token = auth[7:]
         return self._manager.validate_bridge_token(token)
 
+    # Where each integration's credentials may be sent: (base, allowed hosts).
+    #
+    # `base` also REPAIRS the proxy. Callers pass a path — the shipped apps do
+    # `callIntegration('gmail','POST','/gmail/v1/users/me/messages/send',…)` —
+    # and nothing ever resolved it, so httpx got a relative URL and every call
+    # failed. crm-system even carries a "Gmail integration unavailable" fallback
+    # because of it. Resolving against `base` fixes that, and a path can never
+    # escape the base, so it is also the safe form.
+    #
+    # Host matching is exact-or-dot-suffix, so "api.github.com.evil.com" does
+    # not pass as "api.github.com". Omission denies — the safe direction.
+    PROXY_DESTINATIONS: Dict[str, tuple] = {
+        "github": ("https://api.github.com", ("api.github.com",)),
+        "gmail": ("https://www.googleapis.com", ("googleapis.com",)),
+        "google_calendar": ("https://www.googleapis.com", ("googleapis.com",)),
+        "google_docs": ("https://docs.googleapis.com", ("googleapis.com",)),
+        "google_drive": ("https://www.googleapis.com", ("googleapis.com",)),
+        "google_youtube": ("https://www.googleapis.com", ("googleapis.com",)),
+        "google_workspace": ("https://www.googleapis.com", ("googleapis.com",)),
+        "outlook": ("https://graph.microsoft.com", ("graph.microsoft.com",)),
+        "slack": ("https://slack.com", ("slack.com",)),
+        "discord": ("https://discord.com", ("discord.com", "discordapp.com")),
+        "notion": ("https://api.notion.com", ("api.notion.com",)),
+        "hubspot": ("https://api.hubapi.com", ("api.hubapi.com",)),
+        "jira": ("https://api.atlassian.com", ("atlassian.net", "api.atlassian.com")),
+        "linkedin": ("https://api.linkedin.com", ("api.linkedin.com",)),
+        "stripe": ("https://api.stripe.com", ("api.stripe.com",)),
+        "line": ("https://api.line.me", ("api.line.me",)),
+        "lark": ("https://open.feishu.cn", ("open.feishu.cn", "open.larksuite.com")),
+        "lark_calendar": ("https://open.feishu.cn", ("open.feishu.cn", "open.larksuite.com")),
+        "lark_drive": ("https://open.feishu.cn", ("open.feishu.cn", "open.larksuite.com")),
+        "telegram_bot": ("https://api.telegram.org", ("api.telegram.org",)),
+        "telegram_user": ("https://api.telegram.org", ("api.telegram.org",)),
+        "twitter": ("https://api.twitter.com", ("api.twitter.com", "api.x.com")),
+        "whatsapp_business": ("https://graph.facebook.com", ("graph.facebook.com",)),
+    }
+
+    def _project_grants(self, project_id: str, integration: str) -> tuple:
+        """(ok, reason) — does this project declare `integration` in its
+        manifest's `capabilities.integrations`? Fails closed."""
+        import json as _json
+
+        try:
+            project = self._manager.get_project(project_id)
+        except Exception as e:
+            return False, f"unknown project ({e})"
+        if project is None:
+            return False, "unknown project"
+
+        try:
+            manifest_path = Path(project.path) / "manifest.json"
+            manifest = _json.loads(manifest_path.read_text(encoding="utf-8"))
+        except Exception as e:
+            return False, f"manifest unreadable ({e})"
+
+        capabilities = manifest.get("capabilities") or {}
+        declared = capabilities.get("integrations")
+        if not isinstance(declared, list):
+            return False, (
+                "manifest declares no capabilities.integrations — add "
+                f'"capabilities": {{"integrations": ["{integration}"]}} to grant it'
+            )
+        if integration not in declared:
+            return False, f"not in capabilities.integrations {declared}"
+        return True, ""
+
+    def _resolve_destination(self, integration: str, url: str) -> tuple:
+        """(ok, resolved_url_or_reason) for this integration's credentials."""
+        from urllib.parse import urlparse
+
+        entry = self.PROXY_DESTINATIONS.get(integration)
+        if not entry:
+            return False, "integration has no permitted destinations"
+        base, allowed = entry
+
+        raw = (url or "").strip()
+        if not raw:
+            return False, "empty url"
+
+        # A path resolves against the base and cannot escape it. Note "//host"
+        # is protocol-relative, NOT a path — urljoin would happily send it to
+        # another host, so it must be rejected here.
+        if raw.startswith("/") and not raw.startswith("//"):
+            return True, base.rstrip("/") + raw
+
+        try:
+            parsed = urlparse(raw)
+        except Exception:
+            return False, "unparseable url"
+
+        if parsed.scheme != "https":
+            return False, f"scheme {parsed.scheme!r} is not https"
+
+        host = (parsed.hostname or "").lower()
+        if not host:
+            return False, "no host in url"
+
+        for candidate in allowed:
+            if host == candidate or host.endswith("." + candidate):
+                return True, raw
+        return False, f"host {host!r} is not one of {', '.join(allowed)}"
+
     def _get_auth_headers(self, platform_id: str) -> Optional[dict]:
         """
         Get authentication headers from a platform client.