Add MindVisualizer explanation quotas

app.py

@@ -1,311 +1,414 @@
 import json
 import os
 import sys
+import hashlib
+from datetime import datetime, timezone
 from functools import lru_cache
 from pathlib import Path
 from typing import Literal
 
 import numpy as np
-from fastapi import FastAPI, HTTPException
-from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import StreamingResponse
-from pydantic import BaseModel, Field
-
-
-FieldMode = Literal["mean", "gated", "mu1", "mu2", "mu3", "mu4", "mu5", "mu6"]
-
-
-class TrajectoryRequest(BaseModel):
-    seed: list[float] = Field(default_factory=lambda: [0.5, 0.5, 0.5], min_length=3, max_length=3)
-    fieldMode: FieldMode = "mean"
-    steps: int = Field(default=700, ge=8, le=1600)
-    dt: float = Field(default=1.0, gt=0, le=8.0)
-    speedScale: float = Field(default=1.0, gt=0, le=8.0)
-
-
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel, Field
+
+
+FieldMode = Literal["mean", "gated", "mu1", "mu2", "mu3", "mu4", "mu5", "mu6"]
+
+
+class TrajectoryRequest(BaseModel):
+    seed: list[float] = Field(default_factory=lambda: [0.5, 0.5, 0.5], min_length=3, max_length=3)
+    fieldMode: FieldMode = "mean"
+    steps: int = Field(default=700, ge=8, le=1600)
+    dt: float = Field(default=1.0, gt=0, le=8.0)
+    speedScale: float = Field(default=1.0, gt=0, le=8.0)
+
+
 class ExplainRequest(BaseModel):
     trajectory: list[list[float]] = Field(min_length=2)
     fieldMode: FieldMode = "mean"
     callLlm: bool = True
 
 
-def mindvis_repo() -> Path:
-    repo = Path(os.environ.get("MINDVIS_REPO", "/app/mindVisualizer")).resolve()
-    if not repo.exists():
-        raise RuntimeError(f"MINDVIS_REPO does not exist: {repo}")
-    if str(repo) not in sys.path:
-        sys.path.insert(0, str(repo))
-    return repo
-
-
-@lru_cache(maxsize=1)
-def services():
-    repo = mindvis_repo()
-    data = repo / "data"
-
-    from src.field_loader import TriLinearSampler, load_field
-    from src.mesh_overlay import FlowMeshOverlay
-
-    flow = load_field(data / "mdn" / "mdn_particles_rdcim_teacher_edge_hq_grid125_meta.json")
-    mean = flow["mean"]
-    mus = flow["mus"]
-    pi = flow["pi"]
-    ent = flow["ENT"]
-    amin = flow["amin"]
-    amax = flow["amax"]
-    span = amax - amin
-    diag = float(np.linalg.norm(span))
-    target_step = 0.01 * max(diag, 1e-6)
-
-    winner = np.argmax(pi, axis=-1)
-    gated = np.zeros_like(mean, dtype=np.float32)
-    for idx, mu in enumerate(mus):
-        gated += mu * (winner[..., None] == idx).astype(np.float32)
-
-    fields = {"mean": mean, "gated": gated}
-    fields.update({f"mu{idx + 1}": mu for idx, mu in enumerate(mus)})
-    samplers = {name: TriLinearSampler(field, amin, amax) for name, field in fields.items()}
-
-    def vmax(field):
-        mag = np.linalg.norm(field.reshape(-1, 3), axis=1)
-        nz = mag[mag > 0]
-        return float(np.percentile(nz, 100.0)) if nz.size else 1.0
-
-    vmax_by_name = {name: vmax(field) for name, field in fields.items()}
-
-    # The original Python app uses FlowMeshOverlay + its cached label grid for
-    # probe interpretation. This API intentionally reuses that path.
-    import vtk
-
-    ren = vtk.vtkRenderer()
-    win = vtk.vtkRenderWindow()
-    win.OffScreenRenderingOn()
-    win.AddRenderer(ren)
-    mesh = FlowMeshOverlay(ren=ren, win=win, mesh_dir=data / "meshes", alignment_file=data / "brain_alignment.json")
-    cache = data / "label_grid_cache.npz"
-    if not mesh.load_label_grid(cache):
-        raise RuntimeError("Python label_grid_cache is missing; run setup_brain_data.py in mindVisualizer first.")
-
-    # Optional finer parcellation, exactly as the desktop flow mode auto-detects.
-    try:
-        from src.extra_parcellation import ExtraParcellation
-
-        atlas = data / "extra_parcellation" / "combined_atlas.nii.gz"
-        labels = data / "extra_parcellation" / "combined_atlas_labels.json"
-        if atlas.exists():
-            extra = ExtraParcellation(atlas, labels if labels.exists() else None)
-            if extra.load():
-                mesh.set_extra_parcellation(extra)
-    except Exception as exc:
-        print(f"[extra-parcellation] disabled: {exc}")
-
-    ent_sampler = TriLinearSampler(mean, amin, amax) if ent is not None else None
-
-    return {
-        "amin": amin,
-        "amax": amax,
-        "span": span,
-        "samplers": samplers,
-        "vmax": vmax_by_name,
-        "target_step": target_step,
-        "mesh": mesh,
-        "entropy": ent,
-        "entropy_sampler": ent_sampler,
-    }
-
-
-app = FastAPI(title="MindVisualizer API", version="1.0.0")
-
-origins = [origin.strip() for origin in os.environ.get("CORS_ORIGINS", "*").split(",") if origin.strip()]
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=origins,
-    allow_methods=["GET", "POST", "OPTIONS"],
-    allow_headers=["*"],
-)
-
-
-@app.get("/health")
-def health():
-    try:
-        svc = services()
-        return {"ok": True, "fieldMin": svc["amin"].tolist(), "fieldMax": svc["amax"].tolist()}
-    except Exception as exc:
-        raise HTTPException(status_code=503, detail=str(exc)) from exc
-
-
-@app.post("/api/mindvis/trajectory")
-def trajectory(req: TrajectoryRequest):
-    svc = services()
-    world, mags, stopped = integrate_probe(req, svc)
-    transitions = analyze_regions(world, mags, req.callLlm if hasattr(req, "callLlm") else False)
-    return {
-        "fieldMode": req.fieldMode,
-        "trajectory": normalized(world, svc["amin"], svc["span"]).tolist(),
-        "worldTrajectory": world.tolist(),
-        "fieldMagnitudes": mags.tolist(),
-        "stopped": stopped,
-        "transitions": transitions,
-        "regionText": format_transition_text(transitions),
-    }
-
-
+CACHE_DIR = Path(os.environ.get("MINDVIS_CACHE_DIR", "/tmp/mindvis-api-cache"))
+CACHE_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def mindvis_repo() -> Path:
+    repo = Path(os.environ.get("MINDVIS_REPO", "/app/mindVisualizer")).resolve()
+    if not repo.exists():
+        raise RuntimeError(f"MINDVIS_REPO does not exist: {repo}")
+    if str(repo) not in sys.path:
+        sys.path.insert(0, str(repo))
+    return repo
+
+
+@lru_cache(maxsize=1)
+def services():
+    repo = mindvis_repo()
+    data = repo / "data"
+
+    from src.field_loader import TriLinearSampler, load_field
+    from src.mesh_overlay import FlowMeshOverlay
+
+    flow = load_field(data / "mdn" / "mdn_particles_rdcim_teacher_edge_hq_grid125_meta.json")
+    mean = flow["mean"]
+    mus = flow["mus"]
+    pi = flow["pi"]
+    ent = flow["ENT"]
+    amin = flow["amin"]
+    amax = flow["amax"]
+    span = amax - amin
+    diag = float(np.linalg.norm(span))
+    target_step = 0.01 * max(diag, 1e-6)
+
+    winner = np.argmax(pi, axis=-1)
+    gated = np.zeros_like(mean, dtype=np.float32)
+    for idx, mu in enumerate(mus):
+        gated += mu * (winner[..., None] == idx).astype(np.float32)
+
+    fields = {"mean": mean, "gated": gated}
+    fields.update({f"mu{idx + 1}": mu for idx, mu in enumerate(mus)})
+    samplers = {name: TriLinearSampler(field, amin, amax) for name, field in fields.items()}
+
+    def vmax(field):
+        mag = np.linalg.norm(field.reshape(-1, 3), axis=1)
+        nz = mag[mag > 0]
+        return float(np.percentile(nz, 100.0)) if nz.size else 1.0
+
+    vmax_by_name = {name: vmax(field) for name, field in fields.items()}
+
+    # The original Python app uses FlowMeshOverlay + its cached label grid for
+    # probe interpretation. This API intentionally reuses that path.
+    import vtk
+
+    ren = vtk.vtkRenderer()
+    win = vtk.vtkRenderWindow()
+    win.OffScreenRenderingOn()
+    win.AddRenderer(ren)
+    mesh = FlowMeshOverlay(ren=ren, win=win, mesh_dir=data / "meshes", alignment_file=data / "brain_alignment.json")
+    cache = data / "label_grid_cache.npz"
+    if not mesh.load_label_grid(cache):
+        raise RuntimeError("Python label_grid_cache is missing; run setup_brain_data.py in mindVisualizer first.")
+
+    # Optional finer parcellation, exactly as the desktop flow mode auto-detects.
+    try:
+        from src.extra_parcellation import ExtraParcellation
+
+        atlas = data / "extra_parcellation" / "combined_atlas.nii.gz"
+        labels = data / "extra_parcellation" / "combined_atlas_labels.json"
+        if atlas.exists():
+            extra = ExtraParcellation(atlas, labels if labels.exists() else None)
+            if extra.load():
+                mesh.set_extra_parcellation(extra)
+    except Exception as exc:
+        print(f"[extra-parcellation] disabled: {exc}")
+
+    ent_sampler = TriLinearSampler(mean, amin, amax) if ent is not None else None
+
+    return {
+        "amin": amin,
+        "amax": amax,
+        "span": span,
+        "samplers": samplers,
+        "vmax": vmax_by_name,
+        "target_step": target_step,
+        "mesh": mesh,
+        "entropy": ent,
+        "entropy_sampler": ent_sampler,
+    }
+
+
+app = FastAPI(title="MindVisualizer API", version="1.0.0")
+
+origins = [origin.strip() for origin in os.environ.get("CORS_ORIGINS", "*").split(",") if origin.strip()]
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_methods=["GET", "POST", "OPTIONS"],
+    allow_headers=["*"],
+)
+
+
+@app.get("/health")
+def health():
+    try:
+        svc = services()
+        return {"ok": True, "fieldMin": svc["amin"].tolist(), "fieldMax": svc["amax"].tolist()}
+    except Exception as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+
+
+@app.post("/api/mindvis/trajectory")
+def trajectory(req: TrajectoryRequest):
+    svc = services()
+    world, mags, stopped = integrate_probe(req, svc)
+    transitions = analyze_regions(world, mags, req.callLlm if hasattr(req, "callLlm") else False)
+    return {
+        "fieldMode": req.fieldMode,
+        "trajectory": normalized(world, svc["amin"], svc["span"]).tolist(),
+        "worldTrajectory": world.tolist(),
+        "fieldMagnitudes": mags.tolist(),
+        "stopped": stopped,
+        "transitions": transitions,
+        "regionText": format_transition_text(transitions),
+    }
+
+
 @app.post("/api/mindvis/explain")
-def explain(req: ExplainRequest):
-    return build_explanation(req)
+def explain(req: ExplainRequest, request: Request):
+    return build_explanation(req, request=request)
 
 
 @app.post("/api/mindvis/explain/stream")
-def explain_stream(req: ExplainRequest):
+def explain_stream(req: ExplainRequest, request: Request):
     def event_stream():
         def event(payload: dict) -> str:
             return json.dumps(payload) + "\n"
 
         try:
+            cached = cached_explanation(req)
+            if cached is not None:
+                yield event({"type": "log", "message": "Returning a cached path explanation."})
+                yield event({"type": "result", **cached})
+                return
+
             yield event({"type": "log", "message": "Loading MindVisualizer field samplers and brain label grid."})
             svc = services()
             traj = np.asarray(req.trajectory, dtype=np.float32)
             if traj.ndim != 2 or traj.shape[1] != 3:
                 raise HTTPException(status_code=400, detail="trajectory must be an array of [x,y,z] points")
-
-            yield event({"type": "log", "message": f"Received {len(traj)} normalized probe samples in {req.fieldMode} mode."})
-            world = denormalized(np.clip(traj, 0.0, 1.0), svc["amin"], svc["span"])
-            sampler = svc["samplers"].get(req.fieldMode, svc["samplers"]["mean"])
-            mags = np.linalg.norm(sampler.sample_vec(world), axis=1).astype(np.float32)
-
-            yield event({"type": "log", "message": "Sampling flow strength and entropy along the traveled path."})
-            transitions = analyze_regions(world, mags, req.callLlm)
-            yield event({"type": "log", "message": f"Detected {len(transitions)} entered or nearby anatomical regions."})
+
+            yield event({"type": "log", "message": f"Received {len(traj)} normalized probe samples in {req.fieldMode} mode."})
+            world = denormalized(np.clip(traj, 0.0, 1.0), svc["amin"], svc["span"])
+            sampler = svc["samplers"].get(req.fieldMode, svc["samplers"]["mean"])
+            mags = np.linalg.norm(sampler.sample_vec(world), axis=1).astype(np.float32)
+
+            yield event({"type": "log", "message": "Sampling flow strength and entropy along the traveled path."})
+            transitions = analyze_regions(world, mags, req.callLlm)
+            yield event({"type": "log", "message": f"Detected {len(transitions)} entered or nearby anatomical regions."})
 
             text = format_transition_text(transitions)
             if req.callLlm:
+                ensure_free_budget(request)
                 from src.region_analyzer import analyze_with_gpt
 
                 yield event({"type": "log", "message": "Calling the OpenAI model for the neuroscience interpretation."})
                 text = analyze_with_gpt(
                     transitions,
-                    use_rag=os.environ.get("MINDVIS_USE_RAG", "0") == "1",
+                    use_rag=os.environ.get("MINDVIS_USE_RAG", "0") == "1",
                     model=os.environ.get("OPENAI_MODEL", "gpt-5.4-mini"),
                     debug=os.environ.get("MINDVIS_DEBUG", "0") == "1",
                 )
+                ensure_not_budget_error(text)
                 yield event({"type": "log", "message": "LLM interpretation finished."})
 
-            yield event({"type": "result", "text": text, "transitions": transitions, "regionText": format_transition_text(transitions)})
+            result = {"text": text, "transitions": transitions, "regionText": format_transition_text(transitions)}
+            write_explanation_cache(req, result)
+            yield event({"type": "result", **result})
+        except HTTPException as exc:
+            detail = exc.detail if isinstance(exc.detail, dict) else {"error": str(exc.detail)}
+            yield event({"type": "error", "status": exc.status_code, **detail})
         except Exception as exc:
             yield event({"type": "error", "message": str(exc)})
 
     return StreamingResponse(event_stream(), media_type="application/x-ndjson")
 
 
-def build_explanation(req: ExplainRequest, log=None):
+def build_explanation(req: ExplainRequest, log=None, request: Request | None = None):
+    cached = cached_explanation(req)
+    if cached is not None:
+        return cached
+
     log = log or (lambda _message: None)
     log("Loading MindVisualizer field samplers and brain label grid.")
     svc = services()
-    traj = np.asarray(req.trajectory, dtype=np.float32)
-    if traj.ndim != 2 or traj.shape[1] != 3:
-        raise HTTPException(status_code=400, detail="trajectory must be an array of [x,y,z] points")
-
-    log(f"Received {len(traj)} normalized probe samples in {req.fieldMode} mode.")
-    world = denormalized(np.clip(traj, 0.0, 1.0), svc["amin"], svc["span"])
-    sampler = svc["samplers"].get(req.fieldMode, svc["samplers"]["mean"])
-    mags = np.linalg.norm(sampler.sample_vec(world), axis=1).astype(np.float32)
-    log("Sampling flow strength and entropy along the traveled path.")
-    transitions = analyze_regions(world, mags, req.callLlm)
-    log(f"Detected {len(transitions)} entered or nearby anatomical regions.")
-    text = format_transition_text(transitions)
+    traj = np.asarray(req.trajectory, dtype=np.float32)
+    if traj.ndim != 2 or traj.shape[1] != 3:
+        raise HTTPException(status_code=400, detail="trajectory must be an array of [x,y,z] points")
+
+    log(f"Received {len(traj)} normalized probe samples in {req.fieldMode} mode.")
+    world = denormalized(np.clip(traj, 0.0, 1.0), svc["amin"], svc["span"])
+    sampler = svc["samplers"].get(req.fieldMode, svc["samplers"]["mean"])
+    mags = np.linalg.norm(sampler.sample_vec(world), axis=1).astype(np.float32)
+    log("Sampling flow strength and entropy along the traveled path.")
+    transitions = analyze_regions(world, mags, req.callLlm)
+    log(f"Detected {len(transitions)} entered or nearby anatomical regions.")
+    text = format_transition_text(transitions)
 
     if req.callLlm:
+        if request is not None:
+            ensure_free_budget(request)
         from src.region_analyzer import analyze_with_gpt
 
         log("Calling the OpenAI model for the neuroscience interpretation.")
         text = analyze_with_gpt(
             transitions,
-            use_rag=os.environ.get("MINDVIS_USE_RAG", "0") == "1",
+            use_rag=os.environ.get("MINDVIS_USE_RAG", "0") == "1",
             model=os.environ.get("OPENAI_MODEL", "gpt-5.4-mini"),
             debug=os.environ.get("MINDVIS_DEBUG", "0") == "1",
         )
+        ensure_not_budget_error(text)
         log("LLM interpretation finished.")
 
-    return {"text": text, "transitions": transitions, "regionText": format_transition_text(transitions)}
-
-
-def integrate_probe(req: TrajectoryRequest, svc: dict) -> tuple[np.ndarray, np.ndarray, bool]:
-    p = denormalized(np.asarray(req.seed[:3], dtype=np.float32)[None, :], svc["amin"], svc["span"])
-    sampler = svc["samplers"].get(req.fieldMode, svc["samplers"]["mean"])
-    step = req.dt * req.speedScale * (svc["target_step"] / max(svc["vmax"].get(req.fieldMode, 1.0), 1e-9))
-    path = []
-    mags = []
-    stopped = False
+    result = {"text": text, "transitions": transitions, "regionText": format_transition_text(transitions)}
+    write_explanation_cache(req, result)
+    return result
 
-    for _ in range(req.steps):
-        path.append(p[0].copy())
-        velocity = sampler.sample_vec(p)[0]
-        mags.append(float(np.linalg.norm(velocity)))
-        new_pos = np.clip(p[0] + velocity * step, svc["amin"], svc["amax"])
 
-        # Original probe boundary behavior: if a step leaves labelled brain
-        # space, try half-step; otherwise stop instead of wrapping.
-        if not is_valid_probe_position(new_pos, svc["mesh"]):
-            half = np.clip(p[0] + velocity * step * 0.5, svc["amin"], svc["amax"])
-            if is_valid_probe_position(half, svc["mesh"]):
-                new_pos = half
-            else:
-                stopped = True
-                break
+def cache_key(req: ExplainRequest) -> Path:
+    payload = req.model_dump()
+    digest = hashlib.sha256(json.dumps(payload, sort_keys=True).encode("utf-8")).hexdigest()
+    return CACHE_DIR / f"mindvis-{digest}.json"
 
-        if np.linalg.norm(new_pos - p[0]) < 1e-5:
-            stopped = True
-            break
-        p[0] = new_pos.astype(np.float32)
 
-    return np.asarray(path, dtype=np.float32), np.asarray(mags, dtype=np.float32), stopped
-
-
-def is_valid_probe_position(point: np.ndarray, mesh) -> bool:
-    key = mesh.get_region_at_point(point)
-    if key is None:
-        key = mesh.find_nearest_region(point, search_radius=2, max_distance_mm=5.0)
-    return key is not None
-
-
-def analyze_regions(world: np.ndarray, mags: np.ndarray, call_llm: bool) -> list[dict]:
-    svc = services()
-    from src.region_analyzer import analyze_probe_path
-
-    transitions = analyze_probe_path(
-        world,
-        svc["mesh"],
-        sample_every=5,
-        field_mags=mags,
-        entropy_sampler=svc["entropy_sampler"],
-        entropy_field=svc["entropy"],
-    )
-    return make_json_safe(transitions)
-
-
-def format_transition_text(transitions: list[dict]) -> str:
-    from src.region_analyzer import format_transitions_text
-
-    return format_transitions_text(transitions)
+def cached_explanation(req: ExplainRequest) -> dict | None:
+    path = cache_key(req)
+    if not path.exists():
+        return None
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except Exception:
+        return None
 
 
-def denormalized(points: np.ndarray, amin: np.ndarray, span: np.ndarray) -> np.ndarray:
-    return points.astype(np.float32) * span[None, :] + amin[None, :]
+def write_explanation_cache(req: ExplainRequest, result: dict) -> None:
+    if not req.callLlm:
+        return
+    try:
+        cache_key(req).write_text(json.dumps(result, ensure_ascii=False), encoding="utf-8")
+    except Exception:
+        pass
 
 
-def normalized(points: np.ndarray, amin: np.ndarray, span: np.ndarray) -> np.ndarray:
-    return (points.astype(np.float32) - amin[None, :]) / span[None, :]
+def ensure_free_budget(request: Request) -> None:
+    if not client_budget_available(request):
+        raise HTTPException(status_code=429, detail={"error": "client_free_limit_exhausted"})
+    if not daily_budget_available():
+        raise HTTPException(status_code=429, detail={"error": "budget_exhausted"})
 
 
-def make_json_safe(value):
-    if isinstance(value, np.ndarray):
-        return value.tolist()
-    if isinstance(value, np.generic):
-        return value.item()
-    if isinstance(value, dict):
-        return {str(k): make_json_safe(v) for k, v in value.items() if not str(k).startswith("_")}
-    if isinstance(value, list):
-        return [make_json_safe(v) for v in value]
-    if isinstance(value, tuple):
-        return [make_json_safe(v) for v in value]
-    return value
+def daily_budget_available() -> bool:
+    limit = int(os.environ.get("MINDVIS_DAILY_REQUEST_LIMIT", "80"))
+    if limit <= 0:
+        return False
+    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
+    counter_path = CACHE_DIR / f"counter-{today}.txt"
+    try:
+        current = int(counter_path.read_text()) if counter_path.exists() else 0
+    except Exception:
+        current = 0
+    if current >= limit:
+        return False
+    counter_path.write_text(str(current + 1))
+    return True
+
+
+def client_budget_available(request: Request) -> bool:
+    limit = int(os.environ.get("MINDVIS_FREE_EXPLAINS_PER_CLIENT", "2"))
+    if limit <= 0:
+        return False
+    forwarded_for = request.headers.get("x-forwarded-for", "").split(",")[0].strip()
+    host = forwarded_for or (request.client.host if request.client else "unknown")
+    user_agent = request.headers.get("user-agent", "unknown")[:180]
+    client_id = request.headers.get("x-mindvis-client-id", "unknown")[:120]
+    client_hash = hashlib.sha256(f"{host}|{user_agent}|{client_id}".encode("utf-8")).hexdigest()[:16]
+    today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
+    counter_path = CACHE_DIR / f"client-{today}-{client_hash}.txt"
+    try:
+        current = int(counter_path.read_text()) if counter_path.exists() else 0
+    except Exception:
+        current = 0
+    if current >= limit:
+        return False
+    counter_path.write_text(str(current + 1))
+    return True
+
+
+def ensure_not_budget_error(text: str) -> None:
+    msg = (text or "").lower()
+    if any(token in msg for token in ["insufficient_quota", "quota", "billing", "rate limit", "[gpt error]"]):
+        raise HTTPException(status_code=429, detail={"error": "openai_budget_exhausted"})
+
+
+def integrate_probe(req: TrajectoryRequest, svc: dict) -> tuple[np.ndarray, np.ndarray, bool]:
+    p = denormalized(np.asarray(req.seed[:3], dtype=np.float32)[None, :], svc["amin"], svc["span"])
+    sampler = svc["samplers"].get(req.fieldMode, svc["samplers"]["mean"])
+    step = req.dt * req.speedScale * (svc["target_step"] / max(svc["vmax"].get(req.fieldMode, 1.0), 1e-9))
+    path = []
+    mags = []
+    stopped = False
+
+    for _ in range(req.steps):
+        path.append(p[0].copy())
+        velocity = sampler.sample_vec(p)[0]
+        mags.append(float(np.linalg.norm(velocity)))
+        new_pos = np.clip(p[0] + velocity * step, svc["amin"], svc["amax"])
+
+        # Original probe boundary behavior: if a step leaves labelled brain
+        # space, try half-step; otherwise stop instead of wrapping.
+        if not is_valid_probe_position(new_pos, svc["mesh"]):
+            half = np.clip(p[0] + velocity * step * 0.5, svc["amin"], svc["amax"])
+            if is_valid_probe_position(half, svc["mesh"]):
+                new_pos = half
+            else:
+                stopped = True
+                break
+
+        if np.linalg.norm(new_pos - p[0]) < 1e-5:
+            stopped = True
+            break
+        p[0] = new_pos.astype(np.float32)
+
+    return np.asarray(path, dtype=np.float32), np.asarray(mags, dtype=np.float32), stopped
+
+
+def is_valid_probe_position(point: np.ndarray, mesh) -> bool:
+    key = mesh.get_region_at_point(point)
+    if key is None:
+        key = mesh.find_nearest_region(point, search_radius=2, max_distance_mm=5.0)
+    return key is not None
+
+
+def analyze_regions(world: np.ndarray, mags: np.ndarray, call_llm: bool) -> list[dict]:
+    svc = services()
+    from src.region_analyzer import analyze_probe_path
+
+    transitions = analyze_probe_path(
+        world,
+        svc["mesh"],
+        sample_every=5,
+        field_mags=mags,
+        entropy_sampler=svc["entropy_sampler"],
+        entropy_field=svc["entropy"],
+    )
+    return make_json_safe(transitions)
+
+
+def format_transition_text(transitions: list[dict]) -> str:
+    from src.region_analyzer import format_transitions_text
+
+    return format_transitions_text(transitions)
+
+
+def denormalized(points: np.ndarray, amin: np.ndarray, span: np.ndarray) -> np.ndarray:
+    return points.astype(np.float32) * span[None, :] + amin[None, :]
+
+
+def normalized(points: np.ndarray, amin: np.ndarray, span: np.ndarray) -> np.ndarray:
+    return (points.astype(np.float32) - amin[None, :]) / span[None, :]
+
+
+def make_json_safe(value):
+    if isinstance(value, np.ndarray):
+        return value.tolist()
+    if isinstance(value, np.generic):
+        return value.item()
+    if isinstance(value, dict):
+        return {str(k): make_json_safe(v) for k, v in value.items() if not str(k).startswith("_")}
+    if isinstance(value, list):
+        return [make_json_safe(v) for v in value]
+    if isinstance(value, tuple):
+        return [make_json_safe(v) for v in value]
+    return value