Wire --concurrent flag and close concurrent execution gaps

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
 {

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)
 
 

sandbox/egg_lib/orch_cli.py

@@ -30,6 +30,10 @@
     egg-orch container get <pid> <cid>           Get container info
     egg-orch container stop <pid> <cid>          Stop a container
     egg-orch container logs <pid> <cid>          Get container logs
+    egg-orch message send <pid> --to <role> ...  Send inter-agent message (concurrent mode)
+    egg-orch message poll <pid> ...              Poll for messages (concurrent mode)
+    egg-orch message status <pid>                Get message bus status (concurrent mode)
+    egg-orch signal readiness <pid> --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)
 

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)
 

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