Reactive mesh · Phase M3: mesh Modal endpoint + app wiring (flag-gated)

modal_mesh_endpoint.py (new): LLaMA-Mesh on Modal (trikona-mesh app, scale-to-
zero), returns raw mesh text; geometry runs in the backend, not here. DEPLOY
FLAGGED — never run by the loop. app.py: _mesh_emit POSTs to MESH_ENDPOINT_URL
(server-side, hidden) and _codegen_generate injects mesh_fn into pipeline.build
only when TRIKONA_MESH + URL are set. Offline tests via canned _model_emit/
_mesh_emit: routes to a valid mesh contract when on; unchanged (guided, mesh
never called) when off; endpoint module parses. Suite green.

app.py

@@ -40,6 +40,12 @@ MODAL_TIMEOUT = int(os.environ.get("MODAL_TIMEOUT", "150"))
 CODEGEN = bool(os.environ.get("TRIKONA_CODEGEN"))
 _SPEC_CACHE: dict[str, dict] = {}  # prompt -> last build_spec (for model-free regen)
 
+# Reactive mesh fallback (SPEC_REACTIVE_MESH): when codegen can't construct a
+# shape, route to LLaMA-Mesh. Opt-in via TRIKONA_MESH + MESH_ENDPOINT_URL; the
+# mesh endpoint URL stays server-side, never exposed to the browser.
+MESH_URL = os.environ.get("MESH_ENDPOINT_URL")
+MESH_ON = bool(os.environ.get("TRIKONA_MESH"))
+
 EXAMPLE_PROMPTS = [
     "dodecahedron",
     "octahedron",
@@ -99,6 +105,36 @@ def _model_emit(prompt: str, params: dict | None, feedback: str | None = None) -
     return obj if isinstance(obj, dict) else {"error": "model_invalid_json"}
 
 
+def _mesh_emit(prompt: str) -> str | None:
+    """Reactive mesh tier: POST to the LLaMA-Mesh endpoint, return raw mesh text.
+
+    Returns None on any failure (the pipeline then falls through to guided). Keeps
+    the mesh endpoint URL server-side, like the codegen endpoint.
+    """
+    if not MESH_URL or requests is None:
+        return None
+    deadline = time.monotonic() + MODAL_TIMEOUT
+    try:
+        while True:
+            resp = requests.post(MESH_URL, json={"prompt": prompt},
+                                 timeout=MODAL_TIMEOUT, allow_redirects=False)
+            if resp.is_redirect or resp.status_code in (301, 302, 303, 307, 308):
+                if time.monotonic() >= deadline:
+                    return None
+                time.sleep(2)
+                continue
+            data = resp.content.decode("utf-8", "replace")
+            try:
+                obj = json.loads(data)
+                if isinstance(obj, dict) and isinstance(obj.get("text"), str):
+                    return obj["text"]
+            except (json.JSONDecodeError, TypeError):
+                pass
+            return data  # endpoint returned raw text rather than {"text": ...}
+    except Exception:
+        return None
+
+
 def _regen_modelfree(spec: dict, params: dict | None) -> str | None:
     """Re-run a cached generator with new params — no model call. None on failure."""
     from geometry import assembler, metrics, primitives
@@ -123,9 +159,10 @@ def _regen_modelfree(spec: dict, params: dict | None) -> str | None:
 
 def _codegen_generate(prompt: str, params: dict | None = None,
                       cache_key: str | None = None) -> str:
-    """Full codegen path via the pipeline (gate -> retry -> primitive fallback)."""
+    """Full codegen path via the pipeline (gate -> retry -> primitive -> mesh)."""
     from geometry import pipeline
-    result, meta = pipeline.build(_model_emit, prompt, params=params)
+    mesh_fn = _mesh_emit if (MESH_ON and MESH_URL) else None
+    result, meta = pipeline.build(_model_emit, prompt, params=params, mesh_fn=mesh_fn)
     if cache_key is not None and meta.get("spec") is not None:
         _SPEC_CACHE[cache_key] = meta["spec"]
     return json.dumps(result)

modal_mesh_endpoint.py

@@ -0,0 +1,73 @@
+"""Trikona mesh endpoint — LLaMA-Mesh on Modal (SPEC_REACTIVE_MESH §6).
+
+The reactive fallback for ORGANIC / freeform shapes that the codegen path can't
+express. Returns the model's RAW mesh text ({"text": ...}); the Gradio backend
+(geometry.mesh_adapter) parses it into the frozen contract — no geometry runs
+here. Structured like modal_endpoint.py: on-demand (scale-to-zero), weights on a
+Volume, model loaded inside @modal.enter on the GPU container.
+
+FLAGGED — the build loop writes this file but MUST NOT deploy it. To enable:
+    modal deploy modal_mesh_endpoint.py
+then copy the URL into the Space secret MESH_ENDPOINT_URL and set TRIKONA_MESH=1.
+Confirm the exact LLaMA-Mesh HF repo id below before deploying.
+"""
+import json
+from pathlib import Path
+
+import modal
+from pydantic import BaseModel
+
+# LLaMA-Mesh: ~8B (LLaMA-3.1-8B fine-tune), within the 32B cap, fits an A100-40GB.
+# VERIFY this repo id before deploy (e.g. the original "Zhengyi/LLaMA-Mesh" or an
+# NVIDIA mirror).
+MODEL_NAME = "Zhengyi/LLaMA-Mesh"
+MAX_MODEL_LEN = 8192      # meshes are long token sequences
+GPU_MEMORY_UTILIZATION = 0.90
+TEMPERATURE = 0.6         # some diversity helps organic mesh generation
+MAX_TOKENS = 7000         # room for a sizeable mesh; adapter enforces hard limits
+
+app = modal.App("trikona-mesh")
+
+hf_cache = modal.Volume.from_name("trikona-mesh-hf-cache", create_if_missing=True)
+HF_CACHE_DIR = "/root/.cache/huggingface"
+
+image = (
+    modal.Image.debian_slim(python_version="3.11")
+    .pip_install("vllm==0.6.3", "transformers==4.46.3", "fastapi[standard]", "pydantic>=2")
+    .env({"HF_HOME": HF_CACHE_DIR})
+)
+
+
+class MeshRequest(BaseModel):
+    prompt: str
+
+
+@app.cls(
+    gpu="A100",
+    image=image,
+    volumes={HF_CACHE_DIR: hf_cache},
+    scaledown_window=300,
+    timeout=180,
+    max_containers=1,
+    startup_timeout=600,
+)
+class MeshModel:
+    @modal.enter()
+    def load(self):
+        from vllm import LLM, SamplingParams
+
+        self.llm = LLM(
+            model=MODEL_NAME,
+            max_model_len=MAX_MODEL_LEN,
+            gpu_memory_utilization=GPU_MEMORY_UTILIZATION,
+        )
+        self.sampling_params = SamplingParams(temperature=TEMPERATURE, max_tokens=MAX_TOKENS)
+
+    @modal.fastapi_endpoint(method="POST")
+    def generate(self, request: MeshRequest) -> dict:
+        # LLaMA-Mesh emits OBJ-like text when asked to create a 3D model. The
+        # backend (mesh_adapter) extracts v/f lines, so we return raw text.
+        messages = [{"role": "user",
+                     "content": f"Create a 3D obj model of {request.prompt}."}]
+        outputs = self.llm.chat(messages, self.sampling_params)
+        return {"text": outputs[0].outputs[0].text}

tests/test_mesh_integration.py

@@ -0,0 +1,71 @@
+"""Phase M3 — mesh endpoint + app wiring (SPEC_REACTIVE_MESH §6, §7). Offline only.
+
+Uses canned _model_emit / _mesh_emit (no live Modal). Verifies the codegen path
+routes to mesh when the flag is on, stays unchanged when off, and that the mesh
+endpoint module parses. Live deploy + live run are FLAGGED (not tested).
+"""
+import ast
+import importlib.util
+import json
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+import app  # noqa: E402
+
+_spec = importlib.util.spec_from_file_location("schema_mod", ROOT / "tests" / "test_schema.py")
+_schema = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(_schema)
+validate_shape = _schema.validate_shape
+
+CUBE_OBJ = (ROOT / "tests" / "fixtures" / "mesh" / "cube_quads.obj").read_text()
+
+# An organic build_spec whose generator yields a degenerate mesh (fails the gate)
+# with no matching primitive -> codegen exhausts -> mesh tier.
+DEGEN_ORGANIC = {
+    "type": "build_spec", "shape_id": "blob_creature", "regularity": "freeform",
+    "generator": {"lang": "python", "entry": "build",
+                  "code": "def build(p):\n    return [[0,0,0],[1,0,0],[0.5,1e-9,0]], [[0,1,2,'b']]\n"},
+    "components": [{"id": "b", "label": "b", "param_dependencies": []}],
+    "params": [], "constraints": [], "construction_steps": [],
+}
+
+
+def test_mesh_flag_defaults_off():
+    assert app.MESH_ON is False
+
+
+def test_mesh_emit_returns_none_without_url(monkeypatch):
+    monkeypatch.setattr(app, "MESH_URL", None)
+    assert app._mesh_emit("anything") is None
+
+
+def test_codegen_routes_to_mesh_when_enabled(monkeypatch):
+    monkeypatch.setattr(app, "MESH_ON", True)
+    monkeypatch.setattr(app, "MESH_URL", "http://stub")
+    monkeypatch.setattr(app, "_model_emit", lambda prompt, params, feedback=None: DEGEN_ORGANIC)
+    monkeypatch.setattr(app, "_mesh_emit", lambda prompt: CUBE_OBJ)
+    contract = json.loads(app._codegen_generate("a blob creature", cache_key="k"))
+    assert contract.get("type") != "guided_response"
+    assert validate_shape(contract) == []
+    assert contract["params"] == []
+    assert contract["shape_id"].startswith("mesh_")
+
+
+def test_codegen_stays_guided_when_mesh_off(monkeypatch):
+    monkeypatch.setattr(app, "MESH_ON", False)            # flag off
+    monkeypatch.setattr(app, "_model_emit", lambda prompt, params, feedback=None: DEGEN_ORGANIC)
+    called = {"mesh": False}
+    monkeypatch.setattr(app, "_mesh_emit", lambda prompt: called.__setitem__("mesh", True) or CUBE_OBJ)
+    contract = json.loads(app._codegen_generate("a blob creature"))
+    assert contract.get("type") == "guided_response"     # no mesh tier
+    assert called["mesh"] is False                        # _mesh_emit never used
+
+
+def test_mesh_endpoint_module_parses():
+    src = (ROOT / "modal_mesh_endpoint.py").read_text()
+    ast.parse(src)  # syntax-valid; deploy is FLAGGED, not exercised here
+    assert "trikona-mesh" in src and "MeshRequest" in src