diff --git a/docs/guides/sdlc-pipeline.md b/docs/guides/sdlc-pipeline.md index f9e2bb1380..72cb6b1b25 100644 --- a/docs/guides/sdlc-pipeline.md +++ b/docs/guides/sdlc-pipeline.md @@ -1275,7 +1275,20 @@ message bus hosted by the orchestrator. ### Configuration -Enable concurrent execution in the pipeline config: +Enable concurrent execution with the `--concurrent` CLI flag: + +```bash +# Issue mode +egg-sdlc -r egg -i 999 --concurrent + +# Local/prompt mode +egg-sdlc -r egg -p "Add feature X" --concurrent + +# Via egg-orch directly +egg-orch pipeline create --repo owner/repo --issue 999 --branch egg/issue-999 --concurrent +``` + +Or pass it in the pipeline config JSON (e.g. via the API): ```json { diff --git a/sandbox/bin/egg-pipeline-watch b/sandbox/bin/egg-pipeline-watch index b51fee9b3c..825d34ca88 100755 --- a/sandbox/bin/egg-pipeline-watch +++ b/sandbox/bin/egg-pipeline-watch @@ -113,6 +113,20 @@ def _write(text: str, file=None) -> None: file.flush() +_STATE_ICONS = { + "READY": GREEN + "R", + "WORKING": CYAN + "W", + "BLOCKED": YELLOW + "B", + "OBJECTING": RED + "X", +} + + +def _agent_icon(info: object) -> str: + """Return a single colored icon character for an agent's readiness state.""" + state = info.get("state", "WORKING") if isinstance(info, dict) else str(info) + return f"{_STATE_ICONS.get(state, DIM + '?')}{RESET}" + + def format_status_line(pipeline_id: str, data: dict) -> str: """Format a single-line status display.""" status = data.get("status", "unknown") @@ -130,6 +144,15 @@ def format_status_line(pipeline_id: str, data: dict) -> str: line = f"{BOLD}{pipeline_id}{RESET} {color}{display_status}{RESET} phase={phase}" if pending: line += f" {YELLOW}({pending} pending decision{'s' if pending != 1 else ''}){RESET}" + + # Compact concurrent agent states + concurrent = data.get("concurrent") + if concurrent: + agents = concurrent.get("consensus", {}).get("agents", {}) + if agents: + parts = [_agent_icon(info) for _, info in sorted(agents.items())] + line += f" agents=[{''.join(parts)}]" + if time_part: line += f" {DIM}{time_part}{RESET}" @@ -146,6 +169,47 @@ def render_header(pipeline_id: str, event_type: str | None = None) -> str: return "\n".join(lines) +def _render_concurrent_info(concurrent: dict) -> str: + """Render concurrent execution status (agent readiness, messages, consensus).""" + lines = [] + + # Agent readiness states + consensus = concurrent.get("consensus", {}) + agents = consensus.get("agents", {}) + if agents: + state_colors = { + "READY": GREEN, + "WORKING": CYAN, + "BLOCKED": YELLOW, + "OBJECTING": RED, + } + parts = [] + for role, info in sorted(agents.items()): + state = info.get("state", "WORKING") if isinstance(info, dict) else str(info) + color = state_colors.get(state, DIM) + parts.append(f"{color}{role}:{state}{RESET}") + lines.append(f" Agents: {' | '.join(parts)}") + + # Consensus status + is_complete = consensus.get("is_complete", False) + blocking = consensus.get("blocking_agents", []) + if is_complete: + lines.append(f" Consensus: {GREEN}reached{RESET}") + elif blocking: + lines.append(f" Consensus: waiting on {YELLOW}{', '.join(blocking)}{RESET}") + + # Message bus stats + messages = concurrent.get("messages", {}) + total = messages.get("total", 0) + if total: + by_type = messages.get("by_type", {}) + type_parts = [f"{t}={c}" for t, c in sorted(by_type.items())] + type_str = f" ({', '.join(type_parts)})" if type_parts else "" + lines.append(f" Messages: {total}{type_str}") + + return "\n".join(lines) + + def render_event_info(data: dict) -> str: """Render event metadata below the DAG.""" lines = [] @@ -164,6 +228,12 @@ def render_event_info(data: dict) -> str: time_part = _utc_to_local_time(timestamp) lines.append(f" {DIM}Updated: {time_part} Event: {event_type}{RESET}") + # Concurrent execution info + concurrent = data.get("concurrent") + if concurrent: + lines.append("") + lines.append(_render_concurrent_info(concurrent)) + return "\n".join(lines) diff --git a/sandbox/egg_lib/orch_cli.py b/sandbox/egg_lib/orch_cli.py index 510903f1ff..5c373a4b88 100755 --- a/sandbox/egg_lib/orch_cli.py +++ b/sandbox/egg_lib/orch_cli.py @@ -30,6 +30,10 @@ egg-orch container get Get container info egg-orch container stop Stop a container egg-orch container logs Get container logs + egg-orch message send --to ... Send inter-agent message (concurrent mode) + egg-orch message poll ... Poll for messages (concurrent mode) + egg-orch message status Get message bus status (concurrent mode) + egg-orch signal readiness --state ... Signal readiness state (concurrent mode) """ import argparse @@ -402,6 +406,8 @@ def cmd_pipeline_create(args: argparse.Namespace) -> int: data["prompt"] = args.prompt if args.network_mode: data["network_mode"] = args.network_mode + if args.concurrent: + data["config"] = {"concurrent_execution": True} result = orch_request("/api/v1/pipelines", method="POST", data=data) @@ -1178,6 +1184,12 @@ def create_parser() -> argparse.ArgumentParser: choices=["public", "private"], help="Network mode for spawned containers", ) + pl_create.add_argument( + "--concurrent", + action="store_true", + default=False, + help="Enable concurrent agent execution within phases", + ) _add_json_flag(pl_create) pl_create.set_defaults(func=cmd_pipeline_create) diff --git a/sandbox/egg_lib/sdlc_cli.py b/sandbox/egg_lib/sdlc_cli.py index 7b8a003018..55e3130029 100644 --- a/sandbox/egg_lib/sdlc_cli.py +++ b/sandbox/egg_lib/sdlc_cli.py @@ -345,7 +345,12 @@ def watch_pipeline( # --- Local Mode --- -def run_local_mode(client: OrchClient, prompt: str | None = None, repo: str | None = None) -> int: +def run_local_mode( + client: OrchClient, + prompt: str | None = None, + repo: str | None = None, + concurrent: bool = False, +) -> int: """Run egg-sdlc in local (prompt-driven) mode. Args: @@ -386,8 +391,13 @@ def run_local_mode(client: OrchClient, prompt: str | None = None, repo: str | No print(f"\n{DIM}Creating local pipeline...{RESET}") try: + config = {"concurrent_execution": True} if concurrent else None pipeline = client.create_pipeline( - mode="local", prompt=prompt, repo=repo, network_mode=_detect_network_mode() + mode="local", + prompt=prompt, + repo=repo, + network_mode=_detect_network_mode(), + config=config, ) except OrchestratorError as e: _write(f"{RED}Failed to create pipeline: {e}{RESET}\n", file=sys.stderr) @@ -424,6 +434,7 @@ def _restart_pipeline( repo: str, branch: str, network_mode: str | None = None, + config: dict[str, object] | None = None, ) -> None: """Delete an existing pipeline and re-create it. @@ -436,12 +447,18 @@ def _restart_pipeline( branch=branch, mode="issue", network_mode=network_mode, + config=config, ) client.start_pipeline(pipeline_id) print(f" {GREEN}Pipeline restarted.{RESET}") -def run_issue_mode(client: OrchClient, issue_number: int, repo: str | None = None) -> int: +def run_issue_mode( + client: OrchClient, + issue_number: int, + repo: str | None = None, + concurrent: bool = False, +) -> int: """Run egg-sdlc in issue mode.""" if not repo: _write( @@ -459,6 +476,8 @@ def run_issue_mode(client: OrchClient, issue_number: int, repo: str | None = Non print(f" Pipeline: {pipeline_id}") print(f" Branch: {branch}") + config: dict[str, object] | None = {"concurrent_execution": True} if concurrent else None + # Create pipeline print(f"\n{DIM}Creating pipeline...{RESET}") try: @@ -469,6 +488,7 @@ def run_issue_mode(client: OrchClient, issue_number: int, repo: str | None = Non branch=branch, mode="issue", network_mode=network_mode, + config=config, ) except OrchestratorError as e: if e.status_code == 409: @@ -486,7 +506,13 @@ def run_issue_mode(client: OrchClient, issue_number: int, repo: str | None = Non # Terminal — delete and re-create print(f" Pipeline was {status}. Restarting...") _restart_pipeline( - client, pipeline_id, issue_number, repo, branch, network_mode=network_mode + client, + pipeline_id, + issue_number, + repo, + branch, + network_mode=network_mode, + config=config, ) elif status in ("running", "awaiting_human"): print(f" Pipeline status: {status}. Attaching to watch loop...") @@ -509,7 +535,13 @@ def run_issue_mode(client: OrchClient, issue_number: int, repo: str | None = Non ) return 1 _restart_pipeline( - client, pipeline_id, issue_number, repo, branch, network_mode=network_mode + client, + pipeline_id, + issue_number, + repo, + branch, + network_mode=network_mode, + config=config, ) except OrchestratorError as e2: _write(f"{RED}Failed to restart pipeline: {e2}{RESET}\n", file=sys.stderr) @@ -576,6 +608,12 @@ def main() -> None: metavar="TEXT", help="Task prompt for local mode (skips interactive input).", ) + parser.add_argument( + "--concurrent", + action="store_true", + default=False, + help="Enable concurrent agent execution (agents run simultaneously within phases).", + ) args = parser.parse_args() @@ -621,9 +659,11 @@ def main() -> None: f"{YELLOW}Warning: --prompt is ignored in issue mode.{RESET}\n", file=sys.stderr, ) - exit_code = run_issue_mode(client, issue_number, repo) + exit_code = run_issue_mode(client, issue_number, repo, concurrent=args.concurrent) else: - exit_code = run_local_mode(client, prompt=args.prompt, repo=repo) + exit_code = run_local_mode( + client, prompt=args.prompt, repo=repo, concurrent=args.concurrent + ) sys.exit(exit_code) diff --git a/tests/sandbox/test_sdlc_cli.py b/tests/sandbox/test_sdlc_cli.py index 22dc488d49..a6d439bab7 100644 --- a/tests/sandbox/test_sdlc_cli.py +++ b/tests/sandbox/test_sdlc_cli.py @@ -10,10 +10,13 @@ from egg_lib.sdlc_cli import ( _resolve_repo_dir, + _restart_pipeline, _write, parse_sse_stream, render_event_info, render_header, + run_issue_mode, + run_local_mode, watch_pipeline, ) @@ -424,3 +427,94 @@ def parse_side_effect(resp): pipeline_id="issue-42", client=client, ) + + +# --------------------------------------------------------------------------- +# Concurrent config wiring tests +# --------------------------------------------------------------------------- + + +class TestRestartPipelineConcurrentConfig: + """Verify _restart_pipeline passes config to create_pipeline.""" + + def test_concurrent_config_passed(self): + client = MagicMock() + config = {"concurrent_execution": True} + _restart_pipeline( + client, + "issue-1", + 1, + "owner/repo", + "egg/issue-1", + network_mode="public", + config=config, + ) + client.create_pipeline.assert_called_once_with( + issue_number=1, + repo="owner/repo", + branch="egg/issue-1", + mode="issue", + network_mode="public", + config={"concurrent_execution": True}, + ) + + def test_no_concurrent_config(self): + client = MagicMock() + _restart_pipeline( + client, + "issue-2", + 2, + "owner/repo", + "egg/issue-2", + network_mode="private", + ) + client.create_pipeline.assert_called_once_with( + issue_number=2, + repo="owner/repo", + branch="egg/issue-2", + mode="issue", + network_mode="private", + config=None, + ) + + +class TestRunLocalModeConcurrent: + """Verify run_local_mode passes concurrent config to create_pipeline.""" + + @patch("egg_lib.sdlc_cli.watch_pipeline", return_value="complete") + @patch("egg_lib.sdlc_cli._detect_network_mode", return_value="public") + def test_concurrent_true(self, _mock_net, _mock_watch): + client = MagicMock() + client.create_pipeline.return_value = {"id": "local-1"} + run_local_mode(client, prompt="Build feature X", repo="owner/repo", concurrent=True) + _, kwargs = client.create_pipeline.call_args + assert kwargs["config"] == {"concurrent_execution": True} + + @patch("egg_lib.sdlc_cli.watch_pipeline", return_value="complete") + @patch("egg_lib.sdlc_cli._detect_network_mode", return_value="public") + def test_concurrent_false(self, _mock_net, _mock_watch): + client = MagicMock() + client.create_pipeline.return_value = {"id": "local-2"} + run_local_mode(client, prompt="Build feature Y", repo="owner/repo", concurrent=False) + _, kwargs = client.create_pipeline.call_args + assert kwargs["config"] is None + + +class TestRunIssueModeConcurrent: + """Verify run_issue_mode passes concurrent config to create_pipeline.""" + + @patch("egg_lib.sdlc_cli.watch_pipeline", return_value="complete") + @patch("egg_lib.sdlc_cli._detect_network_mode", return_value="public") + def test_concurrent_true_creates_with_config(self, _mock_net, _mock_watch): + client = MagicMock() + run_issue_mode(client, issue_number=99, repo="owner/repo", concurrent=True) + _, kwargs = client.create_pipeline.call_args + assert kwargs["config"] == {"concurrent_execution": True} + + @patch("egg_lib.sdlc_cli.watch_pipeline", return_value="complete") + @patch("egg_lib.sdlc_cli._detect_network_mode", return_value="public") + def test_concurrent_false_creates_without_config(self, _mock_net, _mock_watch): + client = MagicMock() + run_issue_mode(client, issue_number=100, repo="owner/repo", concurrent=False) + _, kwargs = client.create_pipeline.call_args + assert kwargs["config"] is None