feat(backend): view.json snapshots and tier-1 activation ordering

Write runs/{id}/view.json at finalize_run and serve GET /api/runs/{id}
snapshot-first (critic + todos from session state). Record lab_response
before Census activation so approved rows are visible to reverse-ETL;
patch snapshot on experiment approval. Immutable Cache-Control on pinned
artifact fragments.

Author: samartho4

backend/dry_lab/callbacks.py

@@ -22,6 +22,7 @@
 import uuid
 
 from . import run_store
+from . import view_builder
 from .config import settings
 from .dataset_context import pipeline_fallback_summary
 from .sub_agents.investigator.tools import shutdown_kernel
@@ -229,6 +230,13 @@ async def finalize_run(callback_context):
         pass  # already flushed
     except Exception as e:
         logger.warning("finish_run failed (continuing): %s", e)
+
+    # 3b) Persist a read-optimized view snapshot beside the provenance pins (fast GET /api/runs/{id}).
+    try:
+        view_builder.write_snapshot(run_id, view_builder.build_run_view_from_state(state, status=status))
+    except Exception as e:
+        logger.warning("write_snapshot failed for %s (continuing): %s", run_id, e)
+
     # 4) CROSS-RUN MEMORY (Phase D — gated + OFF the critical path). If a real Vertex AI Memory Bank is wired
     #    (settings.memory_enabled, i.e. MEMORY_SERVICE_URI set), persist this session so a RETURNING user's
     #    cohort / prior findings / preferences can be recalled on a later run. UNSET (the default) -> ADK's

backend/dry_lab/server.py

@@ -30,8 +30,8 @@
 from google.adk.cli.utils.service_factory import create_artifact_service_from_options
 
 from . import run_store
+from . import view_builder
 from .config import settings
-from .dataset_context import data_status_dict
 from .sub_agents.pipeline.tools import activate_experiment, activation_configured
 
 # agents_dir = the directory that CONTAINS the `dry_lab` package (so ADK discovers dry_lab/agent.py:root_agent).
@@ -173,8 +173,17 @@ async def emit_experiment(req: dict):
     decision = req.get("decision") or "approved"
 
     if activation_configured():
-        # TIER 1 — real Fivetran Activation (reverse-ETL). Triggers the Census sync; the approved request travels
-        # from BigQuery run_status to the destination. Returns an honest response (channel/sync_run_id/error).
+        # TIER 1 — Census reverse-ETL reads run_status WHERE lab_decision='approved'. Record the decision BEFORE
+        # triggering the sync so the row is visible when the Activation runs.
+        if run_id:
+            try:
+                run_store.record_lab_response(
+                    run_id, decision=decision,
+                    response={"channel": "fivetran_activation", "status": "triggering", "mocked": False},
+                )
+            except Exception as e:
+                logging.getLogger("dry_lab.server").error(
+                    "record_lab_response (pre-activation) failed for run %s: %s", run_id, e)
         lab_response = activate_experiment(req)
     elif settings.lab_endpoint_url:
         # TIER 2 — real webhook.
@@ -196,6 +205,11 @@ async def emit_experiment(req: dict):
         except Exception as e:
             logging.getLogger("dry_lab.server").error(
                 "record_lab_response append failed for run %s: %s", run_id, e)
+        try:
+            view_builder.patch_activation(run_id, proposal=req, lab_response=lab_response)
+        except Exception as e:
+            logging.getLogger("dry_lab.server").warning(
+                "view snapshot activation patch failed for run %s: %s", run_id, e)
     return {"status": "sent", "decision": decision, "lab_response": lab_response}
 
 
@@ -265,45 +279,8 @@ async def get_shared(sid: str):
                                     os.environ.get("DRYLAB_PROVENANCE_BUCKET", ""),
                                     os.environ.get("ARTIFACT_BUCKET", "")) if b}
 
-# The eval harness prepends an internal "[Use skill: …]" routing directive to a run's goal — machinery, never a
-# user-facing title. Strip it server-side so both the read-only run view and the runs list show the real goal.
-# (Mirror of cleanGoal() in frontend/src/components/RunsList.tsx.)
-_USE_SKILL_RE = re.compile(r"^\s*\[Use skill:[^\]]*\]\s*", re.I)
-
-
 def _clean_goal(goal: str | None) -> str:
-    return _USE_SKILL_RE.sub("", goal or "").strip()
-
-
-def _pinned_results(evidence: list[dict]) -> tuple[list[dict], dict]:
-    """From computed evidence rows (each pinned to gs://bucket/runs/<run>/<file>#<gen>) build read-only
-    ResultFile entries served via /pinned, plus a {filename: pinned_uri} map (latest wins)."""
-    import mimetypes  # noqa: F401  (kept local; used implicitly by the frontend kind inference)
-    by_name: dict[str, str] = {}
-    for e in evidence:
-        if e.get("kind") != "computed":
-            continue
-        uri = e.get("result_uri") or ""
-        if not uri.startswith("gs://"):
-            continue
-        base = uri.split("#", 1)[0].split("/")[-1]
-        by_name[base] = uri  # latest row for a name wins (re-runs overwrite within a run)
-    def _kind(n: str) -> str:
-        low = n.lower()
-        if low.endswith((".csv", ".tsv")):
-            return "de_table"
-        if low.endswith((".png", ".jpg", ".jpeg", ".svg")):
-            return "figure"
-        if low.endswith((".md", ".pdf")):
-            return "report"
-        if low.endswith(".ipynb"):
-            return "notebook"
-        return "de_table"
-    from urllib.parse import quote
-    # The pinned uri carries a '#<generation>' fragment + ':' — encode it so the browser keeps it in the query.
-    results = [{"name": n, "uri": f"/api/pinned?uri={quote(n_uri, safe='')}", "kind": _kind(n)}
-               for n, n_uri in by_name.items()]
-    return results, by_name
+    return view_builder.clean_goal(goal)
 
 
 def _fetch_pinned_bytes(pinned_uri: str) -> bytes | None:
@@ -362,65 +339,12 @@ async def get_run_view(run_id: str, request: Request):
     run = run_store.get_run(run_id, user_id=_request_uid(request) or None)
     if not run:
         raise HTTPException(status_code=404, detail=f"run '{run_id}' not found")
-    evidence = run.get("evidence") or []
-    results, by_name = _pinned_results(evidence)
-
-    # report.md text (pinned in save_report) -> the report tab; absent is an honest empty state.
-    report = None
-    if "report.md" in by_name:
-        data = _fetch_pinned_bytes(by_name["report.md"])
-        if data:
-            report = {"markdown": data.decode("utf-8", errors="replace")}
-
-    status = run.get("status") or ""
-    summaries = {r.get("stage"): r.get("summary") for r in (run.get("stage_summaries") or []) if isinstance(r, dict)}
-    proposal = run.get("experiment_request")
-    _LOOP = [("plan", "Plan & approve", "@Planner"), ("pipeline", "Sync & QC data (Fivetran)", "@Pipeline"),
-             ("investigator", "Analyze & write report", "@Investigator"),
-             ("critic", "Reliability review", "@Critic"), ("propose", "Propose next experiment", "@Proposer")]
-    passed = status in ("completed", "proposed")  # finish_run only reaches these when the critic passed
-    todos = []
-    for stage, label, agent in _LOOP:
-        if stage == "plan":
-            done, note = True, "Plan approved"
-        elif stage == "propose":
-            done, note = bool(proposal), None
-        elif stage == "critic":
-            done, note = passed, (f"Passed (recorded run status: {status})" if passed else None)
-        else:
-            done, note = (stage in summaries), summaries.get(stage)
-        todos.append({"id": stage, "label": label, "agent": agent,
-                      "status": "done" if done else "pending", "result_summary": note})
-
-    activation = None
-    if proposal and run.get("lab_response") is not None:
-        activation = {"payload": proposal, "lab_response": run.get("lab_response"), "status": "sent"}
-
-    view = {
-        "research_goal": _clean_goal(run.get("goal")),
-        "messages": [],   # the frontend derives a read-only thread from the grounded fields below
-        "todos": todos,
-        "traces": [],
-        "results": results,
-        "evidence": [{"id": e.get("evidence_id"), "kind": e.get("kind"), "claim": e.get("claim"),
-                      "result_uri": e.get("result_uri"), "source_checksum": e.get("source_checksum"),
-                      "code_cell_ref": e.get("code_cell_ref"), "citation": e.get("citation")} for e in evidence],
-        # Honest per-dataset label: route through dataset_context using the run's RECORDED skill (run_store
-        # persists + returns it) so a trials/survival/dose-response run on /runs shows ITS dataset, not the
-        # GSE206285 default that was previously hardcoded for every reconstructed run.
-        "data_status": data_status_dict(run.get("skill"), run.get("goal") or ""),
-        "report": report,
-        "critic": None,   # the verdict isn't persisted in the index; don't fabricate one for a past run
-        "critic_iterations": [],
-        "proposed_experiment": proposal,
-        "activation": activation,
-        "awaiting_approval": None,
-        # skill_tier is NOT in the thin runs index (it's a live/session signal), so a reconstructed past run leaves
-        # it absent → no unverified banner here. Honest by omission, not fabrication (same rule as critic above):
-        # we never assert "vetted" for a run whose tier we didn't durably record. A SHARED /r/:id view DOES carry
-        # skill_tier because it snapshots the live DryLabView (which had it set by save_report).
-    }
-    return view
+
+    snap = view_builder.read_snapshot(run_id)
+    if snap:
+        return view_builder.enrich_snapshot(snap, run, fetch_report=_fetch_pinned_bytes)
+
+    return view_builder.build_run_view_from_record(run, fetch_report=_fetch_pinned_bytes)
 
 
 @app.get("/api/pinned")
@@ -434,7 +358,11 @@ async def get_pinned(uri: str):
     import mimetypes
     name = uri.split("#", 1)[0].split("/")[-1]
     mime, _ = mimetypes.guess_type(name)
-    return Response(content=data, media_type=mime or "application/octet-stream")
+    return Response(
+        content=data,
+        media_type=mime or "application/octet-stream",
+        headers={"Cache-Control": "public, max-age=31536000, immutable"},
+    )
 
 
 # --- BRING YOUR OWN SKILL (PHASE B; additive, governed). The agent's vetted-skill registry is already data-driven