feat(core): per-state concurrency limits for batch execution (#407)

## Summary
- ConcurrencyConfig with per-status parallelism caps and effective_workers() calculation
- BatchConfig in EnvironmentConfig for YAML persistence
- parse_concurrency_by_status() for CLI string parsing
- Wired into _execute_parallel for actual group dispatch

## Validation
- Demo: All criteria verified (per-status limits, global bound, config roundtrip, fallback, parsing)
- Tests: 21 unit tests, 1755 total core tests passing
- CI: All checks green
- Review: CodeRabbit findings addressed (effective_workers wiring, error messages)

Closes #407

Author: frankbria

codeframe/core/conductor.py

@@ -414,6 +414,76 @@ def _utc_now() -> datetime:
     return datetime.now(timezone.utc)
 
 
+@dataclass
+class ConcurrencyConfig:
+    """Per-state concurrency limits for batch execution.
+
+    When ``by_status`` is non-empty, each status key caps the number of
+    concurrent workers for tasks in that state. Unspecified statuses
+    fall back to ``max_parallel``.
+    """
+
+    max_parallel: int = 4
+    by_status: dict[str, int] = field(default_factory=dict)
+
+    def get_limit_for_status(self, status: str) -> int:
+        """Return the concurrency limit for a given task status."""
+        return self.by_status.get(status, self.max_parallel)
+
+    def effective_workers(
+        self,
+        *,
+        statuses: list[str],
+        group_size: int,
+        global_running: int,
+    ) -> int:
+        """Compute the effective worker count for a group of tasks.
+
+        Takes the minimum of:
+        - Global slots remaining (max_parallel - global_running)
+        - Per-status limit for the most constrained status in the group
+        - Group size
+        """
+        global_slots = max(1, self.max_parallel - global_running)
+        if statuses and self.by_status:
+            per_status = min(self.get_limit_for_status(s) for s in statuses)
+        else:
+            per_status = self.max_parallel
+        return max(1, min(global_slots, per_status, group_size))
+
+
+def parse_concurrency_by_status(value: str | None) -> dict[str, int]:
+    """Parse a --max-parallel-by-status string into a dict.
+
+    Format: "READY=3,IN_PROGRESS=2"
+
+    Raises:
+        ValueError: On invalid status names or format.
+    """
+    from codeframe.core.state_machine import TaskStatus
+
+    if not value:
+        return {}
+
+    valid_statuses = {s.value for s in TaskStatus}
+    result: dict[str, int] = {}
+
+    for pair in value.split(","):
+        pair = pair.strip()
+        if "=" not in pair:
+            raise ValueError(f"Invalid format '{pair}'. Expected STATUS=N")
+        key, val = pair.split("=", 1)
+        key = key.strip().upper()
+        if key not in valid_statuses:
+            raise ValueError(f"Invalid status '{key}'. Valid: {', '.join(sorted(valid_statuses))}")
+        try:
+            result[key] = int(val.strip())
+        except ValueError:
+            raise ValueError(f"Invalid value '{val.strip()}' for status '{key}'. Must be an integer.")
+
+    return result
+
+
 class BatchStatus(str, Enum):
     """Status of a batch execution."""
 
@@ -462,6 +532,7 @@ class BatchRun:
     engine: str = "react"
     stall_timeout_s: int = 300
     stall_action: str = "blocker"
+    concurrency: ConcurrencyConfig = field(default_factory=ConcurrencyConfig)
 
 
 def start_batch(
@@ -476,6 +547,7 @@ def start_batch(
     engine: str = "react",
     stall_timeout_s: int = 300,
     stall_action: str = "blocker",
+    concurrency_by_status: Optional[dict[str, int]] = None,
 ) -> BatchRun:
     """Start a batch execution of multiple tasks.
 
@@ -510,6 +582,11 @@ def start_batch(
     now = _utc_now()
     on_failure_enum = OnFailure(on_failure)
 
+    concurrency = ConcurrencyConfig(
+        max_parallel=max_parallel,
+        by_status=concurrency_by_status or {},
+    )
+
     batch = BatchRun(
         id=batch_id,
         workspace_id=workspace.id,
@@ -524,6 +601,7 @@ def start_batch(
         engine=engine,
         stall_timeout_s=stall_timeout_s,
         stall_action=stall_action,
+        concurrency=concurrency,
     )
 
     # Save to database
@@ -1446,8 +1524,18 @@ def _execute_parallel(
             else:
                 failed_count += 1
         else:
-            # Multiple tasks - run in parallel
-            effective_workers = min(group_size, batch.max_parallel)
+            # Multiple tasks - run in parallel (use per-status limits if configured)
+            if batch.concurrency.by_status:
+                group_statuses = []
+                for tid in group:
+                    t = tasks.get(workspace, tid)
+                    if t:
+                        group_statuses.append(t.status.value)
+                effective_workers = batch.concurrency.effective_workers(
+                    statuses=group_statuses, group_size=group_size, global_running=0,
+                )
+            else:
+                effective_workers = min(group_size, batch.max_parallel)
             print(f"Running {group_size} tasks with {effective_workers} workers")
 
             # Execute group in parallel

codeframe/core/config.py

@@ -81,6 +81,14 @@ class AgentBudgetConfig:
     stall_timeout_s: int = 300
 
 
+@dataclass
+class BatchConfig:
+    """Batch execution configuration."""
+
+    max_parallel: int = 4
+    max_parallel_by_status: dict[str, int] = dataclass_field(default_factory=dict)
+
+
 @dataclass
 class HooksConfig:
     """Workspace lifecycle hooks configuration.
@@ -124,6 +132,9 @@ class EnvironmentConfig:
     # Agent budget
     agent_budget: AgentBudgetConfig = dataclass_field(default_factory=AgentBudgetConfig)
 
+    # Batch execution
+    batch: BatchConfig = dataclass_field(default_factory=BatchConfig)
+
     # Workspace lifecycle hooks
     hooks: HooksConfig = dataclass_field(default_factory=HooksConfig)
 
@@ -281,6 +292,8 @@ def from_dict(cls, data: dict[str, Any]) -> "EnvironmentConfig":
             data["context"] = ContextConfig(**data["context"])
         if "agent_budget" in data and isinstance(data["agent_budget"], dict):
             data["agent_budget"] = AgentBudgetConfig(**data["agent_budget"])
+        if "batch" in data and isinstance(data["batch"], dict):
+            data["batch"] = BatchConfig(**data["batch"])
         if "hooks" in data and isinstance(data["hooks"], dict):
             data["hooks"] = HooksConfig(**data["hooks"])
         return cls(**data)

tests/core/test_concurrency_config.py

@@ -0,0 +1,186 @@
+"""Tests for per-state concurrency limits in batch execution."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+pytestmark = pytest.mark.v2
+
+
+class TestConcurrencyConfig:
+    """Test ConcurrencyConfig dataclass."""
+
+    def test_defaults(self) -> None:
+        from codeframe.core.conductor import ConcurrencyConfig
+
+        cfg = ConcurrencyConfig()
+        assert cfg.max_parallel == 4
+        assert cfg.by_status == {}
+
+    def test_custom_values(self) -> None:
+        from codeframe.core.conductor import ConcurrencyConfig
+
+        cfg = ConcurrencyConfig(max_parallel=8, by_status={"READY": 3, "IN_PROGRESS": 2})
+        assert cfg.max_parallel == 8
+        assert cfg.by_status["READY"] == 3
+
+    def test_get_limit_for_status_configured(self) -> None:
+        from codeframe.core.conductor import ConcurrencyConfig
+
+        cfg = ConcurrencyConfig(max_parallel=4, by_status={"READY": 2})
+        assert cfg.get_limit_for_status("READY") == 2
+
+    def test_get_limit_for_status_fallback_to_global(self) -> None:
+        from codeframe.core.conductor import ConcurrencyConfig
+
+        cfg = ConcurrencyConfig(max_parallel=4, by_status={"READY": 2})
+        assert cfg.get_limit_for_status("IN_PROGRESS") == 4
+
+    def test_get_limit_for_status_empty_by_status(self) -> None:
+        from codeframe.core.conductor import ConcurrencyConfig
+
+        cfg = ConcurrencyConfig(max_parallel=6)
+        assert cfg.get_limit_for_status("READY") == 6
+
+    def test_effective_workers_global_limit(self) -> None:
+        from codeframe.core.conductor import ConcurrencyConfig
+
+        cfg = ConcurrencyConfig(max_parallel=2, by_status={"READY": 5})
+        # Global limit (2) is less than per-status limit (5)
+        workers = cfg.effective_workers(statuses=["READY"], group_size=10, global_running=0)
+        assert workers == 2
+
+    def test_effective_workers_per_status_limit(self) -> None:
+        from codeframe.core.conductor import ConcurrencyConfig
+
+        cfg = ConcurrencyConfig(max_parallel=10, by_status={"READY": 3})
+        # Per-status limit (3) is less than global (10)
+        workers = cfg.effective_workers(statuses=["READY"], group_size=10, global_running=0)
+        assert workers == 3
+
+    def test_effective_workers_group_size_limit(self) -> None:
+        from codeframe.core.conductor import ConcurrencyConfig
+
+        cfg = ConcurrencyConfig(max_parallel=10)
+        workers = cfg.effective_workers(statuses=["READY"], group_size=2, global_running=0)
+        assert workers == 2
+
+    def test_effective_workers_accounts_for_running(self) -> None:
+        from codeframe.core.conductor import ConcurrencyConfig
+
+        cfg = ConcurrencyConfig(max_parallel=4)
+        workers = cfg.effective_workers(statuses=["READY"], group_size=10, global_running=3)
+        assert workers == 1  # Only 1 global slot left
+
+    def test_effective_workers_mixed_statuses(self) -> None:
+        from codeframe.core.conductor import ConcurrencyConfig
+
+        cfg = ConcurrencyConfig(max_parallel=10, by_status={"READY": 3, "IN_PROGRESS": 1})
+        # Mixed group: bottleneck is IN_PROGRESS (1)
+        workers = cfg.effective_workers(statuses=["READY", "IN_PROGRESS"], group_size=5, global_running=0)
+        assert workers == 1
+
+    def test_effective_workers_never_negative(self) -> None:
+        from codeframe.core.conductor import ConcurrencyConfig
+
+        cfg = ConcurrencyConfig(max_parallel=2)
+        workers = cfg.effective_workers(statuses=["READY"], group_size=5, global_running=10)
+        assert workers >= 1  # At least 1 worker
+
+
+class TestBatchConfig:
+    """Test BatchConfig in EnvironmentConfig."""
+
+    def test_defaults(self) -> None:
+        from codeframe.core.config import EnvironmentConfig
+
+        cfg = EnvironmentConfig()
+        assert cfg.batch.max_parallel == 4
+        assert cfg.batch.max_parallel_by_status == {}
+
+    def test_from_dict(self) -> None:
+        from codeframe.core.config import EnvironmentConfig
+
+        cfg = EnvironmentConfig.from_dict({
+            "batch": {
+                "max_parallel": 8,
+                "max_parallel_by_status": {"READY": 3, "IN_PROGRESS": 2},
+            }
+        })
+        assert cfg.batch.max_parallel == 8
+        assert cfg.batch.max_parallel_by_status["READY"] == 3
+
+    def test_roundtrip(self) -> None:
+        from codeframe.core.config import BatchConfig, EnvironmentConfig
+
+        orig = EnvironmentConfig(batch=BatchConfig(max_parallel=6, max_parallel_by_status={"READY": 2}))
+        d = orig.to_dict()
+        restored = EnvironmentConfig.from_dict(d)
+        assert restored.batch.max_parallel == 6
+        assert restored.batch.max_parallel_by_status["READY"] == 2
+
+
+class TestStartBatchConcurrency:
+    """Test start_batch with concurrency_by_status."""
+
+    def test_start_batch_accepts_concurrency_by_status(self) -> None:
+        from codeframe.core.conductor import start_batch
+
+        workspace = MagicMock()
+        workspace.id = "w1"
+
+        mock_task = MagicMock()
+        mock_task.id = "t1"
+        mock_task.title = "Test"
+
+        with patch("codeframe.core.conductor.tasks.get", return_value=mock_task):
+            with patch("codeframe.core.conductor._save_batch"):
+                with patch("codeframe.core.conductor.events.emit_for_workspace"):
+                    with patch("codeframe.core.conductor._execute_serial"):
+                        batch = start_batch(
+                            workspace, ["t1"],
+                            concurrency_by_status={"READY": 2},
+                        )
+
+        assert batch.concurrency.by_status == {"READY": 2}
+        assert batch.concurrency.max_parallel == 4  # default
+
+
+class TestParseConcurrencyString:
+    """Test parsing of --max-parallel-by-status CLI flag."""
+
+    def test_parse_valid_string(self) -> None:
+        from codeframe.core.conductor import parse_concurrency_by_status
+
+        result = parse_concurrency_by_status("READY=3,IN_PROGRESS=2")
+        assert result == {"READY": 3, "IN_PROGRESS": 2}
+
+    def test_parse_single_value(self) -> None:
+        from codeframe.core.conductor import parse_concurrency_by_status
+
+        result = parse_concurrency_by_status("READY=5")
+        assert result == {"READY": 5}
+
+    def test_parse_none_returns_empty(self) -> None:
+        from codeframe.core.conductor import parse_concurrency_by_status
+
+        result = parse_concurrency_by_status(None)
+        assert result == {}
+
+    def test_parse_empty_returns_empty(self) -> None:
+        from codeframe.core.conductor import parse_concurrency_by_status
+
+        result = parse_concurrency_by_status("")
+        assert result == {}
+
+    def test_parse_invalid_status_raises(self) -> None:
+        from codeframe.core.conductor import parse_concurrency_by_status
+
+        with pytest.raises(ValueError, match="Invalid status"):
+            parse_concurrency_by_status("INVALID=3")
+
+    def test_parse_invalid_format_raises(self) -> None:
+        from codeframe.core.conductor import parse_concurrency_by_status
+
+        with pytest.raises(ValueError):
+            parse_concurrency_by_status("READY:3")