feat(brc): live-backed read_peer_artifact + wrapper sync-to-proposal (#3076)

docs/reference/agent-tools.md

@@ -130,7 +130,7 @@ that requires the handler docstring to explain why no CLI exists.
 | `mcp__brc__send_heartbeat` | Emit a structured `HEARTBEAT` (schema-validated, per-role deduped, rate-limited) to the dedicated `/heartbeat` endpoint. Use `state=WAITING_ON_ROLE` + `waiting_on=<peer>` while blocking on BRC. Valid states: `WORKING`, `WAITING_ON_ROLE`, `WAITING_FOR_EVENT`, `PROPOSED`, `IDLE`. | `handlers.message.message_heartbeat` | `egg-orch message heartbeat` |
 
 > **Blocking waits use Bash, not MCP** (#2211). Long-poll waits don't fit the MCP transport — both transports cap tool calls below typical quiet-phase intervals (~30 s streamable-HTTP, ~60 s in-process SDK), and every cap-elapsed return is a wasted LLM turn. Use `egg-orch message wait` / `egg-orch message wait-loop` (sandbox) and `egg-orch pipeline wait-status` (host) via Bash. The §1 idiom in `docs/reference/agent-wait-patterns.md` is the canonical shape.
-| `mcp__brc__read_peer_artifact` | Read entries from `.egg-state/brc-history/<identifier>-<phase>.json` (and the per-slice partition `<identifier>-implement-<slice_id>.json` when `EGG_SLICE_ID` is set and `phase == "implement"`; the sibling `<identifier>-implement-unattributed.json` is merged in by default and disabled via `include_unattributed=False`). Optional filters: `peer_role` / `producer_role` (alias), `message_type` (str or list). `limit` / `cursor` pagination (default `limit=50`, max 500). The identifier is resolved server-side from `EGG_ISSUE_NUMBER` / `EGG_PIPELINE_ID` (agents cannot pass an arbitrary id; path-traversal hardening). Returns `{items: [...], next_cursor: <str|None>, total_available: <int>, skipped_malformed: <int>, hint?: <str>}` — `hint` is present only when no history file exists on disk, the expected state for the phase currently in flight (#3076): brc-history is written at phase *completion* and reaches an agent worktree only at spawn, so a current-phase read is structurally empty and is not evidence that peers have not proposed (live proposals: `pending_reviews[].proposal_commit_sha` + `git show <sha>:<path>`). | `handlers.brc.brc_read_peer_artifact` | `egg-orch brc read-peer-artifact` *(slice-5 of #2908; thin wrapper, registration still `cli_command=None` — see callout below)* |
+| `mcp__brc__read_peer_artifact` | Read the BRC transcript for a phase from TWO merged sources (#3076 / #3077 phase 1): the orchestrator's live `/brc-transcript` route (the message store holds exactly the in-flight phase — a peer's `CONSENSUS_PROPOSE` is visible the moment it is sent, Delphi-redacted server-side for unreviewed reviewers) and the local `.egg-state/brc-history/<identifier>-<phase>.json` files (phases completed before spawn; per-slice partition `<identifier>-implement-<slice_id>.json` when `EGG_SLICE_ID` is set and `phase == "implement"`, with the `unattributed` sibling merged in unless `include_unattributed=False`). Records dedup by message id, sorted by timestamp. Optional filters: `peer_role` / `producer_role` (alias), `message_type` (str or list). `limit` / `cursor` pagination (default `limit=50`, max 500). The identifier / pipeline id are resolved server-side from `EGG_ISSUE_NUMBER` / `EGG_PIPELINE_ID` (agents cannot pass an arbitrary id; path-traversal hardening). Returns `{items: [...], next_cursor: <str|None>, total_available: <int>, skipped_malformed: <int>, live: <bool>, hint?: <str>}` — `live` says whether the live route contributed; `hint` is present only when both sources are empty: with the live route reachable that genuinely means no BRC messages for the phase yet, without it the emptiness is structural (#3076; brc-history is written at phase *completion*) and the hint points at the event-payload fallback (`pending_reviews[].proposal_commit_sha` + `git show <sha>:<path>`). | `handlers.brc.brc_read_peer_artifact` | `egg-orch brc read-peer-artifact` *(slice-5 of #2908; thin wrapper, registration still `cli_command=None` — see callout below)* |
 | `mcp__brc__resolve_obligation` | Mark a reviewer's conditional-ACK obligation as satisfied in-cycle (#2338). Required: `reviewer_role`, `producer_role`. Optional: `commit_sha`, `note`. The matrix keeps the obligation text for audit, but `get_pre_merge_conditions` filters resolved entries — the PR body and HITL gate stop surfacing the obligation. The orchestrator persists a `CONSENSUS_OBLIGATION_RESOLVED` message so the resolution survives orchestrator restart, and rejects `resolver_role == producer_role` so a producer cannot self-resolve their own obligation. Resolution is per-version: any later ACK / NACK / invalidate on the same edge resets the resolved flag. | `handlers.brc.brc_resolve_obligation` | `egg-orch brc resolve-obligation` *(slice-5 of #2908; thin wrapper, registration still `cli_command=None` — see callout below)* |
 
 #### `brc_propose` push behavior
@@ -305,7 +305,7 @@ verbs are:
 - `mcp__sdlc__check_hitl_answers` — no CLI; aggregates HITL state across phases.
 - `mcp__brc__get_state` — no CLI in MCP registry (`cli_command=None`); has a verb-level CLI alias `egg-orch brc get-state` (added in #2908 for the event-pump wrapper). Schema derives from `schemas.py`, not argparse.
 - `mcp__brc__list_blocking` — no CLI in MCP registry (`cli_command=None`); has a verb-level CLI alias `egg-orch brc list-blocking` (added in #2908).
-- `mcp__brc__read_peer_artifact` — registration says no CLI; thin `egg-orch brc read-peer-artifact` wrapper added in slice-5 of #2908. Reads `.egg-state/brc-history/<identifier>-<phase>.json` files from local disk — no HTTP / gateway.
+- `mcp__brc__read_peer_artifact` — registration says no CLI; thin `egg-orch brc read-peer-artifact` wrapper added in slice-5 of #2908. Merges the orchestrator's live `/brc-transcript` route (HTTP, current phase) with the `.egg-state/brc-history/<identifier>-<phase>.json` files on local disk (completed phases) — see #3076.
 - `mcp__brc__resolve_obligation` — registration says no CLI; thin `egg-orch brc resolve-obligation` wrapper added in slice-5 of #2908. Net-new in-cycle conditional-ACK obligation-resolution capability (#2338); producer/tester-driven via the MCP surface and the wrapper.
 - `mcp__phase__get_context` — no CLI in MCP registry (`cli_command=None`); has a verb-level CLI alias `egg-orch phase get-context` (added in #2908 for the event-pump wrapper). Schema derives from `schemas.py`, not argparse.
 - `mcp__phase__get_assigned_tasks` — no CLI; filtered view over `egg-contract show`.

orchestrator/consensus_wrapper.py

@@ -457,6 +457,85 @@
     {agent_command_prefix} "$prompt"
 }}
 
+# Sync-to-proposal (#3076 / #3077 clause 2): before a review invocation
+# (ack/nack), merge each pending producer's proposed commit into this
+# reviewer's worktree so reviewers that must RUN the proposal (tester)
+# have a real checkout. This is deterministic wrapper bash replacing
+# the fetch/merge prose that previously lived in spawn prompts the
+# event pump provably discards (#3033). Fail-soft at every step: an
+# unresolvable SHA, a conflicting merge, or a dirty tree logs and
+# continues — the per-event prompt's `git show <sha>:<path>` reads
+# (#3078) work from the shared object store either way, so the agent
+# is never blocked on this sync succeeding.
+#
+# NB: ``git merge --no-edit`` produces a real merge commit (or fast-
+# forwards) — HEAD advances and is NOT reset between events. Two
+# durable side effects follow:
+#   - Dual-role agents (e.g. tester acting as producer-then-reviewer-
+#     then-producer): a subsequent producer turn commits on top of an
+#     ancestry that contains peer proposal commits. The producer arm
+#     (R11a, see the dispatcher below) intentionally skips the sync,
+#     but it does NOT un-merge syncs from prior reviewer arms.
+#   - Multiple producers per event: SHAs are merged sequentially. A
+#     conflict on the second SHA aborts that merge but leaves the
+#     first merge intact on HEAD.
+# Neither is wrong — the per-event-prompt ``git show <sha>:<path>``
+# fallback covers the failure path either way — but this is durable
+# HEAD mutation, not transient enrichment.
+sync_to_proposals() {{
+    local event_payload="$1"
+    local repo="${{EGG_REPO_PATH:-$PWD}}"
+    local shas sha
+    # Extract pending_reviews[].proposal_commit_sha, strictly hex-
+    # validated (7-64 chars) before any git interpolation. The payload
+    # is orchestrator-composed, but the producer-supplied SHA rides
+    # through it, so revalidate at the consumer — same stance as
+    # event_prompt.py's _extract_proposal_sha_for_producer. The
+    # ProposalPayload writer-side check also admits non-hex sentinels
+    # like RECONSTRUCTED_NO_SHA; the hex requirement here filters those
+    # out (there is nothing to merge for a reconstructed proposal).
+    shas=$(printf '%s' "$event_payload" | python3 -c "
+import sys, json, re
+try:
+    d = json.load(sys.stdin)
+except Exception:
+    sys.exit(0)
+pat = re.compile(r'[0-9a-fA-F]{{7,64}}')
+out = []
+for pr in (d.get('pending_reviews') or []):
+    if not isinstance(pr, dict):
+        continue
+    sha = str(pr.get('proposal_commit_sha') or '')
+    if pat.fullmatch(sha) and sha not in out:
+        out.append(sha)
+print('\n'.join(out))
+" 2>/dev/null)
+    [ -z "$shas" ] && return 0
+    while IFS= read -r sha; do
+        [ -z "$sha" ] && continue
+        if ! git -C "$repo" cat-file -e "$sha^{{commit}}" 2>/dev/null; then
+            # Per-role worktrees share the host repo's object store, so
+            # the SHA normally resolves without network; a best-effort
+            # fetch covers split-object-store runtimes.
+            git -C "$repo" fetch --quiet origin >/dev/null 2>&1 || true
+        fi
+        if ! git -C "$repo" cat-file -e "$sha^{{commit}}" 2>/dev/null; then
+            cw_log "sync-to-proposal: $sha unresolvable in $repo; reviewer falls back to the prompt's git-show reads."
+            continue
+        fi
+        if git -C "$repo" merge-base --is-ancestor "$sha" HEAD 2>/dev/null; then
+            continue
+        fi
+        if git -C "$repo" merge --no-edit "$sha" >/dev/null 2>&1; then
+            cw_log "sync-to-proposal: merged proposal commit $sha into the worktree."
+        else
+            git -C "$repo" merge --abort >/dev/null 2>&1 || true
+            cw_log "sync-to-proposal: merge of $sha failed (conflict or dirty tree); aborted — reviewer reads via git show instead."
+        fi
+    done <<< "$shas"
+    return 0
+}}
+
 # Idle / no-progress safety budget (#2908 task-2-3). Replaces the
 # legacy capped-restart cap (deleted by task-4-2): if no actionable
 # event arrives for the configured idle budget we raise an
@@ -683,6 +762,15 @@
             # PR removed the legacy 3-restart cap; this rc gate is the
             # equivalent ceiling on the action path.
             cw_log "Invoking agent (action=$ACTION)."
+            # R11a (sync_to_proposals docstring anchor): the ``propose`` arm
+            # intentionally falls through this gate without syncing -- a
+            # producer's own commits are already on HEAD in its worktree,
+            # and merging peer proposal commits onto a producer turn is
+            # exactly the dual-role bleed-through the sync is designed to
+            # avoid. Only ``ack`` / ``nack`` (reviewer arms) sync.
+            if [ "$ACTION" = "ack" ] || [ "$ACTION" = "nack" ]; then
+                sync_to_proposals "$EVENT_PAYLOAD"
+            fi
             invoke_agent_for_event "$ACTION" "$EVENT_PAYLOAD"
             agent_rc=$?
             if [ "$agent_rc" -eq 0 ]; then

orchestrator/routes/event_prompt.py

@@ -721,6 +721,14 @@ def _build_delta_entries(
         artifacts = _extract_artifacts_for_producer(event_payload, producer)
         if not artifacts and not proposal_sha:
             continue
+        # Strip backticks from artifact paths before interpolation. The
+        # paths are producer-supplied through ``snapshot["artifacts"]``;
+        # ``proposal_sha`` is hex-validated upstream, but ``a`` is not.
+        # The agent (not bash) is the consumer here so this is not a
+        # shell-injection vector, but a stray backtick in a path would
+        # break the markdown code span the agent renders. Defensive
+        # belt-and-braces rather than trusting producer payloads.
+        artifacts = [a.replace("`", "") for a in artifacts]
         refs_text = "\n".join(f"- `{a}`" for a in artifacts)
         if proposal_sha:
             # First review with a known proposal SHA: render concrete,

orchestrator/routes/messages.py

@@ -399,6 +399,167 @@ def _apply_delphi_filter(
     return filtered_messages
 
 
+# Valid phase names for the live BRC transcript route. Mirrors the
+# sandbox handler's ``_VALID_PHASES`` (sandbox/egg_agent_tools/handlers/
+# brc.py) — the route and its only consumer must agree on the set.
+_BRC_TRANSCRIPT_PHASES: frozenset[str] = frozenset({"refine", "plan", "implement", "pr"})
+
+# Page-size bounds for the live BRC transcript route. The default is
+# deliberately high relative to poll_messages (a phase transcript is a
+# bounded audit log, not a live tail) and the cap matches the writer's
+# own retrieval bound in ``_write_brc_history`` (limit=10000).
+_BRC_TRANSCRIPT_DEFAULT_LIMIT = 1000
+_BRC_TRANSCRIPT_MAX_LIMIT = 10000
+
+
+@messages_bp.route("/<pipeline_id>/brc-transcript", methods=["GET"])
+def get_brc_transcript(pipeline_id: str) -> tuple[Response, int]:
+    """Serve the live BRC transcript for a phase from the message store.
+
+    This is the served-read counterpart of the on-disk
+    ``.egg-state/brc-history/`` files (#3076 / #3077 phase 1): those
+    files are written by ``pipelines._write_brc_history`` at phase
+    COMPLETION into the orchestrator's worktree, so for the phase
+    currently in flight they exist nowhere an agent can read. This
+    route serves the same records — ``Message.to_dict()`` dicts
+    filtered to ``BRC_HISTORY_TYPES`` — straight from the message
+    store, which holds exactly the in-flight phase's messages (the
+    store is cleared on phase transitions by
+    ``phases._clear_concurrent_state``).
+
+    Query params:
+        phase: required — one of refine/plan/implement/pr. Records are
+            matched on ``Message.phase``.
+        role: optional — the calling agent's role. Applied to the same
+            Delphi visibility filter as ``poll_messages`` so this route
+            is not a blinding bypass: a reviewer that has not yet
+            reviewed a producer sees that producer's CONSENSUS_PROPOSE
+            with the payload redacted to version + commit_sha.
+        slice_id: optional — canonical ``slice-<N>`` only. Mirrors the
+            writer's implement-phase attribution (#2548): CONSENSUS_*
+            records must carry a matching canonical
+            ``metadata.slice_id`` (missing/non-canonical ones are
+            dropped, writer parity); other BRC types without canonical
+            slice scope are the "unattributed" bucket.
+        include_unattributed: optional bool (default true) — include
+            the unattributed bucket when ``slice_id`` is given.
+        limit: optional page size (default 1000, max 10000). The most
+            recent ``limit`` records are returned in chronological
+            order.
+
+    Response data::
+
+        {"records": [...], "count": N, "phase": "plan",
+         "truncated": false}
+
+    ``truncated`` reflects post-filter trimming only: it is true iff
+    the phase-filtered record set exceeded ``limit`` and was clipped
+    to the most recent ``limit`` records. It does **not** report
+    upstream message-store overflow. The route reads up to
+    ``_BRC_TRANSCRIPT_MAX_LIMIT`` (10000) raw messages from the store
+    before phase filtering, which matches ``_write_brc_history``'s own
+    retrieval bound — so in steady state both sides see the same
+    window and ``truncated`` is the only loss signal that matters. A
+    pipeline that holds >10000 messages and never transitions phase
+    cleanly could in principle lose the oldest BRC records to the
+    upstream cap; if you need to detect that case, cross-reference
+    ``count`` with the message store's own metrics.
+
+    Auth: agent-facing, same trust surface as ``poll_messages`` (the
+    gateway-enforced NetworkPolicy is the boundary; ``role`` is caller-
+    supplied there too).
+    """
+    try:
+        get_state_store_for_pipeline(pipeline_id)
+    except InvalidPipelineIdError as e:
+        return _make_error(str(e), 400)
+    except PipelineNotFoundError as e:
+        return _make_error(str(e), 404)
+
+    phase = (request.args.get("phase") or "").strip()
+    if phase not in _BRC_TRANSCRIPT_PHASES:
+        return _make_error(
+            f"'phase' must be one of {sorted(_BRC_TRANSCRIPT_PHASES)}; got {phase!r}"
+        )
+
+    role = request.args.get("role") or None
+
+    raw_slice_id = request.args.get("slice_id")
+    slice_id: str | None = None
+    if raw_slice_id:
+        try:
+            slice_id = _extract_slice_id({"slice_id": raw_slice_id})
+        except ValueError as exc:
+            return _make_error(f"Invalid slice_id: {exc}")
+
+    include_unattributed = (request.args.get("include_unattributed") or "true").lower() not in (
+        "false",
+        "0",
+        "no",
+    )
+
+    try:
+        limit = int(request.args.get("limit", str(_BRC_TRANSCRIPT_DEFAULT_LIMIT)))
+    except ValueError, TypeError:
+        return _make_error("Invalid limit parameter: must be an integer")
+    if limit <= 0:
+        return _make_error("Invalid limit parameter: must be > 0")
+    limit = min(limit, _BRC_TRANSCRIPT_MAX_LIMIT)
+
+    # Lazy import: ``routes.pipelines`` is the canonical owner of the
+    # BRC type sets (and is always loaded in a running app), but a
+    # module-level import here would be load-order sensitive and pull
+    # the 16k-line module into any test that imports this blueprint.
+    try:
+        from routes.pipelines import BRC_HISTORY_TYPES, CONSENSUS_BRC_TYPES
+    except ImportError:  # pragma: no cover - package-relative fallback
+        from .pipelines import (  # type: ignore[no-redef]
+            BRC_HISTORY_TYPES,
+            CONSENSUS_BRC_TYPES,
+        )
+    try:
+        from slice_id_validation import SLICE_ID_PATTERN
+    except ImportError:  # pragma: no cover - package-relative fallback
+        from ..slice_id_validation import SLICE_ID_PATTERN  # type: ignore[no-redef]
+
+    message_store = get_message_store()
+    messages = message_store.get_messages(pipeline_id, limit=_BRC_TRANSCRIPT_MAX_LIMIT)
+
+    records = [m for m in messages if m.message_type in BRC_HISTORY_TYPES and m.phase == phase]
+
+    if slice_id is not None:
+        # Mirror the writer's per-slice attribution (#2548): CONSENSUS_*
+        # without a canonical slice_id is a contract violation and is
+        # dropped; non-consensus BRC types without canonical slice scope
+        # form the cross-cutting "unattributed" bucket.
+        bucket: list[Message] = []
+        for m in records:
+            raw_sid = m.metadata.get("slice_id") if isinstance(m.metadata, dict) else None
+            canonical = isinstance(raw_sid, str) and bool(SLICE_ID_PATTERN.match(raw_sid))
+            if canonical:
+                if raw_sid == slice_id:
+                    bucket.append(m)
+            elif m.message_type not in CONSENSUS_BRC_TYPES and include_unattributed:
+                bucket.append(m)
+        records = bucket
+
+    records = _apply_delphi_filter(pipeline_id, role, records)
+
+    truncated = len(records) > limit
+    if truncated:
+        records = records[-limit:]
+
+    return _make_success(
+        "BRC transcript retrieved",
+        data={
+            "records": [m.to_dict() for m in records],
+            "count": len(records),
+            "phase": phase,
+            "truncated": truncated,
+        },
+    )
+
+
 # Message types that are produced *as a side effect* of a producer's
 # own confirm reaching global consensus. A producer in WORKING/PROPOSED
 # that waits on these would be waiting on itself — its own confirm is