debugviewer: live-session sidebar + auto-attach + flashing fix; engine live-trace hoist; pipeline e2e fix

baseworkflow/baseworkflow.py

@@ -163,11 +163,43 @@ def context(self, **kw: Any) -> Context:
             kw.setdefault("services", self._services)
         return super().context(**kw)
 
+    def _trace_name(self) -> str:
+        """Label for the live-trace sidecar: the workflow's YAML stem, which is the
+        name the debug viewer lists a workflow under — so an attached run maps onto
+        the right graph. A subclass engine gets the right label for free via its
+        overridden ``WORKFLOW_PATH`` (``workflows/websitewf.yml`` → ``websitewf``)."""
+        return os.path.splitext(os.path.basename(self.WORKFLOW_PATH))[0]
+
     def run(self, payload: Any = None, ctx: Optional[Context] = None, *, until: str = ""):
         if ctx is None:
             ctx = self.context()
         self._seed(ctx.shelves)
-        return super().run(payload, ctx=ctx, until=until)
+        # Auto-attach the live-trace sidecar observer HERE, in the engine base, so
+        # every workflow built on BaseWorkflow — baseworkflow, websitewf, any future
+        # overlay — is watchable in the debug viewer with zero per-engine wiring (a
+        # consumer overlay must not have to know sidecars exist). Opt-in via
+        # $DISPATCH_LIVE_TRACE[_DIR]; passive telemetry — it can never change the
+        # outcome. We never clobber a caller-supplied observer (the in-process viewer
+        # sets ctx.observer itself and finishes it itself) — we only own, and finish,
+        # an observer we create here.
+        own_observer = None
+        if getattr(ctx, "observer", None) is None:
+            try:
+                from foundation import trace
+                if trace.tracing_enabled():
+                    own_observer = trace.new_observer(self._trace_name())
+                    ctx.observer = own_observer
+            except Exception:  # noqa: BLE001 — tracing is best-effort telemetry
+                own_observer = None
+        try:
+            result = super().run(payload, ctx=ctx, until=until)
+        except BaseException:
+            if own_observer is not None:
+                own_observer.finish("errored")
+            raise
+        if own_observer is not None:
+            own_observer.finish("succeeded" if getattr(result, "ok", False) else "failed")
+        return result
 
 
 def run_mock(
@@ -246,33 +278,16 @@ def run_live(
     svc = services if services is not None else bw_services.build_services()
     wf = BaseWorkflow(factory, job=job, triage=triage, request=request, services=svc)
     ctx = wf.context(dry_run=dry_run)
-
-    # Opt-in live-trace sidecar so an out-of-process watcher (the debug viewer) can
-    # follow this run. Off unless $DISPATCH_LIVE_TRACE[_DIR] is set; never overrides a
-    # caller-supplied observer; passive (an observer cannot change the run's outcome).
-    observer = getattr(ctx, "observer", None)
-    if observer is None:
-        try:
-            from foundation import trace
-            if trace.tracing_enabled():
-                observer = trace.new_observer("baseworkflow")
-                ctx.observer = observer
-        except Exception:  # noqa: BLE001 — tracing is best-effort telemetry
-            observer = None
+    # Live-trace sidecar wiring now lives in BaseWorkflow.run (shared by every engine),
+    # so there is nothing to do here beyond running and tearing down the shelf root.
     try:
         result = wf.run(ctx=ctx)
-        if observer is not None:
-            observer.finish("succeeded" if result.ok else "failed")
         return {
             "result": result,
             "ctx": ctx,
             "workflow": wf,
             "deliverables": wf.shelves.deliverables.snapshot(),
             "interpreter": wf.last_interpreter,
         }
-    except BaseException:
-        if observer is not None:
-            observer.finish("errored")
-        raise
     finally:
         shutil.rmtree(shelf_root, ignore_errors=True)

baseworkflow/bindings/admin.py

@@ -700,7 +700,7 @@ def record(**kw: Any) -> Dict[str, Any]:
                    f"{', '.join(names) or '(none)'}")
         return record(planned=names)
 
-    environ = _env.Environment()
+    environ = _env.Environment.default()
     token = environ.get("SUPABASE_ACCESS_TOKEN") or os.environ.get("SUPABASE_ACCESS_TOKEN")
     ref = environ.get("SUPABASE_PROJECT_REF") or os.environ.get("SUPABASE_PROJECT_REF")
     if not (token and ref):

baseworkflow/bindings/research.py

@@ -331,4 +331,10 @@ def register(reg: Any) -> None:
     reg.register_predicate("questions_satisfied", questions_satisfied)
     reg.register_predicate("research_exhausted", research_exhausted)
     from .architect import LIVE_INFERENCE_WORKERS
-    LIVE_INFERENCE_WORKERS["research_investigate"] = _investigate_worker
+    # Keyed by the action token's LAST SEGMENT ("research:investigate" -> "investigate"),
+    # matching ArchitectFactory._inference_runner's `token.split(":")[-1]` dispatch — NOT
+    # the manifest bind name "research_investigate". They differ only for this token, so
+    # the old key never matched: the live worker was silently bypassed and the raw model
+    # text (prose) hit the JSON adapter, failing the tick. (Every other worker's bind
+    # name equals its token's last segment, so they matched by coincidence.)
+    LIVE_INFERENCE_WORKERS["investigate"] = _investigate_worker

foundation/actions/adapter.py

@@ -19,9 +19,25 @@
 from abc import ABC, abstractmethod
 from typing import Any
 
+from .result import ActionError
+
 VALID_FORMATS = ("text", "json", "yaml")
 
 
+def _salvage_json(s: str) -> Any:
+    """Best-effort extract a JSON object/array embedded in prose or code fences —
+    models often wrap the payload. Returns the parsed value, or ``None`` if nothing
+    parseable is found."""
+    for opener, closer in (("{", "}"), ("[", "]")):
+        start, end = s.find(opener), s.rfind(closer)
+        if 0 <= start < end:
+            try:
+                return json.loads(s[start:end + 1])
+            except ValueError:
+                continue
+    return None
+
+
 class Adapter(ABC):
     """Two-way coercion between text and a structured object."""
 
@@ -60,7 +76,27 @@ def encode(self, obj: Any) -> str:
     def decode(self, text: str) -> Any:
         if not isinstance(text, str):
             return text
-        return json.loads(text)
+        s = text.strip()
+        # An empty model response carries no structured payload. Fail SAFE to an
+        # empty object rather than crashing the whole tick with a bare
+        # JSONDecodeError — the inference layer's failures-as-data contract. (This
+        # is the common real failure: a model that times out / returns nothing.)
+        if not s:
+            return {}
+        try:
+            return json.loads(s)
+        except json.JSONDecodeError:
+            # Models often wrap the payload in prose or ``` fences — salvage the
+            # embedded object/array before giving up.
+            salvaged = _salvage_json(s)
+            if salvaged is not None:
+                return salvaged
+            # Genuinely non-JSON, non-empty output: surface a HANDLED ActionError so
+            # the leaf boundary records a clean failure (not an uncaught crash).
+            raise ActionError(
+                f"json adapter: model output is not valid JSON "
+                f"({len(s)} chars, starts {s[:60]!r})"
+            )
 
 
 class YamlAdapter(Adapter):

foundation/trace.py

@@ -27,12 +27,19 @@
 
 import json
 import os
+import threading
 import time
 from typing import Any, Dict, Optional
 
 _ENV_DIR = "DISPATCH_LIVE_TRACE_DIR"
 _ENV_ON = "DISPATCH_LIVE_TRACE"
 
+# How often the FileObserver bumps its sidecar's mtime while the run is alive, so a
+# watcher can tell a live-but-quiet run (mid-inference — no state transitions for
+# minutes) from a dead one whose writer vanished. Must be well under the watcher's
+# staleness cutoff.
+_HEARTBEAT_SECS = 15.0
+
 
 def trace_dir() -> str:
     """Directory holding live-run sidecars. Override with ``$DISPATCH_LIVE_TRACE_DIR``;
@@ -64,6 +71,20 @@ def __init__(self, path: str, *, workflow: str) -> None:
         except Exception:  # noqa: BLE001
             pass
         self._write({"type": "meta", "workflow": workflow, "started": self._started}, mode="w")
+        # Liveness heartbeat: bump the sidecar's mtime every _HEARTBEAT_SECS so a
+        # watcher can distinguish a live-but-quiet run from a dead writer. Daemon
+        # thread — it dies with the process (a hard kill correctly stops the beat),
+        # and finish() stops it cleanly on a normal end.
+        self._stop = threading.Event()
+        self._beat = threading.Thread(target=self._heartbeat, name="trace-heartbeat", daemon=True)
+        self._beat.start()
+
+    def _heartbeat(self) -> None:
+        while not self._stop.wait(_HEARTBEAT_SECS):
+            try:
+                os.utime(self.path, None)  # touch mtime only; adds no line for the tailer to parse
+            except OSError:
+                return
 
     def _write(self, event: Dict[str, Any], *, mode: str = "a") -> None:
         try:
@@ -84,8 +105,12 @@ def on_event(self, event: Dict[str, Any]) -> None:
         self._write({"type": "event", **event})
 
     def finish(self, status: str) -> None:
-        """Write the terminal status line. Idempotent enough for best-effort use."""
+        """Write the terminal status line and stop the heartbeat. Idempotent enough
+        for best-effort use."""
         self._write({"type": "status", "status": status})
+        stop = getattr(self, "_stop", None)
+        if stop is not None:
+            stop.set()
 
 
 def new_observer(workflow: str, *, run_id: Optional[str] = None) -> "FileObserver":

tools/debugviewer/bus.py

@@ -191,8 +191,14 @@ def list_external(self) -> List[Dict[str, Any]]:
             files = []
         out: List[Dict[str, Any]] = []
         for p in files[:50]:
+            try:
+                mtime = os.path.getmtime(p)
+            except OSError:
+                mtime = 0.0
+            # mtime lets the UI tell a genuinely-active run from a sidecar whose writer
+            # died without a terminal status line (which otherwise reads "running" forever).
             out.append({"id": os.path.splitext(os.path.basename(p))[0],
-                        "path": p, **self._read_sidecar(p)})
+                        "path": p, "mtime": mtime, **self._read_sidecar(p)})
         return out
 
     def attach_external(self, run_id: str) -> Optional[Run]: