diff --git a/docs/reference/agent-tools.md b/docs/reference/agent-tools.md index d60023ca8c..3d6a6b2290 100644 --- a/docs/reference/agent-tools.md +++ b/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=` 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/-.json` (and the per-slice partition `-implement-.json` when `EGG_SLICE_ID` is set and `phase == "implement"`; the sibling `-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: , total_available: , skipped_malformed: , hint?: }` — `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 :`). | `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/-.json` files (phases completed before spawn; per-slice partition `-implement-.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: , total_available: , skipped_malformed: , live: , hint?: }` — `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 :`). | `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/-.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/-.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`. diff --git a/orchestrator/consensus_wrapper.py b/orchestrator/consensus_wrapper.py index 7499b3461c..3a537baa2e 100644 --- a/orchestrator/consensus_wrapper.py +++ b/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 :` 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 :`` +# 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 diff --git a/orchestrator/routes/event_prompt.py b/orchestrator/routes/event_prompt.py index 9f2ec48625..df6a2f6c1a 100644 --- a/orchestrator/routes/event_prompt.py +++ b/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, diff --git a/orchestrator/routes/messages.py b/orchestrator/routes/messages.py index 256901c34a..6ac3c0a009 100644 --- a/orchestrator/routes/messages.py +++ b/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("//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-`` 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 diff --git a/orchestrator/tests/test_compose_event_prompt.py b/orchestrator/tests/test_compose_event_prompt.py index 4ea2ec1b05..5713aeb6fe 100644 --- a/orchestrator/tests/test_compose_event_prompt.py +++ b/orchestrator/tests/test_compose_event_prompt.py @@ -932,6 +932,86 @@ def test_build_delta_entries_first_review_renders_git_show_commands() -> None: assert f"git log {full_sha} --not origin/main -p" in delta_full +def test_build_delta_entries_first_review_strips_backticks_from_artifact_paths() -> None: + """Producer-supplied artifact paths containing a backtick would otherwise + break the markdown code span the agent renders. The first-review + fallback strips backticks before interpolation; ``proposal_sha`` is + hex-validated upstream so it is not the attack surface. Belt-and- + braces only -- the agent (not bash) is the consumer, so this is not + a shell-injection vector.""" + from pathlib import Path + + from orchestrator.routes.event_prompt import _build_delta_entries + + entries = _build_delta_entries( + action="ack", + role="reviewer_code", + base_branch="main", + repo_path=Path("/tmp"), + memory_text="", + event_payload={ + "pending_reviews": [ + { + "producer": "coder", + "current_version": 1, + "artifact_refs": ["src/`evil`.py", "src/ok.py"], + "proposal_commit_sha": "abc1234", + } + ] + }, + ) + assert len(entries) == 1 + delta = entries[0]["delta"] + # The stripped path is what gets interpolated; no raw backtick from + # the artifact value reaches the rendered ``git show`` command, which + # would otherwise terminate the surrounding markdown code span. + assert "git show abc1234:src/evil.py" in delta + assert "git show abc1234:src/ok.py" in delta + assert "`evil`" not in delta + + +def test_build_delta_entries_first_review_no_sha_strips_backticks_from_refs_text() -> None: + """Parallel to the ``if proposal_sha:`` strip test above, but covers + the ``else`` (no-SHA) branch at ``event_prompt.py``. The strip is one + list comp upstream of both branches (line 731), so this is a + belt-and-braces pin on the shared upstream — a regression that + re-introduces unescaped backticks in either rendered path would + surface in both tests.""" + from pathlib import Path + + from orchestrator.routes.event_prompt import _build_delta_entries + + entries = _build_delta_entries( + action="ack", + role="reviewer_code", + base_branch="main", + repo_path=Path("/tmp"), + memory_text="", + event_payload={ + "pending_reviews": [ + { + "producer": "coder", + "current_version": 1, + "artifact_refs": ["src/`evil`.py", "src/ok.py"], + # No proposal_commit_sha -> drives the else branch + # (degraded-baseline ``refs_text`` rendering). + } + ] + }, + ) + assert len(entries) == 1 + delta = entries[0]["delta"] + # ``refs_text`` interpolates each artifact in a backtick-wrapped code + # span (``- `{a}` ``). With the strip applied, ``evil`` lands without + # surrounding raw backticks; without it, the inner backticks would + # terminate the markdown code span early. + assert "- `src/evil.py`" in delta + assert "- `src/ok.py`" in delta + assert "`evil`" not in delta + # Sanity: this is the no-SHA branch, no ``git show`` command rendered. + assert "git show" not in delta + + def test_build_delta_entries_first_review_sha_without_artifacts_still_renders() -> None: """A proposal SHA with an empty artifact list still yields an entry (the full-change git log command) — pre-#3076 the empty artifact diff --git a/orchestrator/tests/test_consensus_wrapper.py b/orchestrator/tests/test_consensus_wrapper.py index 6abbcd93b5..85e1160839 100644 --- a/orchestrator/tests/test_consensus_wrapper.py +++ b/orchestrator/tests/test_consensus_wrapper.py @@ -1317,3 +1317,178 @@ def test_no_effort_omits_flag(self): cmd = build_consensus_wrapped_command("Prompt", model="opus") script = cmd[2] assert "--effort" not in script + + +class TestSyncToProposals: + """Wrapper-performed sync-to-proposal on review actions (#3076 / + #3077 clause 2). + + The designed mid-phase artifact flow used to live as fetch/merge + prose in spawn prompts the event pump provably discards + (``del prompt_text``, #3033) — so reviewers that must RUN a + proposal (tester) never had the producer's commits in their + worktree. The wrapper now performs that sync deterministically: + before an ``ack``/``nack`` invocation it merges each pending + producer's ``proposal_commit_sha`` into the reviewer worktree, + fail-soft at every step. + """ + + def _script(self, monkeypatch) -> str: + monkeypatch.setenv("EGG_BRC_EVENT_PUMP", "true") + return build_consensus_wrapped_command("Prompt")[2] + + def test_template_defines_sync_to_proposals(self, monkeypatch): + script = self._script(monkeypatch) + assert "sync_to_proposals() {" in script + + def test_sync_runs_only_for_review_actions(self, monkeypatch): + """The sync call must be gated on ack/nack — a producer's own + ``propose`` invocation must NOT merge peers' commits into its + worktree (R11a: propose own work first, peer state irrelevant). + """ + script = self._script(monkeypatch) + guard = 'if [ "$ACTION" = "ack" ] || [ "$ACTION" = "nack" ]; then' + assert guard in script + # The call rides inside that guard, with the event payload. + guarded_block = script.split(guard, 1)[1].split("fi", 1)[0] + assert 'sync_to_proposals "$EVENT_PAYLOAD"' in guarded_block + + def test_sync_precedes_agent_invocation(self, monkeypatch): + """Ordering invariant: the worktree must be synced BEFORE the + one-shot agent runs, or the tester still reviews a stale tree. + """ + script = self._script(monkeypatch) + sync_pos = script.index('sync_to_proposals "$EVENT_PAYLOAD"') + invoke_pos = script.index('invoke_agent_for_event "$ACTION" "$EVENT_PAYLOAD"') + assert sync_pos < invoke_pos + + def test_sha_extraction_is_hex_validated(self, monkeypatch): + """The producer-supplied SHA is interpolated into git argv; + the extractor must hex-validate (7-64 chars) so shell + metacharacters and non-hex sentinels (RECONSTRUCTED_NO_SHA) + never reach git. + """ + script = self._script(monkeypatch) + assert "[0-9a-fA-F]{7,64}" in script + assert "fullmatch" in script + + def test_merge_failure_is_fail_soft(self, monkeypatch): + """A conflicting merge must abort and continue — the per-event + prompt's ``git show`` reads (#3078) remain the fallback; the + agent invocation must never be blocked on the sync. + """ + script = self._script(monkeypatch) + assert "merge --abort" in script + # The function never propagates failure into the action arm. + fn_body = script.split("sync_to_proposals() {", 1)[1] + # Take through the function's closing `return 0`. + assert "return 0" in fn_body.split("\n}\n", 1)[0] + + def _extract_sync_harness(self, script: str, repo: str, payload: str) -> str: + """Build a runnable bash harness: cw_log + sync_to_proposals.""" + import re + + cw_match = re.search(r"cw_log\(\) \{.*?\n\}", script, flags=re.DOTALL) + assert cw_match is not None + sync_match = re.search(r"sync_to_proposals\(\) \{.*?\n\}", script, flags=re.DOTALL) + assert sync_match is not None + return ( + "#!/bin/bash\nset -uo pipefail\n" + f"EGG_REPO_PATH={shlex.quote(repo)}\n" + + cw_match.group(0) + + "\n" + + sync_match.group(0) + + "\nsync_to_proposals " + + shlex.quote(payload) + + '\necho "SYNC_RC=$?"\n' + ) + + def test_behavioral_merge_and_metachar_filter(self, tmp_path, monkeypatch): + """End-to-end: a real proposal SHA on a producer branch is + merged into the reviewer's checkout (the proposed artifact + becomes Read-able); a shell-metachar SHA is filtered before + any git command; the function exits 0 regardless. + """ + import json as _json + + repo = tmp_path / "repo" + repo.mkdir() + + def git(*args): + subprocess.run( + ["git", "-C", str(repo), *args], + check=True, + capture_output=True, + env={ + **os.environ, + "GIT_AUTHOR_NAME": "t", + "GIT_AUTHOR_EMAIL": "t@t", + "GIT_COMMITTER_NAME": "t", + "GIT_COMMITTER_EMAIL": "t@t", + }, + ) + + git("init", "-q", "-b", "main") + (repo / "f.txt").write_text("base\n") + git("add", ".") + git("commit", "-qm", "base") + git("checkout", "-qb", "producer") + (repo / "plan.md").write_text("the plan\n") + git("add", ".") + git("commit", "-qm", "plan draft") + sha = subprocess.run( + ["git", "-C", str(repo), "rev-parse", "HEAD"], + check=True, + capture_output=True, + text=True, + ).stdout.strip() + git("checkout", "-qb", "reviewer", "main") + + payload = _json.dumps( + { + "pending_reviews": [ + {"producer": "architect", "proposal_commit_sha": sha}, + {"producer": "evil", "proposal_commit_sha": "abc; rm -rf /"}, + {"producer": "noop", "proposal_commit_sha": ""}, + ] + } + ) + script = self._script(monkeypatch) + harness = self._extract_sync_harness(script, str(repo), payload) + result = subprocess.run( + ["bash", "-c", harness], + capture_output=True, + text=True, + timeout=30, + env={ + **os.environ, + "GIT_AUTHOR_NAME": "t", + "GIT_AUTHOR_EMAIL": "t@t", + "GIT_COMMITTER_NAME": "t", + "GIT_COMMITTER_EMAIL": "t@t", + }, + ) + assert "SYNC_RC=0" in result.stdout, ( + f"sync_to_proposals must exit 0; stdout={result.stdout!r} stderr={result.stderr!r}" + ) + # The proposed artifact is now a real file in the reviewer tree. + assert (repo / "plan.md").read_text() == "the plan\n" + # The metachar SHA was filtered, not executed/attempted. + assert "rm -rf" not in result.stderr + + def test_behavioral_unresolvable_sha_logs_and_continues(self, tmp_path, monkeypatch): + """A well-formed but unknown SHA logs the git-show fallback and + exits 0 — never fails the action arm.""" + import json as _json + + repo = tmp_path / "repo" + repo.mkdir() + subprocess.run(["git", "-C", str(repo), "init", "-q", "-b", "main"], check=True) + payload = _json.dumps( + {"pending_reviews": [{"producer": "x", "proposal_commit_sha": "a" * 40}]} + ) + script = self._script(monkeypatch) + harness = self._extract_sync_harness(script, str(repo), payload) + result = subprocess.run(["bash", "-c", harness], capture_output=True, text=True, timeout=30) + assert "SYNC_RC=0" in result.stdout + assert "unresolvable" in result.stderr diff --git a/orchestrator/tests/test_messages.py b/orchestrator/tests/test_messages.py index f1a30d14bf..a962c5caa9 100644 --- a/orchestrator/tests/test_messages.py +++ b/orchestrator/tests/test_messages.py @@ -20,7 +20,7 @@ from message_store import Message, MessageStore, MessageType, reset_message_store from routes.messages import messages_bp -from state_store import InvalidPipelineIdError +from state_store import InvalidPipelineIdError, PipelineNotFoundError @pytest.fixture @@ -3403,3 +3403,193 @@ def test_heartbeat_non_object_json_body_returns_400(self, client, app, raw_body) body = json.loads(resp.data) assert body["success"] is False assert "json object" in body["message"].lower(), body + + +class TestBrcTranscript: + """Live BRC transcript route (#3076 / #3077 phase 1). + + Serves the in-flight phase's BRC records straight from the message + store — the served-read counterpart of the on-disk brc-history + files, which are written only at phase COMPLETION and therefore + can never cover the current phase from an agent worktree. + """ + + def _msg(self, mt, from_role="coder", phase="plan", metadata=None, body="b"): + return Message( + pipeline_id="test-pipeline", + from_role=from_role, + to_role="all", + message_type=mt, + subject=f"{mt} from {from_role}", + body=body, + metadata=metadata or {}, + phase=phase, + ) + + def _get(self, client, store, query): + with ( + patch("routes.messages.get_message_store", return_value=store), + patch("routes.messages.get_state_store_for_pipeline") as mock_pipeline, + ): + mock_pipeline.return_value = (MagicMock(), _make_pipeline_mock()) + resp = client.get(f"/api/v1/pipelines/test-pipeline/brc-transcript?{query}") + return resp, json.loads(resp.data) + + def test_phase_required(self, client, app): + store = MessageStore() + resp, data = self._get(client, store, "") + assert resp.status_code == 400 + assert "phase" in data["message"] + + def test_invalid_phase_rejected(self, client, app): + store = MessageStore() + resp, data = self._get(client, store, "phase=bogus") + assert resp.status_code == 400 + + def test_invalid_slice_id_rejected(self, client, app): + store = MessageStore() + resp, data = self._get(client, store, "phase=implement&slice_id=../etc") + assert resp.status_code == 400 + assert "slice_id" in data["message"] + + def test_returns_brc_records_for_requested_phase_only(self, client, app): + store = MessageStore() + store.add_message(self._msg("CONSENSUS_PROPOSE", phase="plan")) + store.add_message(self._msg("CONSENSUS_ACK", from_role="risk_analyst", phase="refine")) + store.add_message(self._msg("PROGRESS", phase="plan")) # non-BRC type + resp, data = self._get(client, store, "phase=plan") + assert resp.status_code == 200 + records = data["data"]["records"] + assert len(records) == 1 + assert records[0]["message_type"] == "CONSENSUS_PROPOSE" + assert records[0]["phase"] == "plan" + # Same serialized shape as the on-disk brc-history JSON + # (Message.to_dict()): the handler-side merge depends on it. + assert {"id", "from_role", "message_type", "timestamp"} <= set(records[0]) + + def test_mid_phase_propose_visible_immediately(self, client, app): + """The #3076 regression pin: a producer's CONSENSUS_PROPOSE is + served the moment it is in the store — no phase completion, no + file write, no spawn-fork dependency.""" + store = MessageStore() + store.add_message( + self._msg( + "CONSENSUS_PROPOSE", + from_role="architect", + metadata={"payload": {"version": 1, "commit_sha": "b521d7d3918"}}, + ) + ) + resp, data = self._get(client, store, "phase=plan") + assert data["data"]["count"] == 1 + assert data["data"]["records"][0]["from_role"] == "architect" + + def test_delphi_redaction_applies(self, client, app): + """The live route must not be a Delphi-blinding bypass: an + unreviewed reviewer sees the producer's PROPOSE redacted to + version + commit_sha, same as poll_messages.""" + from peer_consensus import PeerConsensusTracker + from review_graph import ReviewCriticality, ReviewEdge, ReviewGraph + + graph = ReviewGraph([ReviewEdge("reviewer_code", "coder", ReviewCriticality.CRITICAL)]) + tracker = PeerConsensusTracker("test-pipeline", graph, cooldown_seconds=0) + tracker.register_agent("coder") + tracker.register_agent("reviewer_code") + tracker.handle_propose( + "coder", + {"summary": "Implemented auth", "artifacts": ["src/auth.py"], "commit_sha": "abc1234"}, + ) + + store = MessageStore() + store.add_message( + self._msg( + "CONSENSUS_PROPOSE", + body="Detailed self-assessment", + metadata={ + "payload": { + "summary": "Implemented auth", + "artifacts": ["src/auth.py"], + "version": 1, + "commit_sha": "abc1234", + } + }, + ) + ) + with ( + patch("routes.messages.get_message_store", return_value=store), + patch("routes.messages.get_state_store_for_pipeline") as mock_pipeline, + patch("peer_consensus.get_peer_consensus_tracker", return_value=tracker), + ): + mock_pipeline.return_value = (MagicMock(), _make_pipeline_mock()) + resp = client.get( + "/api/v1/pipelines/test-pipeline/brc-transcript?phase=plan&role=reviewer_code" + ) + data = json.loads(resp.data) + rec = data["data"]["records"][0] + assert rec["body"] == "" + assert rec["metadata"]["delphi_redacted"] is True + assert "summary" not in rec["metadata"]["payload"] + assert rec["metadata"]["payload"]["commit_sha"] == "abc1234" + + def test_slice_attribution_mirrors_writer(self, client, app): + """Implement-phase slice scoping (#2548 parity): CONSENSUS_* + records need a matching canonical slice_id (missing one = writer + contract violation, dropped); non-consensus BRC types without + slice scope are the unattributed bucket.""" + store = MessageStore() + store.add_message( + self._msg("CONSENSUS_PROPOSE", phase="implement", metadata={"slice_id": "slice-1"}) + ) + store.add_message( + self._msg("CONSENSUS_PROPOSE", phase="implement", metadata={"slice_id": "slice-2"}) + ) + store.add_message(self._msg("CONSENSUS_ACK", phase="implement")) # no slice: dropped + store.add_message(self._msg("HEARTBEAT", phase="implement")) # unattributed + resp, data = self._get(client, store, "phase=implement&slice_id=slice-1") + types = [r["message_type"] for r in data["data"]["records"]] + slice_ids = [r["metadata"].get("slice_id") for r in data["data"]["records"]] + assert "HEARTBEAT" in types + assert slice_ids.count("slice-2") == 0 + assert types.count("CONSENSUS_PROPOSE") == 1 + assert types.count("CONSENSUS_ACK") == 0 + + resp, data = self._get( + client, store, "phase=implement&slice_id=slice-1&include_unattributed=false" + ) + types = [r["message_type"] for r in data["data"]["records"]] + assert "HEARTBEAT" not in types + + def test_no_slice_filter_returns_all_implement_records(self, client, app): + store = MessageStore() + store.add_message( + self._msg("CONSENSUS_PROPOSE", phase="implement", metadata={"slice_id": "slice-1"}) + ) + store.add_message(self._msg("HEARTBEAT", phase="implement")) + resp, data = self._get(client, store, "phase=implement") + assert data["data"]["count"] == 2 + + def test_limit_keeps_most_recent_in_order(self, client, app): + store = MessageStore() + for i in range(5): + store.add_message(self._msg("CONSENSUS_PROPOSE", body=f"v{i}")) + resp, data = self._get(client, store, "phase=plan&limit=2") + records = data["data"]["records"] + assert data["data"]["truncated"] is True + assert [r["body"] for r in records] == ["v3", "v4"] + + def test_invalid_limit_rejected(self, client, app): + store = MessageStore() + resp, data = self._get(client, store, "phase=plan&limit=abc") + assert resp.status_code == 400 + resp, data = self._get(client, store, "phase=plan&limit=0") + assert resp.status_code == 400 + + def test_unknown_pipeline_returns_404(self, client, app): + with ( + patch("routes.messages.get_message_store", return_value=MessageStore()), + patch( + "routes.messages.get_state_store_for_pipeline", + side_effect=PipelineNotFoundError("nope"), + ), + ): + resp = client.get("/api/v1/pipelines/missing/brc-transcript?phase=plan") + assert resp.status_code == 404 diff --git a/sandbox/egg_agent_tools/handlers/brc.py b/sandbox/egg_agent_tools/handlers/brc.py index a87a4e60dd..64fe9667b3 100644 --- a/sandbox/egg_agent_tools/handlers/brc.py +++ b/sandbox/egg_agent_tools/handlers/brc.py @@ -987,13 +987,65 @@ def _resolve_env_identifier_for_brc_history() -> str: ) +def _fetch_live_brc_transcript( + phase: str, + *, + slice_id: str | None, + include_unattributed: bool, +) -> tuple[list[dict[str, Any]], bool]: + """Query the orchestrator's live BRC transcript route (#3076). + + Returns ``(records, live_ok)``. ``live_ok=False`` means the route + was unreachable or returned an unusable shape (older orchestrator + without the route, transport failure) — the caller degrades to the + on-disk files and says so in the ``hint``. Never raises: the live + read is an enhancement over the disk read, not a new hard + dependency. + """ + pid = get_pipeline_id() + if not pid: + return [], False + params: dict[str, str] = {"phase": phase} + role = get_agent_role() + if role: + params["role"] = role + if slice_id: + params["slice_id"] = slice_id + if not include_unattributed: + params["include_unattributed"] = "false" + try: + result = orchestrator_request(f"/api/v1/pipelines/{pid}/brc-transcript?{urlencode(params)}") + except GatewayError as exc: + _logger.warning("live brc-transcript query failed: %s", exc) + return [], False + data = result.get("data") if isinstance(result, dict) else None + records = data.get("records") if isinstance(data, dict) else None + if not isinstance(records, list): + return [], False + return [r for r in records if isinstance(r, dict)], True + + def brc_read_peer_artifact(req: dict[str, Any]) -> dict[str, Any]: - """Read consensus history for a peer from the local brc-history log. + """Read the BRC transcript for a phase: live store + on-disk history. + + Two sources, merged (#3076 / #3077 phase 1): - No CLI counterpart (decision-8): reads from the local - ``.egg-state/brc-history/`` files written by - ``orchestrator.routes.pipelines._write_brc_history`` so reviewers - never have to hand-grep JSON off disk. + * **Live** — the orchestrator's ``/brc-transcript`` route, serving + the in-flight phase's CONSENSUS_*/BRC records straight from the + message store (which holds exactly the current phase; it is + cleared on phase transitions). This is what makes the tool + truthful mid-phase: a reviewer sees a producer's + CONSENSUS_PROPOSE as soon as it is sent. + * **Disk** — the local ``.egg-state/brc-history/`` files written by + ``orchestrator.routes.pipelines._write_brc_history`` at phase + COMPLETION; they cover phases that completed before this agent + spawned (and survive orchestrator restarts). + + Records are deduplicated by message ``id`` and sorted by timestamp. + A live-route failure degrades gracefully to disk-only with a + ``hint`` saying the live source was unavailable. + + No CLI counterpart (decision-8). File resolution mirrors the writer's per-slice partition (#2548): @@ -1040,14 +1092,20 @@ def brc_read_peer_artifact(req: dict[str, Any]) -> dict[str, Any]: Response: { ok: True, phase: str, items: [...], next_cursor: str|None, - total_available: int, skipped_malformed: int, hint?: str } + total_available: int, skipped_malformed: int, live: bool, + hint?: str } + + ``live`` reports whether the orchestrator's live transcript route + contributed records (``True``) or was unavailable and the result + is disk-only (``False``). - ``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 via the spawn fork point, so an empty result for - the current phase is structural, not evidence that peers have - not proposed. + ``hint`` is present only when both sources yielded zero records. + With a reachable live route that genuinely means no BRC messages + exist for the phase yet; without one, the hint explains the + structural emptiness (#3076): brc-history is written at phase + COMPLETION and reaches an agent worktree only via the spawn fork + point, so a disk-only empty result for the current phase is not + evidence that peers have not proposed. ``skipped_malformed`` counts brc-history records that were silently skipped because they failed isinstance-dict parsing; the @@ -1120,6 +1178,7 @@ def brc_read_peer_artifact(req: dict[str, Any]) -> dict[str, Any]: # cross-cutting `unattributed` sibling. Other phases and non-slice # implement runs read the aggregate file. history_files: list[Path] = [] + slice_id_env: str | None = None if phase == "implement": # Defense-in-depth via the public _gateway helper: resolves # EGG_SLICE_ID, validates against the canonical `^slice-$` @@ -1149,11 +1208,9 @@ def brc_read_peer_artifact(req: dict[str, Any]) -> dict[str, Any]: raise HandlerError("Resolved brc-history path escapes .egg-state/brc-history/") records: list[Any] = [] - any_existed = False for hf in history_files: if not hf.exists(): continue - any_existed = True try: chunk = json.loads(hf.read_text()) except (OSError, json.JSONDecodeError) as exc: @@ -1166,15 +1223,65 @@ def brc_read_peer_artifact(req: dict[str, Any]) -> dict[str, Any]: ) records.extend(chunk) - if not any_existed: - # No history file on disk. This is the EXPECTED state for the - # phase currently in flight (#3076): brc-history files are - # written by the orchestrator at phase COMPLETION, into the - # pipeline work branch — they reach an agent worktree only via - # the spawn fork point, i.e. only for phases that completed - # before this agent spawned. Say so explicitly: a bare empty - # result here reads as "peers produced nothing" and has driven - # reviewers to NACK proposals they simply could not see. + # Live source (#3076 / #3077 phase 1): the orchestrator's message + # store holds exactly the in-flight phase's records (it is cleared + # on phase transitions), which is precisely the window the on-disk + # files structurally cannot cover. Merge the two, dedup by message + # id (overlap is possible when a restarted phase's prior transcript + # was committed and reached this worktree at the spawn fork). + live_records, live_ok = _fetch_live_brc_transcript( + phase, + slice_id=slice_id_env, + include_unattributed=include_unattributed, + ) + # Disk records pass through untouched (the writer's partitions are + # disjoint by construction); only LIVE records are dedup'd against + # what disk already covers — the overlap case is a restarted + # phase whose prior transcript was committed and reached this + # worktree at the spawn fork. + disk_ids: set[str] = { + rec["id"] for rec in records if isinstance(rec, dict) and isinstance(rec.get("id"), str) + } + merged: list[Any] = list(records) + seen_live: set[str] = set() + for rec in live_records: + rid = rec.get("id") + if isinstance(rid, str) and rid: + if rid in disk_ids or rid in seen_live: + continue + seen_live.add(rid) + merged.append(rec) + + if not merged: + # Nothing on disk and nothing live. With a reachable live route + # this now genuinely means "no BRC messages for this phase yet" + # (e.g. no peer has proposed); without one, fall back to the + # structural explanation so an empty result is never read as + # "peers produced nothing" — that misreading drove reviewers to + # NACK proposals they simply could not see (#3076). + if live_ok: + hint = ( + f"No BRC messages recorded for phase {phase!r} yet " + "(live orchestrator store reachable, no on-disk " + "brc-history records). If you expected a peer proposal, " + "the peer has not proposed yet — wait for the " + "CONSENSUS_PROPOSE event rather than concluding the " + "work does not exist." + ) + else: + hint = ( + "No brc-history file exists in this worktree for phase " + f"{phase!r}, and the live orchestrator transcript route " + "was unavailable. brc-history is written at phase " + "COMPLETION; for the phase currently in flight an empty " + "result here is structural and is NOT evidence that " + "peers have not proposed. Live proposals arrive via " + "your event payload (pending_reviews carries each " + "producer's proposal_commit_sha and artifact_refs); " + "read a peer's proposed artifact with `git show " + ":` — the SHA resolves from " + "your worktree via the shared object store." + ) return { "ok": True, "phase": phase, @@ -1182,23 +1289,13 @@ def brc_read_peer_artifact(req: dict[str, Any]) -> dict[str, Any]: "next_cursor": None, "total_available": 0, "skipped_malformed": prior_skipped, - "hint": ( - "No brc-history file exists in this worktree for phase " - f"{phase!r}. brc-history is written at phase COMPLETION; " - "for the phase currently in flight this tool is always " - "empty and that is NOT evidence that peers have not " - "proposed. Live proposals arrive via your event payload " - "(pending_reviews carries each producer's " - "proposal_commit_sha and artifact_refs); read a peer's " - "proposed artifact with `git show " - ":` — the SHA resolves from " - "your worktree via the shared object store." - ), + "live": live_ok, + "hint": hint, } filtered: list[dict[str, Any]] = [] skipped_malformed = 0 - for rec in records: + for rec in merged: if not isinstance(rec, dict): skipped_malformed += 1 continue @@ -1208,11 +1305,10 @@ def brc_read_peer_artifact(req: dict[str, Any]) -> dict[str, Any]: continue filtered.append(rec) - # Re-sort merged records by timestamp so the per-slice transcript - # and the unattributed sibling interleave chronologically. Records + # Re-sort by timestamp so disk transcripts, the unattributed + # sibling, and live records interleave chronologically. Records # without a timestamp sort last, in original order (stable sort). - if len(history_files) > 1: - filtered.sort(key=lambda r: (r.get("timestamp") is None, r.get("timestamp") or "")) + filtered.sort(key=lambda r: (r.get("timestamp") is None, r.get("timestamp") or "")) total = len(filtered) total_skipped = prior_skipped + skipped_malformed @@ -1235,4 +1331,5 @@ def brc_read_peer_artifact(req: dict[str, Any]) -> dict[str, Any]: "next_cursor": next_cursor, "total_available": total, "skipped_malformed": total_skipped, + "live": live_ok, } diff --git a/sandbox/egg_agent_tools/tools/brc.py b/sandbox/egg_agent_tools/tools/brc.py index 5c74111307..ebecee57f1 100644 --- a/sandbox/egg_agent_tools/tools/brc.py +++ b/sandbox/egg_agent_tools/tools/brc.py @@ -407,15 +407,17 @@ async def brc_resolve_obligation(args: dict[str, Any]) -> dict[str, Any]: @tool( "read_peer_artifact", "Read the BRC consensus TRANSCRIPT (message records, not artifact " - "content) for a COMPLETED phase from the local " - "`.egg-state/brc-history/-.json` log. The log is " - "written by the orchestrator at phase completion and reaches your " - "worktree only at spawn — so for the phase currently in flight this " - "tool is always empty (#3076); that is NOT evidence peers have not " - "proposed. For live proposals use your event payload " - "(pending_reviews carries `proposal_commit_sha` + `artifact_refs`) " - "and read artifact content with `git show :`. Paginated " - "via `limit` + opaque `cursor`.", + "content) for a phase. Merges two sources: the orchestrator's LIVE " + "message store (the phase currently in flight — a peer's " + "CONSENSUS_PROPOSE is visible here as soon as it is sent) and the " + "local `.egg-state/brc-history/-.json` log " + "(phases completed before you spawned). The `live` response field " + "says whether the live source was reachable; when it was not, an " + "empty result for the in-flight phase is structural (#3076) and is " + "NOT evidence peers have not proposed. To read artifact CONTENT, " + "use your event payload (pending_reviews carries " + "`proposal_commit_sha` + `artifact_refs`) and `git show " + ":`. Paginated via `limit` + opaque `cursor`.", _READ_PEER_ARTIFACT_SCHEMA, ) async def brc_read_peer_artifact(args: dict[str, Any]) -> dict[str, Any]: diff --git a/tests/sandbox/egg_agent_tools/test_handlers_brc.py b/tests/sandbox/egg_agent_tools/test_handlers_brc.py index d9b96f346e..75934614f0 100644 --- a/tests/sandbox/egg_agent_tools/test_handlers_brc.py +++ b/tests/sandbox/egg_agent_tools/test_handlers_brc.py @@ -2066,3 +2066,161 @@ def test_stale_version_path_still_skips_memory(self, monkeypatch, tmp_path): "Stale-version rejection must not write a memory entry — " "the reviewer has not actually issued a verdict." ) + + +def _live_envelope(records): + """Shape returned by the orchestrator's /brc-transcript route.""" + return { + "success": True, + "message": "BRC transcript retrieved", + "data": {"records": records, "count": len(records), "phase": "plan", "truncated": False}, + } + + +class TestBrcReadPeerArtifactLive: + """Live message-store merge for read_peer_artifact (#3076 / #3077 + phase 1). + + The on-disk brc-history files are written at phase COMPLETION, so + the phase in flight is structurally invisible through them. The + handler now also queries the orchestrator's /brc-transcript route + (the live message store holds exactly the current phase) and merges + the two sources, dedup'd by message id. + """ + + def _set_env(self, monkeypatch, tmp_path, identifier="1917"): + monkeypatch.setenv("EGG_ISSUE_NUMBER", identifier) + monkeypatch.setenv("EGG_REPO_PATH", str(tmp_path)) + # Live queries key off the pipeline id; set it alongside the + # issue number (the issue number wins for FILE naming only). + monkeypatch.setenv("EGG_PIPELINE_ID", "pipeline-2b3d8b0b") + monkeypatch.setenv("EGG_AGENT_ROLE", "risk_analyst") + monkeypatch.delenv("EGG_SLICE_ID", raising=False) + + def test_live_records_returned_without_any_disk_file(self, tmp_path, monkeypatch): + """The #3076 incident shape: producers proposed mid-phase, no + brc-history file exists anywhere — the records must now come + back from the live store, with no misleading hint.""" + self._set_env(monkeypatch, tmp_path) + live = _records(("architect", "CONSENSUS_PROPOSE")) + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + return_value=_live_envelope(live), + ): + resp = brc.brc_read_peer_artifact({"phase": "plan"}) + assert resp["ok"] is True + assert resp["live"] is True + assert [r["from_role"] for r in resp["items"]] == ["architect"] + assert "hint" not in resp + + def test_live_and_disk_merge_dedup_by_id(self, tmp_path, monkeypatch): + self._set_env(monkeypatch, tmp_path) + disk = _records(("coder", "CONSENSUS_PROPOSE"), ("tester", "CONSENSUS_ACK")) + _make_history_file(tmp_path, "1917", "plan", disk) + # Live store holds one duplicate (same id) and one new record. + live = [disk[1], dict(disk[1], id="id-99", from_role="architect")] + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + return_value=_live_envelope(live), + ): + resp = brc.brc_read_peer_artifact({"phase": "plan"}) + ids = [r["id"] for r in resp["items"]] + assert sorted(ids) == ["id-1", "id-2", "id-99"] + assert resp["total_available"] == 3 + assert resp["live"] is True + + def test_live_failure_degrades_to_disk(self, tmp_path, monkeypatch): + self._set_env(monkeypatch, tmp_path) + _make_history_file(tmp_path, "1917", "plan", _records(("coder", "CONSENSUS_PROPOSE"))) + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + side_effect=GatewayError("connection refused"), + ): + resp = brc.brc_read_peer_artifact({"phase": "plan"}) + assert len(resp["items"]) == 1 + assert resp["live"] is False + assert "hint" not in resp + + def test_empty_with_live_reachable_says_not_proposed_yet(self, tmp_path, monkeypatch): + """With a reachable live route, emptiness is a real answer — + the hint must say "not proposed yet", NOT the structural + explanation (which would wrongly suggest the channel is dead).""" + self._set_env(monkeypatch, tmp_path) + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + return_value=_live_envelope([]), + ): + resp = brc.brc_read_peer_artifact({"phase": "plan"}) + assert resp["items"] == [] + assert resp["live"] is True + assert "not proposed yet" in resp["hint"] + assert "phase COMPLETION" not in resp["hint"] + + def test_empty_with_live_down_keeps_structural_hint(self, tmp_path, monkeypatch): + self._set_env(monkeypatch, tmp_path) + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + side_effect=GatewayError("boom"), + ): + resp = brc.brc_read_peer_artifact({"phase": "plan"}) + assert resp["items"] == [] + assert resp["live"] is False + hint = resp["hint"] + assert "phase COMPLETION" in hint + assert "NOT evidence" in hint + assert "git show" in hint + assert "unavailable" in hint + + def test_filters_apply_to_live_records(self, tmp_path, monkeypatch): + self._set_env(monkeypatch, tmp_path) + live = _records( + ("architect", "CONSENSUS_PROPOSE"), + ("risk_analyst", "CONSENSUS_NACK"), + ) + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + return_value=_live_envelope(live), + ): + resp = brc.brc_read_peer_artifact({"phase": "plan", "peer_role": "architect"}) + assert [r["from_role"] for r in resp["items"]] == ["architect"] + + def test_live_query_carries_phase_role_and_slice(self, tmp_path, monkeypatch): + """The query must scope to the requested phase, pass the + caller's role (Delphi filtering server-side) and the env slice + id for implement-phase reads.""" + self._set_env(monkeypatch, tmp_path) + monkeypatch.setenv("EGG_SLICE_ID", "slice-2") + captured = {} + + def _capture(endpoint, **kw): + captured["endpoint"] = endpoint + return _live_envelope([]) + + with patch("egg_agent_tools.handlers.brc.orchestrator_request", side_effect=_capture): + brc.brc_read_peer_artifact({"phase": "implement"}) + ep = captured["endpoint"] + assert ep.startswith("/api/v1/pipelines/pipeline-2b3d8b0b/brc-transcript?") + assert "phase=implement" in ep + assert "role=risk_analyst" in ep + assert "slice_id=slice-2" in ep + + def test_unusable_live_shape_treated_as_unavailable(self, tmp_path, monkeypatch): + self._set_env(monkeypatch, tmp_path) + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + return_value={"success": True, "data": {"records": "bogus"}}, + ): + resp = brc.brc_read_peer_artifact({"phase": "plan"}) + assert resp["live"] is False + + def test_no_pipeline_id_skips_live_query(self, tmp_path, monkeypatch): + """Without EGG_PIPELINE_ID the handler must not attempt HTTP at + all (spawn contexts that only carry EGG_ISSUE_NUMBER).""" + self._set_env(monkeypatch, tmp_path) + monkeypatch.delenv("EGG_PIPELINE_ID") + with patch( + "egg_agent_tools.handlers.brc.orchestrator_request", + side_effect=AssertionError("must not be called"), + ): + resp = brc.brc_read_peer_artifact({"phase": "plan"}) + assert resp["live"] is False