Skip to content
Merged
Show file tree
Hide file tree
Changes from 3 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions docs/reference/agent-tools.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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`.
Expand Down
88 changes: 88 additions & 0 deletions orchestrator/consensus_wrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
8 changes: 8 additions & 0 deletions orchestrator/routes/event_prompt.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
161 changes: 161 additions & 0 deletions orchestrator/routes/messages.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading
Loading