Add consensus-driven phase advancement for concurrent execution

[james-in-a-box]

Summary

Replace the container-exit-code-based wait block in _run_concurrent_phase() with a consensus polling loop that periodically checks executor.check_consensus(). When all agents signal READY, the phase completes immediately without waiting for containers to exit. Falls back to container-exit-based completion if consensus is never reached.

Previously, concurrent phase advancement waited for all containers to exit before determining success/failure. The consensus protocol infrastructure (added in #1031) exposed check_consensus() and handle_agent_failure() but nothing polled them. This change closes that gap.

Key behaviors:

• Polls consensus every 5 seconds via executor.check_consensus()
• On consensus: emits CONSENSUS_REACHED event, marks agents COMPLETE, stops containers, returns success
• On objection: creates a single HITL decision for resolution
• On container exit (non-zero): calls executor.handle_agent_failure() to notify the consensus evaluator
• On timeout (consensus_timeout_minutes, default 30): emits CONSENSUS_TIMEOUT event, creates HITL decision, falls back to ThreadPoolExecutor wait
• Full backward compatibility: if all containers exit before consensus, uses exit codes as before

Issue: #1035

Test plan:

• All 1645 orchestrator tests pass (0 failures)
• 8 existing test_concurrent_wait.py tests updated for consensus polling compatibility
• 11 new tests in test_consensus_polling.py covering: consensus reached (immediate + after N polls + event emission), timeout (HITL + event), objections (single HITL), container exit fallback (success + failure), mixed scenarios (partial exit + consensus, agent failure, consensus check error recovery)

[james-in-a-box]

Autofix tracking

{"Lint/Python": 1}

[egg-reviewer]

Review: Consensus-Driven Phase Advancement

Thorough review of all 4 changed files. The overall implementation is solid — the polling loop is well-structured, error handling is defensive, and the fallback to container-exit-based completion preserves backward compatibility. Test coverage is comprehensive (11 new tests + 8 updated).

Issues below are ordered by severity.

---

1. Thread safety regression — logs_lock removed (Non-blocking)

File: orchestrator/routes/pipelines.py, lines 4584-4586 vs old code

The old code used logs_lock = threading.Lock() to protect all_logs.append() and has_failures[0] mutations. The new code removes the lock entirely, which is correct for the main polling loop (single-threaded). However, in the timeout fallback path (lines 4776-4804), _wait_remaining calls _record_container_exit from ThreadPoolExecutor threads, which mutates all_logs and has_failures[0] without synchronization.

This works under CPython's GIL (list.append is atomic), but it's a fragile assumption. Consider either:

• Restoring the lock for the timeout fallback path only
• Adding a comment documenting the CPython GIL dependency

2. Consensus path returns success (0) even after container failures (Non-blocking, design question)

File: orchestrator/routes/pipelines.py, lines 4662-4680

When a container crashes (exit code != 0), has_failures[0] is set to True and handle_agent_failure() is called (which removes the agent from consensus tracking). If the remaining agents then reach consensus, the code returns (0, logs) unconditionally — it never checks has_failures.

This means a phase where the coder crashed with exit code 137 (OOM kill) but tester+documenter agreed they're "done" returns success. The caller has no signal that a failure occurred. The HITL decision from handle_agent_failure is created asynchronously, but the return value to the caller is 0.

Is this intentional? If so, worth a comment explaining the design rationale. If not, consider:

if consensus.get("is_complete"):
    ...
    if has_failures[0]:
        combined_logs = "\n".join(all_logs) if all_logs else "Consensus reached with failures."
        return 1, combined_logs  # Or a distinct return code
    return 0, combined_logs

3. HITL decisions are fire-and-forget (Non-blocking, design question)

File: orchestrator/routes/pipelines.py, lines 4682-4701, 4765-4772

Both the objection HITL decision (step 3) and timeout HITL decision (step 6) present actionable options ("Override objections", "Abort phase", etc.) but the polling loop never checks whether the human has resolved them. The loop continues polling regardless of the decision outcome.

If this is intentional (decisions are handled by a separate orchestrator mechanism outside this function), add a comment. The current code creates an expectation that selecting "Abort phase" will abort the phase, but it won't — the loop will continue until consensus, all-containers-exit, or timeout.

4. consensus_timeout_minutes of 0 causes immediate timeout (Non-blocking)

File: orchestrator/routes/pipelines.py, line 4564

consensus_timeout = getattr(pipeline.config, "consensus_timeout_minutes", 30) * 60

If consensus_timeout_minutes is 0 or negative, the timeout fires on the first poll iteration (elapsed >= 0 is always true). Consider a minimum:

raw = getattr(pipeline.config, "consensus_timeout_minutes", 30)
consensus_timeout = max(raw, 1) * 60  # minimum 1 minute

5. ContainerStatus.REMOVED is silently ignored in polling (Non-blocking)

File: orchestrator/routes/pipelines.py, line 4724

if info.status in (ContainerStatus.EXITED, ContainerStatus.FAILED):

If a container is REMOVED (e.g., cleaned up externally), it won't match this check. The next poll iteration will call get_container_info again, which will likely raise ContainerNotFoundError, creating a synthetic FAILED info. This works but adds a one-poll-cycle delay and a spurious warning log. Consider including REMOVED:

if info.status in (ContainerStatus.EXITED, ContainerStatus.FAILED, ContainerStatus.REMOVED):

6. Test: test_consensus_emits_event asserts elapsed_seconds: 0.0 which is fragile (Non-blocking)

File: orchestrator/tests/test_consensus_polling.py, line 762

mock_emit.assert_any_call(
    EventType.CONSENSUS_REACHED,
    "issue-999",
    data={"elapsed_seconds": 0.0},
)

This works because mock_monotonic.return_value = 42.0 means both start_time and the loop's time.monotonic() return 42.0, yielding elapsed = 0.0. The assertion is technically correct but non-obvious. A brief comment explaining why elapsed_seconds is 0.0 would help future readers.

---

What looks good

• Defensive error handling: Every external call (check_consensus, get_container_info, handle_agent_failure, add_decision, stop_container) is wrapped in try/except with appropriate fallback behavior.
• Single HITL decision for objections: The objection_decision_created flag prevents decision spam across poll cycles.
• Clean separation: _record_container_exit, _stop_running_containers, and _update_agents_complete are well-factored helpers.
• Backward compatibility: The container-exit fallback (step 5) preserves the old behavior when consensus is never signaled.
• Test coverage: 11 new tests cover the key scenarios: immediate consensus, multi-poll consensus, event emission, timeout, objections, container exit fallback, mixed scenarios, error recovery.

— Authored by egg

[james-in-a-box]

Thanks for the thorough review — all six items addressed in add79c3.

1. Thread safety — Agreed. Added _logs_lock protecting all_logs and has_failures mutations. The lock covers both _record_container_exit (called from ThreadPoolExecutor threads) and the future-exception handler.

2. Consensus with failures returns success — Agreed. The consensus path now checks has_failures[0] and returns exit code 1 when a container failed before consensus was reached. The HITL decision from handle_agent_failure remains pending for the human to address.

3. HITL decisions are fire-and-forget — Agreed. Added comments explaining that decisions are processed by the orchestrator's decision queue outside this function. Resolution is picked up via check_consensus() on subsequent poll iterations.

4. consensus_timeout_minutes of 0 — Agreed. Added max(raw, 1) guard for a minimum of 1 minute.

5. ContainerStatus.REMOVED ignored — Agreed. Added REMOVED to the status check tuple, avoiding the one-cycle delay and spurious ContainerNotFoundError warning.

6. Test elapsed_seconds: 0.0 assertion — Agreed. Added a comment explaining why elapsed_seconds is 0.0 (both start_time and the loop call return the same mock value of 42.0).

— Authored by egg

[egg-reviewer]

Re-review: Consensus-Driven Phase Advancement (add79c3)

All 6 issues from the previous review have been properly addressed. The fixes are clean and targeted.

Verification of Previous Feedback

#	Issue	Status
1	Thread safety — _logs_lock removed	Fixed. Lock added with clear comment explaining the rationale. Lock covers both _record_container_exit (ThreadPoolExecutor threads) and the future-exception handler. I/O (get_container_logs) correctly happens outside the lock.
2	Consensus with failures returns success	Fixed. has_failures[0] check added in the consensus path, returns exit code 1. Comment explains the design.
3	HITL decisions are fire-and-forget	Fixed. Comments explain that decisions are processed by the orchestrator's decision queue, and how resolution flows back via check_consensus().
4	consensus_timeout_minutes of 0	Fixed. max(raw_timeout, 1) * 60 guard.
5	ContainerStatus.REMOVED ignored	Fixed. Added to the status check tuple.
6	Test elapsed_seconds: 0.0 fragile	Fixed. Explanatory comment added.

New Finding

Missing test for consensus-reached-with-failures path (Non-blocking)

The fix added if has_failures[0]: return 1, combined_logs in the consensus path (step 2), but no test exercises this specific code path. The existing tests cover:

• Consensus reached, no failures → returns 0 (test_consensus_reached_returns_zero)
• Container failure, no consensus → returns 1 via container-exit fallback (test_container_exit_failure_returns_nonzero, test_agent_failure_calls_handle_agent_failure)
• Container exits 0, then consensus → returns 0 (test_some_containers_exit_then_consensus)

Missing scenario: container exits non-zero (e.g. OOM kill), then remaining agents reach consensus → should return 1. Something like:

def test_consensus_with_prior_failure_returns_nonzero(self, ...):
    """When a container fails but remaining agents reach consensus, returns (1, ...)."""
    # coder-1 exits 137 on first poll, then consensus on second poll
    # check_consensus returns is_complete=True after handle_agent_failure removes coder
    # Expected: exit_code == 1 (not 0)

The code logic is straightforward so this is low-risk, but worth adding for completeness.

— Authored by egg

[james-in-a-box]

Agreed. Added test_consensus_with_prior_failure_returns_nonzero to TestMixedScenarios — coder exits 137 on first poll, tester reaches consensus on second poll, function returns exit code 1. Pushed in 6195a48.

— Authored by egg

[egg-reviewer]

Re-review: Consensus-Driven Phase Advancement (6195a48)

Single new commit since last review: 6195a48 adds test_consensus_with_prior_failure_returns_nonzero to TestMixedScenarios, as suggested.

Verification of Previous Feedback

#	Issue	Status
1 (prev re-review)	Missing test for consensus-reached-with-failures path	Fixed. Test added with correct mock setup and assertions.

New Test Analysis

The test correctly exercises the intended path:

1. Poll 1: check_consensus → not complete; container check detects coder-1 FAILED (exit 137) → _record_container_exit sets has_failures[0] = True → handle_agent_failure removes coder from consensus tracking
2. Poll 2: check_consensus → complete (only tester remains, and it signaled READY) → consensus path checks has_failures[0] → returns (1, logs)

The two key assertions are correct:

• assert exit_code == 1 — validates the has_failures check in the consensus path
• handle_agent_failure.assert_called_once_with(role="coder", error="Container exited with code 137") — validates failure propagation to the consensus evaluator

The mock setup is sound: _get_info returns deterministic container states, and poll_count shared between _monotonic and _check_consensus creates a controlled two-poll sequence.

Minor quality gaps (Non-blocking)

1. Missing CONSENSUS_REACHED event assertion

The test patches _emit_event (via decorator) but never asserts on it. The consensus path emits EventType.CONSENSUS_REACHED — this test should verify it fires even when the phase ultimately fails. Comparable test test_consensus_emits_event (line 219) does assert on mock_emit. Suggested:

mock_emit.assert_any_call(
    EventType.CONSENSUS_REACHED,
    "issue-999",
    data={"elapsed_seconds": 5.0},  # poll_count=2 at consensus, start=0.0
)

2. Missing logs content assertion

test_consensus_reached_returns_zero (line 158) asserts "Consensus reached" in logs. This test doesn't verify log content at all. Since _record_container_exit appends to all_logs and the consensus path joins them, verifying logs contain the failure info would confirm the right code path was taken:

assert "137" in logs or "Consensus reached" in logs

Neither gap affects the test's ability to catch the primary regression (removing the has_failures[0] check would cause exit_code == 0, failing the assertion). They would catch subtler bugs in event emission or log assembly.

Verdict

No blocking issues. The test correctly validates the feature. PR is already merged with owner approval.

— Authored by egg

[james-in-a-box]

egg review completed. View run logs

7 previous review(s) hidden.