Skip to content
Merged
Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
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
7 changes: 5 additions & 2 deletions docs/guides/concurrent-execution.md
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,10 @@ Relevant `PipelineConfig` fields:
| `start_phase` | `null` | Skip earlier phases and begin execution from `"plan"` or `"implement"` |
| `max_concurrent_agents` | `6` | Maximum agents per phase |
| `message_poll_hint_seconds` | `30` | Suggested polling interval for agents |
| `consensus_timeout_minutes` | `30` | Consensus timeout before escalation or auto-advance |
| `consensus_timeout_minutes` | `null` | Legacy global consensus timeout in minutes. When set, applies to every phase and overrides the phase-aware defaults below. When `null`, each phase falls back to its calibrated default. |
| `consensus_timeout_minutes_refine` | `null` (effective `30`) | Per-phase override for refine. Wins over the legacy global. |
| `consensus_timeout_minutes_plan` | `null` (effective `60`) | Per-phase override for plan. Wins over the legacy global. |
| `consensus_timeout_minutes_implement` | `null` (effective `90`) | Per-phase override for implement. Wins over the legacy global. |
| `brc_consensus_progress_gate_seconds` | `300` | Defer the consensus-timeout HITL decision while BRC bus activity (proposals, ACKs/NACKs) or container heartbeats have fired within this window. Set to `0` to disable. |
| `agent_idle_timeout_minutes` | `60` | Idle agent timeout before termination |

Expand Down Expand Up @@ -720,7 +723,7 @@ If any agent is in the `OBJECTING` readiness state (separate from BRC phase), th

### Timeout Handling

If consensus is not reached within `consensus_timeout_minutes`, the orchestrator first checks the **BRC progress gate** before opening a HITL decision. While any of the following have fired within `brc_consensus_progress_gate_seconds` (default 300 s), the orchestrator continues polling rather than escalating immediately:
If consensus is not reached within the resolved per-phase timeout (per-phase override > legacy global > calibrated default — refine 30, plan 60, implement 90; see issue #2263), the orchestrator first checks the **BRC progress gate** before opening a HITL decision. While any of the following have fired within `brc_consensus_progress_gate_seconds` (default 300 s), the orchestrator continues polling rather than escalating immediately:

- A `CONSENSUS_PROPOSE` or ACK/NACK on the BRC bus
- A container heartbeat from any active role in the current phase
Expand Down
15 changes: 10 additions & 5 deletions docs/guides/sdlc-pipeline.md
Original file line number Diff line number Diff line change
Expand Up @@ -1249,7 +1249,10 @@ Or pass it in the pipeline config JSON (e.g. via the API):
| `start_phase` | str | `null` | Skip earlier phases and start execution from `"plan"` or `"implement"`. When set to `"implement"`, pass top-level `analysis`/`plan` fields to seed the contract (see Short-flow pipelines above). |
| `max_concurrent_agents` | int | `6` | Maximum agents running simultaneously |
| `message_poll_hint_seconds` | int | `30` | Suggested polling interval for agents |
| `consensus_timeout_minutes` | int | `30` | Timeout before HITL escalation |
| `consensus_timeout_minutes` | int \| null | `null` | Global consensus timeout. When set, applies to every phase. When `null` (the default), each phase falls back to the calibrated per-phase default below. |
| `consensus_timeout_minutes_refine` | int \| null | `null` (effective `30`) | Per-phase consensus timeout for refine. Wins over the legacy global. |
| `consensus_timeout_minutes_plan` | int \| null | `null` (effective `60`) | Per-phase consensus timeout for plan. Wins over the legacy global. |
| `consensus_timeout_minutes_implement` | int \| null | `null` (effective `90`) | Per-phase consensus timeout for implement. Wins over the legacy global. |
| `brc_consensus_progress_gate_seconds` | int | `300` | Defer consensus-timeout HITL while BRC bus or container heartbeats are active. Set to `0` to disable. |
| `agent_idle_timeout_minutes` | int | `60` | Agent idle timeout |
| `overseer_enabled` | bool | `true` | Enable the overseer agent for pipeline health monitoring |
Expand Down Expand Up @@ -1316,7 +1319,7 @@ Phase completion in concurrent mode uses a consensus-based approach:
- The orchestrator polls every 5 seconds and stops containers immediately on consensus
4. Any agent can object (signal `OBJECTING`) to block completion
- A HITL decision is created with options: **Override objections**, **Wait for resolution**, **Abort phase**
5. Timeout (`consensus_timeout_minutes`, default 30) triggers HITL escalation
5. Timeout (per-phase: refine 30 / plan 60 / implement 90 by default; configurable via `consensus_timeout_minutes_<phase>` or the legacy global `consensus_timeout_minutes`) triggers HITL escalation
- Options: **Continue waiting**, **Accept current state**, **Abort phase**
- Phase falls back to exit-code-based completion while awaiting the decision
6. If a container exits cleanly without signaling `READY`, the consensus wrapper restarts it with a recovery prompt (up to `MAX_CONSENSUS_RESTARTS`, default 2). After exhausting restarts, the wrapper performs a final consensus check — if consensus has already been reached (`is_complete=True`), it exits with code 0 (success). Only if consensus is genuinely incomplete does it exit with code 1, triggering the single-agent failure path (HITL decision: retry, abort, or continue without). See [Concurrent Execution: Consensus Wrapper](concurrent-execution.md#consensus-wrapper).
Expand Down Expand Up @@ -1443,9 +1446,11 @@ egg-checkpoint show ckpt-<id>
role. Messages are filtered by `to_role` — only targeted messages and broadcasts
(`to_role: "all"`) are returned.

**Consensus timeout**: If agents don't reach consensus within `consensus_timeout_minutes`,
a HITL decision is created. Check agent states via `egg-orch pipeline status` to
identify blocked or stuck agents.
**Consensus timeout**: If agents don't reach consensus within the resolved per-phase
budget (`consensus_timeout_minutes_<phase>` if set, else the legacy global
`consensus_timeout_minutes`, else the calibrated default — refine 30 / plan 60 /
implement 90), a HITL decision is created. Check agent states via
`egg-orch pipeline status` to identify blocked or stuck agents.

**Message bus empty**: Verify the pipeline has `concurrent_execution: true` in its
config. The message bus is only active for concurrent pipelines.
Expand Down
64 changes: 62 additions & 2 deletions orchestrator/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,34 @@
from egg_contracts.models import PipelinePhase
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

# Phase-aware fallback defaults for consensus timeout. Calibrated against
# producer/reviewer fan-out and iteration profile per phase — see #2263.
PHASE_CONSENSUS_TIMEOUT_DEFAULTS_MIN: dict[str, int] = {
"refine": 30,
"plan": 60,
"implement": 90,
}


def resolve_consensus_timeout_minutes(config: "PipelineConfig", phase: str) -> int:
"""Resolve the consensus timeout (minutes) for *phase*.

Resolution order, highest priority first:

1. The phase-specific override field (``consensus_timeout_minutes_<phase>``).
2. The legacy global field (``consensus_timeout_minutes``), if explicitly
set — preserves the AC clause that pipelines passing only the global
continue to behave identically across all three phases.
3. The phase-aware default from :data:`PHASE_CONSENSUS_TIMEOUT_DEFAULTS_MIN`,
falling back to 30 for any unknown phase string.
"""
override = getattr(config, f"consensus_timeout_minutes_{phase}", None)
if override is not None:
return override
if config.consensus_timeout_minutes is not None:
return config.consensus_timeout_minutes
return PHASE_CONSENSUS_TIMEOUT_DEFAULTS_MIN.get(phase, 30)


class PipelineStatus(StrEnum):
"""Overall status of a pipeline."""
Expand Down Expand Up @@ -374,8 +402,40 @@ class PipelineConfig(BaseModel):
message_poll_hint_seconds: int = Field(
default=30, ge=1, description="Suggested message polling interval for agents"
)
consensus_timeout_minutes: int = Field(
default=30, ge=1, description="Timeout for consensus before HITL escalation"
consensus_timeout_minutes: int | None = Field(
default=None,
ge=1,
description=(
"Global consensus timeout in minutes before HITL escalation. When set, "
"applies to every phase and overrides phase-aware defaults. When None "
"(the default), each phase uses its calibrated default from "
"PHASE_CONSENSUS_TIMEOUT_DEFAULTS_MIN unless a per-phase field below "
"is set. (#2263)"
),
)
consensus_timeout_minutes_refine: int | None = Field(
default=None,
ge=1,
description=(
"Per-phase consensus timeout for refine. When set, wins over the legacy "
"global and the phase-aware default. (#2263)"
),
)
consensus_timeout_minutes_plan: int | None = Field(
default=None,
ge=1,
description=(
"Per-phase consensus timeout for plan. When set, wins over the legacy "
"global and the phase-aware default. (#2263)"
),
)
consensus_timeout_minutes_implement: int | None = Field(
default=None,
ge=1,
description=(
"Per-phase consensus timeout for implement. When set, wins over the "
"legacy global and the phase-aware default. (#2263)"
),
)
post_consensus_iteration_budget_seconds: int = Field(
default=3600,
Expand Down
3 changes: 2 additions & 1 deletion orchestrator/routes/pipelines.py
Original file line number Diff line number Diff line change
Expand Up @@ -10373,6 +10373,7 @@ def _run_concurrent_phase(
ContainerInfo,
ContainerStatus,
PipelinePhase,
resolve_consensus_timeout_minutes,
)

try:
Expand Down Expand Up @@ -10697,7 +10698,7 @@ def _run_concurrent_phase(
_logs_lock = threading.Lock()

poll_interval = 5 # seconds
raw_timeout = getattr(pipeline.config, "consensus_timeout_minutes", 30)
raw_timeout = resolve_consensus_timeout_minutes(pipeline.config, phase_str)
consensus_timeout = max(raw_timeout, 1) * 60 # minimum 1 minute
start_time = time.monotonic()
objection_decision_created = False
Expand Down
54 changes: 54 additions & 0 deletions orchestrator/tests/test_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from datetime import UTC, datetime

from models import (
PHASE_CONSENSUS_TIMEOUT_DEFAULTS_MIN,
AgentExecution,
AgentExecutionStatus,
AgentExitInfo,
Expand All @@ -18,6 +19,7 @@
PipelineConfig,
PipelinePhase,
PipelineStatus,
resolve_consensus_timeout_minutes,
)


Expand Down Expand Up @@ -410,6 +412,58 @@ def test_custom_config(self):
assert config.max_review_cycles == 5


class TestResolveConsensusTimeoutMinutes:
"""Tests for resolve_consensus_timeout_minutes (issue #2263).

The resolver picks a per-phase timeout in this order:
1. Per-phase override field if explicitly set.
2. Legacy global ``consensus_timeout_minutes`` if explicitly set
(preserves the back-compat clause: pipelines that pass only the
global behave identically across all three phases).
3. Phase-aware default from PHASE_CONSENSUS_TIMEOUT_DEFAULTS_MIN.
"""

def test_phase_aware_defaults_when_nothing_set(self):
config = PipelineConfig()
assert resolve_consensus_timeout_minutes(config, "refine") == 30
assert resolve_consensus_timeout_minutes(config, "plan") == 60
assert resolve_consensus_timeout_minutes(config, "implement") == 90

def test_phase_defaults_match_constant(self):
config = PipelineConfig()
for phase, expected in PHASE_CONSENSUS_TIMEOUT_DEFAULTS_MIN.items():
assert resolve_consensus_timeout_minutes(config, phase) == expected

def test_legacy_global_applies_to_all_phases(self):
config = PipelineConfig(consensus_timeout_minutes=45)
assert resolve_consensus_timeout_minutes(config, "refine") == 45
assert resolve_consensus_timeout_minutes(config, "plan") == 45
assert resolve_consensus_timeout_minutes(config, "implement") == 45

def test_per_phase_override_wins_over_legacy_global(self):
config = PipelineConfig(
consensus_timeout_minutes=45,
consensus_timeout_minutes_implement=120,
)
assert resolve_consensus_timeout_minutes(config, "refine") == 45
assert resolve_consensus_timeout_minutes(config, "plan") == 45
assert resolve_consensus_timeout_minutes(config, "implement") == 120

def test_per_phase_override_alone_uses_phase_defaults_for_others(self):
config = PipelineConfig(consensus_timeout_minutes_plan=15)
assert resolve_consensus_timeout_minutes(config, "refine") == 30
assert resolve_consensus_timeout_minutes(config, "plan") == 15
assert resolve_consensus_timeout_minutes(config, "implement") == 90

def test_unknown_phase_falls_back_to_30(self):
config = PipelineConfig()
assert resolve_consensus_timeout_minutes(config, "totally-unknown") == 30

def test_unknown_phase_still_honors_legacy_global(self):
config = PipelineConfig(consensus_timeout_minutes=45)
assert resolve_consensus_timeout_minutes(config, "totally-unknown") == 45


class TestStartPhaseValidator:
"""Tests for PipelineConfig.start_phase validation."""

Expand Down
Loading