bearlike
diff --git a/‎apps/mewbo_api/CLAUDE.md‎
Lines changed: 11 additions & 1 deletion b/‎apps/mewbo_api/CLAUDE.md‎
Lines changed: 11 additions & 1 deletion
diff --git a/‎apps/mewbo_api/src/mewbo_api/agentic_search/CLAUDE.md‎
Lines changed: 14 additions & 1 deletion b/‎apps/mewbo_api/src/mewbo_api/agentic_search/CLAUDE.md‎
Lines changed: 14 additions & 1 deletion
diff --git a/‎apps/mewbo_api/src/mewbo_api/agentic_search/scg/CLAUDE.md‎
Lines changed: 47 additions & 0 deletions b/‎apps/mewbo_api/src/mewbo_api/agentic_search/scg/CLAUDE.md‎
Lines changed: 47 additions & 0 deletions
diff --git a/‎apps/mewbo_api/src/mewbo_api/agentic_search/scg/orchestrated_runner.py‎
Lines changed: 182 additions & 26 deletions b/‎apps/mewbo_api/src/mewbo_api/agentic_search/scg/orchestrated_runner.py‎
Lines changed: 182 additions & 26 deletions
@@ -14,7 +14,17 @@ Scope: this file applies to the `apps/mewbo_api/` package. It captures runtime b
 ## Runtime flow (what actually happens)
 - Entry point: `apps/mewbo_api/src/mewbo_api/backend.py` (HTTP API framework).
 - Session endpoints:
-  - `POST /api/sessions` create session
+  - `POST /api/sessions` create session — accepts an optional external `cwd`
+    (top-level or `context.cwd`; also on `POST .../query`) gated behind
+    `api.allow_external_cwd` (default OFF). `ExternalCwdPolicy` (backend.py) is
+    the one seam: flag off + cwd present → structured 403; flag on → must be an
+    existing directory (else 400); explicit cwd WINS over project-derived and is
+    persisted as `context_payload["cwd"]` so `/message` re-engagement and
+    `_resolve_session_cwd` (diff endpoints) keep resolving it (#91 — external
+    workspace managers like Grove anchor sessions in their own worktrees;
+    registering provided-path v_projects is NOT a substitute: the reaper
+    permanently deletes childless provided-path parents). Docker rule applies:
+    the path must be visible in the api container at the identical path.
   - `GET /api/sessions` list sessions (each summary carries `origin` — `user|wiki|search|channel` provenance computed in core `summarize_session`, forwarded verbatim; the console badges/filters on it)
   - `POST /api/sessions/{session_id}/query` enqueue run or core command
   - `GET /api/sessions/{session_id}/events?after=...` poll events
 
@@ -134,6 +134,11 @@ reducer switches on). Rules:
   `result` events drive arrival order. The prototype's `finish_delay_ms`
   / `t_ms` are deprecated decorative fields — real ordering comes from
   event arrival, never a client timer.
+- Orchestrated runs now populate ALL of it (#95): `agent_*` includes a
+  synthetic **coordinator lane** for root-inlined tool activity, `result`
+  events come from the agent's `scg_results` emission, and
+  `total_ms`/`confidence`/`sources_count` are computed, never defaulted —
+  see `scg/CLAUDE.md` → "Coordinator lane + result cards + honest settle".
 
 ## `store.py` is the substitution boundary
 
@@ -203,7 +208,15 @@ behavior). `WorkspaceSourceSync.on_workspace_saved` (`source_sync.py`) is the
 POST/PATCH hook: it refreshes the virtual config, then auto-maps newly-enabled
 **live** sources (idempotent — skips already-mapped/in-flight; a terminal/failed
 job does NOT block a re-map, so a previously-unreachable source re-maps once its
-URL is fixed). It ALSO re-maps already-mapped enabled sources whose live tool
+URL is fixed). **The hook is register-and-return (#97):** only step 1 (virtual
+config refresh + NL fingerprint) runs on the request thread; the whole auto-map
+fan-out (mappable/drifted/re-enrich resolution + live descriptor builds +
+`MapSourceJob.start`) runs on one named daemon thread
+(`workspace-automap-<id>`) — the descriptor build is a live MCP handshake per
+source, and doing it in-request blocked "Create Workspace" for N handshakes
+(the user read it as browser-dependent indexing). The method returns the
+`Thread | None` so tests join deterministically; routes ignore it. Never move
+the fan-out back in-request, and never let the thread body raise. It ALSO re-maps already-mapped enabled sources whose live tool
 list drifted from the stamped `ManifestHash` (#81-C), and carries the workspace
 `instructions`/`desc` as untrusted `nl_context` to seed the map-time enrich step
 (#81-B — see scg/CLAUDE.md). **Workspace editing IS a graph-lifecycle event
 
@@ -88,6 +88,53 @@ now echoed as an additive `summary` on that event (core `spawn_agent`). So:
   evidence panel is empty (the same "mirror the real engine shape" rule as the
   `completion` payload above).
 
+## Coordinator lane + result cards + honest settle (#95)
+
+A fast-tier run can legitimately spawn ZERO probes (the root inlines every
+tool call — verified live, run `run-02ad562781`), and the original projection
+keyed everything off `sub_agent` events → blank trace, `results=[]`,
+`total_ms=0` beside a correct synthesis. The fixes, all additive on the wire
+(no `OUTPUT_CONTRACT_VERSION` bump):
+
+- **Coordinator lane** — root `tool_result` events project into ONE synthetic
+  lane (id `coordinator`, name `scg-search`, `source_id ""`).
+  `CoordinatorTrace` (run_streamer.py) is the pure shared projection (the
+  `ProbeTrace` stance: live `_project` and settle render byte-identically).
+  SECURITY: a line is a digest only — tool_id + scalar-input hint capped at
+  120 chars; the tool's RESULT payload is never read (connector returns can
+  carry secrets/PII). The `scg_results` call renders as `emitted N results`,
+  never its entries. The lane has no `stop` lifecycle — its `agent_done`
+  fires at settle, `empty = no data-bearing probe AND no results`. Slots come
+  from merged first-seen order (`_assign_lane_slots`) so settle reproduces
+  the live interleaving. Probe `tool_result`s never appear here — probes run
+  in their own sessions; this session's bus only carries the root's.
+- **`scg_results` = transcript-as-transport.** The tool
+  (`mewbo_graph.plugins.scg.results`, granted via `TRAVERSAL_TOOLS`)
+  validates (≤50 entries, `extra="forbid"`, relevance/confidence 0..1) and
+  returns `{ok, count}` — it writes NOTHING (layering: the library never
+  touches the api run store; no MapPhaseSink-style DI needed because the
+  transcript already reaches the api). `ResultsProjection` maps entries →
+  `SearchResult` with stable ids `r-<run_id8>-<n>` — that id is the
+  live↔settle dedup key. Entry `confidence` folds into `relevance` only when
+  `relevance` is absent (no wire field for it). The playbook's terminal
+  discipline: the ROOT emits once, before synthesis; probes never call it.
+- **Metrics provenance** (`_synthesis_metrics`): `sources_count` = distinct
+  grounding sources = data-bearing probe lanes (keyed by `agent_id` — the
+  wire `source_id` is the shared parent grouping key, useless for
+  distinctness) ∪ emitted results' `source` fields. `confidence` =
+  data-bearing/probes when probes ran, else mean folded entry score, else
+  `(0.0, 0)` (console suppresses). The coordinator lane is NEVER a probe.
+  `total_ms` = `started_at→settle` wall clock, stamped on `run_done`,
+  `RunPayload`, and the `_fail` payload — never hardcoded 0 again.
+- **`resolve_entity` trap (open, #95-D):** it reaches search-run sessions via
+  the scg plugin manifest + #84's capability-driven `build_for`, but a search
+  RUN never satisfies `resolve_qa_ctx` (no QA answer, no
+  `structured_workspace` event) → it always errors `wiki ctx not found`.
+  Exclusion is NOT one-seam: the same registration serves
+  `scg-search-structured`, where the ctx DOES resolve. The clean future fix
+  is gating entity tools on wiki-ctx presence rather than the bare `scg`
+  capability — don't hack the manifest.
+
 ## Two correctness traps (both silent — no exception)
 
 1. **Two different `StructureProvider` protocols share a name root and nothing
 
@@ -46,6 +46,7 @@
 from __future__ import annotations
 
 import threading
+from datetime import datetime
 from typing import Any, Literal
 
 from mewbo_core.common import get_logger
@@ -61,12 +62,18 @@
     RunRecord,
     SearchResult,
     TraceAgent,
+    TraceLine,
     Workspace,
     utc_now_iso,
 )
 from .config import ScgConfig
 from .playbooks import load_playbook
-from .run_streamer import ProbeTrace, RunEventStreamer
+from .run_streamer import (
+    CoordinatorTrace,
+    ProbeTrace,
+    ResultsProjection,
+    RunEventStreamer,
+)
 from .workspace_binding import WorkspaceGraphBinding
 
 logging = get_logger(name="api.agentic_search.scg.orchestrated_runner")
@@ -299,8 +306,29 @@ def _settle(
             return None
 
         trace = self._build_trace(records)
+        # The root coordinator's own tool activity (#95): a root-inline run (fast
+        # tier, no probe sub-agents) streams nothing through the probe lanes, so
+        # the root's ``tool_result`` events are projected as one extra lane. Its
+        # slot is its first-seen ordinal across the MERGED stream (probes +
+        # coordinator) so the settle reconcile honours transcript order exactly as
+        # the live path does (coordinator opens when its first tool_result lands).
+        # This re-slots the probe lanes in place to leave room for it.
+        coordinator_slot = self._assign_lane_slots(records, trace)
+        coordinator_lines = self._build_coordinator_lines(records)
+        # ``scg_results`` emits the discrete result cards (transcript-as-transport):
+        # the root calls it once before synthesis; the api projects its entries.
+        results = self._build_results(run.run_id, records)
+        # A run is data-bearing iff any probe returned data OR any result emitted.
+        with_data_probes = [a for a in trace if not ProbeTrace.is_dead_end(a.result)]
+        has_data = bool(with_data_probes) or bool(results)
+
         if streamer is not None:
             streamer.reconcile_missing(trace)
+            streamer.reconcile_results(results)
+            if coordinator_lines or streamer.coordinator_opened():
+                streamer.reconcile_coordinator(
+                    coordinator_lines, slot=coordinator_slot, has_data=has_data
+                )
         else:
             # No live streamer (defensive / legacy call): emit the full trace.
             for agent in trace:
@@ -323,19 +351,16 @@ def _settle(
                         agent_id=agent.agent_id, results_count=0, empty=not agent.lines
                     ),
                 )
-
-        # Results: the SCG search synthesizes a cited answer rather than emitting
-        # per-source result cards (the connector return is the verifier, not a
-        # normalized hit list); ``results`` stays empty until a probe contract
-        # carries them. The trace + answer are the live surfaces today.
-        results: list[SearchResult] = []
+            for item in results:
+                store.append_run_event(run.run_id, events.result(item=item))
 
         status = self._run_status(summary)
         answer_text, err = self._task_result(records)
-        # Confidence + sources_count come from the live trace's data-bearing
-        # probes, not a fixture (the echo-era fields read 0% / 0 sources on real
-        # runs). The trace is the single source of both signals.
-        confidence, sources_count = self._synthesis_metrics(trace)
+        # Confidence + sources_count come from REAL probe + result signals, not a
+        # fixture (the echo-era fields read 0% / 0 sources on real runs). Probes
+        # AND emitted results both ground the answer (a root-inline run has only
+        # results); the coordinator lane is NOT a probe — never in the ratio.
+        confidence, sources_count = self._synthesis_metrics(trace, results)
         answer = AnswerSynthesis(
             tldr=answer_text, confidence=confidence, sources_count=sources_count
         )
@@ -355,8 +380,11 @@ def _settle(
                     store.append_run_event(run.run_id, events.answer_delta(text=chunk))
             store.append_run_event(run.run_id, events.answer_ready(answer=answer))
 
+        # Honest elapsed: started_at/created_at ISO → settle, in ms (#95). The
+        # old hardcoded 0 read ``0ms`` next to a ~3-minute run.
+        total_ms = self._elapsed_ms(run)
         store.append_run_event(
-            run.run_id, events.run_done(status=status, total_ms=0)
+            run.run_id, events.run_done(status=status, total_ms=total_ms)
         )
         payload = RunPayload(
             run_id=run.run_id,
@@ -366,7 +394,7 @@ def _settle(
             status=status,
             tier=run.tier,
             model=run.model,
-            total_ms=0,
+            total_ms=total_ms,
             answer=answer if status == "completed" else AnswerSynthesis(),
             results=results,
             trace=trace,
@@ -460,21 +488,148 @@ def _build_trace(records: list[dict[str, Any]]) -> list[TraceAgent]:
         return [agents[a] for a in order]
 
     @staticmethod
-    def _synthesis_metrics(trace: list[TraceAgent]) -> tuple[float, int]:
-        """Derive ``(confidence, sources_count)`` from the live probe trace.
-
-        Honest, defined provenance (see :class:`AnswerSynthesis`): a probe is
-        *data-bearing* iff its terminal evidence isn't a ``NO DATA`` dead-end.
-        ``sources_count`` = the count of data-bearing probes (each walks one
-        qualified pathway, so it is the breadth of grounding behind the answer);
-        ``confidence`` = data-bearing / probes run. An empty trace yields
-        ``(0.0, 0)`` so the console suppresses the chip rather than render an
-        unearned ``0%``.
+    def _assign_lane_slots(
+        records: list[dict[str, Any]], trace: list[TraceAgent]
+    ) -> int | None:
+        """Re-slot probe lanes by MERGED first-seen order; return coordinator slot.
+
+        The live streamer assigns a lane's slot from ``len(self._order)`` at its
+        first event, so probe and coordinator slots interleave by transcript
+        arrival. The settle path must reproduce that exact ordering. This scans
+        the transcript once, assigning each lane (probe ``agent_id`` or the
+        sentinel coordinator, opened by its first root ``tool_result``) its
+        first-seen ordinal, then writes the probe ordinals back onto *trace* in
+        place. Returns the coordinator's ordinal, or ``None`` when the root
+        issued no tool_result (no coordinator lane).
         """
-        if not trace:
+        slots: dict[str, int] = {}
+        coordinator_key = "__coordinator__"
+        for rec in records:
+            kind = rec.get("type")
+            payload = rec.get("payload") or {}
+            if kind == "sub_agent":
+                key = str(payload.get("agent_id") or "")
+                if not key:
+                    continue
+            elif kind == "tool_result":
+                if not CoordinatorTrace.is_root_event(payload):
+                    continue
+                key = coordinator_key
+            else:
+                continue
+            if key not in slots:
+                slots[key] = len(slots)
+        for agent in trace:
+            if agent.agent_id in slots:
+                agent.slot = slots[agent.agent_id]
+        return slots.get(coordinator_key)
+
+    @staticmethod
+    def _build_coordinator_lines(records: list[dict[str, Any]]) -> list[TraceLine]:
+        """Project the root's ``tool_result`` events into coordinator trace lines.
+
+        The probes run in their OWN sessions, so a ``tool_result`` in THIS
+        session's transcript is always a ROOT tool call — no agent-id filter is
+        needed. Each becomes one secret-free :class:`CoordinatorTrace` digest line
+        (the SAME projection the live streamer uses — DRY), so a reconciled
+        coordinator lane is byte-identical to one that streamed live. Empty when
+        the root spawned probes for everything (only ``sub_agent`` events).
+        """
+        lines: list[TraceLine] = []
+        for rec in records:
+            if rec.get("type") != "tool_result":
+                continue
+            payload = rec.get("payload") or {}
+            if CoordinatorTrace.is_root_event(payload):
+                lines.append(CoordinatorTrace.line(payload))
+        return lines
+
+    @staticmethod
+    def _build_results(
+        run_id: str, records: list[dict[str, Any]]
+    ) -> list[SearchResult]:
+        """Collect the run's result cards from every ``scg_results`` emit.
+
+        Transcript-as-transport (#95): the root calls ``scg_results`` once (the
+        playbook's terminal-discipline step) carrying the discrete cards behind
+        the answer; the api projects each entry onto a stable-id
+        :class:`SearchResult` via :class:`ResultsProjection` (the SAME read the
+        live stream uses, so ids — and thus dedup — agree). Multiple emits are
+        concatenated in order; the stable id (``r-<run_id8>-<n>`` per emit) keeps
+        a re-emit from the same call idempotent across live + settle.
+        """
+        out: list[SearchResult] = []
+        seen: set[str] = set()
+        for rec in records:
+            if rec.get("type") != "tool_result":
+                continue
+            payload = rec.get("payload") or {}
+            if not ResultsProjection.is_results_event(payload):
+                continue
+            for item in ResultsProjection.parse(run_id, payload):
+                if item.id in seen:
+                    continue
+                seen.add(item.id)
+                out.append(item)
+        return out
+
+    @staticmethod
+    def _elapsed_ms(run: RunRecord) -> int:
+        """Honest run duration: ``started_at``/``created_at`` ISO → now, in ms.
+
+        Parses the run's start timestamp (``started_at`` preferred, ``created_at``
+        fallback) and measures to settle time. ``0`` only when neither parses (a
+        record with no start stamp) — never the old hardcoded ``0`` that read
+        ``0ms`` beside a multi-minute run.
+        """
+        start_iso = run.started_at or run.created_at
+        if not start_iso:
+            return 0
+        try:
+            started = datetime.fromisoformat(start_iso)
+            now = datetime.fromisoformat(utc_now_iso())
+        except ValueError:
+            return 0
+        delta_ms = int((now - started).total_seconds() * 1000)
+        return max(delta_ms, 0)
+
+    @staticmethod
+    def _synthesis_metrics(
+        trace: list[TraceAgent], results: list[SearchResult]
+    ) -> tuple[float, int]:
+        """Derive ``(confidence, sources_count)`` from probes AND emitted results.
+
+        Honest, defined provenance (see :class:`AnswerSynthesis`):
+
+        * ``sources_count`` = the count of DISTINCT grounding sources = the union
+          of (each data-bearing probe lane — keyed by its ``agent_id``, since one
+          lane is one qualified-pathway walk and its wire ``source_id`` is the
+          shared parent grouping key, not the connector) ∪ the emitted results'
+          ``source`` fields. So a probe-only run reports its breadth, a
+          root-inline run reports its distinct connectors, and a mixed run reports
+          the union.
+        * ``confidence`` — when probes ran: ``data-bearing probes / probes run``.
+          When NO probe ran but results were emitted (the root-inline path): the
+          mean of the emitted entries' folded score (``relevance``, carrying
+          ``confidence`` when no explicit rank), rounded 2dp. Nothing ran AND
+          nothing emitted → ``(0.0, 0)`` so the console keeps suppressing the
+          chip. The coordinator lane is NOT a probe — never in the probe ratio.
+        """
+        with_data_probes = [a for a in trace if not ProbeTrace.is_dead_end(a.result)]
+        sources = {f"probe:{a.agent_id}" for a in with_data_probes}
+        sources |= {r.source for r in results if r.source}
+        sources_count = len(sources)
+
+        if trace:
+            confidence = round(len(with_data_probes) / len(trace), 2)
+        elif results:
+            # No probes ran (root-inline): mean of the entries' folded score
+            # (``relevance`` carries ``confidence`` when no explicit rank — see
+            # ``ResultsProjection._to_result``).
+            confidence = round(sum(r.relevance for r in results) / len(results), 2)
+        else:
             return 0.0, 0
-        with_data = [a for a in trace if not ProbeTrace.is_dead_end(a.result)]
-        return round(len(with_data) / len(trace), 2), len(with_data)
+        return confidence, sources_count
 
     @staticmethod
     def _task_result(
@@ -519,6 +674,7 @@ def _fail(
             status="failed",
             tier=run.tier,
             model=run.model,
+            total_ms=self._elapsed_ms(run),
             error=message,
         )
         if self._already_settled(store, run.run_id):