[eric] apps: publishing refuses to silently drop a generated app's FastAPI backend

backend/apps/outputs/outputs.py

@@ -17,6 +17,7 @@ from backend.apps.outputs.models import (
 )
 from backend.apps.outputs.code_safety import get_code_warnings
 from backend.apps.outputs.executor import execute_backend_code
+from backend.apps.outputs.publish_capability import check_publish_capability
 from backend.apps.outputs.publish_common import slugify, PublishError
 from backend.apps.outputs.publish_scan import scan_for_publish, quick_ast_gate
 from backend.apps.outputs.publish_build import build_static, collect_bundle
@@ -772,12 +773,16 @@ async def publish_output(body: PublishRequest):
     output = load(body.output_id)
     settings = load_settings()
     if not body.force:
+        capability = check_publish_capability(output).findings
         ast = quick_ast_gate(output)
-        if ast:
+        if capability or ast:
             return PublishResult(
                 ok=False,
                 blocked=True,
-                review=PublishReview(verdict="warn", findings=ast),
+                review=PublishReview(
+                    verdict="block" if capability else "warn",
+                    findings=capability + ast,
+                ),
             ).model_dump()
 
     output.publish_status = "publishing"

backend/apps/outputs/publish_capability.py

@@ -0,0 +1,123 @@
+"""Catch the publish cliff before it ships: an app whose frontend calls its own
+FastAPI backend works in preview and breaks on its public URL.
+
+Publishing uploads a STATIC bundle. The edge serves that bundle plus exactly two
+runtime bridges (`/__compute`, which runs a single sandboxed `backend.py`, and
+`/__llm`); there is no `/api/*` route, so every `/api/...` fetch falls through to
+the static catch-all and 404s. Nothing else in the publish path notices, because
+`publish_scan` is a SECURITY scan. This module is the capability scan."""
+from __future__ import annotations
+
+import os
+import re
+from typing import List
+
+from pydantic import BaseModel, ConfigDict
+from typeguard import typechecked
+
+from backend.apps.outputs.models import Output, PublishReview
+from backend.apps.outputs.publish_common import is_webapp, workspace_dir
+from backend.apps.outputs.workspace_io import WALK_SKIP_DIRS
+
+P_FRONTEND_EXTS = (".ts", ".tsx", ".js", ".jsx", ".vue", ".svelte", ".html")
+P_MAX_FILE_BYTES = 512 * 1024
+P_MAX_LISTED = 8
+# Matches /api/foo, "/api", '/api' and `/api` but not /apiary or /rapid.
+P_API_CALL = re.compile(r"/api(?:/|[\"'`]|$)")
+
+
+class PublishCapabilityReport(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    backend_enabled: bool = False
+    backend_port: str = ""
+    api_callers: List[str] = []
+    findings: List[str] = []
+
+
+@typechecked
+def p_backend_port(root: str) -> str:
+    """The workspace's BACKEND_PORT, or "" when the backend was never enabled."""
+    env_path = os.path.join(root, ".env")
+    try:
+        with open(env_path, "r", encoding="utf-8", errors="replace") as fh:
+            for line in fh:
+                key, _, value = line.partition("=")
+                if key.strip() != "BACKEND_PORT":
+                    continue
+                port = value.split("#", 1)[0].strip()
+                return "" if port.upper() in ("", "NONE") else port
+    except OSError:
+        return ""
+    return ""
+
+
+@typechecked
+def p_api_callers(root: str) -> List[str]:
+    """Frontend files that reach for /api/..., relative to the workspace root."""
+    hits: List[str] = []
+    for base, dirs, fnames in os.walk(root):
+        dirs[:] = [d for d in dirs if d not in WALK_SKIP_DIRS and d != "backend"]
+        for fn in fnames:
+            if not fn.lower().endswith(P_FRONTEND_EXTS):
+                continue
+            full = os.path.join(base, fn)
+            if os.path.islink(full):
+                continue
+            try:
+                if os.path.getsize(full) > P_MAX_FILE_BYTES:
+                    continue
+                with open(full, "r", encoding="utf-8", errors="replace") as fh:
+                    if P_API_CALL.search(fh.read()):
+                        hits.append(os.path.relpath(full, root))
+            except OSError:
+                continue
+    return sorted(hits)
+
+
+@typechecked
+def check_publish_capability(output: Output) -> PublishCapabilityReport:
+    """Does this app depend on something publishing cannot carry?"""
+    if not is_webapp(output):
+        return PublishCapabilityReport()
+    root = workspace_dir(output)
+    port = p_backend_port(root)
+    has_backend = bool(port) or os.path.isfile(os.path.join(root, "backend", "main.py"))
+    if not has_backend:
+        return PublishCapabilityReport()
+    callers = p_api_callers(root)
+    if not callers:
+        return PublishCapabilityReport(backend_enabled=True, backend_port=port)
+    shown = ", ".join(callers[:P_MAX_LISTED])
+    if len(callers) > P_MAX_LISTED:
+        shown += f", and {len(callers) - P_MAX_LISTED} more"
+    return PublishCapabilityReport(
+        backend_enabled=True,
+        backend_port=port,
+        api_callers=callers,
+        findings=[
+            "This app has a FastAPI backend, and publishing does not upload it. "
+            f"{len(callers)} frontend file(s) call /api/... ({shown}); those requests "
+            "will 404 on the published URL even though they work in preview.",
+            "A published app gets static files plus two same-origin bridges: "
+            "window.OUTPUT_COMPUTE(input), which runs a single sandboxed backend.py "
+            "(pure compute, no network, no disk, 30s limit), and window.OUTPUT_LLM(body). "
+            "Move the server-side logic into backend.py to use OUTPUT_COMPUTE, or keep "
+            "this app local instead of publishing it.",
+        ],
+    )
+
+
+@typechecked
+def merge_capability(output: Output, review: PublishReview) -> PublishReview:
+    """Capability findings ride OUTSIDE the security memo, which is keyed on a
+    source hash that never sees .env, so a backend_init.sh run would otherwise
+    return a cached all-clear."""
+    report = check_publish_capability(output)
+    if not report.findings:
+        return review
+    return PublishReview(
+        verdict="block",
+        findings=report.findings + review.findings,
+        scanned_files=review.scanned_files,
+    )

backend/apps/outputs/publish_scan.py

@@ -17,6 +17,7 @@ from typing import Literal
 
 from backend.apps.outputs.code_safety import get_code_warnings
 from backend.apps.outputs.models import Output, PublishReview
+from backend.apps.outputs.publish_capability import merge_capability
 from backend.apps.outputs.publish_common import is_webapp, workspace_dir
 from backend.apps.outputs.workspace_io import WALK_SKIP_DIRS
 
@@ -151,7 +152,7 @@ async def scan_for_publish(output: Output, settings) -> PublishReview:
     cached = memo.get(key)
     if cached is not None:
         memo.move_to_end(key)
-        return cached
+        return merge_capability(output, cached)
     ast_findings, scanned = p_ast_findings(src)
     llm_list, llm_sev = await llm_findings(src, settings)
     findings = ast_findings + llm_list
@@ -169,7 +170,7 @@ async def scan_for_publish(output: Output, settings) -> PublishReview:
     memo.move_to_end(key)
     while len(memo) > P_MEMO_MAX:
         memo.popitem(last=False)
-    return review
+    return merge_capability(output, review)
 
 
 def quick_ast_gate(output: Output) -> list[str]:

backend/tests/test_publish_capability.py

@@ -0,0 +1,208 @@
+"""The publish cliff: an app with a FastAPI backend works in preview and 404s on
+its public URL, because publishing uploads a static bundle and the edge has no
+/api/* route. Before this gate, nothing in the publish path noticed."""
+import uuid
+
+import pytest
+
+from backend.apps.outputs import publish_common
+from backend.apps.outputs.models import Output, PublishReview
+from backend.apps.outputs.publish_capability import (
+    check_publish_capability,
+    merge_capability,
+)
+
+
+@pytest.fixture
+def p_ws_root(tmp_path, monkeypatch):
+    root = tmp_path / "ws"
+    root.mkdir()
+    monkeypatch.setattr(publish_common, "OUTPUTS_WORKSPACE_DIR", str(root))
+    return root
+
+
+def p_app(ws_root, *, workspace: bool = True) -> Output:
+    wsid = uuid.uuid4().hex if workspace else None
+    if wsid:
+        (ws_root / wsid).mkdir()
+    return Output(
+        name="Demo", description="", icon="view_quilt",
+        input_schema={"type": "object", "properties": {}, "required": []},
+        files={}, workspace_id=wsid, session_id=None,
+    )
+
+
+def p_seed(ws_root, output, *, env: str, frontend: str = "", backend_main: bool = False):
+    root = ws_root / (output.workspace_id or "")
+    (root / ".env").write_text(env)
+    if frontend:
+        fe = root / "frontend" / "src"
+        fe.mkdir(parents=True)
+        (fe / "api.ts").write_text(frontend)
+    if backend_main:
+        be = root / "backend"
+        be.mkdir()
+        (be / "main.py").write_text("app = 1\n")
+    return root
+
+
+def test_flat_app_has_no_capability_problem(p_ws_root):
+    out = p_app(p_ws_root, workspace=False)
+    assert check_publish_capability(out).findings == []
+
+
+def test_frontend_only_workspace_is_clean(p_ws_root):
+    out = p_app(p_ws_root)
+    p_seed(p_ws_root, out, env="BACKEND_PORT=NONE\n", frontend="fetch('/data.json')\n")
+    report = check_publish_capability(out)
+    assert report.backend_enabled is False
+    assert report.findings == []
+
+
+def test_backend_plus_api_calls_is_blocked(p_ws_root):
+    out = p_app(p_ws_root)
+    p_seed(
+        p_ws_root, out,
+        env="BACKEND_PORT=8123 # chosen by backend_init.sh\nFRONTEND_PORT=4949\n",
+        frontend="export const JOBS = '/api/jobs';\nfetch(JOBS);\n",
+        backend_main=True,
+    )
+    report = check_publish_capability(out)
+    assert report.backend_enabled is True
+    assert report.backend_port == "8123"
+    assert report.api_callers == ["frontend/src/api.ts"]
+    assert len(report.findings) == 2
+    joined = " ".join(report.findings)
+    assert "frontend/src/api.ts" in joined
+    assert "OUTPUT_COMPUTE" in joined
+
+
+def test_backend_with_no_callers_loses_nothing(p_ws_root):
+    """A backend nothing calls is dead weight, not broken functionality."""
+    out = p_app(p_ws_root)
+    p_seed(
+        p_ws_root, out, env="BACKEND_PORT=8123\n",
+        frontend="const x = 1;\n", backend_main=True,
+    )
+    report = check_publish_capability(out)
+    assert report.backend_enabled is True
+    assert report.findings == []
+
+
+def test_backend_dir_without_port_still_counts(p_ws_root):
+    """backend_init.sh calls this state inconsistent; publishing must not shrug."""
+    out = p_app(p_ws_root)
+    p_seed(
+        p_ws_root, out, env="BACKEND_PORT=NONE\n",
+        frontend="fetch('/api/things')\n", backend_main=True,
+    )
+    assert check_publish_capability(out).findings != []
+
+
+def test_apiary_is_not_an_api_call(p_ws_root):
+    """Prefix matching would flag /apiary and /rapid; the boundary is load-bearing."""
+    out = p_app(p_ws_root)
+    p_seed(
+        p_ws_root, out, env="BACKEND_PORT=8123\n",
+        frontend="fetch('/apiary/bees'); fetch('/rapid');\n", backend_main=True,
+    )
+    assert check_publish_capability(out).findings == []
+
+
+def test_backend_dir_is_not_scanned_for_callers(p_ws_root):
+    """The backend's own source mentioning /api must not count as a frontend caller."""
+    out = p_app(p_ws_root)
+    root = p_seed(p_ws_root, out, env="BACKEND_PORT=8123\n", backend_main=True)
+    (root / "backend" / "routes.js").write_text("// mounts /api/jobs\n")
+    assert check_publish_capability(out).findings == []
+
+
+def test_merge_escalates_a_clean_security_review_to_block(p_ws_root):
+    out = p_app(p_ws_root)
+    p_seed(
+        p_ws_root, out, env="BACKEND_PORT=8123\n",
+        frontend="fetch('/api/x')\n", backend_main=True,
+    )
+    merged = merge_capability(out, PublishReview(verdict="clean", findings=[]))
+    assert merged.verdict == "block"
+    assert len(merged.findings) == 2
+
+
+def test_merge_preserves_security_findings_and_order(p_ws_root):
+    out = p_app(p_ws_root)
+    p_seed(
+        p_ws_root, out, env="BACKEND_PORT=8123\n",
+        frontend="fetch('/api/x')\n", backend_main=True,
+    )
+    merged = merge_capability(
+        out, PublishReview(verdict="warn", findings=["reads os.environ"], scanned_files=["a.py"]),
+    )
+    assert merged.findings[-1] == "reads os.environ"
+    assert merged.scanned_files == ["a.py"]
+
+
+def test_merge_is_a_passthrough_when_nothing_is_lost(p_ws_root):
+    out = p_app(p_ws_root)
+    p_seed(p_ws_root, out, env="BACKEND_PORT=NONE\n")
+    review = PublishReview(verdict="warn", findings=["something else"])
+    assert merge_capability(out, review) is review
+
+
+def test_missing_env_file_does_not_explode(p_ws_root):
+    """A half-seeded workspace must read as 'no backend', not raise."""
+    out = p_app(p_ws_root)
+    report = check_publish_capability(out)
+    assert report.backend_enabled is False
+    assert report.findings == []
+
+
+@pytest.mark.asyncio
+async def test_publish_route_refuses_to_ship_a_broken_app(p_ws_root, monkeypatch):
+    """The route is where the loss was silent: it built and uploaded regardless."""
+    from backend.apps.outputs import outputs as outputs_mod
+    from backend.apps.outputs.models import PublishRequest
+
+    out = p_app(p_ws_root)
+    p_seed(
+        p_ws_root, out, env="BACKEND_PORT=8123\n",
+        frontend="fetch('/api/jobs')\n", backend_main=True,
+    )
+    built = []
+    monkeypatch.setattr(outputs_mod, "load", lambda _: out)
+    monkeypatch.setattr(outputs_mod, "load_settings", lambda: None)
+    monkeypatch.setattr(outputs_mod, "build_static", lambda o: built.append(o))
+
+    res = await outputs_mod.publish_output(PublishRequest(output_id=out.id))
+
+    assert res["ok"] is False
+    assert res["blocked"] is True
+    assert res["review"]["verdict"] == "block"
+    assert built == [], "publish must not build once the gate has fired"
+
+
+@pytest.mark.asyncio
+async def test_force_is_still_the_escape_hatch(p_ws_root, monkeypatch):
+    """A user who read the finding can still ship; the gate informs, it does not trap."""
+    from backend.apps.outputs import outputs as outputs_mod
+    from backend.apps.outputs.models import PublishRequest
+
+    out = p_app(p_ws_root)
+    p_seed(
+        p_ws_root, out, env="BACKEND_PORT=8123\n",
+        frontend="fetch('/api/jobs')\n", backend_main=True,
+    )
+    reached = []
+
+    async def p_boom(_):
+        reached.append(True)
+        raise publish_common.PublishError("stopped past the gate")
+
+    monkeypatch.setattr(outputs_mod, "load", lambda _: out)
+    monkeypatch.setattr(outputs_mod, "save", lambda _: None)
+    monkeypatch.setattr(outputs_mod, "load_settings", lambda: None)
+    monkeypatch.setattr(outputs_mod, "build_static", p_boom)
+
+    res = await outputs_mod.publish_output(PublishRequest(output_id=out.id, force=True))
+
+    assert reached == [True], "force must skip the gate and reach the build"
+    assert res["ok"] is False