bearlike
diff --git a/‎apps/mewbo_api/src/mewbo_api/agentic_search/scg/CLAUDE.md‎
Lines changed: 28 additions & 12 deletions b/‎apps/mewbo_api/src/mewbo_api/agentic_search/scg/CLAUDE.md‎
Lines changed: 28 additions & 12 deletions
diff --git a/‎apps/mewbo_api/src/mewbo_api/agentic_search/scg/orchestrated_runner.py‎
Lines changed: 45 additions & 24 deletions b/‎apps/mewbo_api/src/mewbo_api/agentic_search/scg/orchestrated_runner.py‎
Lines changed: 45 additions & 24 deletions
diff --git a/‎apps/mewbo_api/src/mewbo_api/agentic_search/scg/run_streamer.py‎
Lines changed: 73 additions & 20 deletions b/‎apps/mewbo_api/src/mewbo_api/agentic_search/scg/run_streamer.py‎
Lines changed: 73 additions & 20 deletions
diff --git a/‎apps/mewbo_api/src/mewbo_api/agentic_search/schemas.py‎
Lines changed: 4 additions & 0 deletions b/‎apps/mewbo_api/src/mewbo_api/agentic_search/schemas.py‎
Lines changed: 4 additions & 0 deletions
@@ -106,18 +106,34 @@ keyed everything off `sub_agent` events → blank trace, `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.
+  the live interleaving.
+- **Probe tool_results ARE on this transcript/bus (#102 — the #95
+  "own-sessions" premise was WRONG, verified live in Mongo).** A child loop
+  inherits the parent's `event_logger` (core `AgentContext.child`), and every
+  `tool_result` payload carries the emitting `agent_id` — so EVERY
+  `tool_result` must be classified `agent_id ∈ probe lanes` before
+  projection (live `_probe_emitter` / settle `probe_ids` from the trace; a
+  spawn's `sub_agent` `start` always precedes the child's first tool call, so
+  the lane is known in time). Unclassified, probe tool calls mislabel into
+  the coordinator lane. A probe's `scg_results` projects as ITS result
+  cards; its other tool calls project NOWHERE (the probe lane stays
+  lifecycle-only — the #86 evidence rides the stop summary).
+- **`scg_results` = transcript-as-transport, EVERY search agent emits.** The
+  tool (`mewbo_graph.plugins.scg.results`, granted via `TRAVERSAL_TOOLS` +
+  bound to probes by the capability gate) 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>` (root) / `r-<run_id8>-<agent8>-<n>` (probe emit, salted
+  so concurrent emitters never collide) — that id is the live↔settle dedup
+  key. Entry `confidence` rides the wire verbatim (`SearchResult.confidence`,
+  #102) AND folds into `relevance` only when `relevance` is absent. Playbook
+  discipline: each probe emits ONCE before its evidence block; the ROOT
+  emits once, before synthesis, ONLY for hits it grounded inline (never
+  re-emitting a probe's — duplicate cards have distinct ids, the playbook is
+  the dedup). Emitting is NOT terminal for anyone — a probe's terminal stays
+  the stop-summary evidence block (#86).
 - **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
 
@@ -306,18 +306,25 @@ def _settle(
             return None
 
         trace = self._build_trace(records)
+        # Probe lanes are the classification key for tool_result events (#102):
+        # a child loop inherits the parent's event_logger, so probe tool calls
+        # land on THIS transcript stamped with the probe's agent_id — they must
+        # never read as root/coordinator activity. Same classifier as the live
+        # streamer (agent_id ∈ probe lanes), so live and settle agree.
+        probe_ids = {agent.agent_id for agent in trace}
         # 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)
+        coordinator_slot = self._assign_lane_slots(records, trace, probe_ids)
+        coordinator_lines = self._build_coordinator_lines(records, probe_ids)
         # ``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)
+        # the root emits once before synthesis; each probe may emit its own cards
+        # (#102). The api projects every emit, ids salted by the emitting probe.
+        results = self._build_results(run.run_id, records, probe_ids)
         # 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)
@@ -489,15 +496,18 @@ def _build_trace(records: list[dict[str, Any]]) -> list[TraceAgent]:
 
     @staticmethod
     def _assign_lane_slots(
-        records: list[dict[str, Any]], trace: list[TraceAgent]
+        records: list[dict[str, Any]],
+        trace: list[TraceAgent],
+        probe_ids: set[str],
     ) -> 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
+        sentinel coordinator, opened by its first ROOT ``tool_result`` — a
+        probe's own tool_result, classified by *probe_ids*, never opens it) 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).
@@ -514,6 +524,8 @@ def _assign_lane_slots(
             elif kind == "tool_result":
                 if not CoordinatorTrace.is_root_event(payload):
                     continue
+                if str(payload.get("agent_id") or "") in probe_ids:
+                    continue
                 key = coordinator_key
             else:
                 continue
@@ -525,38 +537,45 @@ def _assign_lane_slots(
         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).
+    def _build_coordinator_lines(
+        records: list[dict[str, Any]], probe_ids: set[str]
+    ) -> list[TraceLine]:
+        """Project the ROOT's ``tool_result`` events into coordinator trace lines.
+
+        A ``tool_result`` in this transcript is NOT always the root's: probe
+        tool calls ride the same transcript stamped with the probe's
+        ``agent_id`` (the child loop inherits the parent's event_logger — #102
+        corrected the #95 own-sessions premise), so any payload whose
+        ``agent_id`` is a probe lane is excluded here exactly as the live
+        streamer excludes it. Each remaining event 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 issued no tool calls.
         """
         lines: list[TraceLine] = []
         for rec in records:
             if rec.get("type") != "tool_result":
                 continue
             payload = rec.get("payload") or {}
+            if str(payload.get("agent_id") or "") in probe_ids:
+                continue
             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]]
+        run_id: str, records: list[dict[str, Any]], probe_ids: set[str]
     ) -> 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.
+        Transcript-as-transport (#95/#102): the root emits once before the
+        synthesis; each probe may emit its own grounded cards. The api projects
+        every entry onto a stable-id :class:`SearchResult` via
+        :class:`ResultsProjection` (the SAME read the live stream uses, salted
+        by the emitting probe's ``agent_id``, so ids — and thus dedup — agree).
+        Emits are concatenated in transcript order; the stable id keeps a
+        re-emit of the same call idempotent across live + settle.
         """
         out: list[SearchResult] = []
         seen: set[str] = set()
@@ -566,7 +585,9 @@ def _build_results(
             payload = rec.get("payload") or {}
             if not ResultsProjection.is_results_event(payload):
                 continue
-            for item in ResultsProjection.parse(run_id, payload):
+            agent_id = str(payload.get("agent_id") or "")
+            emitter = agent_id if agent_id in probe_ids else None
+            for item in ResultsProjection.parse(run_id, payload, emitter=emitter):
                 if item.id in seen:
                     continue
                 seen.add(item.id)
 
@@ -242,10 +242,12 @@ class ResultsProjection:
 
     The ``scg_results`` SessionTool is transcript-as-transport (#95): it validates
     + echoes, the api projects. This static maps each emitted entry onto a
-    :class:`SearchResult` with a STABLE id (``r-<run_id8>-<n>``) so the live
-    stream and settle reconciliation mint the same ids — the dedup key that keeps
-    a result from being emitted twice. Both the live ``_project`` and settle read
-    through here (the ProbeTrace stance), so a result is byte-identical either way.
+    STABLE id — ``r-<run_id8>-<n>`` for the root's emit, ``r-<run_id8>-<agent8>-<n>``
+    for a probe's (#102: probes emit their own cards; the agent suffix keeps
+    concurrent emitters collision-free) — so the live stream and settle
+    reconciliation mint the same ids — the dedup key that keeps a result from
+    being emitted twice. Both the live ``_project`` and settle read through here
+    (the ProbeTrace stance), so a result is byte-identical either way.
     """
 
     @staticmethod
@@ -254,22 +256,32 @@ def is_results_event(payload: dict[str, Any]) -> bool:
         return str(payload.get("tool_id") or "") == "scg_results"
 
     @staticmethod
-    def parse(run_id: str, payload: dict[str, Any]) -> list[SearchResult]:
+    def parse(
+        run_id: str, payload: dict[str, Any], *, emitter: str | None = None
+    ) -> list[SearchResult]:
         """Map the emit's entries onto stable-id :class:`SearchResult`s.
 
+        ``emitter`` is the probe ``agent_id`` when the emit came from a probe
+        lane (``None`` for the root / a legacy payload): it salts the stable id
+        so two agents' emits never collide — and because the same transcript
+        event carries the same ``agent_id`` live and at settle, the ids agree
+        across both reads (the dedup invariant).
+
         Drops any entry that fails the wire model (an ungrounded/malformed card)
         rather than failing the whole projection — the tool already validated on
         the way in, this is the lenient read side. Fields map honestly onto
-        ``SearchResult``; the entry ``confidence`` has no wire home, so it is
-        folded into ``relevance`` ONLY when ``relevance`` was not supplied (never
-        overwriting an explicit rank) — keeping the strongest available signal on
-        the projected card so the settle metrics can read it.
+        ``SearchResult``: ``confidence`` is surfaced verbatim on the wire (#102)
+        AND folded into ``relevance`` ONLY when ``relevance`` was not supplied
+        (never overwriting an explicit rank) — keeping the strongest available
+        signal on the projected card so the settle metrics can read it.
         """
         tool_input = payload.get("tool_input")
         entries = tool_input.get("results") if isinstance(tool_input, dict) else None
         if not isinstance(entries, list):
             return []
         prefix = run_id[:8]
+        if emitter:
+            prefix = f"{prefix}-{emitter[:8]}"
         out: list[SearchResult] = []
         for n, entry in enumerate(entries):
             if not isinstance(entry, dict):
@@ -286,22 +298,25 @@ def _to_result(result_id: str, entry: dict[str, Any]) -> SearchResult:
 
         ``relevance`` carries the entry's explicit rank; absent (or 0), the
         entry's ``confidence`` is folded in as the best available signal so the
-        card and the settle metrics aren't blind. ``confidence`` itself has no
-        wire field, so it is not surfaced separately.
+        card and the settle metrics aren't blind. ``confidence`` ALSO rides its
+        own wire field verbatim (#102) so the console can render the emitting
+        agent's per-card certainty beside the relevance rank.
         """
         raw_kind = str(entry.get("kind") or "docs")
         if raw_kind not in _RESULT_KINDS:
             raise ValueError(f"unknown result kind: {raw_kind}")
         kind = cast(ResultKindLiteral, raw_kind)
         url = entry.get("url")
+        raw_confidence = entry.get("confidence")
         relevance = float(entry.get("relevance") or 0.0)
-        if relevance <= 0.0 and entry.get("confidence") is not None:
-            relevance = float(entry.get("confidence") or 0.0)
+        if relevance <= 0.0 and raw_confidence is not None:
+            relevance = float(raw_confidence or 0.0)
         return SearchResult(
             id=result_id,
             source=str(entry.get("source") or ""),
             kind=kind,
             relevance=relevance,
+            confidence=float(raw_confidence) if raw_confidence is not None else None,
             title=str(entry.get("title") or ""),
             url=str(url) if url else "",
             snippet=str(entry.get("snippet") or ""),
@@ -409,20 +424,58 @@ def _consume(self) -> None:
     def _project(self, record: EventRecord) -> None:
         """Project one transcript event onto the run event log (live).
 
-        Two lanes share this seam: a probe's ``sub_agent`` lifecycle (the probes
-        run in their OWN sessions, but the run streamer subscribes to THIS
-        session's bus — so what arrives here as a ``sub_agent`` is the parent's
-        view of its children) and the ROOT coordinator's ``tool_result`` events
-        (#95 — the root agent's own tool activity on this session's bus). The
-        probe tool calls never appear here; they live in the probes' sessions.
+        Three event classes share this seam: a probe's ``sub_agent`` lifecycle
+        (the parent's view of its children), the ROOT coordinator's
+        ``tool_result`` events (#95), and a PROBE's own ``tool_result`` events.
+        The last class exists because a child loop INHERITS the parent's
+        ``event_logger`` (core ``AgentContext.child``) — probe tool calls land
+        on THIS session's transcript/bus stamped with the probe's ``agent_id``
+        (#102; the original #95 premise that they "live in the probes' own
+        sessions" was wrong, verified live). A probe ``tool_result`` must
+        therefore be classified by ``payload.agent_id`` against the known probe
+        lanes (the spawn's ``sub_agent`` ``start`` always precedes the child's
+        first tool call, so the lane is known by the time its tools fire) —
+        only its ``scg_results`` emit projects (as the probe's result cards);
+        everything else stays off the coordinator lane.
         """
         kind = record.get("type")
         raw = record.get("payload")
         payload: dict[str, Any] = raw if isinstance(raw, dict) else {}
         if kind == "sub_agent":
             self._project_sub_agent(payload)
         elif kind == "tool_result":
-            self._project_coordinator(payload)
+            emitter = self._probe_emitter(payload)
+            if emitter is not None:
+                self._project_probe_tool(emitter, payload)
+            else:
+                self._project_coordinator(payload)
+
+    def _probe_emitter(self, payload: dict[str, Any]) -> str | None:
+        """The probe ``agent_id`` when this tool_result came from a probe lane.
+
+        ``None`` for the root's own tool calls (its ``agent_id`` never opens a
+        probe lane) and for legacy payloads with no ``agent_id`` — both keep
+        the historical coordinator-lane path.
+        """
+        agent_id = str(payload.get("agent_id") or "")
+        if not agent_id:
+            return None
+        with self._lock:
+            return agent_id if agent_id in self._agents else None
+
+    def _project_probe_tool(self, emitter: str, payload: dict[str, Any]) -> None:
+        """Project a probe's own ``tool_result`` (#102).
+
+        Only the probe's ``scg_results`` emit reaches the run log — its entries
+        become ``result`` events with probe-salted stable ids. Every other probe
+        tool call is deliberately dropped here: the probe lane stays
+        lifecycle-only (#86 — its evidence rides the ``stop`` summary), and the
+        coordinator lane must not absorb child activity (the #95 mislabel).
+        """
+        if ResultsProjection.is_results_event(payload):
+            self._emit_results(
+                ResultsProjection.parse(self._run_id, payload, emitter=emitter)
+            )
 
     def _project_sub_agent(self, payload: dict[str, Any]) -> None:
         """Project one probe ``sub_agent`` lifecycle event onto a probe lane."""
 
@@ -284,6 +284,10 @@ class SearchResult(_Wire):
     source: str
     kind: ResultKindLiteral
     relevance: float = 0.0
+    # How sure the emitting agent is this hit answers the query (0..1) —
+    # carried verbatim from an ``scg_results`` entry (agent-emitted cards
+    # only). ``None`` when the emitter offered no defensible confidence.
+    confidence: float | None = None
     title: str
     url: str = ""
     snippet: str = ""